Merge pull request #244 from localshred/abrandoned/broker_probs

Abrandoned/broker probs

Author: abrandoned

.rubocop.yml

@@ -6,6 +6,9 @@ Lint/EndAlignment:
 Lint/Loop:
   Enabled: false
 
+Metrics/ClassLength:
+  Enabled: false
+
 Style/CaseIndentation:
   IndentWhenRelativeTo: end
 

lib/protobuf/rpc/connectors/zmq.rb

@@ -59,11 +59,25 @@ def send_request
         # Private Instance methods
         #
         def check_available_rcv_timeout
-          @check_available_rcv_timeout ||= [ENV["PB_ZMQ_CLIENT_CHECK_AVAILABLE_RCV_TIMEOUT"].to_i, 200].max
+          @check_available_rcv_timeout ||= begin
+            case
+            when ENV.key?("PB_ZMQ_CLIENT_CHECK_AVAILABLE_RCV_TIMEOUT") then
+              ENV["PB_ZMQ_CLIENT_CHECK_AVAILABLE_RCV_TIMEOUT"].to_i
+            else
+              200 # ms
+            end
+          end
         end
 
         def check_available_snd_timeout
-          @check_available_snd_timeout ||= [ENV["PB_ZMQ_CLIENT_CHECK_AVAILABLE_SND_TIMEOUT"].to_i, 200].max
+          @check_available_snd_timeout ||= begin
+            case
+            when ENV.key?("PB_ZMQ_CLIENT_CHECK_AVAILABLE_SND_TIMEOUT") then
+              ENV["PB_ZMQ_CLIENT_CHECK_AVAILABLE_SND_TIMEOUT"].to_i
+            else
+              200 # ms
+            end
+          end
         end
 
         def close_connection

lib/protobuf/rpc/servers/zmq/broker.rb

@@ -23,6 +23,7 @@ def initialize(server)
 
         def run
           @idle_workers = []
+          @running = true
 
           loop do
             process_local_queue
@@ -39,10 +40,11 @@ def run
           end
         ensure
           teardown
+          @running = false
         end
 
         def running?
-          @server.running?
+          @running && @server.running?
         end
 
         private

lib/protobuf/rpc/servers/zmq/server.rb

@@ -76,6 +76,10 @@ def broadcast_beacons?
           !brokerless? && options[:broadcast_beacons]
         end
 
+        def broadcast_busy?
+          broadcast_beacons? && options[:broadcast_busy]
+        end
+
         def broadcast_flatline
           flatline = ::Protobuf::Rpc::DynamicDiscovery::Beacon.new(
             :beacon_type => ::Protobuf::Rpc::DynamicDiscovery::BeaconType::FLATLINE,
@@ -176,15 +180,11 @@ def reaping_interval
 
         def run
           @running = true
-
-          start_broker unless brokerless?
-          start_missing_workers
-
           yield if block_given? # runs on startup
           wait_for_shutdown_signal
           broadcast_flatline if broadcast_beacons?
           Thread.pass until reap_dead_workers.empty?
-          @broker.join unless brokerless?
+          @broker_thread.join unless brokerless?
         ensure
           @running = false
           teardown
@@ -246,14 +246,13 @@ def wait_for_shutdown_signal
           loop do
             break if IO.select([@shutdown_r], nil, nil, timeout)
 
-            if reap_dead_workers?
-              reap_dead_workers
-              start_missing_workers
-            end
+            start_broker unless brokerless?
+            reap_dead_workers if reap_dead_workers?
+            start_missing_workers
 
             next unless broadcast_heartbeat?
 
-            if options[:broadcast_busy] && all_workers_busy?
+            if broadcast_busy? && all_workers_busy?
               broadcast_flatline
             else
               broadcast_heartbeat
@@ -285,15 +284,29 @@ def init_zmq_context
         end
 
         def start_broker
-          @broker = Thread.new(self) do |server|
-            ::Protobuf::Rpc::Zmq::Broker.new(server).run
+          return if @broker && @broker.running? && !@broker_thread.stop?
+          if @broker && !@broker.running?
+            broadcast_flatline if broadcast_busy?
+            @broker_thread.join if @broker_thread
+            init_zmq_context # need a new context to restart the broker
+          end
+
+          @broker = ::Protobuf::Rpc::Zmq::Broker.new(self)
+          @broker_thread = Thread.new(@broker) do |broker|
+            begin
+              broker.run
+            rescue => e
+              message = "Broker failed: #{e.inspect}\n #{e.backtrace.join($INPUT_RECORD_SEPARATOR)}"
+              $stderr.puts(message)
+              logger.error { message }
+            end
           end
         end
 
         def start_worker
-          @workers << Thread.new(self) do |server|
+          @workers << Thread.new(self, @broker) do |server, broker|
             begin
-              ::Protobuf::Rpc::Zmq::Worker.new(server).run
+              ::Protobuf::Rpc::Zmq::Worker.new(server, broker).run
             rescue => e
               message = "Worker failed: #{e.inspect}\n #{e.backtrace.join($INPUT_RECORD_SEPARATOR)}"
               $stderr.puts(message)

lib/protobuf/rpc/servers/zmq/worker.rb

@@ -12,8 +12,9 @@ class Worker
         ##
         # Constructor
         #
-        def initialize(server)
+        def initialize(server, broker)
           @server = server
+          @broker = broker
 
           init_zmq_context
           init_backend_socket
@@ -61,7 +62,7 @@ def run
         end
 
         def running?
-          @server.running?
+          @broker.running? && @server.running?
         end
 
         private