add configuration ENV vars for the values that have been hard-coded

in the connector for zmq and the broker for zmq

the poll weight influences how many messages will be pulled from the
polling socket before it moves on to check the other socket, this is
currently 2/1 weighted in favor of the backend socket (the work
complete)

Also made a configuration value of the number of attempts to make on
getting the server uri on the client side

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,26 @@ def run
             # Something went wrong
             break if rc == -1
 
-            process_backend if @poller.readables.include?(@backend_socket)
+            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 = message_count_read_from_backend + 1
+              process_backend
+              @poller.poll_nonblock
+              readables_include_backend = @poller.readables.include?(@backend_socket)
+            end
+
             process_local_queue # Fair ordering so queued requests get in before new requests
-            process_frontend if @poller.readables.include?(@frontend_socket)
+            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 = message_count_read_from_frontend + 1
+              process_frontend
+              @poller.poll_nonblock
+              readables_include_frontend = @poller.readables.include?(@frontend_socket)
+            end
           end
         ensure
           teardown
@@ -47,6 +64,14 @@ def running?
 
         private
 
+        def backend_poll_weight
+          @backend_poll_weight ||= [ENV["PB_ZMQ_SERVER_BACKEND_POLL_WEIGHT"].to_i, 2].max
+        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))