Make varz thread-aware

Thread-safety is enabled by calling `VCAP::Component.varz.threadsafe!`.

This requires threads to synchronize access to the hash.

Change-Id: I72e9b2e634d0337f995b05b04237605aac22203b

lib/vcap/component.rb

@@ -1,10 +1,11 @@
 # Copyright (c) 2009-2011 VMware, Inc.
+require "base64"
 require "eventmachine"
-require 'thin'
-require "yajl"
+require "monitor"
 require "nats/client"
-require "base64"
-require 'set'
+require "set"
+require "thin"
+require "yajl"
 
 module VCAP
 
@@ -48,34 +49,79 @@ class Component
     CONFIG_SUPPRESS = Set.new([:mbus, :service_mbus, :keys, :database_environment, :password, :pass, :token])
 
     class << self
+      class SafeHash < BasicObject
+        def initialize(hash = {})
+          @hash = hash
+        end
+
+        def threadsafe!
+          @monitor = ::Monitor.new
+        end
+
+        def synchronize
+          if @monitor
+            @monitor.synchronize do
+              begin
+                @thread = ::Thread.current
+                yield
+              ensure
+                @thread = nil
+              end
+            end
+          else
+            yield
+          end
+        end
+
+        def method_missing(sym, *args, &blk)
+          if @monitor && @thread != ::Thread.current
+            ::Kernel.raise "Lock required"
+          end
+
+          @hash.__send__(sym, *args, &blk)
+        end
+      end
+
+      def varz
+        @varz ||= SafeHash.new
+      end
 
-      attr_reader   :varz
       attr_accessor :healthz
 
       def updated_varz
         @last_varz_update ||= 0
-        if Time.now.to_f - @last_varz_update >= 1
-          # Snapshot uptime
-          @varz[:uptime] = VCAP.uptime_string(Time.now - @varz[:start])
 
-          # Grab current cpu and memory usage.
+        if Time.now.to_f - @last_varz_update >= 1
+          # Grab current cpu and memory usage
           rss, pcpu = `ps -o rss=,pcpu= -p #{Process.pid}`.split
-          @varz[:mem] = rss.to_i
-          @varz[:cpu] = pcpu.to_f
 
-          @last_varz_update = Time.now.to_f
+          # Update varz
+          varz.synchronize do
+            @last_varz_update = Time.now.to_f
+
+            varz[:uptime] = VCAP.uptime_string(Time.now - varz[:start])
+            varz[:mem] = rss.to_i
+            varz[:cpu] = pcpu.to_f
+
+            # Return duplicate while holding lock
+            return varz.dup
+          end
+        else
+          # Return duplicate while holding lock
+          varz.synchronize do
+            return varz.dup
+          end
         end
-        varz
       end
 
       def updated_healthz
         @last_healthz_update ||= 0
+
         if Time.now.to_f - @last_healthz_update >= 1
-          # ...
           @last_healthz_update = Time.now.to_f
         end
 
-        healthz
+        healthz.dup
       end
 
       def start_http_server(host, port, auth, logger)
@@ -120,9 +166,11 @@ def register(opts)
         }
 
         # Varz is customizable
-        @varz = @discover.dup
-        @varz[:num_cores] = VCAP.num_cores
-        @varz[:config] = sanitize_config(opts[:config]) if opts[:config]
+        varz.synchronize do
+          varz.merge!(@discover.dup)
+          varz[:num_cores] = VCAP.num_cores
+          varz[:config] = sanitize_config(opts[:config]) if opts[:config]
+        end
 
         @healthz = "ok\n".freeze
 

spec/unit/component_spec.rb

@@ -0,0 +1,55 @@
+# Copyright (c) 2009-2012 VMware, Inc.
+
+require "spec_helper"
+
+require "vcap/component"
+
+describe VCAP::Component do
+  def cleanup
+    VCAP::Component.instance_eval do
+      if instance_variables.include?(:@varz)
+        remove_instance_variable(:@varz)
+      end
+
+      if instance_variables.include?(:@healthz)
+        remove_instance_variable(:@healthz)
+      end
+    end
+  end
+
+  before do
+    cleanup
+  end
+
+  after do
+    cleanup
+  end
+
+  describe "regular #varz" do
+    it "should not raise on get" do
+      expect do
+        VCAP::Component.varz[:foo]
+      end.to_not raise_error
+    end
+  end
+
+  describe "thread-safe #varz" do
+    before do
+      VCAP::Component.varz.threadsafe!
+    end
+
+    it "should raise on get when the lock is not held" do
+      expect do
+        VCAP::Component.varz[:foo]
+      end.to raise_error(/lock/i)
+    end
+
+    it "should not raise on get when the lock is held" do
+      VCAP::Component.varz.synchronize do
+        expect do
+          VCAP::Component.varz[:foo]
+        end.to_not raise_error
+      end
+    end
+  end
+end