Class: Mongo::Server::ConnectionPool

# File 'lib/mongo/server/connection_pool.rb', line 233

attr_reader :generation_manager

# File 'lib/mongo/server/connection_pool.rb', line 342

attr_reader :max_connecting

# File 'lib/mongo/server/connection_pool.rb', line 180

attr_reader :options

# File 'lib/mongo/server/connection_pool.rb', line 246

def paused?
  raise_if_closed!

  @lock.synchronize do
    !@ready
  end
end

# File 'lib/mongo/server/connection_pool.rb', line 59

attr_reader :populate_semaphore

# File 'lib/mongo/server/connection_pool.rb', line 339

attr_reader :populator

# File 'lib/mongo/server/connection_pool.rb', line 311

def ready?
  @lock.synchronize do
    @ready
  end
end

# File 'lib/mongo/server/connection_pool.rb', line 183

attr_reader :server

# File 'lib/mongo/server/connection_pool.rb', line 402

def check_in(connection)
  check_invariants

  @lock.synchronize do
    do_check_in(connection)
  end
ensure
  check_invariants
end

# File 'lib/mongo/server/connection_pool.rb', line 1010

def check_invariants
  return unless Lint.enabled?

  # Server summary calls pool summary which requires pool lock -> deadlock.
  # Obtain the server summary ahead of time.
  server_summary = @server.summary

  @lock.synchronize do
    @available_connections.each do |connection|
      if connection.closed?
        raise Error::LintError, "Available connection is closed: #{connection} for #{server_summary}"
      end
    end

    @pending_connections.each do |connection|
      if connection.closed?
        raise Error::LintError, "Pending connection is closed: #{connection} for #{server_summary}"
      end
    end
  end
end

# File 'lib/mongo/server/connection_pool.rb', line 366

def check_out(connection_global_id: nil, context: nil)
  check_invariants

  publish_cmap_event(
    Monitoring::Event::Cmap::ConnectionCheckOutStarted.new(@server.address)
  )

  raise_if_pool_closed!
  raise_if_pool_paused_locked!

  connection = retrieve_and_connect_connection(
    connection_global_id, context
  )

  publish_cmap_event(
    Monitoring::Event::Cmap::ConnectionCheckedOut.new(@server.address, connection.id, self),
  )

  if Lint.enabled?
    unless connection.connected?
      raise Error::LintError, "Connection pool for #{address} checked out a disconnected connection #{connection.generation}:#{connection.id}"
    end
  end

  connection
ensure
  check_invariants
end

# File 'lib/mongo/server/connection_pool.rb', line 520

def clear(options = nil)
  raise_if_closed!

  if Lint.enabled? && !@server.unknown?
    raise Error::LintError, "Attempting to clear pool for server #{@server.summary} which is known"
  end

  do_clear(options)
end

# File 'lib/mongo/server/connection_pool.rb', line 1074

def clear_pending_connections
  until @pending_connections.empty?
    connection = @pending_connections.take(1).first
    connection.disconnect!
    @pending_connections.delete(connection)
  end
end

# File 'lib/mongo/server/connection_pool.rb', line 644

def close(options = nil)
  return if closed?

  options ||= {}

  stop_populator

  @lock.synchronize do
    until @available_connections.empty?
      connection = @available_connections.pop
      connection.disconnect!(reason: :pool_closed)
    end

    if options[:force]
      until @checked_out_connections.empty?
        connection = @checked_out_connections.take(1).first
        connection.disconnect!(reason: :pool_closed)
        @checked_out_connections.delete(connection)
      end
    end

    unless options && options[:stay_ready]
      # mark pool as closed before releasing lock so
      # no connections can be created, checked in, or checked out
      @closed = true
      @ready = false
    end

    @max_connecting_cv.broadcast
    @size_cv.broadcast
  end

  publish_cmap_event(
    Monitoring::Event::Cmap::PoolClosed.new(@server.address, self)
  )

  true
end

# File 'lib/mongo/server/connection_pool.rb', line 1036

def close_available_connections(service_id)
  if @server.load_balancer? && service_id
    loop do
      conn = @available_connections.detect do |conn|
        conn.service_id == service_id &&
        conn.generation < @generation_manager.generation(service_id: service_id)
      end
      if conn
        @available_connections.delete(conn)
        conn.disconnect!(reason: :stale, interrupted: true)
        @populate_semaphore.signal
      else
        break
      end
    end
  else
    @available_connections.delete_if do |conn|
      if conn.generation < @generation_manager.generation(service_id: service_id)
        conn.disconnect!(reason: :stale, interrupted: true)
        @populate_semaphore.signal
        true
      end
    end
  end
end

# File 'lib/mongo/server/connection_pool.rb', line 734

def close_idle_sockets
  return if closed?
  return unless max_idle_time

  @lock.synchronize do
    i = 0
    while i < @available_connections.length
      connection = @available_connections[i]
      if last_checkin = connection.last_checkin
        if (Time.now - last_checkin) > max_idle_time
          connection.disconnect!(reason: :idle)
          @available_connections.delete_at(i)
          @populate_semaphore.signal
          next
        end
      end
      i += 1
    end
  end
end

# File 'lib/mongo/server/connection_pool.rb', line 994

def connect_connection(connection, context = nil)
  begin
    connection.connect!(context)
  rescue Exception
    connection.disconnect!(reason: :error)
    raise
  end
rescue Error::SocketError, Error::SocketTimeoutError => exc
  @server.unknown!(
    generation: exc.generation,
    service_id: exc.service_id,
    stop_push_monitor: true,
  )
  raise
end

# File 'lib/mongo/server/connection_pool.rb', line 1348

def connect_or_raise(connection, context)
  connect_connection(connection, context)
rescue Exception
  # Handshake or authentication failed
  @lock.synchronize do
    if @pending_connections.include?(connection)
      @pending_connections.delete(connection)
    end
    @max_connecting_cv.signal
    @size_cv.signal
  end
  @populate_semaphore.signal
  publish_cmap_event(
    Monitoring::Event::Cmap::ConnectionCheckOutFailed.new(
      @server.address,
      Monitoring::Event::Cmap::ConnectionCheckOutFailed::CONNECTION_ERROR
    ),
  )
  raise
end

# File 'lib/mongo/server/connection_pool.rb', line 957

def connection_stale_unlocked?(connection)
  connection.generation != generation_unlocked(service_id: connection.service_id) &&
  !connection.pinned?
end

# File 'lib/mongo/server/connection_pool.rb', line 872

def create_and_add_connection
  connection = nil

  @lock.synchronize do
    if !closed? && @ready &&
      (unsynchronized_size + @connection_requests) < min_size &&
      @pending_connections.length < @max_connecting
    then
      connection = create_connection
      @pending_connections << connection
    else
      return true if remove_interrupted_connections
      return true if remove_stale_connection
      return false
    end
  end

  begin
    connect_connection(connection)
  rescue Exception
    @lock.synchronize do
      @pending_connections.delete(connection)
      @max_connecting_cv.signal
      @size_cv.signal
    end
    raise
  end

  @lock.synchronize do
    @available_connections << connection
    @pending_connections.delete(connection)
    @max_connecting_cv.signal
    @size_cv.signal
  end

  true
end

# File 'lib/mongo/server/connection_pool.rb', line 852

def create_connection
  r, _ = @generation_manager.pipe_fds(service_id: server.description.service_id)
  opts = options.merge(
    connection_pool: self,
    pipe: r
    # Do not pass app metadata - this will be retrieved by the connection
    # based on the auth needs.
  )
  unless @server.load_balancer?
    opts[:generation] = generation
  end
  Connection.new(@server, opts)
end

# File 'lib/mongo/server/connection_pool.rb', line 1372

def decrement_connection_requests_and_signal
  @connection_requests -= 1
  @max_connecting_cv.signal
  @size_cv.signal
end

# File 'lib/mongo/server/connection_pool.rb', line 537

def disconnect!(options = nil)
  do_clear(options)
rescue Error::PoolClosedError
  # The "disconnected" state is between closed and paused.
  # When we are trying to disconnect the pool, permit the pool to be
  # already closed.
end

# File 'lib/mongo/server/connection_pool.rb', line 415

def do_check_in(connection)
  # When a connection is interrupted it is checked back into the pool
  # and closed. The operation that was using the connection before it was
  # interrupted will attempt to check it back into the pool, and we
  # should ignore it since its already been closed and removed from the pool.
  return if connection.closed? && connection.interrupted?

  unless connection.connection_pool == self
    raise ArgumentError, "Trying to check in a connection which was not checked out by this pool: #{connection} checked out from pool #{connection.connection_pool} (for #{self})"
  end

  unless @checked_out_connections.include?(connection)
    raise ArgumentError, "Trying to check in a connection which is not currently checked out by this pool: #{connection} (for #{self})"
  end

  # Note: if an event handler raises, resource will not be signaled.
  # This means threads waiting for a connection to free up when
  # the pool is at max size may time out.
  # Threads that begin waiting after this method completes (with
  # the exception) should be fine.

  @checked_out_connections.delete(connection)
  @size_cv.signal

  publish_cmap_event(
    Monitoring::Event::Cmap::ConnectionCheckedIn.new(@server.address, connection.id, self)
  )

  if connection.interrupted?
    connection.disconnect!(reason: :stale)
    return
  end

  if connection.error?
    connection.disconnect!(reason: :error)
    return
  end

  if closed?
    connection.disconnect!(reason: :pool_closed)
    return
  end

  if connection.closed?
    # Connection was closed - for example, because it experienced
    # a network error. Nothing else needs to be done here.
    @populate_semaphore.signal
  elsif connection.generation != generation(service_id: connection.service_id) && !connection.pinned?
    # If connection is marked as pinned, it is used by a transaction
    # or a series of cursor operations in a load balanced setup.
    # In this case connection should not be disconnected until
    # unpinned.
    connection.disconnect!(reason: :stale)
    @populate_semaphore.signal
  else
    connection.record_checkin!
    @available_connections << connection

    @max_connecting_cv.signal
  end
end

# File 'lib/mongo/server/connection_pool.rb', line 545

def do_clear(options = nil)
  check_invariants

  service_id = options && options[:service_id]

  @lock.synchronize do
    # Generation must be bumped before emitting pool cleared event.
    @generation_manager.bump(service_id: service_id)

    unless options && options[:lazy]
      close_available_connections(service_id)
    end

    if options && options[:interrupt_in_use_connections]
      schedule_for_interruption(@checked_out_connections, service_id)
      schedule_for_interruption(@pending_connections, service_id)
    end

    if @ready
      publish_cmap_event(
        Monitoring::Event::Cmap::PoolCleared.new(
          @server.address,
          service_id: service_id,
          interrupt_in_use_connections: options&.[](:interrupt_in_use_connections)
        )
      )
      # Only pause the connection pool if the server was marked unknown,
      # otherwise, allow the retry to be attempted with a ready pool.
      do_pause if !@server.load_balancer? && @server.unknown?
    end

    # Broadcast here to cause all of the threads waiting on the max
    # connecting to break out of the wait loop and error.
    @max_connecting_cv.broadcast
    # Broadcast here to cause all of the threads waiting on the pool size
    # to break out of the wait loop and error.
    @size_cv.broadcast
  end

  # "Schedule the background thread" after clearing. This is responsible
  # for cleaning up stale threads, and interrupting in use connections.
  @populate_semaphore.signal
  true
ensure
  check_invariants
end

# File 'lib/mongo/server/connection_pool.rb', line 493

def do_pause
  if Lint.enabled? && !@server.unknown?
    raise Error::LintError, "Attempting to pause pool for server #{@server.summary} which is known"
  end

  return if !@ready

  @ready = false
end

# File 'lib/mongo/server/connection_pool.rb', line 1209

def get_connection(pid, connection_global_id)
  if connection = next_available_connection(connection_global_id)
    unless valid_available_connection?(connection, pid, connection_global_id)
      return nil
    end

    # We've got a connection, so we decrement the number of connection
    # requests.
    # We do not need to signal condition variable here, because
    # because the execution will continue, and we signal later.
    @connection_requests -= 1

    # If the connection is connected, it's not considered a
    # "pending connection". The pending_connections list represents
    # the set of connections that are awaiting connection.
    unless connection.connected?
      @pending_connections << connection
    end
    return connection
  elsif connection_global_id && @server.load_balancer?
    # A particular connection is requested, but it is not available.
    # If it is nether available not checked out, we should stop here.
    @checked_out_connections.detect do |conn|
      conn.global_id == connection_global_id
    end.tap do |conn|
      if conn.nil?
        publish_cmap_event(
          Monitoring::Event::Cmap::ConnectionCheckOutFailed.new(
            @server.address,
            Monitoring::Event::Cmap::ConnectionCheckOutFailed::CONNECTION_ERROR
          ),
        )
        # We're going to raise, so we need to decrement the number of
        # connection requests.
        decrement_connection_requests_and_signal
        raise Error::MissingConnection.new
      end
    end
    # We need a particular connection, and if it is not available
    # we can wait for an in-progress operation to return
    # such a connection to the pool.
    nil
  else
    connection = create_connection
    @connection_requests -= 1
    @pending_connections << connection
    return connection
  end
end

# File 'lib/mongo/server/connection_pool.rb', line 691

def inspect
  if closed?
    "#<Mongo::Server::ConnectionPool:0x#{object_id} min_size=#{min_size} max_size=#{max_size} " +
      "wait_timeout=#{wait_timeout} closed>"
  elsif !ready?
    "#<Mongo::Server::ConnectionPool:0x#{object_id} min_size=#{min_size} max_size=#{max_size} " +
      "wait_timeout=#{wait_timeout} paused>"
  else
    "#<Mongo::Server::ConnectionPool:0x#{object_id} min_size=#{min_size} max_size=#{max_size} " +
      "wait_timeout=#{wait_timeout} current_size=#{size} available=#{available_count}>"
  end
end

# File 'lib/mongo/server/connection_pool.rb', line 228

def max_idle_time
  @max_idle_time ||= options[:max_idle_time]
end

# File 'lib/mongo/server/connection_pool.rb', line 193

def max_size
  @max_size ||= options[:max_size] || [DEFAULT_MAX_SIZE, min_size].max
end

# File 'lib/mongo/server/connection_pool.rb', line 981

def maybe_raise_pool_cleared!(connection, e)
  if connection&.interrupted?
    err = Error::PoolClearedError.new(connection.server.address, connection.server.pool_internal).tap do |err|
      e.labels.each { |l| err.add_label(l) }
    end
    raise err
  else
    raise e
  end
end

# File 'lib/mongo/server/connection_pool.rb', line 202

def min_size
  @min_size ||= options[:min_size] || DEFAULT_MIN_SIZE
end

# File 'lib/mongo/server/connection_pool.rb', line 836

def next_available_connection(connection_global_id)
  raise_unless_locked!

  if @server.load_balancer? && connection_global_id
    conn = @available_connections.detect do |conn|
      conn.global_id == connection_global_id
    end
    if conn
      @available_connections.delete(conn)
    end
    conn
  else
    @available_connections.pop
  end
end

# File 'lib/mongo/server/connection_pool.rb', line 478

def pause
  raise_if_closed!

  check_invariants

  @lock.synchronize do
    do_pause
  end
ensure
  check_invariants
end

# File 'lib/mongo/server/connection_pool.rb', line 792

def populate
  return false if closed?

  begin
    return create_and_add_connection
  rescue Error::SocketError, Error::SocketTimeoutError => e
    # an error was encountered while connecting the connection,
    # ignore this first error and try again.
    log_warn("Populator failed to connect a connection for #{address}: #{e.class}: #{e}. It will retry.")
  end

  return create_and_add_connection
end

# File 'lib/mongo/server/connection_pool.rb', line 1083

def raise_check_out_timeout!(connection_global_id)
  raise_unless_locked!

  publish_cmap_event(
    Monitoring::Event::Cmap::ConnectionCheckOutFailed.new(
      @server.address,
      Monitoring::Event::Cmap::ConnectionCheckOutFailed::TIMEOUT,
    ),
  )

  connection_global_id_msg = if connection_global_id
    " for connection #{connection_global_id}"
  else
    ''
  end

  msg = "Timed out attempting to check out a connection " +
    "from pool for #{@server.address}#{connection_global_id_msg} after #{wait_timeout} sec. " +
    "Connections in pool: #{@available_connections.length} available, " +
    "#{@checked_out_connections.length} checked out, " +
    "#{@pending_connections.length} pending, " +
    "#{@connection_requests} connections requests " +
    "(max size: #{max_size})"
  raise Error::ConnectionCheckOutTimeout.new(msg, address: @server.address)
end

# File 'lib/mongo/server/connection_pool.rb', line 1109

def raise_check_out_timeout_locked!(connection_global_id)
  @lock.synchronize do
    raise_check_out_timeout!(connection_global_id)
  end
end

# File 'lib/mongo/server/connection_pool.rb', line 967

def raise_if_closed!
  if closed?
    raise Error::PoolClosedError.new(@server.address, self)
  end
end

# File 'lib/mongo/server/connection_pool.rb', line 1150

def raise_if_not_ready!
  raise_unless_locked!
  raise_if_pool_closed!
  raise_if_pool_paused!
end

# File 'lib/mongo/server/connection_pool.rb', line 1115

def raise_if_pool_closed!
  if closed?
    publish_cmap_event(
      Monitoring::Event::Cmap::ConnectionCheckOutFailed.new(
        @server.address,
        Monitoring::Event::Cmap::ConnectionCheckOutFailed::POOL_CLOSED
      ),
    )
    raise Error::PoolClosedError.new(@server.address, self)
  end
end

# File 'lib/mongo/server/connection_pool.rb', line 1127

def raise_if_pool_paused!
  raise_unless_locked!

  if !@ready
    publish_cmap_event(
      Monitoring::Event::Cmap::ConnectionCheckOutFailed.new(
        @server.address,
        # CMAP spec decided to conflate pool paused with all the other
        # possible non-timeout errors.
        Monitoring::Event::Cmap::ConnectionCheckOutFailed::CONNECTION_ERROR,
      ),
    )
    raise Error::PoolPausedError.new(@server.address, self)
  end
end

# File 'lib/mongo/server/connection_pool.rb', line 1143

def raise_if_pool_paused_locked!
  @lock.synchronize do
    raise_if_pool_paused!
  end
end

# File 'lib/mongo/server/connection_pool.rb', line 1156

def raise_unless_locked!
  unless @lock.owned?
    raise ArgumentError, "the lock must be owned when calling this method"
  end
end

# File 'lib/mongo/server/connection_pool.rb', line 593

def ready
  raise_if_closed!

  # TODO: Add this back in RUBY-3174.
  # if Lint.enabled?
  #   unless @server.connected?
  #     raise Error::LintError, "Attempting to ready a pool for server #{@server.summary} which is disconnected"
  #   end
  # end

  @lock.synchronize do
    return if @ready

    @ready = true
  end

  # Note that the CMAP spec demands serialization of CMAP events for a
  # pool. In order to implement this, event publication must be done into
  # a queue which is synchronized, instead of subscribers being invoked
  # from the trigger method like this one here inline. On MRI, assuming
  # the threads yield to others when they stop having work to do, it is
  # likely that the events would in practice always be published in the
  # required order. JRuby, being truly concurrent with OS threads,
  # would not offers such a guarantee.
  publish_cmap_event(
    Monitoring::Event::Cmap::PoolReady.new(@server.address, options, self)
  )

  if options.fetch(:populator_io, true)
    if @populator.running?
      @populate_semaphore.signal
    else
      @populator.run!
    end
  end
end

# File 'lib/mongo/server/connection_pool.rb', line 920

def remove_interrupted_connections
  return false if @interrupt_connections.empty?

  gens = Set.new
  while conn = @interrupt_connections.pop
    if @checked_out_connections.include?(conn)
      # If the connection has been checked out, mark it as interrupted and it will
      # be disconnected on check in.
      conn.interrupted!
      do_check_in(conn)
    elsif @pending_connections.include?(conn)
      # If the connection is pending, disconnect with the interrupted flag.
      conn.disconnect!(reason: :stale, interrupted: true)
      @pending_connections.delete(conn)
    end
    gens << [ conn.generation, conn.service_id ]
  end

  # Close the write side of the pipe. Pending connections might be
  # hanging on the Kernel#select call, so in order to interrupt that,
  # we also listen for the read side of the pipe in Kernel#select and
  # close the write side of the pipe here, which will cause select to
  # wake up and raise an IOError now that the socket is closed.
  # The read side of the pipe will be scheduled for closing on the next
  # generation bump.
  gens.each do |gen, service_id|
    @generation_manager.remove_pipe_fds(gen, service_id: service_id)
  end

  true
end

# File 'lib/mongo/server/connection_pool.rb', line 911

def remove_stale_connection
  if conn = @available_connections.detect(&method(:connection_stale_unlocked?))
    conn.disconnect!(reason: :stale)
    @available_connections.delete(conn)
    return true
  end
end

# File 'lib/mongo/server/connection_pool.rb', line 1271

def retrieve_and_connect_connection(connection_global_id, context =  nil)
  deadline = Utils.monotonic_time + wait_timeout(context)
  connection = nil

  @lock.synchronize do
    # The first gate to checking out a connection. Make sure the number of
    # unavailable connections is less than the max pool size.
    until max_size == 0 || unavailable_connections < max_size
      wait = deadline - Utils.monotonic_time
      raise_check_out_timeout!(connection_global_id) if wait <= 0
      @size_cv.wait(wait)
      raise_if_not_ready!
    end
    @connection_requests += 1
    connection = wait_for_connection(connection_global_id, deadline)
  end

  connect_or_raise(connection, context) unless connection.connected?

  @lock.synchronize do
    @checked_out_connections << connection
    if @pending_connections.include?(connection)
      @pending_connections.delete(connection)
    end
    @max_connecting_cv.signal
    # no need to signal size_cv here since the number of unavailable
    # connections is unchanged.
  end

  connection
end

# File 'lib/mongo/server/connection_pool.rb', line 1066

def schedule_for_interruption(connections, service_id)
  @interrupt_connections += connections.select do |conn|
    (!server.load_balancer? || conn.service_id == service_id) &&
    conn.generation < @generation_manager.generation(service_id: service_id)
  end
end

# File 'lib/mongo/server/connection_pool.rb', line 261

def size
  raise_if_closed!

  @lock.synchronize do
    unsynchronized_size
  end
end

# File 'lib/mongo/server/connection_pool.rb', line 764

def stop_populator
  @populator.stop!

  @lock.synchronize do
    # If stop_populator is called while populate is running, there may be
    # connections waiting to be connected, connections which have not yet
    # been moved to available_connections, or connections moved to available_connections
    # but not deleted from pending_connections. These should be cleaned up.
    clear_pending_connections
  end
end

# File 'lib/mongo/server/connection_pool.rb', line 321

def summary
  @lock.synchronize do
    state = if closed?
      'closed'
    elsif !@ready
      'paused'
    else
      'ready'
    end
    "#<ConnectionPool size=#{unsynchronized_size} (#{min_size}-#{max_size}) " +
      "used=#{@checked_out_connections.length} avail=#{@available_connections.length} pending=#{@pending_connections.length} #{state}>"
  end
end

# File 'lib/mongo/server/connection_pool.rb', line 282

def unavailable_connections
  @checked_out_connections.length + @pending_connections.length + @connection_requests
end

# File 'lib/mongo/server/connection_pool.rb', line 273

def unsynchronized_size
  @available_connections.length + @checked_out_connections.length + @pending_connections.length
end

# File 'lib/mongo/server/connection_pool.rb', line 1162

def valid_available_connection?(connection, pid, connection_global_id)
  if connection.pid != pid
    log_warn("Detected PID change - Mongo client should have been reconnected (old pid #{connection.pid}, new pid #{pid}")
    connection.disconnect!(reason: :stale)
    @populate_semaphore.signal
    return false
  end

  if !connection.pinned?
    # If connection is marked as pinned, it is used by a transaction
    # or a series of cursor operations in a load balanced setup.
    # In this case connection should not be disconnected until
    # unpinned.
    if connection.generation != generation(
      service_id: connection.service_id
    )
      # Stale connections should be disconnected in the clear
      # method, but if any don't, check again here
      connection.disconnect!(reason: :stale)
      @populate_semaphore.signal
      return false
    end

    if max_idle_time && connection.last_checkin &&
      Time.now - connection.last_checkin > max_idle_time
    then
      connection.disconnect!(reason: :idle)
      @populate_semaphore.signal
      return false
    end
  end
  true
end

# File 'lib/mongo/server/connection_pool.rb', line 1310

def wait_for_connection(connection_global_id, deadline)
  connection = nil
  while connection.nil?
    # The second gate to checking out a connection. Make sure 1) there
    # exists an available connection and 2) we are under max_connecting.
    until @available_connections.any? || @pending_connections.length < @max_connecting
      wait = deadline - Utils.monotonic_time
      if wait <= 0
        # We are going to raise a timeout error, so the connection
        # request is not going to be fulfilled. Decrement the counter
        # here.
        decrement_connection_requests_and_signal
        raise_check_out_timeout!(connection_global_id)
      end
      @max_connecting_cv.wait(wait)
      # We do not need to decrement the connection_requests counter
      # or signal here because the pool is not ready yet.
      raise_if_not_ready!
    end

    connection = get_connection(Process.pid, connection_global_id)
    wait = deadline - Utils.monotonic_time
    if connection.nil? && wait <= 0
      # connection is nil here, it means that get_connection method
      # did not create a new connection; therefore, it did not decrease
      # the connection_requests counter. We need to do it here.
      decrement_connection_requests_and_signal
      raise_check_out_timeout!(connection_global_id)
    end
  end

  connection
end

# File 'lib/mongo/server/connection_pool.rb', line 214

def wait_timeout(context = nil)
  if context&.remaining_timeout_sec.nil?
    options[:wait_timeout] || DEFAULT_WAIT_TIMEOUT
  else
    context&.remaining_timeout_sec
  end
end

# File 'lib/mongo/server/connection_pool.rb', line 714

def with_connection(connection_global_id: nil, context: nil)
  raise_if_closed!

  connection = check_out(
    connection_global_id: connection_global_id,
    context: context
  )
  yield(connection)
rescue Error::SocketError, Error::SocketTimeoutError, Error::ConnectionPerished => e
  maybe_raise_pool_cleared!(connection, e)
ensure
  if connection
    check_in(connection)
  end
end