Merge pull request #213 from localshred/abrandoned/zmq_queue_work

allow queue size to be configurable and use an array as local queue as t...

Author: abrandoned

lib/protobuf/rpc/connectors/zmq.rb

@@ -123,6 +123,8 @@ def lookup_server_uri
         def host_alive?(host)
           return true unless ping_port_enabled?
           socket = TCPSocket.new(host, ping_port.to_i)
+          socket.setsockopt(::Socket::IPPROTO_TCP, ::Socket::TCP_NODELAY, 1)
+          socket.setsockopt(::Socket::SOL_SOCKET, ::Socket::SO_LINGER, [1,0].pack('ii'))
 
           true
         rescue

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

@@ -67,7 +67,7 @@ def init_frontend_socket
         end
 
         def init_local_queue
-          @local_queue = ::Queue.new
+          @local_queue = []
         end
 
         def init_poller
@@ -88,6 +88,10 @@ def inproc?
           !!@server.try(:inproc?)
         end
 
+        def local_queue_max_size
+          @local_queue_max_size ||= [ENV["PB_ZMQ_SERVER_QUEUE_MAX_SIZE"].to_i, 5].max
+        end
+
         def process_backend
           worker, ignore, *frames = read_from_backend
 
@@ -102,16 +106,16 @@ def process_frontend
           address, _, message, *frames = read_from_frontend
 
           if message == ::Protobuf::Rpc::Zmq::CHECK_AVAILABLE_MESSAGE
-            if @idle_workers.any? || local_queue.size < 5 # Should make queue a SizedQueue and allow users to configure queue size
+            if local_queue.size < local_queue_max_size
               write_to_frontend([address, "", ::Protobuf::Rpc::Zmq::WORKERS_AVAILABLE])
             else
               write_to_frontend([address, "", ::Protobuf::Rpc::Zmq::NO_WORKERS_AVAILABLE])
             end
           else
-            if @idle_workers.any?
-              write_to_backend([@idle_workers.shift, ""] + [address, "", message ] + frames)
-            else
+            if @idle_workers.empty?
               local_queue.push([address, "", message ] + frames)
+            else
+              write_to_backend([@idle_workers.shift, ""] + [address, "", message ] + frames)
             end
           end
         end