Merge pull request #218 from localshred/abrandoned/config_zmq

Abrandoned/config zmq

Author: abrandoned

lib/protobuf/rpc/connectors/zmq.rb

@@ -75,10 +75,7 @@ def create_socket
 
             if socket # Make sure the context builds the socket
               socket.setsockopt(::ZMQ::LINGER, 0)
-
-              logger.debug { sign_message("Establishing connection: #{server_uri}") }
               zmq_error_check(socket.connect(server_uri), :socket_connect)
-              logger.debug { sign_message("Connection established to #{server_uri}") }
 
               if first_alive_load_balance?
                 begin
@@ -110,7 +107,7 @@ def error?
         # to the host and port in the options
         #
         def lookup_server_uri
-          15.times do
+          server_lookup_attempts.times do
             service_directory.all_listings_for(service).each do |listing|
               host = listing.try(:address)
               port = listing.try(:port)
@@ -121,7 +118,7 @@ def lookup_server_uri
             port = options[:port]
             return "tcp://#{host}:#{port}" if host_alive?(host)
 
-            sleep (5.0/100.0)
+            sleep (1.0/100.0)
           end
 
           raise "Host not found for service #{service}"
@@ -131,7 +128,7 @@ def host_alive?(host)
           return true unless ping_port_enabled?
 
           if (last_response = self.class.ping_port_responses[host])
-            if (Time.now.to_i - last_response[:at]) <= 2
+            if (Time.now.to_i - last_response[:at]) <= host_alive_check_interval
               return last_response[:ping_port_open]
             end
           end
@@ -144,6 +141,10 @@ def host_alive?(host)
           ping_port_open
         end
 
+        def host_alive_check_interval
+          @host_alive_check_interval ||= [ENV["PB_ZMQ_CLIENT_HOST_ALIVE_CHECK_INTERVAL"].to_i, 1].max
+        end
+
         def ping_port_open?(host)
           socket = TCPSocket.new(host, ping_port.to_i)
           socket.setsockopt(::Socket::IPPROTO_TCP, ::Socket::TCP_NODELAY, 1)
@@ -197,6 +198,10 @@ def send_request_with_timeout(timeout, attempt = 0)
           logger.debug { sign_message("Socket closed")  }
         end
 
+        def server_lookup_attempts
+          @server_lookup_attempts ||= [ENV["PB_ZMQ_CLIENT_SERVER_LOOKUP_ATTEMPTS"].to_i, 5].max
+        end
+
         # The service we're attempting to connect to
         #
         def service

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

@@ -33,9 +33,9 @@ def run
             # Something went wrong
             break if rc == -1
 
-            process_backend if @poller.readables.include?(@backend_socket)
+            check_and_process_backend
             process_local_queue # Fair ordering so queued requests get in before new requests
-            process_frontend if @poller.readables.include?(@frontend_socket)
+            check_and_process_frontend
           end
         ensure
           teardown
@@ -47,6 +47,39 @@ def running?
 
         private
 
+        def backend_poll_weight
+          @backend_poll_weight ||= [ENV["PB_ZMQ_SERVER_BACKEND_POLL_WEIGHT"].to_i, 1].max
+        end
+
+        def check_and_process_backend
+          readables_include_backend = @poller.readables.include?(@backend_socket)
+          message_count_read_from_backend = 0
+
+          while readables_include_backend && message_count_read_from_backend < backend_poll_weight do
+            message_count_read_from_backend += 1
+            process_backend
+            @poller.poll_nonblock
+            readables_include_backend = @poller.readables.include?(@backend_socket)
+          end
+        end
+
+        def check_and_process_frontend
+          readables_include_frontend = @poller.readables.include?(@frontend_socket)
+          message_count_read_from_frontend = 0
+
+          while readables_include_frontend && message_count_read_from_frontend < frontend_poll_weight do
+            message_count_read_from_frontend += 1
+            process_frontend
+            break unless local_queue_available? # no need to read frontend just to throw away messages, will prioritize backend when full
+            @poller.poll_nonblock
+            readables_include_frontend = @poller.readables.include?(@frontend_socket)
+          end
+        end
+
+        def frontend_poll_weight
+          @frontend_poll_weight ||= [ENV["PB_ZMQ_SERVER_FRONTEND_POLL_WEIGHT"].to_i, 1].max
+        end
+
         def init_backend_socket
           @backend_socket = @zmq_context.socket(ZMQ::ROUTER)
           zmq_error_check(@backend_socket.bind(@server.backend_uri))
@@ -79,6 +112,10 @@ def inproc?
           !!@server.try(:inproc?)
         end
 
+        def local_queue_available?
+          local_queue.size < local_queue_max_size
+        end
+
         def local_queue_max_size
           @local_queue_max_size ||= [ENV["PB_ZMQ_SERVER_QUEUE_MAX_SIZE"].to_i, 5].max
         end
@@ -97,7 +134,7 @@ def process_frontend
           address, _, message, *frames = read_from_frontend
 
           if message == ::Protobuf::Rpc::Zmq::CHECK_AVAILABLE_MESSAGE
-            if local_queue.size < local_queue_max_size
+            if local_queue_available?
               write_to_frontend([address, ::Protobuf::Rpc::Zmq::EMPTY_STRING, ::Protobuf::Rpc::Zmq::WORKERS_AVAILABLE])
             else
               write_to_frontend([address, ::Protobuf::Rpc::Zmq::EMPTY_STRING, ::Protobuf::Rpc::Zmq::NO_WORKERS_AVAILABLE])