WIP: first attempt at #776 / #941

lib/dalli/pipelined_getter.rb

@@ -17,68 +17,26 @@ def process(keys, &block)
       return {} if keys.empty?
 
       @ring.lock do
-        servers = setup_requests(keys)
-        start_time = Time.now
-        servers = fetch_responses(servers, start_time, @ring.socket_timeout, &block) until servers.empty?
+        requests = setup_requests(keys)
+        fetch_responses(requests, @ring.socket_timeout, &block)
       end
     rescue NetworkError => e
       Dalli.logger.debug { e.inspect }
       Dalli.logger.debug { 'retrying pipelined gets because of timeout' }
       retry
     end
 
-    def setup_requests(keys)
-      groups = groups_for_keys(keys)
-      make_getkq_requests(groups)
-
-      # TODO: How does this exit on a NetworkError
-      finish_queries(groups.keys)
-    end
-
-    ##
-    # Loop through the server-grouped sets of keys, writing
-    # the corresponding getkq requests to the appropriate servers
-    #
-    # It's worth noting that we could potentially reduce bytes
-    # on the wire by switching from getkq to getq, and using
-    # the opaque value to match requests to responses.
-    ##
-    def make_getkq_requests(groups)
-      groups.each do |server, keys_for_server|
-        server.request(:pipelined_get, keys_for_server)
-      rescue DalliError, NetworkError => e
-        Dalli.logger.debug { e.inspect }
-        Dalli.logger.debug { "unable to get keys for server #{server.name}" }
+    def setup_requests(all_keys)
+      groups_for_keys(all_keys).to_h do |server, keys|
+        # It's worth noting that we could potentially reduce bytes
+        # on the wire by switching from getkq to getq, and using
+        # the opaque value to match requests to responses.
+        [server, server.pipelined_get_request(keys)]
       end
     end
 
-    ##
-    # This loops through the servers that have keys in
-    # our set, sending the noop to terminate the set of queries.
-    ##
-    def finish_queries(servers)
-      deleted = []
-
-      servers.each do |server|
-        next unless server.alive?
-
-        begin
-          finish_query_for_server(server)
-        rescue Dalli::NetworkError
-          raise
-        rescue Dalli::DalliError
-          deleted.append(server)
-        end
-      end
-
-      servers.delete_if { |server| deleted.include?(server) }
-    rescue Dalli::NetworkError
-      abort_without_timeout(servers)
-      raise
-    end
-
     def finish_query_for_server(server)
-      server.pipeline_response_setup
+      server.finish_pipeline_request
     rescue Dalli::NetworkError
       raise
     rescue Dalli::DalliError => e
@@ -92,29 +50,101 @@ def abort_without_timeout(servers)
       servers.each(&:pipeline_abort)
     end
 
-    def fetch_responses(servers, start_time, timeout, &block)
+    def fetch_responses(requests, timeout, &block)
+      # FIXME: this was here. why. where should it go?
       # Remove any servers which are not connected
-      servers.delete_if { |s| !s.connected? }
-      return [] if servers.empty?
+      # servers.delete_if { |s| !s.connected? }
+
+      start_time = Time.now
+      servers = requests.keys
+
+      # FIXME: this was executed before the finish request was sent. Why?
+      servers.delete_if { |s| !s.alive? }
+
+      # could be postponed to after the first write
+      servers.each(&:pipeline_response_setup)
 
-      time_left = remaining_time(start_time, timeout)
-      readable_servers = servers_with_response(servers, time_left)
-      if readable_servers.empty?
+      until servers.empty?
+        time_left = remaining_time(start_time, timeout)
+        servers = read_write_select(servers, requests, time_left, &block)
+      end
+    rescue NetworkError
+      # Abort and raise if we encountered a network error.  This triggers
+      # a retry at the top level.
+      abort_without_timeout(servers)
+      raise
+    end
+
+    def read_write_select(servers, requests, time_left, &block)
+      # TODO: - This is a bit challenging.  Essentially the PipelinedGetter
+      # is a reactor, but without the benefit of a Fiber or separate thread.
+      # My suspicion is that we may want to try and push this down into the
+      # individual servers, but I'm not sure.  For now, we keep the
+      # mapping between the alerted object (the socket) and the
+      # corrresponding server here.
+      server_map = servers.each_with_object({}) { |s, h| h[s.sock] = s }
+
+      readable, writable, = IO.select(server_map.keys, server_map.keys,
+                                      nil, time_left)
+
+      if readable.nil?
         abort_with_timeout(servers)
         return []
       end
 
-      # Loop through the servers with responses, and
-      # delete any from our list that are finished
-      readable_servers.each do |server|
+      writable.each do |socket|
+        server = server_map[socket]
+        process_writable(server, servers, requests)
+      end
+
+      readable.each do |socket|
+        server = server_map[socket]
+
         servers.delete(server) if process_server(server, &block)
       end
+
       servers
-    rescue NetworkError
-      # Abort and raise if we encountered a network error.  This triggers
-      # a retry at the top level.
+    end
+
+    def process_writable(server, servers, requests)
+      request = requests[server]
+      return unless request
+
+      new_request = server_pipelined_get(server, request)
+
+      if new_request.empty?
+        requests.delete(server)
+
+        begin
+          finish_query_for_server(server)
+        rescue Dalli::NetworkError
+          raise
+        rescue Dalli::DalliError
+          servers.delete(server)
+        end
+      else
+        requests[server] = new_request
+      end
+    rescue Dalli::NetworkError
       abort_without_timeout(servers)
       raise
+    rescue DalliError => e
+      Dalli.logger.debug { e.inspect }
+      Dalli.logger.debug { "unable to get keys for server #{server.name}" }
+    end
+
+    def server_pipelined_get(server, request)
+      buffer_size = server.socket_sndbuf
+      chunk = request[0..buffer_size]
+      written = server.request(:pipelined_get, chunk)
+      return if written == :wait_writable
+
+      request[written..]
+    rescue Dalli::NetworkError
+      raise
+    rescue DalliError => e
+      Dalli.logger.debug { e.inspect }
+      Dalli.logger.debug { "unable to get keys for server #{server.name}" }
     end
 
     def remaining_time(start, timeout)
@@ -144,23 +174,6 @@ def process_server(server)
       server.pipeline_complete?
     end
 
-    def servers_with_response(servers, timeout)
-      return [] if servers.empty?
-
-      # TODO: - This is a bit challenging.  Essentially the PipelinedGetter
-      # is a reactor, but without the benefit of a Fiber or separate thread.
-      # My suspicion is that we may want to try and push this down into the
-      # individual servers, but I'm not sure.  For now, we keep the
-      # mapping between the alerted object (the socket) and the
-      # corrresponding server here.
-      server_map = servers.each_with_object({}) { |s, h| h[s.sock] = s }
-
-      readable, = IO.select(server_map.keys, nil, nil, timeout)
-      return [] if readable.nil?
-
-      readable.map { |sock| server_map[sock] }
-    end
-
     def groups_for_keys(*keys)
       keys.flatten!
       keys.map! { |a| @key_manager.validate_key(a.to_s) }

lib/dalli/protocol/base.rb

@@ -18,7 +18,8 @@ class Base
 
       def_delegators :@value_marshaller, :serializer, :compressor, :compression_min_size, :compress_by_default?
       def_delegators :@connection_manager, :name, :sock, :hostname, :port, :close, :connected?, :socket_timeout,
-                     :socket_type, :up!, :down!, :write, :reconnect_down_server?, :raise_down_error
+                     :socket_type, :socket_sndbuf, :up!, :down!, :write, :write_nonblock, :reconnect_down_server?,
+                     :raise_down_error
 
       def initialize(attribs, client_options = {})
         hostname, port, socket_type, @weight, user_creds = ServerConfigParser.parse(attribs)
@@ -59,16 +60,17 @@ def lock!; end
 
       def unlock!; end
 
-      # Start reading key/value pairs from this connection. This is usually called
-      # after a series of GETKQ commands. A NOOP is sent, and the server begins
-      # flushing responses for kv pairs that were found.
+      # Get ready to read key/value pairs from this connection.
+      # This is usually called before or after the first GETKQ command.
       #
       # Returns nothing.
       def pipeline_response_setup
         verify_state(:getkq)
-        write_noop
         response_buffer.reset
-        @connection_manager.start_request!
+      end
+
+      def finish_pipeline_request
+        write_noop
       end
 
       # Attempt to receive and parse as many key/value pairs as possible
@@ -143,6 +145,14 @@ def quiet?
       end
       alias multi? quiet?
 
+      def pipelined_get_request(keys)
+        req = +String.new(capacity: pipelined_get_capacity(keys))
+        keys.each do |key|
+          req << quiet_get_request(key)
+        end
+        req
+      end
+
       # NOTE: Additional public methods should be overridden in Dalli::Threadsafe
 
       private
@@ -201,13 +211,14 @@ def connect
         raise
       end
 
-      def pipelined_get(keys)
-        req = +''
-        keys.each do |key|
-          req << quiet_get_request(key)
-        end
-        # Could send noop here instead of in pipeline_response_setup
-        write(req)
+      def pipelined_get(bytes)
+        write_nonblock(bytes)
+      rescue SystemCallError, Timeout::Error, EOFError => e
+        @connection_manager.error_on_request!(e)
+      end
+
+      def pipelined_get_capacity(keys)
+        (keys.size * request_header_size) + keys.reduce(0) { |acc, k| acc + k.size }
       end
 
       def response_buffer

lib/dalli/protocol/binary.rb

@@ -163,6 +163,10 @@ def write_noop
         write(req)
       end
 
+      def request_header_size
+        24
+      end
+
       require_relative 'binary/request_formatter'
       require_relative 'binary/response_header'
       require_relative 'binary/response_processor'

lib/dalli/protocol/connection_manager.rb

@@ -169,6 +169,10 @@ def read_nonblock
         @sock.read_available
       end
 
+      def write_nonblock(bytes)
+        @sock.write_nonblock(bytes, exception: false)
+      end
+
       def max_allowed_failures
         @max_allowed_failures ||= @options[:socket_max_failures] || 2
       end
@@ -247,6 +251,12 @@ def log_up_detected
         time = Time.now - @down_at
         Dalli.logger.warn { format('%<name>s is back (downtime was %<time>.3f seconds)', name: name, time: time) }
       end
+
+      def socket_sndbuf
+        @socket_sndbuf ||=
+          @sock&.getsockopt(::Socket::SOL_SOCKET, ::Socket::SO_SNDBUF)&.int ||
+          32_768
+      end
     end
   end
 end

lib/dalli/protocol/meta.rb

@@ -170,6 +170,10 @@ def authenticate_connection
         raise Dalli::DalliError, 'Authentication not supported for the meta protocol.'
       end
 
+      def request_header_size
+        17
+      end
+
       require_relative 'meta/key_regularizer'
       require_relative 'meta/request_formatter'
       require_relative 'meta/response_processor'

test/integration/test_pipelined_get.rb

@@ -79,6 +79,27 @@
         end
       end
 
+      it 'does not block for a large number of existing keys' do
+        memcached_persistent(p) do |dc|
+          dc.close
+          dc.flush
+
+          key_count = 200_000
+          range = 0...key_count
+          dc.quiet do
+            range.each { |i| dc.set(i, "foobar_#{i}") }
+          end
+
+          Timeout.timeout 60 do
+            resp = dc.get_multi(range.to_a)
+
+            assert_equal key_count, resp.count
+          end
+        rescue Timeout::Error
+          flunk "timed out while getting #{key_count} keys with get_multi"
+        end
+      end
+
       describe 'pipeline_next_responses' do
         it 'raises NetworkError when called before pipeline_response_setup' do
           memcached_persistent(p) do |dc|