diff --git a/CHANGELOG.md b/CHANGELOG.md index 603559f0..151be551 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,15 @@ Unreleased - Add experimental per-request `exp_async_timeout_recovery` option: reads wait with explicit deadlines and poll the socket once more before timing out, so replies delayed only by a busy fiber scheduler are not discarded. (nherson) +- `exp_async_timeout_recovery` reads values of 8 KB or more straight from the socket in reads of up to 64 KiB + instead of through the 8 KB line buffer, cutting a 2 MB get from ~259 to ~37 syscalls. (nherson) + +- Fix `exp_async_timeout_recovery` replaying the previous reply instead of reporting EOF after a TLS server + closes the connection cleanly. (nherson) + +- Fix `exp_async_timeout_recovery` failing a read whose data is ready when the poll after a timed-out wait + misses it, which a zero-timeout poll can do while the thread has a pending interrupt. (nherson) + - Add per-request opaque-correlation controls and a client `opaque_correlation_request_only` mode for opt-in rollouts, with eager PRNG initialization and unchanged unflagged reads. (mrattle) diff --git a/bin/benchmark b/bin/benchmark index 8a9ba396..bb3150f7 100755 --- a/bin/benchmark +++ b/bin/benchmark @@ -7,11 +7,14 @@ # run with: # bundle exec bin/benchmark # RUBY_YJIT_ENABLE=1 BENCH_TARGET=get bundle exec bin/benchmark +# BENCH_TARGET=async_get bundle exec bin/benchmark (not part of BENCH_TARGET=all) require 'bundler/inline' require 'json' gemfile do source 'https://rubygems.org' + # Only async_get needs it; other runs keep the same gems as before that target existed. + gem 'async' if ENV['BENCH_TARGET'] == 'async_get' gem 'benchmark-ips' gem 'logger' gem 'opentelemetry-sdk' @@ -234,3 +237,16 @@ if %w[all delete_multi].include?(bench_target) x.compare! end end + +# Large-value gets under a fiber scheduler, through the default read path and the +# exp_async_timeout_recovery reader. Runs only when selected, so BENCH_TARGET=all is unchanged. +if bench_target == 'async_get' + Sync do + Benchmark.ips do |x| + x.config(warmup: bench_warmup, time: bench_time) + x.report('get dalli async') { client.get('key') } + x.report('get dalli async timeout recovery') { client.get('key', exp_async_timeout_recovery: true) } + x.compare! + end + end +end diff --git a/lib/dalli/protocol/timeout_recovery_reader.rb b/lib/dalli/protocol/timeout_recovery_reader.rb index e5d7b7d9..c2c00c31 100644 --- a/lib/dalli/protocol/timeout_recovery_reader.rb +++ b/lib/dalli/protocol/timeout_recovery_reader.rb @@ -8,6 +8,9 @@ module Protocol ## class TimeoutRecoveryReader CHUNK_SIZE = 8192 + # Most a single large-value read asks for. Bounds the per-connection scratch + # buffer at 64 KiB while cutting a 2 MB value from ~257 reads to ~35. + LARGE_READ_SIZE = 65_536 def initialize(sock, timeout) @sock = sock @@ -36,7 +39,13 @@ def read(count) data = String.new(capacity: count, encoding: Encoding::BINARY) data << @buffer.byteslice(@offset, available) if available.positive? @offset = @buffer.bytesize - append_from_socket(data, count - available) + remaining = count - available + # Large values bypass the 8 KB line buffer: each read asks the socket for at + # most what is still owed, so a value costs a few syscalls rather than one per + # CHUNK_SIZE bytes plus a copy out of @buffer, and never reads past its end. + return append_directly(data, remaining) if remaining >= CHUNK_SIZE + + append_from_socket(data, remaining) end private @@ -59,31 +68,56 @@ def append_from_socket(data, remaining) data end + def append_directly(data, remaining) + while remaining.positive? + chunk = if data.empty? + read_partial(remaining, data) + else + read_partial([remaining, LARGE_READ_SIZE].min, scratch) + end + return data.empty? ? nil : data unless chunk + + data << chunk unless chunk.equal?(data) + remaining -= chunk.bytesize + end + data + end + + def scratch + @scratch ||= String.new(encoding: Encoding::BINARY) + end + + # Reads at most maxlen bytes into outbuf, waiting while the socket has none; nil at EOF. + def read_partial(maxlen, outbuf) + last_try = false + while (result = @sock.read_nonblock(maxlen, outbuf, exception: false)).is_a?(Symbol) + raise IO::TimeoutError, "IO timeout after #{@timeout}s" if last_try + + last_try = !wait_for(result) + end + result + end + # read_nonblock overwrites the buffer, so unconsumed bytes are set aside and restored - # after the read; @offset only moves once the read returns. + # after the read; @offset only moves once the read returns. At EOF a socket empties the + # buffer but OpenSSL leaves it untouched, so it is cleared here to avoid replaying old bytes. def fill partial = @buffer.byteslice(@offset, @buffer.bytesize - @offset) if pending? - filled = read_chunk + filled = !read_partial(CHUNK_SIZE, @buffer).nil? + @buffer.clear unless filled @offset = 0 - if partial - filled ? @buffer.prepend(partial) : @buffer.replace(partial) - end + @buffer.prepend(partial) if partial filled end - # Waits on :wait_readable or :wait_writable, which name the socket method to call. - def read_chunk - while (result = @sock.read_nonblock(CHUNK_SIZE, @buffer, exception: false)).is_a?(Symbol) - wait_for(result) - end - !result.nil? - end - + # Waits on :wait_readable or :wait_writable, which name the socket method to call. After a + # wait that times out, polls once outside the fiber scheduler. Returns false when that poll + # finds nothing too, and the caller then reads once more before failing: a zero-timeout poll + # can report nothing without checking the socket while the thread has a pending interrupt. def wait_for(event) - return if @sock.public_send(event, @timeout) - return if Fiber.blocking { @sock.public_send(event, 0) } + return true if @sock.public_send(event, @timeout) - raise IO::TimeoutError, "IO timeout after #{@timeout}s" + Fiber.blocking { @sock.public_send(event, 0) } ? true : false end end end diff --git a/test/integration/test_async_timeout_recovery.rb b/test/integration/test_async_timeout_recovery.rb index 01ae7a0c..4088b446 100644 --- a/test/integration/test_async_timeout_recovery.rb +++ b/test/integration/test_async_timeout_recovery.rb @@ -7,8 +7,22 @@ # Replies from a separate process, so they arrive while a fiber in this process holds the thread. # A key named "delay-..." delays its reply; every get writes one byte to #signal on receipt. +# A key containing "size" gets an n-byte value; with "parts" and "gap" the value is +# written in n parts with a pause before each later part. Halfway through each pause, once the +# client has drained the previous part and is waiting, a byte is written to #signal. class DelayedReplyServer - SCRIPT = <<~'RUBY' + VALUE_OF = <<~'RUBY' + def value_of(key) + size = key[/size(\d+)/, 1]&.to_i + return "value-of-#{key}" unless size + + pattern = "EN\r\nVA 3 k\r\n\0\xFF".b + (pattern * ((size / pattern.bytesize) + 1)).byteslice(0, size) + end + RUBY + singleton_class.class_eval(VALUE_OF) + + SCRIPT = VALUE_OF + <<~'RUBY' require 'socket' listener = UNIXServer.new(ARGV.fetch(0)) signal = IO.for_fd(3, autoclose: false) @@ -24,8 +38,19 @@ class DelayedReplyServer when 'mg' signal.write('.') sleep(key[/\Adelay(\d+)ms/, 1].to_i / 1000.0) - value = "value-of-#{key}" - connection.write("VA #{value.bytesize}#{" k#{key}" if flags.include?('k')}\r\n#{value}\r\n") + value = value_of(key) + parts = key[/parts(\d+)/, 1]&.to_i || 1 + part_size = (value.bytesize + parts - 1) / parts + first, *rest = Array.new(parts) { |i| value.byteslice(i * part_size, part_size) } + connection.write("VA #{value.bytesize}#{" k#{key}" if flags.include?('k')}\r\n", first, + rest.empty? ? "\r\n" : '') + pause = key[/gap(\d+)ms/, 1].to_i / 2000.0 + rest.each_with_index do |part, index| + sleep(pause) + signal.write('.') + sleep(pause) + connection.write(part, index == rest.size - 1 ? "\r\n" : '') + end else raise "Unexpected command: #{command}" end end @@ -144,6 +169,106 @@ def elapsed end end +module LargeValueHelpers + # More than a UNIX socket buffer holds, so the server's write of a first half returns only once the + # client is reading inside the value. + LARGE_SIZE = 900_000 + + ## + # Records reads and waits on a connection's socket. + ## + class SocketLog + def self.now + Process.clock_gettime(Process::CLOCK_MONOTONIC) + end + + def initialize(sock) + events = @events = [] + sock.singleton_class.prepend(Module.new do + define_method(:read_nonblock) do |*args, **kwargs| + result = super(*args, **kwargs) + events << [:read, result.is_a?(String) ? result.bytesize : 0, SocketLog.now] + result + end + + define_method(:wait_readable) do |timeout = nil| + result = super(timeout) + events << [:wait, timeout, result ? true : false] + result + end + end) + end + + def last_data_at + @events.reverse_each.find { |kind, bytes| kind == :read && bytes.positive? }&.last + end + + # Bytes received before each wait that reported a timeout after which the reader still found + # data: through the poll that follows or, when that poll misses it, the read after the poll. + def recoveries(timeout) + received = 0 + @events.each_cons(3).filter_map do |event, poll, after| + received += event[1] if event.first == :read + received if event == [:wait, timeout, false] && found_data?(poll, after) + end + end + + private + + def found_data?(poll, after) + poll == [:wait, 0, true] || (poll == [:wait, 0, false] && after.first == :read && after[1].positive?) + end + end + + def in_scheduler(selector, &) + return yield unless selector + + with_selector(selector, &) + end + + def assert_reads_through_pauses(client, selector) + key = "size#{LARGE_SIZE}-parts6-gap30ms" + value = nil + duration = in_scheduler(selector) do + client.alive! + elapsed { value = client.get(key, recover) } + end + + assert_equal DelayedReplyServer.value_of(key), value + # Five 30ms pauses: none outlasts the timeout, together they do. + assert_operator duration, :>, 2 * socket_timeout + end + + def assert_fails_within_timeout_of_last_byte(client, selector) + error = hang = nil + in_scheduler(selector) do + client.alive! + log = SocketLog.new(connection_manager(client).sock) + client.get("size#{LARGE_SIZE}-parts2-gap500ms", recover) + rescue Dalli::NetworkError => e + error = e + hang = SocketLog.now - log.last_data_at + end + + assert_kind_of Dalli::NetworkError, error + assert_match(/IO::TimeoutError/, error.message) + assert_operator hang, :>=, socket_timeout + assert_operator hang, :<, socket_timeout + 0.1 + key = "size#{LARGE_SIZE}-after" + + assert_equal DelayedReplyServer.value_of(key), in_scheduler(selector) { client.get(key, recover) } + end + + # A fiber that wakes on the server's second signal, sent while the reader waits inside the value + # for its second half, and holds the thread until after that half arrives. + def start_busy_fiber_inside_value(task, signal) + task.async do + 2.times { await_signal(signal) } + busy + end + end +end + module RecoveryParityHelpers MULTI_KEYS = (Array.new(40) { |i| "multi-#{i}" } << 'missing').freeze @@ -188,6 +313,7 @@ def assert_same_results(client) describe 'exp_async_timeout_recovery' do include AsyncTimeoutRecoveryHelpers include RecoveryParityHelpers + include LargeValueHelpers describe 'without a fiber scheduler' do it 'returns the same results as default reads over TCP' do @@ -247,6 +373,14 @@ def assert_same_results(client) assert_equal 'value-of-b', client.get('b', recover) end end + + it 'reads a large value delivered in pauses that are each shorter than the timeout' do + with_delayed_reply_client { |client| assert_reads_through_pauses(client, nil) } + end + + it 'fails a large value within one timeout once it stops arriving' do + with_delayed_reply_client { |client| assert_fails_within_timeout_of_last_byte(client, nil) } + end end AsyncTimeoutRecoveryHelpers::SELECTORS.each do |selector| @@ -297,6 +431,51 @@ def assert_same_results(client) end end + it 'keeps reading a large value whose rest arrives while another fiber holds the thread' do + with_delayed_reply_client do |client, server| + key = "size#{LargeValueHelpers::LARGE_SIZE}-parts2-gap40ms" + value = sock = log = nil + duration = with_selector(selector) do |task| + client.alive! + sock = connection_manager(client).sock + log = LargeValueHelpers::SocketLog.new(sock) + start_busy_fiber_inside_value(task, server.signal) + elapsed { value = client.get(key, recover) } + end + recoveries = log.recoveries(socket_timeout) + + assert_equal DelayedReplyServer.value_of(key), value + assert_operator duration, :>, socket_timeout + assert_same sock, connection_manager(client).sock + assert recoveries.any? { |received| received.between?(100, LargeValueHelpers::LARGE_SIZE) }, + "expected a wait inside the value to time out and find data; recovered after #{recoveries} bytes" + end + end + + it 'fails a large value without the option when another fiber holds the thread inside it' do + with_delayed_reply_client do |client, server| + error = with_selector(selector) do |task| + client.alive! + start_busy_fiber_inside_value(task, server.signal) + client.get("size#{LargeValueHelpers::LARGE_SIZE}-parts2-gap40ms") + nil + rescue Dalli::NetworkError => e + e + end + + assert_kind_of Dalli::NetworkError, error + assert_match(/IO::TimeoutError/, error.message) + end + end + + it 'reads a large value delivered in pauses that are each shorter than the timeout' do + with_delayed_reply_client { |client| assert_reads_through_pauses(client, selector) } + end + + it 'fails a large value within one timeout once it stops arriving' do + with_delayed_reply_client { |client| assert_fails_within_timeout_of_last_byte(client, selector) } + end + it 'still fails a read when no reply arrives before the timeout' do with_delayed_reply_client do |client| error = nil diff --git a/test/protocol/test_timeout_recovery_reader.rb b/test/protocol/test_timeout_recovery_reader.rb index 9e96d7e0..bef81355 100644 --- a/test/protocol/test_timeout_recovery_reader.rb +++ b/test/protocol/test_timeout_recovery_reader.rb @@ -2,6 +2,7 @@ require_relative '../helper' require 'objspace' +require_relative '../utils/ssl_socket_pair' describe Dalli::Protocol::TimeoutRecoveryReader do def with_reader(timeout: 0.05) @@ -12,6 +13,18 @@ def with_reader(timeout: 0.05) peer&.close unless peer&.closed? end + def elapsed + started = Process.clock_gettime(Process::CLOCK_MONOTONIC) + yield + Process.clock_gettime(Process::CLOCK_MONOTONIC) - started + end + + # Default UNIX socket buffers (8 KB on macOS) cap each read regardless of the reader. + def enlarge_buffers(sock, peer) + peer.setsockopt(Socket::SOL_SOCKET, Socket::SO_SNDBUF, 1 << 20) + sock.setsockopt(Socket::SOL_SOCKET, Socket::SO_RCVBUF, 1 << 20) + end + def write_in_background(peer, *chunks) Thread.new do chunks.each { |chunk| peer.write(chunk) } @@ -81,6 +94,19 @@ def write_in_background(peer, *chunks) end end + it 'reads data that is ready when the poll after a timed-out wait misses it' do + with_reader do |reader, peer, sock| + # A zero-timeout poll can report nothing without checking the socket while the thread has a + # pending interrupt, such as one from Thread#wakeup or a queue push. + sock.define_singleton_method(:wait_readable) do |timeout| + peer.write("HD\r\n") if timeout.positive? + nil + end + + assert_equal "HD\r\n", reader.gets("\r\n") + end + end + it 'keeps its buffer at the chunk size after a large read' do value = SecureRandom.random_bytes(1_000_000) with_reader do |reader, peer| @@ -95,6 +121,109 @@ def write_in_background(peer, *chunks) end end + it 'reads a large value in a few socket reads and stops at its end' do + chunk = Dalli::Protocol::TimeoutRecoveryReader::CHUNK_SIZE + value = SecureRandom.random_bytes((chunk * 40) + 3) + with_reader do |reader, peer, sock| + reads = 0 + counter = Module.new do + define_method(:read_nonblock) do |*args, **kwargs| + reads += 1 + super(*args, **kwargs) + end + end + sock.singleton_class.prepend(counter) + enlarge_buffers(sock, peer) + half = value.bytesize / 2 + write_in_background(peer, "VA #{value.bytesize}\r\n", value.byteslice(0, half), + value.byteslice(half..), "\r\nEN\r\n").join + + assert_equal "VA #{value.bytesize}\r\n", reader.gets("\r\n") + assert_equal value, reader.read(value.bytesize) + assert_operator reads, :<, 20 + assert_equal "\r\n", reader.read(2) + assert_equal "EN\r\n", reader.gets("\r\n") + assert_nil reader.gets("\r\n") + end + end + + it 'returns the bytes received when EOF cuts a large value short' do + chunk = Dalli::Protocol::TimeoutRecoveryReader::CHUNK_SIZE + value = SecureRandom.random_bytes(chunk * 3) + with_reader do |reader, peer| + writer = write_in_background(peer, value) + + assert_equal value, reader.read(chunk * 5) + assert_nil reader.read(chunk * 5) + writer.join + end + end + + it 'keeps reading a large value when a wait in the middle of it reports a timeout while data is ready' do + chunk = Dalli::Protocol::TimeoutRecoveryReader::CHUNK_SIZE + value = SecureRandom.random_bytes(chunk * 8) + with_reader(timeout: 0.05) do |reader, peer, sock| + enlarge_buffers(sock, peer) + peer.write(value.byteslice(0, chunk * 3)) + timed_out = [] + # A scheduler that resumes the reader after the deadline reports a timeout even though the + # rest of the value has arrived by then. + sock.define_singleton_method(:wait_readable) do |timeout| + return super(timeout) if timeout.zero? + + timed_out << timeout + peer.write(value.byteslice((chunk * 3)..)) if timed_out.one? + nil + end + + assert_equal value, reader.read(value.bytesize) + assert_equal [0.05], timed_out + end + end + + it 'fails within one timeout once a large value stops arriving' do + chunk = Dalli::Protocol::TimeoutRecoveryReader::CHUNK_SIZE + with_reader(timeout: 0.2) do |reader, peer, sock| + enlarge_buffers(sock, peer) + peer.write(SecureRandom.random_bytes(chunk * 3)) + duration = elapsed { Timeout.timeout(5) { assert_raises(IO::TimeoutError) { reader.read(chunk * 8) } } } + + # One wait for the timeout, then a poll that does not wait. + assert_operator duration, :>=, 0.2 + assert_operator duration, :<, 0.3 + end + end + + it 'gives each wait a full timeout while a large value arrives in pauses' do + chunk = Dalli::Protocol::TimeoutRecoveryReader::CHUNK_SIZE + parts = Array.new(6) { SecureRandom.random_bytes(chunk * 2) } + with_reader(timeout: 0.1) do |reader, peer| + writer = Thread.new do + parts.each do |part| + peer.write(part) + sleep(0.03) + end + end + duration = elapsed { assert_equal parts.join, reader.read(parts.sum(&:bytesize)) } + + assert_operator duration, :>, 0.1 + writer.join + end + end + + it 'returns nil at EOF instead of replaying earlier bytes when a TLS peer closes' do + sock, peer = SSLSocketPair.open + reader = Dalli::Protocol::TimeoutRecoveryReader.new(sock, 0.05) + peer.write("HD\r\n") + peer.close + + assert_equal "HD\r\n", reader.gets("\r\n") + assert_nil reader.gets("\r\n") + assert_nil reader.read(1) + ensure + sock&.close + end + it 'allocates one string per buffered line' do with_reader do |reader, peer| peer.write("HD\r\n" * 101) diff --git a/test/protocol/test_timeout_recovery_reader_fuzz.rb b/test/protocol/test_timeout_recovery_reader_fuzz.rb new file mode 100644 index 00000000..529382cf --- /dev/null +++ b/test/protocol/test_timeout_recovery_reader_fuzz.rb @@ -0,0 +1,605 @@ +# frozen_string_literal: true + +require_relative '../helper' +require 'objspace' +require 'zlib' +require_relative '../utils/ssl_socket_pair' +require 'async' if Gem::Version.new(RUBY_VERSION) >= Gem::Version.new('3.3') + +# Randomized checks of TimeoutRecoveryReader. Each case streams generated bytes from a thread in +# random pieces while the reader's socket injects short reads, spurious waits, and waits that report +# a timeout although data is ready (what a fiber scheduler does when it resumes the reader late). +# Results are compared with IO#gets/IO#read semantics over the same bytes. +# +# FUZZ_SEED= replays a run, FUZZ_ITERATIONS= scales it, and FUZZ_STATS=1 prints coverage. +module TimeoutRecoveryReaderFuzz + Reader = Dalli::Protocol::TimeoutRecoveryReader + CHUNK = Reader::CHUNK_SIZE + LARGE = Reader::LARGE_READ_SIZE + SEED = Integer(ENV.fetch('FUZZ_SEED') { Random.new_seed.to_s }) + ITERATIONS = Integer(ENV.fetch('FUZZ_ITERATIONS', '8')) + STATS = Hash.new(0) + + TRANSPORTS = %i[unix tcp ssl].freeze + SELECTORS = + if defined?(IO::Event::Selector) + %i[URing EPoll KQueue Select].select { |name| IO::Event::Selector.const_defined?(name) } + else + [] + end + SCHEDULERS = [nil, *SELECTORS].freeze + + # :exact never stalls, :stall stops the stream for good, :pauses stops it for under a timeout at a time. + TIMEOUTS = { exact: 1.0, stall: 0.1, pauses: 0.2 }.freeze + PAUSES = 3 + PAUSE_FRACTION = 0.4 + # How long past the timeout a stalled read may take to fail. + TIMEOUT_SLACK = 0.1 + CASE_LIMIT = 30 + + SIZES = [0, 1, 2, CHUNK - 1, CHUNK, CHUNK + 1, (2 * CHUNK) + 1, LARGE - 1, LARGE, LARGE + 1, + (3 * LARGE) + CHUNK - 1].freeze + PATTERNS = ["\r\n", "\r", "\n", "EN\r\n", "VA 12 k\r\n", "MN\r\n", "\0", "\xFF\xFE", 'é😀'].map(&:b).freeze + SEPARATORS = ["\r\n", "\n"].freeze + + def self.now + Process.clock_gettime(Process::CLOCK_MONOTONIC) + end + + ## + # IO#gets / IO#read semantics over a fixed byte string. + ## + class Model + attr_reader :pos + + def initialize(stream) + @stream = stream + @pos = 0 + end + + def remaining + @stream.bytesize - @pos + end + + def rest + @stream.byteslice(@pos..) + end + + # Stream offset an operation must see before it can return, or nil when it needs EOF. + def need(kind, arg) + if kind == :gets + index = @stream.byteindex(arg, @pos) + index && (index + arg.bytesize) + elsif arg.zero? + @pos + elsif arg <= remaining + @pos + arg + end + end + + def apply(kind, arg) + if kind == :gets + return nil if remaining.zero? + + index = @stream.byteindex(arg, @pos) + take(index ? index + arg.bytesize - @pos : remaining) + else + return take(0) if arg.zero? + return nil if remaining.zero? + + take([arg, remaining].min) + end + end + + private + + def take(length) + data = @stream.byteslice(@pos, length) + @pos += length + data + end + end + + ## + # Fault injection and bookkeeping for one reader socket. + ## + class Faults + # Recent reads and waits, for failure messages. + HISTORY = 24 + + attr_reader :waits, :delivered, :last_data_at + + def initialize(rng) + @rng = rng + @spurious = [0.0, 0.05, 0.2].sample(random: rng) + @writable = [0.0, 0.02].sample(random: rng) + @short = [0.0, 0.2, 0.5].sample(random: rng) + @starve = [0.0, 0.3, 1.0].sample(random: rng) + @waits = [] + @history = [] + @delivered = 0 + mark + end + + def attach(sock) + sock.singleton_class.prepend(Hooks) + sock.instance_variable_set(:@fuzz_faults, self) + end + + def mark + @last_data_at = TimeoutRecoveryReaderFuzz.now + end + + # The length to request, or the Symbol a spurious EAGAIN would return. The read after a poll + # that found nothing is the reader's last check before failing, so it gets no spurious EAGAIN: + # that would fail a read whose data is there, which no socket does. + def read_length(sock, maxlen) + STATS[:large_reads] += 1 if maxlen > CHUNK + last_check = @last_check + @last_check = false + roll = @rng.rand + spurious = spurious_result(sock, roll) unless last_check + return spurious.tap { record(:read, maxlen, :spurious, spurious) } if spurious + + roll > 1 - @short ? @rng.rand(1..maxlen) : maxlen + end + + # A TLS socket holding decrypted bytes always returns them, so it never gets a spurious EAGAIN. + def spurious_result(sock, roll) + return if sock.respond_to?(:pending) && sock.pending.positive? + return :wait_readable if roll < @spurious + + :wait_writable if roll < @spurious + @writable + end + + def read(maxlen, length) + result = yield + record(:read, maxlen, length, result.is_a?(String) ? result.bytesize : result.inspect) + return result unless result.is_a?(String) + + @delivered += result.bytesize + mark + result + end + + # Reports a timeout without waiting when the socket is ready and the roll says the scheduler + # resumed the reader late; otherwise waits. + def wait(sock, event, timeout) + started = TimeoutRecoveryReaderFuzz.now + @waits << [timeout, Fiber.blocking?, !Fiber.scheduler.nil?] + starved = starve?(sock, event, timeout) + result = starved ? nil : yield + @last_check = timeout&.zero? && !result + record(event, timeout, starved ? :starved : :waited, result ? :ready : result.inspect, + (TimeoutRecoveryReaderFuzz.now - started).round(4)) + result + end + + def history + @history.map { |event| event.join(' ') }.join("\n") + end + + private + + def starve?(sock, event, timeout) + return false if timeout.nil? || timeout.zero? || @rng.rand >= @starve + return false unless Fiber.blocking { IO.instance_method(event).bind_call(sock.to_io, 0) } + + STATS[:starved_waits] += 1 + true + end + + def record(*event) + @history << [(TimeoutRecoveryReaderFuzz.now - @last_data_at).round(4), *event] + @history.shift if @history.size > HISTORY + end + end + + ## + # Prepended to the reader's socket. + ## + module Hooks + def read_nonblock(maxlen, buf = nil, exception: true) + length = @fuzz_faults.read_length(self, maxlen) + return length if length.is_a?(Symbol) + + @fuzz_faults.read(maxlen, length) { super(length, buf, exception: exception) } + end + + def wait_readable(timeout = nil) + @fuzz_faults.wait(self, :wait_readable, timeout) { super } + end + + def wait_writable(timeout = nil) + @fuzz_faults.wait(self, :wait_writable, timeout) { super } + end + end + + ## + # Byte streams and the reads to run against them. + ## + module Scripts + module_function + + def payload(rng) + size = case rng.rand(6) + when 0, 1 then SIZES.sample(random: rng) + when 2 then rng.rand(1..(4 * CHUNK)) + when 3, 4 then rng.rand(1..(4 * LARGE)) + else rng.rand(1..1_500_000) + end + return rng.bytes(size) if rng.rand(3).zero? + + unit = PATTERNS.sample(rng.rand(1..3), random: rng).join.b + (unit * ((size / unit.bytesize) + 1)).byteslice(0, size) + end + + # meta-protocol-shaped replies: header line, value, terminator, then EN and EOF. + def protocol_script(rng, with_large_value: false) + stream = String.new(encoding: Encoding::BINARY) + ops = protocol_values(rng, with_large_value).flat_map do |value| + stream << "VA #{value.bytesize} t#{rng.rand(1000)}\r\n" << value << "\r\n" + [[:gets, SEPARATORS.sample(random: rng)], [:read, value.bytesize], [:read, 2]] + end + stream << "EN\r\n" + [stream, ops + [[:gets, "\r\n"], [:gets, "\r\n"], [:read, 1]]] + end + + def protocol_values(rng, with_large_value) + values = Array.new(rng.rand(1..4)) { payload(rng) } + return values unless with_large_value && values.none? { |value| value.bytesize >= 3 * CHUNK } + + values << rng.bytes(rng.rand((3 * CHUNK)..(3 * LARGE))) + end + + # Arbitrary bytes and operations, including reads of exactly the rest and past EOF. + def random_script(rng) + stream = Array.new(rng.rand(1..4)) { payload(rng) }.join.b + model = Model.new(stream) + ops = Array.new(rng.rand(4..24)) do + op = rng.rand(3).zero? ? [:gets, SEPARATORS.sample(random: rng)] : [:read, read_size(rng, model.remaining)] + model.apply(*op) + op + end + [stream, ops] + end + + def read_size(rng, remaining) + case rng.rand(4) + when 0 then remaining + when 1 then remaining + rng.rand(1..CHUNK) + when 2 then SIZES.sample(random: rng) + else rng.rand(0..(3 * LARGE)) + end + end + end + + ## + # Writer steps: [:write, bytes], [:pause, seconds], or [:stall] until released. + ## + module Plans + module_function + + # The steps, plus the offset where a :stall stream stops. + def build(mode, rng, stream, ops, index) + case mode + when :exact then [exact(rng, stream), nil] + when :pauses then [pauses(rng, stream), nil] + when :stall + stall_at = index.even? ? stall_inside_large_read(rng, stream, ops) : rng.rand(0..stream.bytesize) + [pieces(rng, stream.byteslice(0, stall_at)) + [[:stall]] + pieces(rng, stream.byteslice(stall_at..)), stall_at] + end + end + + def exact(rng, stream) + pieces(rng, stream).flat_map { |step| rng.rand(8).zero? ? [[:pause, rng.rand * 0.002], step] : [step] } + end + + # Pauses at distinct offsets, each shorter than the timeout. + def pauses(rng, stream) + cuts = Set.new + cuts << rng.rand(1...stream.bytesize) while cuts.size < PAUSES + pause = [:pause, TIMEOUTS[:pauses] * PAUSE_FRACTION] + segments = [0, *cuts.sort, stream.bytesize].each_cons(2) + segments.flat_map { |from, to| pieces(rng, stream.byteslice(from...to)) << pause }[0...-1] + end + + # An offset at least two chunks into a value of three or more, so the reader is past its line + # buffer and reading the value straight from the socket when the stream stops. + def stall_inside_large_read(rng, stream, ops) + model = Model.new(stream) + reads = ops.filter_map do |kind, arg| + start = model.pos + model.apply(kind, arg) + [start, arg] if kind == :read && arg >= 3 * CHUNK + end + start, length = reads.sample(random: rng) + start + rng.rand((2 * CHUNK)...length) + end + + def pieces(rng, bytes) + steps = [] + offset = 0 + while offset < bytes.bytesize + length = case rng.rand(5) + when 0 then rng.rand(1..16) + when 1 then [CHUNK - 1, CHUNK, CHUNK + 1, LARGE - 1, LARGE + 1].sample(random: rng) + when 2 then rng.rand(1..(2 * LARGE)) + else rng.rand(1..400_000) + end + steps << [:write, bytes.byteslice(offset, length)] + offset += length + end + steps + end + end + + ## + # Socket pairs and the thread that writes to the peer. + ## + module Wire + module_function + + def socket_pair(transport) + case transport + when :unix then UNIXSocket.pair + when :tcp + server = TCPServer.new('127.0.0.1', 0) + client = TCPSocket.new('127.0.0.1', server.local_address.ip_port) + [client, server.accept] + when :ssl then SSLSocketPair.open + end + ensure + server&.close + end + + def start_writer(peer, steps, release) + Thread.new do + Thread.current.report_on_exception = false + steps.each do |step, arg| + case step + when :write then peer.write(arg) + when :pause then sleep(arg) + when :stall then release.pop + end + end + rescue IOError, SystemCallError, OpenSSL::SSL::SSLError + nil # the reader gave up and closed its end + ensure + close_quietly(peer) + end + end + + def close_quietly(io) + io&.close + rescue StandardError + nil + end + end + + ## + # One case: the reader under test, its faults, the model, and where the stream stalls. + ## + Case = Struct.new(:label, :reader, :faults, :model, :ops, :stall_at, :timeout) do + def returned + @returned ||= 0 + end + + def returned!(bytes) + @returned = returned + (bytes ? bytes.bytesize : 0) + end + + def buffered + faults.delivered - returned + end + + def stalls_before?(kind, arg) + return false unless stall_at + + need = model.need(kind, arg) + need.nil? || need > stall_at + end + end + + ## + # Assertions shared by every case. + ## + module Checks + def assert_same_bytes(expected, actual, context) + return assert_nil(actual, context) if expected.nil? + + refute_nil actual, "#{context}: expected #{expected.bytesize} bytes, got nil" + assert_equal Encoding::BINARY, actual.encoding, context + return pass if expected == actual + + shorter = [expected.bytesize, actual.bytesize].min + index = (0...shorter).find { |i| expected.getbyte(i) != actual.getbyte(i) } || shorter + + flunk "#{context}: expected #{expected.bytesize} bytes, got #{actual.bytesize}; first difference at byte #{index}" + end + + # finish_request! discards the connection when pending? is true, so it must reflect exactly the + # bytes taken from the socket but not yet returned. A read that owed CHUNK_SIZE or more after + # draining the buffer must not take bytes past its end. + def assert_buffer_consistent(test_case, large_read, context) + buffered = test_case.buffered + + assert_equal buffered.positive?, test_case.reader.pending?, "#{context}: pending? with #{buffered} bytes buffered" + assert_equal 0, buffered, "#{context}: large read took #{buffered} bytes past its end" if large_read + end + + def assert_fails_within_timeout(test_case, kind, arg, context) + assert_raises(IO::TimeoutError, context) { test_case.reader.public_send(kind, arg) } + hang = TimeoutRecoveryReaderFuzz.now - test_case.faults.last_data_at + timeout = test_case.timeout + STATS[:timeouts] += 1 + STATS[:max_hang_past_timeout] = [STATS[:max_hang_past_timeout], hang - timeout].max + events = "; recent socket events (seconds since data, event, ...):\n#{test_case.faults.history}" + + assert_operator hang, :>=, timeout * 0.9, "#{context}: failed #{hang.round(3)}s after the last byte#{events}" + assert_operator hang, :<=, timeout + TIMEOUT_SLACK, + "#{context}: failed #{hang.round(3)}s after the last byte with a #{timeout}s timeout#{events}" + end + + # Every pause is shorter than the timeout, but together they outlast it. + def assert_paused_through(started, label) + paused = PAUSES * TIMEOUTS[:pauses] * PAUSE_FRACTION + + assert_operator TimeoutRecoveryReaderFuzz.now - started, :>=, paused * 0.9, "#{label}: finished before its pauses" + end + + # Every wait is bounded by the reader's timeout and, under a fiber scheduler, lets other fibers + # run. The only other wait is the zero-timeout poll that bypasses the scheduler after a wait + # reports a timeout. + def assert_bounded_waits(test_case) + test_case.faults.waits.each do |wait_timeout, blocking, scheduled| + bounded = wait_timeout&.zero? ? blocking : wait_timeout == test_case.timeout && !(scheduled && blocking) + + assert bounded, "#{test_case.label}: waited with timeout #{wait_timeout.inspect} " \ + "(blocking: #{blocking.inspect}, scheduler: #{scheduled})" + end + end + + def assert_bounded_scratch(test_case) + scratch = test_case.reader.instance_variable_get(:@scratch) + return unless scratch + + assert_operator ObjectSpace.memsize_of(scratch), :<=, LARGE + 256, "#{test_case.label}: scratch buffer grew" + end + end + + ## + # Runs cases; included into the spec. + ## + module Cases + include Checks + + def fuzz(transport, scheduler, mode, count) + with_scheduler(scheduler) do + count.times do |index| + label = "FUZZ_SEED=#{SEED} #{transport}/#{scheduler || 'no scheduler'}/#{mode} case #{index}" + rng = Random.new(Zlib.crc32(label)) + Timeout.timeout(CASE_LIMIT, nil, "#{label} still running after #{CASE_LIMIT}s") do + fuzz_case(transport, mode, index, rng, label) + end + end + end + end + + private + + def with_scheduler(selector) + return yield unless selector + + reactor = Async::Reactor.new(selector: IO::Event::Selector.const_get(selector).new(Fiber.current)) + reactor.run(finished: false) do |task| + yield + ensure + task.children&.each(&:stop) + end.wait + ensure + Fiber.set_scheduler(nil) if selector + end + + def fuzz_case(transport, mode, index, rng, label) + stream, ops = script(rng, mode) + steps, stall_at = Plans.build(mode, rng, stream, ops, index) + sock, peer = Wire.socket_pair(transport) + release = Thread::Queue.new + writer = Wire.start_writer(peer, steps, release) + faults = Faults.new(Random.new(rng.rand(2**32))) + faults.attach(sock) + timeout = TIMEOUTS.fetch(mode) + check(Case.new(label, Reader.new(sock, timeout), faults, Model.new(stream), ops, stall_at, timeout), mode, rng) + ensure + Wire.close_quietly(sock) + release&.push(true) + assert writer.join(10), "#{label}: writer still running" if writer + end + + def script(rng, mode) + return Scripts.protocol_script(rng, with_large_value: mode == :stall) unless mode == :exact + return Scripts.random_script(rng) if rng.rand(3).zero? + + stream, ops = Scripts.protocol_script(rng) + # EOF partway through a reply. + stream = stream.byteslice(0, rng.rand(0..stream.bytesize)) if rng.rand(4).zero? + [stream, ops] + end + + def check(test_case, mode, rng) + started = TimeoutRecoveryReaderFuzz.now + test_case.faults.mark + STATS[:cases] += 1 + unless run_ops(test_case) == :stalled + assert_same_bytes(test_case.model.rest, drain(test_case, rng), "#{test_case.label} drain") + assert_paused_through(started, test_case.label) if mode == :pauses + end + + assert_bounded_waits(test_case) + assert_bounded_scratch(test_case) + end + + def run_ops(test_case) + test_case.ops.each_with_index do |(kind, arg), index| + context = "#{test_case.label} op #{index} #{kind}(#{arg.inspect})" + if test_case.stalls_before?(kind, arg) + assert_fails_within_timeout(test_case, kind, arg, context) + return :stalled + end + + large_read = kind == :read && arg - test_case.buffered >= CHUNK + actual = read_reporting_history(test_case, kind, arg, context) + test_case.returned!(actual) + + assert_same_bytes(test_case.model.apply(kind, arg), actual, context) + assert_buffer_consistent(test_case, large_read, context) + STATS[:ops] += 1 + end + end + + def drain(test_case, rng) + rest = String.new(encoding: Encoding::BINARY) + size = -> { [1, CHUNK, LARGE + 1, 3 * LARGE].sample(random: rng) } + while (chunk = read_reporting_history(test_case, :read, size.call, test_case.label)) + rest << chunk + end + rest + end + + def read_reporting_history(test_case, kind, arg, context) + test_case.reader.public_send(kind, arg) + rescue IO::TimeoutError => e + flunk "#{context}: unexpected #{e.message}; recent socket events " \ + "(seconds since data, event, ...):\n#{test_case.faults.history}" + end + end +end + +if ENV['FUZZ_STATS'] + Minitest.after_run do + puts "TimeoutRecoveryReader fuzz FUZZ_SEED=#{TimeoutRecoveryReaderFuzz::SEED}: #{TimeoutRecoveryReaderFuzz::STATS}" + end +end + +describe 'TimeoutRecoveryReader fuzz' do + include TimeoutRecoveryReaderFuzz::Cases + + TimeoutRecoveryReaderFuzz::TRANSPORTS.each do |transport| + TimeoutRecoveryReaderFuzz::SCHEDULERS.each do |scheduler| + describe "over #{transport} #{scheduler ? "under Async with #{scheduler}" : 'without a fiber scheduler'}" do + it 'matches IO#gets and IO#read on randomized streams and faults' do + fuzz(transport, scheduler, :exact, TimeoutRecoveryReaderFuzz::ITERATIONS) + end + + it 'fails within one timeout of the last byte when the stream stops' do + fuzz(transport, scheduler, :stall, [TimeoutRecoveryReaderFuzz::ITERATIONS / 2, 2].max) + end + + it 'keeps reading through pauses that are each shorter than the timeout' do + fuzz(transport, scheduler, :pauses, [TimeoutRecoveryReaderFuzz::ITERATIONS / 8, 1].max) + end + end + end + end +end diff --git a/test/utils/ssl_socket_pair.rb b/test/utils/ssl_socket_pair.rb new file mode 100644 index 00000000..134e426c --- /dev/null +++ b/test/utils/ssl_socket_pair.rb @@ -0,0 +1,34 @@ +# frozen_string_literal: true + +require 'openssl' +require 'socket' +require_relative 'certificate_generator' + +## +# A connected TLS client/server pair over loopback TCP, using the test certificates. The client is a +# Dalli::Socket::SSLSocket, as Dalli would create. +## +module SSLSocketPair + def self.open + listener = TCPServer.new('127.0.0.1', 0) + client_tcp = TCPSocket.new('127.0.0.1', listener.local_address.ip_port) + server = OpenSSL::SSL::SSLSocket.new(listener.accept, server_context) + server.sync_close = true + accepting = Thread.new { server.accept } + client = Dalli::Socket::SSLSocket.new(client_tcp, CertificateGenerator.ssl_context) + client.hostname = 'localhost' + client.sync_close = true + client.connect + accepting.join + [client, server] + ensure + listener&.close + end + + def self.server_context + context = OpenSSL::SSL::SSLContext.new + context.cert = OpenSSL::X509::Certificate.new(File.read(CertificateGenerator::MEMCACHED_CERT_PATH)) + context.key = OpenSSL::PKey.read(File.read(CertificateGenerator::MEMCACHED_PK_PATH)) + context + end +end