From d656ac2b495263e925c4e1a362c0b3d5ff27bf4e Mon Sep 17 00:00:00 2001 From: Andrew Jones Date: Tue, 7 Mar 2023 15:53:30 -0800 Subject: [PATCH 1/3] fix get_multi socket corruption: verify/lock connection first Currently there is a race condition: if something happens between make_getkq_requests and finish_queries, the getkq request responses will be written to the socket, but the sockets @request_in_progress flag will not be set. This means that future gets may read corrupted/incorrect socket data. Instead we should conceive of the "request" as starting as soon as we go to write the getkq to the servers. Therefore we should verify the op and mark start_request! prior to sending the getkq ops. --- lib/dalli/options.rb | 6 ++++++ lib/dalli/pipelined_getter.rb | 1 + lib/dalli/protocol/base.rb | 9 +++++++-- 3 files changed, 14 insertions(+), 2 deletions(-) diff --git a/lib/dalli/options.rb b/lib/dalli/options.rb index a48cd4ba..b30b7630 100644 --- a/lib/dalli/options.rb +++ b/lib/dalli/options.rb @@ -31,6 +31,12 @@ def close end end + def pipeline_get_setup + @lock.synchronize do + super + end + end + def pipeline_response_setup @lock.synchronize do super diff --git a/lib/dalli/pipelined_getter.rb b/lib/dalli/pipelined_getter.rb index cc34bc5c..26261770 100644 --- a/lib/dalli/pipelined_getter.rb +++ b/lib/dalli/pipelined_getter.rb @@ -45,6 +45,7 @@ def setup_requests(keys) ## def make_getkq_requests(groups) groups.each do |server, keys_for_server| + server.pipeline_get_setup server.request(:pipelined_get, keys_for_server) rescue DalliError, NetworkError => e Dalli.logger.debug { e.inspect } diff --git a/lib/dalli/protocol/base.rb b/lib/dalli/protocol/base.rb index 89c4fd25..cd167186 100644 --- a/lib/dalli/protocol/base.rb +++ b/lib/dalli/protocol/base.rb @@ -59,16 +59,21 @@ def lock!; end def unlock!; end + # verify and start request before sending GETKQ commands, + # otherwise the socket could get corrupted if we fail to reach pipline_response stage + def pipeline_get_setup + verify_state(:getkq) + @connection_manager.start_request! + 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. # # Returns nothing. def pipeline_response_setup - verify_state(:getkq) write_noop response_buffer.reset - @connection_manager.start_request! end # Attempt to receive and parse as many key/value pairs as possible From 4395df52514c6e7728bad5b0b93ed42e7528d941 Mon Sep 17 00:00:00 2001 From: Andrew Jones Date: Tue, 7 Mar 2023 16:05:32 -0800 Subject: [PATCH 2/3] introduce write_nonblock for pipelined_get so we can start_request! before get_multi's write operations --- lib/dalli/protocol/base.rb | 2 +- lib/dalli/protocol/connection_manager.rb | 10 ++++++++++ 2 files changed, 11 insertions(+), 1 deletion(-) diff --git a/lib/dalli/protocol/base.rb b/lib/dalli/protocol/base.rb index cd167186..835bc1ff 100644 --- a/lib/dalli/protocol/base.rb +++ b/lib/dalli/protocol/base.rb @@ -210,7 +210,7 @@ def pipelined_get(keys) req << quiet_get_request(key) end # Could send noop here instead of in pipeline_response_setup - write(req) + write_nonblock(req) end def response_buffer diff --git a/lib/dalli/protocol/connection_manager.rb b/lib/dalli/protocol/connection_manager.rb index 5a0b6ff0..e2ee1ab0 100644 --- a/lib/dalli/protocol/connection_manager.rb +++ b/lib/dalli/protocol/connection_manager.rb @@ -171,6 +171,16 @@ def read_nonblock @sock.read_available end + # Non-blocking write. Should only be used in the context + # of a caller who has called start_request!, but not yet + # called finish_request!. Here to support the operation + # of the get_multi operation. + def write_nonblock + @sock.write(bytes) + rescue SystemCallError, Timeout::Error => e + error_on_request!(e) + end + def max_allowed_failures @max_allowed_failures ||= @options[:socket_max_failures] || 2 end From ea158770291456457eb7cd7012c726c9fdae4557 Mon Sep 17 00:00:00 2001 From: Andrew Jones Date: Tue, 7 Mar 2023 16:53:24 -0800 Subject: [PATCH 3/3] move setup and noop into pipelined_get to avoid blocking issues --- lib/dalli/protocol/base.rb | 7 +++---- lib/dalli/protocol/binary.rb | 8 ++++++-- lib/dalli/protocol/connection_manager.rb | 2 +- lib/dalli/protocol/meta.rb | 8 ++++++-- 4 files changed, 16 insertions(+), 9 deletions(-) diff --git a/lib/dalli/protocol/base.rb b/lib/dalli/protocol/base.rb index 835bc1ff..8f8b7d65 100644 --- a/lib/dalli/protocol/base.rb +++ b/lib/dalli/protocol/base.rb @@ -18,7 +18,7 @@ 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, :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) @@ -63,7 +63,6 @@ def unlock!; end # otherwise the socket could get corrupted if we fail to reach pipline_response stage def pipeline_get_setup verify_state(:getkq) - @connection_manager.start_request! end # Start reading key/value pairs from this connection. This is usually called @@ -72,7 +71,6 @@ def pipeline_get_setup # # Returns nothing. def pipeline_response_setup - write_noop response_buffer.reset end @@ -209,8 +207,9 @@ def pipelined_get(keys) keys.each do |key| req << quiet_get_request(key) end - # Could send noop here instead of in pipeline_response_setup + @connection_manager.start_request! write_nonblock(req) + write_noop(nonblock: true) end def response_buffer diff --git a/lib/dalli/protocol/binary.rb b/lib/dalli/protocol/binary.rb index 66f71516..4d549d20 100644 --- a/lib/dalli/protocol/binary.rb +++ b/lib/dalli/protocol/binary.rb @@ -158,9 +158,13 @@ def version response_processor.version end - def write_noop + def write_noop(nonblock: false) req = RequestFormatter.standard_request(opkey: :noop) - write(req) + if nonblock + write_nonblock(req) + else + write(req) + end end require_relative 'binary/request_formatter' diff --git a/lib/dalli/protocol/connection_manager.rb b/lib/dalli/protocol/connection_manager.rb index e2ee1ab0..0ec18239 100644 --- a/lib/dalli/protocol/connection_manager.rb +++ b/lib/dalli/protocol/connection_manager.rb @@ -175,7 +175,7 @@ def read_nonblock # of a caller who has called start_request!, but not yet # called finish_request!. Here to support the operation # of the get_multi operation. - def write_nonblock + def write_nonblock(bytes) @sock.write(bytes) rescue SystemCallError, Timeout::Error => e error_on_request!(e) diff --git a/lib/dalli/protocol/meta.rb b/lib/dalli/protocol/meta.rb index b2e66c37..8dd293d0 100644 --- a/lib/dalli/protocol/meta.rb +++ b/lib/dalli/protocol/meta.rb @@ -162,8 +162,12 @@ def version response_processor.version end - def write_noop - write(RequestFormatter.meta_noop) + def write_noop(nonblock: false) + if nonblock + write_nonblock(RequestFormatter.meta_noop) + else + write(RequestFormatter.meta_noop) + end end def authenticate_connection