From 24567f47c124af913c52fb6f4131fafda967cdf1 Mon Sep 17 00:00:00 2001 From: Matthew Hirst Date: Thu, 10 Sep 2026 06:40:33 +0200 Subject: [PATCH 1/4] Stop reactor leaks, TLS handshake hangs and reactor spin Reactor lifecycle * Cluster#close_async stops the ione IO reactor itself in addition to the control connection doing so, and only resolves once the reactor has stopped, so a closed cluster can never leave an io_reactor thread behind. * ControlConnection#close_async used to return early when its status was :closed, which is also the initial status. Closing a cluster whose control connection never connected returned a future that never resolved and left the reactor running. It now keys off whether the close already completed. Cassandra::Cluster::IoReactor (subclass of Ione::Io::IoReactor) * The connect timeout now bounds the TLS handshake. ione only applied it to the TCP connect, so a proxy that accepts TCP but hangs TLS hung connects forever. * A TLS socket waiting for server bytes is selected for readability. ione selected every connecting socket for writability; a connected TCP socket is always writable, so the loop spun at 100% CPU while a handshake hung. * Closing a TLS connection closes the raw TCP socket too (sync_close). * The reactor sleeps until IO, the next timer, or an unblock instead of ticking every second; sockets whose fd was closed underneath it are evicted rather than re-selected. * The reactor thread is named io_reactor so leaked reactors can be counted. Reconnection * Reconnection::Policies::Exponential accepts jitter: (fraction) so a fleet does not reconnect in lockstep; the ceiling is always honoured. Requires ione ~> 1.3 (the reactor subclass relies on 1.3 internals). Specs only use loopback sockets opened by the spec itself. Co-Authored-By: Claude Fable 5.1 --- CHANGELOG.md | 12 + Gemfile.lock | 2 +- cassandra-driver.gemspec | 2 +- lib/cassandra/cluster.rb | 18 +- lib/cassandra/cluster/control_connection.rb | 10 +- lib/cassandra/cluster/io_reactor.rb | 276 ++++++++++++++++++ lib/cassandra/driver.rb | 2 +- .../reconnection/policies/exponential.rb | 47 ++- lib/cassandra/version.rb | 2 +- spec/cassandra/cluster/io_reactor_spec.rb | 212 ++++++++++++++ spec/cassandra/cluster_spec.rb | 56 ++++ .../reconnection/policies/exponential_spec.rb | 71 +++++ 12 files changed, 693 insertions(+), 17 deletions(-) create mode 100644 lib/cassandra/cluster/io_reactor.rb create mode 100644 spec/cassandra/cluster/io_reactor_spec.rb create mode 100644 spec/cassandra/reconnection/policies/exponential_spec.rb diff --git a/CHANGELOG.md b/CHANGELOG.md index 321656d70..03b8f27e0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,15 @@ +# 3.2.6 +Bug Fixes: +* `Cluster#close` / `Cluster#close_async` now always stop the ione IO reactor, so a closed cluster can never leave its reactor thread behind. +* Closing a cluster whose control connection never connected used to return a future that never resolved and left the reactor running; `ControlConnection#close_async` now stops the reactor in that state too. +* The connect timeout now bounds the TLS handshake as well as the TCP connect. A peer that accepts TCP but never answers the handshake used to hang connects forever. +* A TLS socket waiting for the server's handshake bytes is selected for readability. Selecting it for writability made the reactor spin at 100% CPU for as long as the handshake was pending. +* Closing a TLS connection now closes the underlying TCP socket too, instead of leaking the file descriptor until GC. +* The reactor sleeps until there is IO, a timer is due or it is unblocked, instead of waking every second, and evicts sockets whose file descriptor was closed underneath it. +* The reactor thread is named `io_reactor` so leaked reactors can be counted per process. +* `Reconnection::Policies::Exponential` accepts a `jitter:` fraction (e.g. `Exponential.new(1, 60, 2, jitter: 0.25)`) so a fleet of processes does not reconnect in lockstep; the ceiling is always honoured. +* Requires ione 1.3. + # 3.2.5 Bug Fixes: * [RUBY-293](https://datastax-oss.atlassian.net/browse/RUBY-293) Infinite loop when connecting with allow_beta_protocol diff --git a/Gemfile.lock b/Gemfile.lock index 14944100f..61e6b6126 100644 --- a/Gemfile.lock +++ b/Gemfile.lock @@ -9,7 +9,7 @@ PATH remote: . specs: cassandra-driver (3.2.5) - ione + ione (~> 1.3) sorted_set GEM diff --git a/cassandra-driver.gemspec b/cassandra-driver.gemspec index 0eda62aca..47d686a05 100644 --- a/cassandra-driver.gemspec +++ b/cassandra-driver.gemspec @@ -30,7 +30,7 @@ Gem::Specification.new do |s| s.files << 'ext/cassandra_murmur3/cassandra_murmur3.c' end - s.add_runtime_dependency 'ione' + s.add_runtime_dependency 'ione', '~> 1.3' s.add_runtime_dependency 'sorted_set' s.add_development_dependency 'bundler' diff --git a/lib/cassandra/cluster.rb b/lib/cassandra/cluster.rb index 7ae041034..3fb75d28e 100644 --- a/lib/cassandra/cluster.rb +++ b/lib/cassandra/cluster.rb @@ -286,13 +286,18 @@ def close_async promise = @futures.promise @control_connection.close_async.on_complete do |f| - if f.resolved? - promise.fulfill(self) - else - f.on_failure {|e| promise.break(e)} - end + # The control connection stops the reactor as part of closing, but stop + # it here as well so that a closed cluster can never leave its reactor + # thread behind, whatever state the control connection was in. + @io_reactor.stop.on_complete do |_| + if f.resolved? + promise.fulfill(self) + else + f.on_failure {|e| promise.break(e)} + end - @executor.shutdown + @executor.shutdown + end end promise.future @@ -325,6 +330,7 @@ def inspect require 'cassandra/cluster/connector' require 'cassandra/cluster/control_connection' require 'cassandra/cluster/failed_connection' +require 'cassandra/cluster/io_reactor' require 'cassandra/cluster/metadata' require 'cassandra/cluster/options' require 'cassandra/cluster/registry' diff --git a/lib/cassandra/cluster/control_connection.rb b/lib/cassandra/cluster/control_connection.rb index 5396f2564..39155a17f 100644 --- a/lib/cassandra/cluster/control_connection.rb +++ b/lib/cassandra/cluster/control_connection.rb @@ -128,7 +128,14 @@ def host_down(host) def close_async synchronize do - return @closed_promise.future if @status == :closing || @status == :closed + # The status starts out as :closed before the first connect, so it + # cannot be used on its own to tell whether closing already happened. + # Closing a never-connected control connection must still stop the + # reactor and resolve, otherwise the reactor thread leaks and + # Cluster#close blocks forever. + if @status == :closing || @closed_promise.future.completed? + return @closed_promise.future + end @status = :closing end f = @io_reactor.stop @@ -140,6 +147,7 @@ def close_async end def connection_closed(cause) + synchronize { @status = :closed } @closed_promise.fulfill end diff --git a/lib/cassandra/cluster/io_reactor.rb b/lib/cassandra/cluster/io_reactor.rb new file mode 100644 index 000000000..5391fe24a --- /dev/null +++ b/lib/cassandra/cluster/io_reactor.rb @@ -0,0 +1,276 @@ +# encoding: utf-8 + +#-- +# Copyright DataStax, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +#++ + +module Cassandra + class Cluster + # An {Ione::Io::IoReactor} tuned for the way the driver uses it. + # + # It differs from the stock ione reactor in the following ways: + # + # * The TLS handshake is bounded by the connect timeout. The stock reactor + # only applies the timeout to the TCP connect, so a peer that accepts TCP + # but never answers the ClientHello hangs the connect forever. + # * A TLS socket waiting for bytes from the server is selected for + # readability. The stock reactor selects every connecting socket for + # writability, and a connected TCP socket is always writable, so the + # reactor loop spins at 100% CPU for as long as the handshake is pending. + # * Closing a TLS connection also closes the underlying TCP socket, instead + # of leaving the file descriptor open until garbage collection. + # * The reactor sleeps in select until there is IO, a timer is due, or it is + # unblocked, instead of waking up every second regardless. Connect + # attempts still get a bounded select so their timeouts can be enforced. + # * A socket whose file descriptor was closed underneath the reactor is + # evicted instead of being re-selected forever. + # * The reactor thread is named so leaked reactors can be counted per + # process (see {THREAD_NAME}). + # + # @private + class IoReactor < Ione::Io::IoReactor + # Name given to the reactor thread; count threads with this name to detect + # leaked reactors. + THREAD_NAME = 'io_reactor'.freeze + + def initialize(options = {}) + super + @scheduler = Scheduler.new(@options) + @io_loop = IoLoop.new(@unblocker, @scheduler, @options) + end + + # Same contract as {Ione::Io::IoReactor#connect}, but the connect timeout + # covers the TLS handshake as well as the TCP connect. + def connect(host, port, options = nil, &block) + if options.is_a?(::Numeric) || options.nil? + timeout = options || 5 + ssl = false + else + timeout = options[:timeout] || 5 + ssl = options[:ssl] + end + + deadline = @clock.now + timeout + connection = Ione::Io::Connection.new(host, port, timeout, @unblocker, @clock) + f = connection.connect + @io_loop.add_socket(connection) + @unblocker.unblock if running? + + if ssl + f = f.flat_map do + ssl_context = ssl == true ? nil : ssl + upgraded = SslConnection.new(host, + port, + connection.to_io, + @unblocker, + ssl_context, + deadline, + @clock) + ff = upgraded.connect + @io_loop.remove_socket(connection) + @io_loop.add_socket(upgraded) + @unblocker.unblock + ff + end + end + + f = f.map(&block) if block_given? + f + end + + # The reactor sleeps until its next timer is due, so a timer scheduled + # while it is asleep has to wake it up. + def schedule_timer(timeout) + timer = super + @unblocker.unblock if running? + timer + end + + # @private + class Scheduler < Ione::Io::Scheduler + def initialize(options = {}) + super + @clock = options[:clock] || ::Time + end + + # @return [Numeric, nil] seconds until the earliest pending timer is + # due (zero when overdue), or nil when there are no timers + def next_timeout + timer = @lock.synchronize { @timer_queue.peek } + return nil unless timer + + remaining = timer.time - @clock.now + remaining > 0 ? remaining : 0 + end + end + + # @private + class IoLoop < Ione::Io::IoLoopBody + def initialize(unblocker, scheduler, options = {}) + super(unblocker, options) + @scheduler = scheduler + @clock = options[:clock] || ::Time + @tick_resolution = options[:tick_resolution] || 1 + @drain_timeout = options[:drain_timeout] || 5 + end + + # @param max_timeout [Numeric, nil] upper bound on how long to block in + # select, regardless of timers + def tick(max_timeout = nil) + name_thread + + readables = [] + writables = [] + connecting = [] + + @sockets.each do |s| + if s.connected? + readables << s + elsif s.connecting? + connecting << s + # A TLS handshake waiting on the server must be selected for + # readability; selecting it for writability returns immediately + # and spins the loop. + if s.respond_to?(:handshake_wants_read?) && s.handshake_wants_read? + readables << s + next + end + end + + writables << s if s.connecting? || s.writable? + end + + timeout = @scheduler.next_timeout + # Connect and handshake timeouts are checked from #connect, which only + # runs when the loop wakes up, so keep ticking while connecting. + timeout = [timeout, @tick_resolution].compact.min unless connecting.empty? + timeout = [timeout, max_timeout].compact.min if max_timeout + + begin + r, w, _ = @selector.select(readables, writables, nil, timeout) + connecting.each(&:connect) + r && r.each {|s| s.read if s.connected?} + w && w.each(&:flush) + rescue ::IOError, ::Errno::EBADF => e + evict_dead_sockets(e) + end + end + + def drain_sockets + threshold = @clock.now + @drain_timeout + until @clock.now >= threshold || @sockets.none?(&:writable?) + @sockets.each(&:drain) + tick(@tick_resolution) + @lock.synchronize { @sockets = @sockets.reject(&:closed?) } + end + if @clock.now >= threshold + raise Ione::Io::ReactorError, + format('Socket drain timeout after %p s', @drain_timeout) + end + end + + private + + def name_thread + thread = ::Thread.current + thread.name = THREAD_NAME if thread.name.nil? + end + + # select raised because a file descriptor in the set is closed. Close + # and drop the offending sockets so the loop cannot spin on them. + def evict_dead_sockets(error) + @sockets.each do |s| + next if s.closed? || !s.is_a?(Ione::Io::BaseConnection) + io = s.to_io + next if io.nil? || !io.closed? + begin + s.close(error) + rescue + nil + end + end + @lock.synchronize { @sockets = @sockets.reject(&:closed?) } + end + end + + # @private + class SslConnection < Ione::Io::SslConnection + def initialize(host, port, io, unblocker, ssl_context, deadline, clock) + super(host, port, io, unblocker, ssl_context) + @deadline = deadline + @clock = clock + @wants_read = false + end + + # @return [Boolean] true while the handshake is waiting for bytes from + # the server + def handshake_wants_read? + @wants_read + end + + def connect + if @io.nil? + @io = if @ssl_context + @socket_impl.new(@raw_io, @ssl_context) + else + @socket_impl.new(@raw_io) + end + @io.sync_close = true if @io.respond_to?(:sync_close=) + end + @io.connect_nonblock + @wants_read = false + @state = CONNECTED_STATE + @connected_promise.fulfill(self) + @connected_promise.future + rescue ::IO::WaitReadable + @wants_read = true + fail_if_past_deadline + @connected_promise.future + rescue ::IO::WaitWritable + @wants_read = false + fail_if_past_deadline + @connected_promise.future + rescue => e + close(e) + @connected_promise.future + end + + def close(cause = nil) + closed = super + if closed + # The SSL socket closes the raw socket when it exists (sync_close), + # but a handshake that never started leaves only the raw socket. + begin + @raw_io.close unless @raw_io.closed? + rescue ::SystemCallError, ::IOError + nil + end + end + closed + end + + private + + def fail_if_past_deadline + return if @clock.now < @deadline + close(Ione::Io::ConnectionTimeoutError.new( + "Could not complete TLS handshake with #{@host}:#{@port} " \ + 'within the connect timeout' + )) + end + end + end + end +end diff --git a/lib/cassandra/driver.rb b/lib/cassandra/driver.rb index f99df3e34..642122935 100644 --- a/lib/cassandra/driver.rb +++ b/lib/cassandra/driver.rb @@ -28,7 +28,7 @@ def self.let(name, &block) define_method(:"#{name}=") { |object| @instances[name] = object } end - let(:io_reactor) { Ione::Io::IoReactor.new } + let(:io_reactor) { Cluster::IoReactor.new } let(:cluster_registry) { Cluster::Registry.new(logger) } let(:cluster_schema) { Cluster::Schema.new } let(:cluster_metadata) do diff --git a/lib/cassandra/reconnection/policies/exponential.rb b/lib/cassandra/reconnection/policies/exponential.rb index ffd84fb84..3ff7945af 100644 --- a/lib/cassandra/reconnection/policies/exponential.rb +++ b/lib/cassandra/reconnection/policies/exponential.rb @@ -20,18 +20,23 @@ module Cassandra module Reconnection module Policies # A reconnection policy that returns a constant exponentially growing - # reconnection interval up to a given maximum + # reconnection interval up to a given maximum, optionally randomised by + # a jitter fraction so that many clients do not reconnect in lockstep. class Exponential < Policy # @private class Schedule - def initialize(start, max, exponent) + def initialize(start, max, exponent, jitter, random) @interval = start @max = max @exponent = exponent + @jitter = jitter + @random = random end def next - @interval.tap { backoff if @interval < @max } + interval = @interval + backoff if @interval < @max + randomize(interval) end private @@ -45,11 +50,26 @@ def backoff new_interval end end + + # Spreads the interval by up to +/- jitter, never above the maximum. + def randomize(interval) + return interval if @jitter.zero? + + value = interval + interval * @jitter * (@random.rand * 2 - 1) + value > @max ? @max : value + end end # @param start [Numeric] beginning interval - # @param max [Numeric] maximum reconnection interval + # @param max [Numeric] maximum reconnection interval; never + # exceeded, even with jitter # @param exponent [Numeric] (2) interval exponent to use + # @param jitter [Numeric] (0) fraction in `0...1` by which each + # interval is randomised, e.g. `0.25` for +/- 25% + # @param random [#rand] (Random::DEFAULT) source of randomness for + # jitter + # + # @raise [ArgumentError] if jitter is not in `0...1` # # @example Using this policy # policy = Cassandra::Reconnection::Policies::Exponential.new(0.5, 10, 2) @@ -62,16 +82,31 @@ def backoff # schedule.next # 10.0 # schedule.next # 10.0 # schedule.next # 10.0 - def initialize(start, max, exponent = 2) + # + # @example With jitter, so that a fleet does not reconnect in lockstep + # policy = Cassandra::Reconnection::Policies::Exponential.new(1, 60, 2, jitter: 0.25) + # schedule = policy.schedule + # schedule.next # somewhere in 0.75..1.25 + # schedule.next # somewhere in 1.5..2.5 + # # ... + # schedule.next # somewhere in 45..60 + def initialize(start, max, exponent = 2, jitter: 0, random: ::Random) + jitter = Float(jitter) + unless jitter >= 0 && jitter < 1 + raise ::ArgumentError, "jitter must be in 0...1, #{jitter.inspect} given" + end + @start = start @max = max @exponent = exponent + @jitter = jitter + @random = random end # @return [Cassandra::Reconnection::Schedule] an exponential # reconnection schedule def schedule - Schedule.new(@start, @max, @exponent) + Schedule.new(@start, @max, @exponent, @jitter, @random) end end end diff --git a/lib/cassandra/version.rb b/lib/cassandra/version.rb index ed8d2183c..714ee21e5 100644 --- a/lib/cassandra/version.rb +++ b/lib/cassandra/version.rb @@ -17,5 +17,5 @@ #++ module Cassandra - VERSION = '3.2.5'.freeze + VERSION = '3.2.6'.freeze end diff --git a/spec/cassandra/cluster/io_reactor_spec.rb b/spec/cassandra/cluster/io_reactor_spec.rb new file mode 100644 index 000000000..ae3758995 --- /dev/null +++ b/spec/cassandra/cluster/io_reactor_spec.rb @@ -0,0 +1,212 @@ +# encoding: utf-8 + +#-- +# Copyright DataStax, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +#++ + +require 'spec_helper' +require 'socket' +require 'openssl' +require 'timeout' + +# These specs only ever talk to loopback sockets opened by the spec itself. +module Cassandra + class Cluster + describe(IoReactor) do + # Wraps IO.select so the spec can see how often, and for how long, the + # reactor loop goes to sleep. + class CountingSelector + attr_reader :calls, :timeouts + + def initialize + @calls = 0 + @timeouts = [] + end + + def select(readables, writables, errorables, timeout) + @calls += 1 + @timeouts << timeout + ::IO.select(readables, writables, errorables, timeout) + end + end + + let(:selector) { CountingSelector.new } + let(:reactor) { IoReactor.new(selector: selector) } + + after do + begin + reactor.stop.value + rescue + nil + end + end + + def free_port + server = ::TCPServer.new('127.0.0.1', 0) + port = server.addr[1] + server.close + port + end + + def reactor_threads + ::Thread.list.select {|t| t.name == IoReactor::THREAD_NAME} + end + + describe('thread') do + it 'is named io_reactor while running and exits on stop' do + before = ::Thread.list.size + + reactor.start.value + await { reactor_threads.size == 1 } + + reactor.stop.value + await { ::Thread.list.size <= before } + expect(reactor_threads).to be_empty + end + end + + describe('when idle') do + it 'sleeps in select without a timeout instead of ticking' do + reactor.start.value + sleep(0.3) + + # start plus draining the unblocker; a ticking reactor would have + # woken up a few hundred times by now with a small tick resolution + expect(selector.calls).to be <= 3 + expect(selector.timeouts.last).to be_nil + end + + it 'wakes up for a timer scheduled while it is asleep' do + reactor.start.value + sleep(0.1) + + started = ::Time.now + reactor.schedule_timer(0.05).value + expect(::Time.now - started).to be < 1 + end + + it 'sleeps only until the next timer is due' do + reactor.start.value + sleep(0.1) + + reactor.schedule_timer(0.2) + await { selector.timeouts.last && selector.timeouts.last <= 0.2 } + expect(selector.timeouts.last).to be > 0 + end + end + + describe('TLS handshake') do + let(:server) { ::TCPServer.new('127.0.0.1', 0) } + let(:port) { server.addr[1] } + + after { server.close unless server.closed? } + + context('when the server accepts TCP but never answers the handshake') do + it 'fails within the connect timeout instead of hanging' do + reactor.start.value + + started = ::Time.now + future = reactor.connect('127.0.0.1', port, timeout: 0.5, ssl: true) + accepted = server.accept + + expect { future.value }.to raise_error(Ione::Io::ConnectionTimeoutError, /TLS handshake/) + expect(::Time.now - started).to be < 3 + + # the half-open socket is closed, not left in the loop: reading + # drains the ClientHello and then hits EOF instead of blocking + expect { ::Timeout.timeout(2) { accepted.read } }.not_to raise_error + accepted.close + end + + it 'does not spin the reactor loop while the handshake is pending' do + reactor.start.value + + future = reactor.connect('127.0.0.1', port, timeout: 1, ssl: true) + accepted = server.accept + begin + future.value + rescue Ione::Io::ConnectionTimeoutError + nil + end + accepted.close + + # the stock reactor selects the socket for writability, which + # returns immediately, and goes round tens of thousands of times + expect(selector.calls).to be < 20 + end + end + + context('against a local TLS server') do + let(:key) { ::OpenSSL::PKey::RSA.new(2048) } + let(:cert) do + cert = ::OpenSSL::X509::Certificate.new + cert.version = 2 + cert.serial = 1 + cert.subject = ::OpenSSL::X509::Name.parse('/CN=127.0.0.1') + cert.issuer = cert.subject + cert.public_key = key.public_key + cert.not_before = ::Time.now - 60 + cert.not_after = ::Time.now + 3600 + cert.sign(key, ::OpenSSL::Digest::SHA256.new) + cert + end + let(:server_context) do + ctx = ::OpenSSL::SSL::SSLContext.new + ctx.cert = cert + ctx.key = key + ctx + end + let(:client_context) do + ctx = ::OpenSSL::SSL::SSLContext.new + ctx.verify_mode = ::OpenSSL::SSL::VERIFY_NONE + ctx + end + + it 'completes the handshake and receives data' do + ssl_server = ::OpenSSL::SSL::SSLServer.new(server, server_context) + server_thread = ::Thread.new do + socket = ssl_server.accept + socket.write('hello') + socket.flush + sleep(0.2) + socket.close + end + + reactor.start.value + connection = reactor.connect('127.0.0.1', port, timeout: 5, ssl: client_context).value + expect(connection).to be_connected + + received = +'' + connection.on_data {|data| received << data} + await { received == 'hello' } + + server_thread.join(5) + end + end + end + + describe('plain TCP') do + it 'fails fast when nothing is listening' do + reactor.start.value + + started = ::Time.now + future = reactor.connect('127.0.0.1', free_port, timeout: 5) + expect { future.value }.to raise_error(Ione::Io::ConnectionError) + expect(::Time.now - started).to be < 2 + end + end + end + end +end diff --git a/spec/cassandra/cluster_spec.rb b/spec/cassandra/cluster_spec.rb index 74b572d01..3a0123d34 100644 --- a/spec/cassandra/cluster_spec.rb +++ b/spec/cassandra/cluster_spec.rb @@ -107,5 +107,61 @@ module Cassandra end end end + + # Regression specs for closed and failed clusters releasing their reactor + # threads. These use a real reactor and only touch loopback sockets opened + # by the spec. + context('with a real io reactor') do + def free_port + server = ::TCPServer.new('127.0.0.1', 0) + port = server.addr[1] + server.close + port + end + + def reactor_threads + ::Thread.list.select {|t| t.name == Cluster::IoReactor::THREAD_NAME} + end + + describe('#close') do + it 'stops the reactor thread and the executor threads' do + before = ::Thread.list.size + driver = Driver.new(io_reactor: Cluster::IoReactor.new) + cluster = driver.cluster + + driver.io_reactor.start.value + await { reactor_threads.size == 1 } + expect(::Thread.list.size).to be > before + + cluster.close + + await { ::Thread.list.size <= before } + expect(reactor_threads).to be_empty + end + + it 'can be called more than once' do + driver = Driver.new(io_reactor: Cluster::IoReactor.new) + cluster = driver.cluster + driver.io_reactor.start.value + + expect(cluster.close).to eq(cluster) + expect(cluster.close).to eq(cluster) + expect(reactor_threads).to be_empty + end + end + + describe('Cassandra.cluster') do + it 'leaves no threads behind when no host can be reached' do + before = ::Thread.list.size + + expect do + Cassandra.cluster(hosts: ['127.0.0.1'], port: free_port, connect_timeout: 1, timeout: 1) + end.to raise_error(Errors::NoHostsAvailable) + + await { ::Thread.list.size <= before } + expect(reactor_threads).to be_empty + end + end + end end end diff --git a/spec/cassandra/reconnection/policies/exponential_spec.rb b/spec/cassandra/reconnection/policies/exponential_spec.rb new file mode 100644 index 000000000..a0d0ed67e --- /dev/null +++ b/spec/cassandra/reconnection/policies/exponential_spec.rb @@ -0,0 +1,71 @@ +# encoding: utf-8 + +#-- +# Copyright DataStax, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +#++ + +require 'spec_helper' + +module Cassandra + module Reconnection + module Policies + describe(Exponential) do + describe('#schedule') do + it 'grows exponentially up to the ceiling' do + schedule = Exponential.new(0.5, 10, 2).schedule + expect(Array.new(7) { schedule.next }).to eq([0.5, 1.0, 2.0, 4.0, 8.0, 10, 10]) + end + + it 'has no jitter by default' do + intervals = Array.new(20) { Exponential.new(1, 60, 2).schedule.next } + expect(intervals.uniq).to eq([1]) + end + + context('with jitter') do + let(:policy) { Exponential.new(1, 60, 2, jitter: 0.25) } + + it 'randomises each interval within the requested fraction' do + first = Array.new(50) { policy.schedule.next } + expect(first).to all(be_between(0.75, 1.25)) + expect(first.uniq.size).to be > 1 + + schedule = policy.schedule + schedule.next + expect(schedule.next).to be_between(1.5, 2.5) + end + + it 'never exceeds the ceiling' do + schedule = policy.schedule + intervals = Array.new(20) { schedule.next } + expect(intervals).to all(be <= 60) + expect(intervals.last(5)).to all(be_between(45, 60)) + end + + it 'uses the given source of randomness' do + random = double('random', rand: 1.0) + schedule = Exponential.new(1, 60, 2, jitter: 0.25, random: random).schedule + expect(schedule.next).to eq(1.25) + end + + it 'rejects a jitter outside 0...1' do + expect { Exponential.new(1, 60, 2, jitter: 1) }.to raise_error(::ArgumentError) + expect { Exponential.new(1, 60, 2, jitter: -0.1) }.to raise_error(::ArgumentError) + end + end + end + end + end + end +end From 4f6c7a548cd8af215b26cfbe2f85f927aded355c Mon Sep 17 00:00:00 2001 From: Matthew Hirst Date: Thu, 10 Sep 2026 07:21:09 +0200 Subject: [PATCH 2/4] Fix reactor lifecycle and timeout review findings --- CHANGELOG.md | 6 +- Gemfile.lock | 4 +- cassandra-driver.gemspec | 2 +- lib/cassandra/cluster.rb | 17 +- lib/cassandra/cluster/control_connection.rb | 31 +- lib/cassandra/cluster/io_reactor.rb | 125 ++++++-- .../reconnection/policies/exponential.rb | 13 +- .../cluster/control_connection_spec.rb | 56 ++++ spec/cassandra/cluster/io_reactor_spec.rb | 282 +++++++++++++++++- spec/cassandra/cluster_spec.rb | 15 + .../reconnection/policies/exponential_spec.rb | 21 ++ 11 files changed, 507 insertions(+), 65 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 03b8f27e0..bbaf2e085 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,7 +8,11 @@ Bug Fixes: * The reactor sleeps until there is IO, a timer is due or it is unblocked, instead of waking every second, and evicts sockets whose file descriptor was closed underneath it. * The reactor thread is named `io_reactor` so leaked reactors can be counted per process. * `Reconnection::Policies::Exponential` accepts a `jitter:` fraction (e.g. `Exponential.new(1, 60, 2, jitter: 0.25)`) so a fleet of processes does not reconnect in lockstep; the ceiling is always honoured. -* Requires ione 1.3. +* Reactor restarts restore the wake-up pipe; shutdown drains use a fixed tick even when timers are overdue. +* Connection attempts sleep until their nearest deadline, and timers scheduled on the reactor thread avoid redundant wake-ups. +* `connect_timeout: Float::INFINITY` continues to allow unbounded TCP connects and TLS handshakes. +* Reconnection jitter samples within the bounded window, avoiding a concentration of retries at the maximum interval. +* Requires ione 1.3.x (`~> 1.3.0`). # 3.2.5 Bug Fixes: diff --git a/Gemfile.lock b/Gemfile.lock index 61e6b6126..2e0375ff4 100644 --- a/Gemfile.lock +++ b/Gemfile.lock @@ -8,8 +8,8 @@ GIT PATH remote: . specs: - cassandra-driver (3.2.5) - ione (~> 1.3) + cassandra-driver (3.2.6) + ione (~> 1.3.0) sorted_set GEM diff --git a/cassandra-driver.gemspec b/cassandra-driver.gemspec index 47d686a05..8e609f8e1 100644 --- a/cassandra-driver.gemspec +++ b/cassandra-driver.gemspec @@ -30,7 +30,7 @@ Gem::Specification.new do |s| s.files << 'ext/cassandra_murmur3/cassandra_murmur3.c' end - s.add_runtime_dependency 'ione', '~> 1.3' + s.add_runtime_dependency 'ione', '~> 1.3.0' s.add_runtime_dependency 'sorted_set' s.add_development_dependency 'bundler' diff --git a/lib/cassandra/cluster.rb b/lib/cassandra/cluster.rb index 3fb75d28e..160a9ceb2 100644 --- a/lib/cassandra/cluster.rb +++ b/lib/cassandra/cluster.rb @@ -286,18 +286,13 @@ def close_async promise = @futures.promise @control_connection.close_async.on_complete do |f| - # The control connection stops the reactor as part of closing, but stop - # it here as well so that a closed cluster can never leave its reactor - # thread behind, whatever state the control connection was in. - @io_reactor.stop.on_complete do |_| - if f.resolved? - promise.fulfill(self) - else - f.on_failure {|e| promise.break(e)} - end - - @executor.shutdown + if f.resolved? + promise.fulfill(self) + else + f.on_failure {|e| promise.break(e)} end + + @executor.shutdown end promise.future diff --git a/lib/cassandra/cluster/control_connection.rb b/lib/cassandra/cluster/control_connection.rb index 39155a17f..4932e910b 100644 --- a/lib/cassandra/cluster/control_connection.rb +++ b/lib/cassandra/cluster/control_connection.rb @@ -40,7 +40,7 @@ def initialize(logger, io_reactor, cluster_registry, cluster_schema, @schema_fetcher = schema_fetcher @refreshing_statuses = ::Hash.new(false) @refresh_schema_future = nil - @status = :closed + @status = :disconnected @refreshing_hosts = false @refreshing_host = ::Hash.new(false) @closed_promise = Ione::Promise.new @@ -59,6 +59,10 @@ def on_close(&block) def connect_async synchronize do return Ione::Future.resolved if @status == :connecting || @status == :connected + if @status == :closing + return Ione::Future.failed(Errors::ClientError.new('Control connection is closing')) + end + @closed_promise = Ione::Promise.new if @status == :closed @status = :connecting end @@ -87,7 +91,7 @@ def host_up(host) @io_reactor.cancel_timer(timer) if timer unless @connection || - (@status == :closing || @status == :closed) || + (@status == :disconnected || @status == :closing || @status == :closed) || @load_balancing_policy.distance(host) == :ignore return connect_to_first_available( @load_balancing_policy.plan(nil, VOID_STATEMENT, VOID_OPTIONS) @@ -127,28 +131,27 @@ def host_down(host) end def close_async - synchronize do - # The status starts out as :closed before the first connect, so it - # cannot be used on its own to tell whether closing already happened. - # Closing a never-connected control connection must still stop the - # reactor and resolve, otherwise the reactor thread leaks and - # Cluster#close blocks forever. - if @status == :closing || @closed_promise.future.completed? + promise = synchronize do + if @status == :closing || @status == :closed return @closed_promise.future end @status = :closing + @closed_promise end f = @io_reactor.stop f.on_value(&method(:connection_closed)) f.on_failure(&method(:connection_closed)) - @closed_promise.future + promise.future end def connection_closed(cause) - synchronize { @status = :closed } - @closed_promise.fulfill + promise = synchronize do + @status = :closed + @closed_promise + end + promise.fulfill end def inspect @@ -641,9 +644,7 @@ def connect_to_first_available(plan, errors = nil) synchronize do if connection == @connection - if @status == :closing - @status = :closed - else + unless @status == :closing || @status == :closed @status = :reconnecting reconnect = true end diff --git a/lib/cassandra/cluster/io_reactor.rb b/lib/cassandra/cluster/io_reactor.rb index 5391fe24a..bd34f6f5d 100644 --- a/lib/cassandra/cluster/io_reactor.rb +++ b/lib/cassandra/cluster/io_reactor.rb @@ -47,10 +47,22 @@ class IoReactor < Ione::Io::IoReactor def initialize(options = {}) super + @unblocker.close + @unblocker = Unblocker.new @scheduler = Scheduler.new(@options) @io_loop = IoLoop.new(@unblocker, @scheduler, @options) end + def start + @lock.synchronize do + if (@state == STOPPED_STATE || @state == CRASHED_STATE) && @unblocker.closed? + @unblocker.reopen + @io_loop.add_socket(@unblocker) + end + end + super + end + # Same contract as {Ione::Io::IoReactor#connect}, but the connect timeout # covers the TLS handshake as well as the TCP connect. def connect(host, port, options = nil, &block) @@ -62,8 +74,7 @@ def connect(host, port, options = nil, &block) ssl = options[:ssl] end - deadline = @clock.now + timeout - connection = Ione::Io::Connection.new(host, port, timeout, @unblocker, @clock) + connection = Connection.new(host, port, timeout, @unblocker, @clock) f = connection.connect @io_loop.add_socket(connection) @unblocker.unblock if running? @@ -76,7 +87,7 @@ def connect(host, port, options = nil, &block) connection.to_io, @unblocker, ssl_context, - deadline, + connection.deadline, @clock) ff = upgraded.connect @io_loop.remove_socket(connection) @@ -94,10 +105,42 @@ def connect(host, port, options = nil, &block) # while it is asleep has to wake it up. def schedule_timer(timeout) timer = super - @unblocker.unblock if running? + @unblocker.unblock if running? && @io_loop.thread != ::Thread.current timer end + # @private + class Unblocker < Ione::Io::Unblocker + # Retain the object so connections queued before a restart also use + # the new pipe when they need to wake the reactor. + def reopen + initialize if closed? + end + end + + # @private + class Connection < Ione::Io::Connection + attr_reader :deadline + + def initialize(*args) + super + # Time cannot represent infinity; nil leaves the connect unbounded. + @deadline = @clock.now + @connection_timeout unless @connection_timeout == ::Float::INFINITY + end + + def connect + return @connected_promise.future if closed? + + if @deadline && !connected? && @clock.now >= @deadline + close(Ione::Io::ConnectionTimeoutError.new( + "Could not connect to #{@host}:#{@port} within #{@connection_timeout}s" + )) + return @connected_promise.future + end + super + end + end + # @private class Scheduler < Ione::Io::Scheduler def initialize(options = {}) @@ -118,6 +161,8 @@ def next_timeout # @private class IoLoop < Ione::Io::IoLoopBody + attr_reader :thread + def initialize(unblocker, scheduler, options = {}) super(unblocker, options) @scheduler = scheduler @@ -126,9 +171,9 @@ def initialize(unblocker, scheduler, options = {}) @drain_timeout = options[:drain_timeout] || 5 end - # @param max_timeout [Numeric, nil] upper bound on how long to block in - # select, regardless of timers - def tick(max_timeout = nil) + # @param timeout [Numeric, nil] fixed select timeout while draining; + # otherwise derived from pending timers and connect deadlines + def tick(timeout = nil) name_thread readables = [] @@ -152,20 +197,27 @@ def tick(max_timeout = nil) writables << s if s.connecting? || s.writable? end - timeout = @scheduler.next_timeout - # Connect and handshake timeouts are checked from #connect, which only - # runs when the loop wakes up, so keep ticking while connecting. - timeout = [timeout, @tick_resolution].compact.min unless connecting.empty? - timeout = [timeout, max_timeout].compact.min if max_timeout + unless timeout + deadlines = connecting.map do |s| + if s.respond_to?(:deadline) + deadline = s.deadline + [deadline - @clock.now, 0].max if deadline + else + @tick_resolution + end + end + timeout = [@scheduler.next_timeout, *deadlines].compact.min + end begin r, w, _ = @selector.select(readables, writables, nil, timeout) - connecting.each(&:connect) - r && r.each {|s| s.read if s.connected?} - w && w.each(&:flush) - rescue ::IOError, ::Errno::EBADF => e - evict_dead_sockets(e) + rescue ::IOError, ::Errno::EBADF, ::TypeError => e + raise unless evict_dead_sockets(readables + writables, e) + return end + connecting.each(&:connect) + r && r.each {|s| s.read if s.connected?} + w && w.each(&:flush) end def drain_sockets @@ -184,29 +236,42 @@ def drain_sockets private def name_thread - thread = ::Thread.current - thread.name = THREAD_NAME if thread.name.nil? + @thread = ::Thread.current + @thread.name = THREAD_NAME if @thread.name.nil? end # select raised because a file descriptor in the set is closed. Close # and drop the offending sockets so the loop cannot spin on them. - def evict_dead_sockets(error) - @sockets.each do |s| - next if s.closed? || !s.is_a?(Ione::Io::BaseConnection) - io = s.to_io - next if io.nil? || !io.closed? + def evict_dead_sockets(sockets, error) + dead = sockets.uniq.select do |s| + next true if s.closed? + begin + io = s.to_io + next true if io.nil? || io.closed? + # A descriptor closed through another IO wrapper can still + # report closed? == false. Probe the actual descriptor. + ::IO.select([io], nil, nil, 0) + false + rescue ::IOError, ::Errno::EBADF, ::TypeError + true + end + end + dead.each do |s| begin - s.close(error) + s.is_a?(Ione::Io::BaseConnection) ? s.close(error) : s.close rescue nil end end - @lock.synchronize { @sockets = @sockets.reject(&:closed?) } + @lock.synchronize { @sockets = @sockets.reject {|s| s.closed? || dead.include?(s)} } + !dead.empty? end end # @private class SslConnection < Ione::Io::SslConnection + attr_reader :deadline + def initialize(host, port, io, unblocker, ssl_context, deadline, clock) super(host, port, io, unblocker, ssl_context) @deadline = deadline @@ -221,6 +286,10 @@ def handshake_wants_read? end def connect + return @connected_promise.future if closed? + fail_if_past_deadline + return @connected_promise.future if closed? + if @io.nil? @io = if @ssl_context @socket_impl.new(@raw_io, @ssl_context) @@ -253,7 +322,7 @@ def close(cause = nil) # The SSL socket closes the raw socket when it exists (sync_close), # but a handshake that never started leaves only the raw socket. begin - @raw_io.close unless @raw_io.closed? + @raw_io.close if @raw_io && !@raw_io.closed? rescue ::SystemCallError, ::IOError nil end @@ -264,7 +333,7 @@ def close(cause = nil) private def fail_if_past_deadline - return if @clock.now < @deadline + return if @deadline.nil? || @clock.now < @deadline close(Ione::Io::ConnectionTimeoutError.new( "Could not complete TLS handshake with #{@host}:#{@port} " \ 'within the connect timeout' diff --git a/lib/cassandra/reconnection/policies/exponential.rb b/lib/cassandra/reconnection/policies/exponential.rb index 3ff7945af..98740fe80 100644 --- a/lib/cassandra/reconnection/policies/exponential.rb +++ b/lib/cassandra/reconnection/policies/exponential.rb @@ -55,8 +55,9 @@ def backoff def randomize(interval) return interval if @jitter.zero? - value = interval + interval * @jitter * (@random.rand * 2 - 1) - value > @max ? @max : value + lower = interval * (1 - @jitter) + upper = [interval * (1 + @jitter), @max].min + lower + (upper - lower) * @random.rand end end @@ -66,7 +67,7 @@ def randomize(interval) # @param exponent [Numeric] (2) interval exponent to use # @param jitter [Numeric] (0) fraction in `0...1` by which each # interval is randomised, e.g. `0.25` for +/- 25% - # @param random [#rand] (Random::DEFAULT) source of randomness for + # @param random [#rand] (Random) source of randomness for # jitter # # @raise [ArgumentError] if jitter is not in `0...1` @@ -91,7 +92,11 @@ def randomize(interval) # # ... # schedule.next # somewhere in 45..60 def initialize(start, max, exponent = 2, jitter: 0, random: ::Random) - jitter = Float(jitter) + begin + jitter = Float(jitter) + rescue ::TypeError, ::ArgumentError + raise ::ArgumentError, "jitter must be in 0...1, #{jitter.inspect} given" + end unless jitter >= 0 && jitter < 1 raise ::ArgumentError, "jitter must be in 0...1, #{jitter.inspect} given" end diff --git a/spec/cassandra/cluster/control_connection_spec.rb b/spec/cassandra/cluster/control_connection_spec.rb index aeecb4c51..966ae2f5e 100644 --- a/spec/cassandra/cluster/control_connection_spec.rb +++ b/spec/cassandra/cluster/control_connection_spec.rb @@ -1059,6 +1059,62 @@ def handle_request(&handler) end describe "#close_async" do + it 'stops the reactor even before the first connect' do + expect(io_reactor).to receive(:stop).once.and_return(Ione::Future.resolved) + expect(control_connection.close_async).to be_resolved + expect(control_connection.close_async).to be_resolved + end + + it 'does not connect on a host-up notification before the first connect' do + expect(io_reactor).not_to receive(:connect) + control_connection.host_up(cluster_registry.hosts.first).value + end + + it 'can close again after an explicit reconnect' do + control_connection.connect_async.value + first_close = control_connection.close_async + expect(first_close).to be_resolved + + control_connection.connect_async.value + stopped = Ione::Promise.new + expect(io_reactor).to receive(:stop).once.and_return(stopped.future) + second_close = control_connection.close_async + expect(second_close).not_to equal(first_close) + expect(second_close).not_to be_completed + stopped.fulfill + expect(second_close).to be_resolved + end + + it 'keeps the original close future when a listener reconnects' do + control_connection.connect_async.value + first_close = nil + control_connection.on_close do + first_close = control_connection.close_async + control_connection.connect_async.value + end + returned = control_connection.close_async + + expect(returned).to equal(first_close) + expect(first_close).to be_resolved + expect(io_reactor).to be_running + end + + it 'stops only once when a socket closes before the stop future resolves' do + control_connection.connect_async.value + stopped = Ione::Promise.new + expect(io_reactor).to receive(:stop).once do + last_connection.close + stopped.future + end + + closing = control_connection.close_async + expect(control_connection.close_async).to equal(closing) + expect(closing).not_to be_completed + expect { control_connection.connect_async.value }.to raise_error(Errors::ClientError, /closing/) + stopped.fulfill + expect(closing).to be_resolved + end + context 'when connected' do before do control_connection.connect_async.value diff --git a/spec/cassandra/cluster/io_reactor_spec.rb b/spec/cassandra/cluster/io_reactor_spec.rb index ae3758995..bbb8dd2bf 100644 --- a/spec/cassandra/cluster/io_reactor_spec.rb +++ b/spec/cassandra/cluster/io_reactor_spec.rb @@ -47,9 +47,9 @@ def select(readables, writables, errorables, timeout) after do begin - reactor.stop.value + ::Timeout.timeout(3) { reactor.stop.value } rescue - nil + reactor.instance_variable_get(:@io_loop).thread.kill if reactor.running? end end @@ -75,6 +75,50 @@ def reactor_threads await { ::Thread.list.size <= before } expect(reactor_threads).to be_empty end + + it 'wakes for timers and stops after repeated restarts' do + ::Timeout.timeout(3) do + 3.times do + reactor.start.value + reactor.schedule_timer(0.01).value + reactor.stop.value + end + end + expect(reactor_threads).to be_empty + end + + it 'restarts when start is requested while stopping' do + ::Timeout.timeout(3) do + reactor.start.value + restarted = reactor.schedule_timer(0).flat_map do + reactor.stop + reactor.start + end + expect(restarted.value).to eq(reactor) + reactor.schedule_timer(0.01).value + reactor.stop.value + end + end + + it 'restores the unblocker after a reactor crash' do + crashed = Ione::Promise.new + reactor.on_error {|error| crashed.fulfill(error)} + allow(selector).to receive(:select).and_wrap_original do |original, *args| + if crashed.future.completed? + original.call(*args) + else + raise 'selector failure' + end + end + + ::Timeout.timeout(3) do + reactor.start.value + expect(crashed.future.value.message).to eq('selector failure') + reactor.start.value + reactor.schedule_timer(0.01).value + reactor.stop.value + end + end end describe('when idle') do @@ -105,6 +149,19 @@ def reactor_threads await { selector.timeouts.last && selector.timeouts.last <= 0.2 } expect(selector.timeouts.last).to be > 0 end + + it 'does not wake itself when scheduling a timer on the reactor thread' do + reactor.start.value + await { selector.calls > 0 } + unblocker = reactor.instance_variable_get(:@unblocker) + allow(unblocker).to receive(:unblock).and_call_original + + ::Timeout.timeout(3) do + reactor.schedule_timer(0).flat_map { reactor.schedule_timer(0.01) }.value + end + + expect(unblocker).to have_received(:unblock).once + end end describe('TLS handshake') do @@ -122,7 +179,7 @@ def reactor_threads accepted = server.accept expect { future.value }.to raise_error(Ione::Io::ConnectionTimeoutError, /TLS handshake/) - expect(::Time.now - started).to be < 3 + expect(::Time.now - started).to be < 1 # the half-open socket is closed, not left in the loop: reading # drains the ClientHello and then hits EOF instead of blocking @@ -146,6 +203,25 @@ def reactor_threads # returns immediately, and goes round tens of thousands of times expect(selector.calls).to be < 20 end + + it 'keeps an infinite timeout pending while allowing timers and shutdown' do + accepted = nil + ::Timeout.timeout(3) do + reactor.start.value + future = reactor.connect('127.0.0.1', port, timeout: Float::INFINITY, ssl: true) + accepted = server.accept + + reactor.schedule_timer(0.05).value + expect(future).not_to be_completed + expect(selector.calls).to be < 20 + + reactor.stop.value + expect(future).to be_completed + expect(accepted.read).not_to be_empty + end + ensure + accepted.close if accepted + end end context('against a local TLS server') do @@ -206,6 +282,206 @@ def reactor_threads expect { future.value }.to raise_error(Ione::Io::ConnectionError) expect(::Time.now - started).to be < 2 end + + it 'can write through a connection queued before restarting' do + server = ::TCPServer.new('127.0.0.1', 0) + reactor.start.value + reactor.stop.value + connected = reactor.connect('127.0.0.1', server.addr[1], timeout: 1) + accepted = server.accept + + ::Timeout.timeout(3) do + reactor.start.value + connected.value.write('hello') + expect(accepted.read(5)).to eq('hello') + reactor.stop.value + end + ensure + accepted.close if accepted + server.close if server + end + + it 'connects and writes with an infinite timeout' do + accepted = nil + server = ::TCPServer.new('127.0.0.1', 0) + ::Timeout.timeout(3) do + reactor.start.value + connection = reactor.connect('127.0.0.1', server.addr[1], Float::INFINITY).value + accepted = server.accept + connection.write('hello') + expect(accepted.read(5)).to eq('hello') + end + ensure + accepted.close if accepted + server.close if server + end + end + end + + describe(IoReactor::IoLoop) do + let(:clock) { double('clock', now: 100.0) } + let(:selector) { double('selector') } + let(:unblocker) { IoReactor::Unblocker.new } + let(:scheduler) { IoReactor::Scheduler.new(clock: clock) } + let(:io_loop) do + IoReactor::IoLoop.new(unblocker, scheduler, + clock: clock, selector: selector, tick_resolution: 1, drain_timeout: 3) + end + + before { @thread_name = ::Thread.current.name } + after do + io_loop.close_sockets + ::Thread.current.name = @thread_name + end + + it 'uses a fixed timeout while draining with an overdue timer' do + timer = scheduler.schedule_timer(-1) + socket = double('socket', connected?: false, connecting?: false, + writable?: true, closed?: false, drain: nil, close: nil) + io_loop.add_socket(socket) + now = 100.0 + allow(selector).to receive(:select) do |_, _, _, timeout| + expect(timeout).to eq(1) + now += timeout + allow(clock).to receive(:now).and_return(now) + nil + end + + expect { io_loop.drain_sockets }.to raise_error(Ione::Io::ReactorError, /drain timeout/) + expect(selector).to have_received(:select).exactly(3).times + expect(timer).not_to be_completed + end + + it 'selects only until the nearest connect deadline' do + [100.25, nil, 102.0].each do |deadline| + socket = double('connection', connected?: false, connecting?: true, + closed?: false, deadline: deadline, connect: nil, close: nil) + io_loop.add_socket(socket) + end + scheduler.schedule_timer(0.5) + + expect(selector).to receive(:select).with(anything, anything, nil, 0.25) + io_loop.tick + end + + it 'selects without a deadline for an infinite timeout' do + socket = double('connection', connected?: false, connecting?: true, + closed?: false, deadline: nil, connect: nil, close: nil) + io_loop.add_socket(socket) + + expect(selector).to receive(:select).with(anything, anything, nil, nil) + io_loop.tick + end + + it 'honours timers while a connection has an infinite timeout' do + socket = double('connection', connected?: false, connecting?: true, + closed?: false, deadline: nil, connect: nil, close: nil) + io_loop.add_socket(socket) + scheduler.schedule_timer(0.25) + + expect(selector).to receive(:select).with(anything, anything, nil, 0.25) + io_loop.tick + end + + it 'lets an earlier timer bound select while connecting' do + socket = double('connection', connected?: false, connecting?: true, + closed?: false, deadline: 105.0, connect: nil, close: nil) + io_loop.add_socket(socket) + scheduler.schedule_timer(0.1) + + expect(selector).to receive(:select) do |_, _, _, timeout| + expect(timeout).to be_within(0.0001).of(0.1) + nil + end + io_loop.tick + end + + def add_connection(io) + connection = Ione::Io::BaseConnection.new('127.0.0.1', 9042, unblocker) + connection.instance_variable_set(:@io, io) + connection.instance_variable_set(:@state, Ione::Io::BaseConnection::CONNECTED_STATE) + io_loop.add_socket(connection) + connection + end + + [::IOError, ::Errno::EBADF, ::TypeError].each do |error_class| + it "evicts an invalid socket after #{error_class} and retains healthy sockets" do + reader, writer = ::IO.pipe + dead_io = reader.dup + dead = add_connection(dead_io) + healthy = add_connection(reader) + if error_class == ::IOError + dead_io.close + elsif error_class == ::Errno::EBADF + # Close the fd through another wrapper: closed? still returns false. + ::IO.for_fd(dead_io.fileno).close + expect(dead_io).not_to be_closed + else + dead_io.close + dead.instance_variable_set(:@io, nil) + end + expect { ::IO.select([dead], nil, nil, 0) }.to raise_error(error_class) + allow(selector).to receive(:select) do |*args| + ::IO.select(*args) + end + + io_loop.tick + + expect(dead).to be_closed + expect(healthy).not_to be_closed + received = nil + healthy.on_data {|data| received = data} + writer.write('hello') + io_loop.tick + expect(received).to eq('hello') + ensure + writer.close if writer + begin + dead_io.close if dead_io && !dead_io.closed? + rescue ::Errno::EBADF + nil + end + end + end + + it 'propagates select errors unrelated to dead sockets' do + expect(selector).to receive(:select).and_raise(::TypeError, 'bad selector argument') + expect { io_loop.tick }.to raise_error(::TypeError, 'bad selector argument') + end + end + + describe(IoReactor::Connection) do + it 'fails at the connect deadline without waiting for another polling tick' do + clock = double('clock', now: 100.0) + connection = IoReactor::Connection.new('127.0.0.1', 9042, 0.25, nil, clock) + allow(clock).to receive(:now).and_return(100.25) + + expect { connection.connect.value }.to raise_error(Ione::Io::ConnectionTimeoutError) + expect(connection).to be_closed + end + end + + describe(IoReactor::SslConnection) do + it 'fails the handshake without raising when the raw socket has disappeared' do + connection = IoReactor::SslConnection.new('127.0.0.1', 9042, nil, nil, nil, + ::Time.now + 1, ::Time) + future = connection.connect + + expect(future).to be_failed + expect { future.value }.to raise_error(Ione::Io::ConnectionError) + expect(connection).to be_closed + expect(connection.close).to eq(false) + end + + it 'closes the raw socket when closed before the handshake starts' do + reader, writer = ::IO.pipe + connection = IoReactor::SslConnection.new('127.0.0.1', 9042, reader, nil, nil, + ::Time.now + 1, ::Time) + expect(connection.close).to eq(true) + expect(reader).to be_closed + ensure + reader.close if reader && !reader.closed? + writer.close if writer end end end diff --git a/spec/cassandra/cluster_spec.rb b/spec/cassandra/cluster_spec.rb index 3a0123d34..161f43e4b 100644 --- a/spec/cassandra/cluster_spec.rb +++ b/spec/cassandra/cluster_spec.rb @@ -83,10 +83,25 @@ module Cassandra end it 'closes control connection' do + expect(io_reactor).not_to receive(:stop) expect(control_connection).to receive(:close_async).once.and_return(Ione::Future.resolved) expect(cluster.close_async).to eq(promise) expect(promise).to have_received(:fulfill).once.with(cluster) end + + it 'waits for the control connection to finish stopping the reactor' do + stopped = Ione::Promise.new + expect(io_reactor).to receive(:stop).once.and_return(stopped.future) + allow(executor).to receive(:shutdown) + + cluster.close_async + expect(promise).not_to have_received(:fulfill) + expect(executor).not_to have_received(:shutdown) + + stopped.fulfill + expect(promise).to have_received(:fulfill).once.with(cluster) + expect(executor).to have_received(:shutdown).once + end end [ diff --git a/spec/cassandra/reconnection/policies/exponential_spec.rb b/spec/cassandra/reconnection/policies/exponential_spec.rb index a0d0ed67e..661c9d063 100644 --- a/spec/cassandra/reconnection/policies/exponential_spec.rb +++ b/spec/cassandra/reconnection/policies/exponential_spec.rb @@ -59,10 +59,31 @@ module Policies expect(schedule.next).to eq(1.25) end + it 'spreads draws uniformly across the bounded window at the ceiling' do + random = double('random') + allow(random).to receive(:rand).and_return(0.0, 0.25, 0.5, 0.75) + schedule = Exponential.new(60, 60, 2, jitter: 0.25, random: random).schedule + + expect(Array.new(4) { schedule.next }).to eq([45, 48.75, 52.5, 56.25]) + end + + it 'bounds the window before drawing when approaching the ceiling' do + random = double('random', rand: 0.5) + schedule = Exponential.new(50, 60, 2, jitter: 0.25, random: random).schedule + + expect(schedule.next).to eq(48.75) + end + it 'rejects a jitter outside 0...1' do expect { Exponential.new(1, 60, 2, jitter: 1) }.to raise_error(::ArgumentError) expect { Exponential.new(1, 60, 2, jitter: -0.1) }.to raise_error(::ArgumentError) end + + it 'rejects invalid jitter values with ArgumentError' do + [nil, Object.new, 'invalid', Float::NAN].each do |jitter| + expect { Exponential.new(1, 60, 2, jitter: jitter) }.to raise_error(::ArgumentError) + end + end end end end From a950db14aac35454a7765490db12b97b7c62c847 Mon Sep 17 00:00:00 2001 From: Matthew Hirst Date: Thu, 10 Sep 2026 07:30:00 +0200 Subject: [PATCH 3/4] Make reactor restart transitions atomic --- CHANGELOG.md | 2 +- lib/cassandra/cluster/io_reactor.rb | 59 +++++++++++-- spec/cassandra/cluster/io_reactor_spec.rb | 101 ++++++++++++++++++++++ 3 files changed, 156 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index bbaf2e085..34c3abb82 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,7 +8,7 @@ Bug Fixes: * The reactor sleeps until there is IO, a timer is due or it is unblocked, instead of waking every second, and evicts sockets whose file descriptor was closed underneath it. * The reactor thread is named `io_reactor` so leaked reactors can be counted per process. * `Reconnection::Policies::Exponential` accepts a `jitter:` fraction (e.g. `Exponential.new(1, 60, 2, jitter: 0.25)`) so a fleet of processes does not reconnect in lockstep; the ceiling is always honoured. -* Reactor restarts restore the wake-up pipe; shutdown drains use a fixed tick even when timers are overdue. +* Reactor restarts restore the wake-up pipe atomically with the start transition, including when shutdown completes during a restart request; shutdown drains use a fixed tick even when timers are overdue. * Connection attempts sleep until their nearest deadline, and timers scheduled on the reactor thread avoid redundant wake-ups. * `connect_timeout: Float::INFINITY` continues to allow unbounded TCP connects and TLS handshakes. * Reconnection jitter samples within the bounded window, avoiding a concentration of retries at the maximum interval. diff --git a/lib/cassandra/cluster/io_reactor.rb b/lib/cassandra/cluster/io_reactor.rb index bd34f6f5d..71de98936 100644 --- a/lib/cassandra/cluster/io_reactor.rb +++ b/lib/cassandra/cluster/io_reactor.rb @@ -54,13 +54,30 @@ def initialize(options = {}) end def start - @lock.synchronize do - if (@state == STOPPED_STATE || @state == CRASHED_STATE) && @unblocker.closed? - @unblocker.reopen - @io_loop.add_socket(@unblocker) + stopping = @lock.synchronize do + return @started_promise.future if @state == RUNNING_STATE + + if @state == STOPPING_STATE + @stopped_promise.future + else + # Restore the pipe in the same transition that starts the reactor. + # Shutdown cannot close it between this check and a later #start. + if @unblocker.closed? + @unblocker.reopen + @io_loop.add_socket(@unblocker) + end + started = @started_promise = Ione::Promise.new + stopped = @stopped_promise = Ione::Promise.new + @error_listeners.each {|listener| stopped.future.on_failure(&listener)} + @state = RUNNING_STATE + ::Thread.start { run(started, stopped) } + return started.future end end - super + + # A completed future runs its callbacks immediately, so subscribe only + # after releasing the state lock; restarting needs to acquire it again. + stopping.flat_map { start }.fallback { start } end # Same contract as {Ione::Io::IoReactor#connect}, but the connect timeout @@ -109,6 +126,38 @@ def schedule_timer(timeout) timer end + private + + def run(started, stopped) + error = nil + begin + started.fulfill(self) + while @state == RUNNING_STATE + @io_loop.tick + @scheduler.tick + end + rescue => e + error = e + ensure + begin + begin + @io_loop.drain_sockets + rescue => e + error ||= e + end + @io_loop.close_sockets + @scheduler.cancel_timers + rescue => e + error ||= e + ensure + @lock.synchronize { @state = error ? CRASHED_STATE : STOPPED_STATE } + # A new run may start as soon as the state lock is released. Finish + # this run's promise, and invoke its callbacks outside the lock. + error ? stopped.fail(error) : stopped.fulfill(self) + end + end + end + # @private class Unblocker < Ione::Io::Unblocker # Retain the object so connections queued before a restart also use diff --git a/spec/cassandra/cluster/io_reactor_spec.rb b/spec/cassandra/cluster/io_reactor_spec.rb index bbb8dd2bf..c6308d336 100644 --- a/spec/cassandra/cluster/io_reactor_spec.rb +++ b/spec/cassandra/cluster/io_reactor_spec.rb @@ -100,6 +100,66 @@ def reactor_threads end end + it 'restarts when shutdown completes immediately after start observes stopping' do + entered_select = ::Queue.new + release_select = ::Queue.new + observed_stopping = ::Queue.new + resume_start = ::Queue.new + begin_start = ::Queue.new + requester = nil + first_select = true + paused_start = false + + allow(selector).to receive(:select).and_wrap_original do |original, *args| + if first_select + first_select = false + entered_select << true + release_select.pop + nil + else + original.call(*args) + end + end + + lock = reactor.instance_variable_get(:@lock) + allow(lock).to receive(:synchronize).and_wrap_original do |original, &block| + result = original.call(&block) + if ::Thread.current == requester && !paused_start + paused_start = true + observed_stopping << true + resume_start.pop + end + result + end + + ::Timeout.timeout(3) do + reactor.start.value + entered_select.pop + stopped = reactor.stop + requester = ::Thread.new do + begin_start.pop + reactor.start + end + begin_start << true + + # Let shutdown close the pipe after start releases the state lock, + # before it can continue starting or subscribe to the stop future. + observed_stopping.pop + release_select << true + stopped.value + resume_start << true + + expect(requester.value.value).to eq(reactor) + reactor.schedule_timer(0.01).value + reactor.stop.value + end + ensure + release_select << true + resume_start << true + begin_start << true + requester.kill.join if requester && requester.alive? + end + it 'restores the unblocker after a reactor crash' do crashed = Ione::Promise.new reactor.on_error {|error| crashed.fulfill(error)} @@ -119,6 +179,47 @@ def reactor_threads reactor.stop.value end end + + it 'completes the old stop future when a new run has already started' do + shutdown_finished = ::Queue.new + complete_stop = ::Queue.new + paused_completion = false + previous_thread = nil + + ::Timeout.timeout(3) do + reactor.start.value + await { reactor.instance_variable_get(:@io_loop).thread } + previous_thread = reactor.instance_variable_get(:@io_loop).thread + lock = reactor.instance_variable_get(:@lock) + allow(lock).to receive(:synchronize).and_wrap_original do |original, &block| + result = original.call(&block) + if ::Thread.current == previous_thread && + reactor.instance_variable_get(:@state) == IoReactor::STOPPED_STATE && + !paused_completion + paused_completion = true + shutdown_finished << true + complete_stop.pop + end + result + end + + old_stop = reactor.stop + shutdown_finished.pop + reactor.start.value + new_stop = reactor.instance_variable_get(:@stopped_promise).future + expect(old_stop).not_to be_completed + complete_stop << true + + expect(old_stop.value).to eq(reactor) + expect(new_stop).not_to be_completed + reactor.schedule_timer(0.01).value + expect(reactor.stop).to equal(new_stop) + new_stop.value + end + ensure + complete_stop << true + previous_thread.join(1) if previous_thread + end end describe('when idle') do From d70bb801731bcc57e2b520d392bc4b3af780fa0e Mon Sep 17 00:00:00 2001 From: Matthew Hirst Date: Mon, 14 Sep 2026 14:29:15 +0200 Subject: [PATCH 4/4] Mirror socket lifecycle and deadline fixes from ione --- CHANGELOG.md | 3 + lib/cassandra/cluster/io_reactor.rb | 108 +++++++++-- spec/cassandra/cluster/io_reactor_spec.rb | 224 +++++++++++++++++++++- 3 files changed, 312 insertions(+), 23 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 34c3abb82..b30ef7bdc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,9 @@ Bug Fixes: * `Reconnection::Policies::Exponential` accepts a `jitter:` fraction (e.g. `Exponential.new(1, 60, 2, jitter: 0.25)`) so a fleet of processes does not reconnect in lockstep; the ceiling is always honoured. * Reactor restarts restore the wake-up pipe atomically with the start transition, including when shutdown completes during a restart request; shutdown drains use a fixed tick even when timers are overdue. * Connection attempts sleep until their nearest deadline, and timers scheduled on the reactor thread avoid redundant wake-ups. +* Closing or draining connections and listeners wakes the reactor so idle sockets and listening ports are released promptly. +* Socket `IOError` and `EBADF` failures during connect, read, or flush close only the affected socket; transient accept errors leave the listener available for retry. +* Completed TCP connections and TLS handshakes take precedence over expired deadlines, including connections queued while the reactor is stopped. * `connect_timeout: Float::INFINITY` continues to allow unbounded TCP connects and TLS handshakes. * Reconnection jitter samples within the bounded window, avoiding a concentration of retries at the maximum interval. * Requires ione 1.3.x (`~> 1.3.0`). diff --git a/lib/cassandra/cluster/io_reactor.rb b/lib/cassandra/cluster/io_reactor.rb index 71de98936..f69d05167 100644 --- a/lib/cassandra/cluster/io_reactor.rb +++ b/lib/cassandra/cluster/io_reactor.rb @@ -167,8 +167,60 @@ def reopen end end + # @private + module ConnectionLifecycle + # Release the descriptor and wake select before notifying listeners, + # matching the close ordering in ione's reactor fix. + def close(cause = nil) + @lock.synchronize do + return false if @state == Ione::Io::BaseConnection::CLOSED_STATE + @state = Ione::Io::BaseConnection::CLOSED_STATE + @writable = false + end + if @io + begin + @io.close + @io = nil + rescue ::SystemCallError, ::IOError + # The descriptor may already have been closed by another thread. + end + end + @unblocker.unblock + if cause && !cause.is_a?(Ione::IoError) + cause = Ione::Io::ConnectionClosedError.new(cause.message) + end + cause ? @closed_promise.fail(cause) : @closed_promise.fulfill(self) + true + end + end + + # @private + module AcceptorLifecycle + def close + closed = super + @unblocker.unblock if closed + closed + end + + # The inherited drain alias bypasses an overridden close method. + def drain + close + end + + def read + super + rescue ::IOError, ::Errno::EBADF + close + rescue ::SystemCallError + # A transient accept error leaves the listener available for retry. + nil + end + end + # @private class Connection < Ione::Io::Connection + include ConnectionLifecycle + attr_reader :deadline def initialize(*args) @@ -180,13 +232,18 @@ def initialize(*args) def connect return @connected_promise.future if closed? - if @deadline && !connected? && @clock.now >= @deadline + # Let a completed kernel handshake win over the deadline. The parent + # leaves the connection pending only after EINPROGRESS or EALREADY. + future = super + if @deadline && connecting? && @clock.now >= @deadline close(Ione::Io::ConnectionTimeoutError.new( "Could not connect to #{@host}:#{@port} within #{@connection_timeout}s" )) - return @connected_promise.future end - super + future + rescue ::IOError => e + close(e) + @connected_promise.future end end @@ -220,6 +277,17 @@ def initialize(unblocker, scheduler, options = {}) @drain_timeout = options[:drain_timeout] || 5 end + def add_socket(socket) + # bind/accept create ione's server sockets. Apply the same lifecycle + # fixes to those instances without modifying the dependency globally. + if socket.is_a?(Ione::Io::ServerConnection) + socket.extend(ConnectionLifecycle) + elsif socket.is_a?(Ione::Io::Acceptor) + socket.extend(AcceptorLifecycle) + end + super + end + # @param timeout [Numeric, nil] fixed select timeout while draining; # otherwise derived from pending timers and connect deadlines def tick(timeout = nil) @@ -264,9 +332,9 @@ def tick(timeout = nil) raise unless evict_dead_sockets(readables + writables, e) return end - connecting.each(&:connect) - r && r.each {|s| s.read if s.connected?} - w && w.each(&:flush) + connecting.each {|s| dispatch(s, :connect)} + r && r.each {|s| dispatch(s, :read) if s.connected?} + w && w.each {|s| dispatch(s, :flush)} end def drain_sockets @@ -289,6 +357,19 @@ def name_thread @thread.name = THREAD_NAME if @thread.name.nil? end + def dispatch(socket, method) + socket.__send__(method) + rescue ::IOError, ::Errno::EBADF => e + close_socket(socket, e) + end + + def close_socket(socket, error) + socket.is_a?(Ione::Io::BaseConnection) ? socket.close(error) : socket.close + rescue + # The descriptor may already be closed. + nil + end + # select raised because a file descriptor in the set is closed. Close # and drop the offending sockets so the loop cannot spin on them. def evict_dead_sockets(sockets, error) @@ -305,13 +386,7 @@ def evict_dead_sockets(sockets, error) true end end - dead.each do |s| - begin - s.is_a?(Ione::Io::BaseConnection) ? s.close(error) : s.close - rescue - nil - end - end + dead.each {|s| close_socket(s, error)} @lock.synchronize { @sockets = @sockets.reject {|s| s.closed? || dead.include?(s)} } !dead.empty? end @@ -319,6 +394,8 @@ def evict_dead_sockets(sockets, error) # @private class SslConnection < Ione::Io::SslConnection + include ConnectionLifecycle + attr_reader :deadline def initialize(host, port, io, unblocker, ssl_context, deadline, clock) @@ -335,10 +412,9 @@ def handshake_wants_read? end def connect - return @connected_promise.future if closed? - fail_if_past_deadline - return @connected_promise.future if closed? + return @connected_promise.future if closed? || connected? + # Check the deadline only if the handshake still needs IO. if @io.nil? @io = if @ssl_context @socket_impl.new(@raw_io, @ssl_context) diff --git a/spec/cassandra/cluster/io_reactor_spec.rb b/spec/cassandra/cluster/io_reactor_spec.rb index c6308d336..71908e19f 100644 --- a/spec/cassandra/cluster/io_reactor_spec.rb +++ b/spec/cassandra/cluster/io_reactor_spec.rb @@ -263,6 +263,73 @@ def reactor_threads expect(unblocker).to have_received(:unblock).once end + + [:close, :drain].each do |operation| + it "wakes when an outgoing connection receives #{operation} from another thread" do + server = ::TCPServer.new('127.0.0.1', 0) + reactor.start.value + connection = reactor.connect('127.0.0.1', server.addr[1], 1).value + accepted = server.accept + unblocker = reactor.instance_variable_get(:@unblocker) + events = [] + allow(unblocker).to receive(:unblock).and_wrap_original do |original| + events << :wakeup + original.call + end + connection.on_closed { events << :closed } + sleep(0.1) + calls = selector.calls + + connection.public_send(operation) + + expect(unblocker).to have_received(:unblock) + expect(events.index(:wakeup)).to be < events.index(:closed) + await { selector.calls > calls } + expect(::Timeout.timeout(2) { accepted.read }).to eq('') + ensure + accepted.close if accepted && !accepted.closed? + server.close if server && !server.closed? + end + + it "releases a listening port when its acceptor receives #{operation}" do + reactor.start.value + acceptor = reactor.bind('127.0.0.1', 0, 5).value + port = acceptor.to_io.local_address.ip_port + unblocker = reactor.instance_variable_get(:@unblocker) + allow(unblocker).to receive(:unblock).and_call_original + sleep(0.1) + calls = selector.calls + + acceptor.public_send(operation) + + expect(unblocker).to have_received(:unblock) + await { selector.calls > calls } + rebound = ::TCPServer.new('127.0.0.1', port) + ensure + rebound.close if rebound + end + + it "wakes when an accepted connection receives #{operation} from another thread" do + reactor.start.value + acceptor = reactor.bind('127.0.0.1', 0, 5).value + accepted = Ione::Promise.new + acceptor.on_accept {|connection| accepted.fulfill(connection)} + peer = ::TCPSocket.new('127.0.0.1', acceptor.to_io.local_address.ip_port) + connection = ::Timeout.timeout(2) { accepted.future.value } + unblocker = reactor.instance_variable_get(:@unblocker) + allow(unblocker).to receive(:unblock).and_call_original + sleep(0.1) + calls = selector.calls + + connection.public_send(operation) + + expect(unblocker).to have_received(:unblock) + await { selector.calls > calls } + expect(::Timeout.timeout(2) { peer.read }).to eq('') + ensure + peer.close if peer && !peer.closed? + end + end end describe('TLS handshake') do @@ -279,7 +346,7 @@ def reactor_threads future = reactor.connect('127.0.0.1', port, timeout: 0.5, ssl: true) accepted = server.accept - expect { future.value }.to raise_error(Ione::Io::ConnectionTimeoutError, /TLS handshake/) + expect { ::Timeout.timeout(3) { future.value } }.to raise_error(Ione::Io::ConnectionTimeoutError, /TLS handshake/) expect(::Time.now - started).to be < 1 # the half-open socket is closed, not left in the loop: reading @@ -294,7 +361,7 @@ def reactor_threads future = reactor.connect('127.0.0.1', port, timeout: 1, ssl: true) accepted = server.accept begin - future.value + ::Timeout.timeout(3) { future.value } rescue Ione::Io::ConnectionTimeoutError nil end @@ -374,7 +441,47 @@ def reactor_threads end end + describe('accept errors') do + [::IOError, ::Errno::EBADF, ::Errno::ECONNABORTED].each do |error_class| + it "handles #{error_class} at the listener without stopping the reactor" do + reactor.start.value + acceptor = reactor.bind('127.0.0.1', 0, 5).value + allow(acceptor.to_io).to receive(:accept_nonblock).and_raise(error_class) + + expect { acceptor.read }.not_to raise_error + + expect(acceptor.closed?).to eq(error_class != ::Errno::ECONNABORTED) + ::Timeout.timeout(2) { reactor.schedule_timer(0.01).value } + expect(reactor).to be_running + end + end + end + describe('plain TCP') do + context('when TCP completes while the reactor is stopped') do + let(:clock) { double('clock', now: 100.0) } + let(:reactor) { IoReactor.new(selector: selector, clock: clock) } + + it 'uses the completed connection even when the deadline has elapsed before restart' do + server = ::TCPServer.new('127.0.0.1', 0) + reactor.start.value + reactor.stop.value + connected = reactor.connect('127.0.0.1', server.addr[1], timeout: 0.25) + peer = server.accept + allow(clock).to receive(:now).and_return(100.25) + + ::Timeout.timeout(3) do + reactor.start.value + connected.value.write('hello') + expect(peer.read(5)).to eq('hello') + reactor.stop.value + end + ensure + peer.close if peer + server.close if server + end + end + it 'fails fast when nothing is listening' do reactor.start.value @@ -505,6 +612,47 @@ def add_connection(io) connection end + [:connect, :read, :flush].each do |operation| + [::IOError, ::Errno::EBADF].each do |error_class| + it "isolates #{error_class} during #{operation} and continues reading healthy connections" do + reader, writer = ::IO.pipe + bad = add_connection(reader.dup) + healthy = add_connection(reader) + if operation == :connect + bad.instance_variable_set(:@state, Ione::Io::BaseConnection::CONNECTING_STATE) + elsif operation == :flush + allow(bad).to receive(:writable?).and_return(true) + end + allow(bad).to receive(operation).and_raise(error_class, 'closed stream') + selected_readers = operation == :read ? [bad, healthy] : [healthy] + selected_writers = operation == :flush ? [bad] : nil + allow(selector).to receive(:select).and_return([selected_readers, selected_writers, nil]) + received = nil + healthy.on_data {|data| received = data} + writer.write('hello') + + expect { io_loop.tick }.not_to raise_error + + expect(bad).to be_closed + expect(healthy).not_to be_closed + expect(received).to eq('hello') + ensure + writer.close if writer + end + end + end + + it 'propagates dispatch errors unrelated to closed sockets' do + reader, writer = ::IO.pipe + connection = add_connection(reader) + allow(connection).to receive(:read).and_raise(ArgumentError, 'invalid handler') + allow(selector).to receive(:select).and_return([[connection], nil, nil]) + + expect { io_loop.tick }.to raise_error(ArgumentError, 'invalid handler') + ensure + writer.close if writer + end + [::IOError, ::Errno::EBADF, ::TypeError].each do |error_class| it "evicts an invalid socket after #{error_class} and retains healthy sockets" do reader, writer = ::IO.pipe @@ -552,19 +700,81 @@ def add_connection(io) end describe(IoReactor::Connection) do + let(:clock) { double('clock', now: 100.0) } + let(:socket) { double('socket', close: nil) } + let(:socket_impl) do + impl = double('socket_impl') + allow(impl).to receive(:getaddrinfo).and_return([[nil, 9042, nil, '127.0.0.1', ::Socket::AF_INET, ::Socket::SOCK_STREAM]]) + allow(impl).to receive(:sockaddr_in).and_return('SOCKADDR') + allow(impl).to receive(:new).and_return(socket) + impl + end + let(:connection) do + IoReactor::Connection.new('127.0.0.1', 9042, 0.25, double('unblocker', unblock: nil), clock, socket_impl) + end + it 'fails at the connect deadline without waiting for another polling tick' do - clock = double('clock', now: 100.0) - connection = IoReactor::Connection.new('127.0.0.1', 9042, 0.25, nil, clock) + allow(socket).to receive(:connect_nonblock).and_raise(Errno::EINPROGRESS) + future = connection.connect allow(clock).to receive(:now).and_return(100.25) - expect { connection.connect.value }.to raise_error(Ione::Io::ConnectionTimeoutError) + connection.connect + expect { future.value }.to raise_error(Ione::Io::ConnectionTimeoutError) + expect(connection).to be_closed + end + + [nil, Errno::EISCONN].each do |result| + it "prefers a completed connect (#{result || 'success'}) over the deadline" do + allow(socket).to receive(:connect_nonblock).and_raise(Errno::EINPROGRESS) + future = connection.connect + if result + allow(socket).to receive(:connect_nonblock).and_raise(result) + else + allow(socket).to receive(:connect_nonblock).and_return(0) + end + allow(clock).to receive(:now).and_return(100.25) + + connection.connect + + expect(future.value).to eq(connection) + expect(connection).to be_connected + end + end + + it 'fails only its own connection when the first connect attempt raises IOError' do + allow(socket).to receive(:connect_nonblock).and_raise(IOError, 'closed stream') + future = nil + + expect { future = connection.connect }.not_to raise_error + + expect { future.value }.to raise_error(Ione::Io::ConnectionError, 'closed stream') expect(connection).to be_closed end end describe(IoReactor::SslConnection) do + it 'accepts a handshake that completes at the deadline and preserves the completed future' do + clock = double('clock', now: 100.0) + raw_socket = double('raw socket', closed?: false, close: nil) + ssl_socket = double('SSL socket', close: nil) + socket_impl = double('SSL socket implementation', new: ssl_socket) + connection = IoReactor::SslConnection.new('127.0.0.1', 9042, raw_socket, double('unblocker', unblock: nil), nil, 100.25, clock) + connection.instance_variable_set(:@socket_impl, socket_impl) + allow(ssl_socket).to receive(:connect_nonblock).and_raise(::IO::EAGAINWaitReadable) + future = connection.connect + allow(clock).to receive(:now).and_return(100.25) + allow(ssl_socket).to receive(:connect_nonblock).and_return(ssl_socket) + + connection.connect + + expect(future.value).to eq(connection) + expect(connection).to be_connected + expect(connection.connect).to equal(future) + expect(connection).not_to be_closed + end + it 'fails the handshake without raising when the raw socket has disappeared' do - connection = IoReactor::SslConnection.new('127.0.0.1', 9042, nil, nil, nil, + connection = IoReactor::SslConnection.new('127.0.0.1', 9042, nil, double('unblocker', unblock: nil), nil, ::Time.now + 1, ::Time) future = connection.connect @@ -576,7 +786,7 @@ def add_connection(io) it 'closes the raw socket when closed before the handshake starts' do reader, writer = ::IO.pipe - connection = IoReactor::SslConnection.new('127.0.0.1', 9042, reader, nil, nil, + connection = IoReactor::SslConnection.new('127.0.0.1', 9042, reader, double('unblocker', unblock: nil), nil, ::Time.now + 1, ::Time) expect(connection.close).to eq(true) expect(reader).to be_closed