Class ActiveRecord::ConnectionAdapters::ConnectionPool::Reaper < Object

# File activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb, line 290
def initialize(pool, frequency)
  @pool      = pool
  @frequency = frequency
end

# File activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb, line 786
def acquire_connection(checkout_timeout)
  # NOTE: we rely on <tt>@available.poll</tt> and +try_to_checkout_new_connection+ to
  # +conn.lease+ the returned connection (and to do this in a +synchronized+
  # section). This is not the cleanest implementation, as ideally we would
  # <tt>synchronize { conn.lease }</tt> in this method, but by leaving it to <tt>@available.poll</tt>
  # and +try_to_checkout_new_connection+ we can piggyback on +synchronize+ sections
  # of the said methods and avoid an additional +synchronize+ overhead.
  if conn = @available.poll || try_to_checkout_new_connection
    conn
  else
    reap
    @available.poll(checkout_timeout)
  end
end

# File activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb, line 388
def active_connection?
  @thread_cached_conns[connection_cache_key(Thread.current)]
end

# File activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb, line 846
def adopt_connection(conn)
  conn.pool = self
  @connections << conn
end

# File activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb, line 680
    def attempt_to_checkout_all_existing_connections(raise_on_acquisition_timeout = true)
      collected_conns = synchronize do
        # account for our own connections
        @connections.select { |conn| conn.owner == Thread.current }
      end

      newly_checked_out = []
      timeout_time      = Time.now + (@checkout_timeout * 2)

      @available.with_a_bias_for(Thread.current) do
        loop do
          synchronize do
            return if collected_conns.size == @connections.size && @now_connecting == 0
            remaining_timeout = timeout_time - Time.now
            remaining_timeout = 0 if remaining_timeout < 0
            conn = checkout_for_exclusive_access(remaining_timeout)
            collected_conns   << conn
            newly_checked_out << conn
          end
        end
      end
    rescue ExclusiveConnectionTimeoutError
      # <tt>raise_on_acquisition_timeout == false</tt> means we are directed to ignore any
      # timeouts and are expected to just give up: we've obtained as many connections
      # as possible, note that in a case like that we don't return any of the
      # +newly_checked_out+ connections.

      if raise_on_acquisition_timeout
        release_newly_checked_out = true
        raise
      end
    rescue Exception # if something else went wrong
      # this can't be a "naked" rescue, because we have should return conns
      # even for non-StandardErrors
      release_newly_checked_out = true
      raise
    ensure
      if release_newly_checked_out && newly_checked_out
        # releasing only those conns that were checked out in this method, conns
        # checked outside this method (before it was called) are not for us to release
        newly_checked_out.each { |conn| checkin(conn) }
      end
    end

    #--
    # Must be called in a synchronize block.
    def checkout_for_exclusive_access(checkout_timeout)
      checkout(checkout_timeout)
    rescue ConnectionTimeoutError
      # this block can't be easily moved into attempt_to_checkout_all_existing_connections's
      # rescue block, because doing so would put it outside of synchronize section, without
      # being in a critical section thread_report might become inaccurate
      msg = "could not obtain ownership of all database connections in #{checkout_timeout} seconds".dup

      thread_report = []
      @connections.each do |conn|
        unless conn.owner == Thread.current
          thread_report << "#{conn} is owned by #{conn.owner}"
        end
      end

      msg << " (#{thread_report.join(', ')})" if thread_report.any?

      raise ExclusiveConnectionTimeoutError, msg
    end

    def with_new_connections_blocked
      synchronize do
        @threads_blocking_new_connections += 1
      end

      yield
    ensure
      num_new_conns_required = 0

      synchronize do
        @threads_blocking_new_connections -= 1

        if @threads_blocking_new_connections.zero?
          @available.clear

          num_new_conns_required = num_waiting_in_queue

          @connections.each do |conn|
            next if conn.in_use?

            @available.add conn
            num_new_conns_required -= 1
          end
        end
      end

      bulk_make_new_connections(num_new_conns_required) if num_new_conns_required > 0
    end

    # Acquire a connection by one of 1) immediately removing one
    # from the queue of available connections, 2) creating a new
    # connection if the pool is not at capacity, 3) waiting on the
    # queue for a connection to become available.
    #
    # Raises:
    # - ActiveRecord::ConnectionTimeoutError if a connection could not be acquired
    #
    #--
    # Implementation detail: the connection returned by +acquire_connection+
    # will already be "+connection.lease+ -ed" to the current thread.
    def acquire_connection(checkout_timeout)
      # NOTE: we rely on <tt>@available.poll</tt> and +try_to_checkout_new_connection+ to
      # +conn.lease+ the returned connection (and to do this in a +synchronized+
      # section). This is not the cleanest implementation, as ideally we would
      # <tt>synchronize { conn.lease }</tt> in this method, but by leaving it to <tt>@available.poll</tt>
      # and +try_to_checkout_new_connection+ we can piggyback on +synchronize+ sections
      # of the said methods and avoid an additional +synchronize+ overhead.
      if conn = @available.poll || try_to_checkout_new_connection
        conn
      else
        reap
        @available.poll(checkout_timeout)
      end
    end

    #--
    # if owner_thread param is omitted, this must be called in synchronize block
    def remove_connection_from_thread_cache(conn, owner_thread = conn.owner)
      @thread_cached_conns.delete_pair(connection_cache_key(owner_thread), conn)
    end
    alias_method :release, :remove_connection_from_thread_cache

    def new_connection
      Base.send(spec.adapter_method, spec.config).tap do |conn|
        conn.schema_cache = schema_cache.dup if schema_cache
      end
    end

    # If the pool is not at a <tt>@size</tt> limit, establish new connection. Connecting
    # to the DB is done outside main synchronized section.
    #--
    # Implementation constraint: a newly established connection returned by this
    # method must be in the +.leased+ state.
    def try_to_checkout_new_connection
      # first in synchronized section check if establishing new conns is allowed
      # and increment @now_connecting, to prevent overstepping this pool's @size
      # constraint
      do_checkout = synchronize do
        if @threads_blocking_new_connections.zero? && (@connections.size + @now_connecting) < @size
          @now_connecting += 1
        end
      end
      if do_checkout
        begin
          # if successfully incremented @now_connecting establish new connection
          # outside of synchronized section
          conn = checkout_new_connection
        ensure
          synchronize do
            if conn
              adopt_connection(conn)
              # returned conn needs to be already leased
              conn.lease
            end
            @now_connecting -= 1
          end
        end
      end
    end

    def adopt_connection(conn)
      conn.pool = self
      @connections << conn
    end

    def checkout_new_connection
      raise ConnectionNotEstablished unless @automatic_reconnect
      new_connection
    end

    def checkout_and_verify(c)
      c._run_checkout_callbacks do
        c.verify!
      end
      c
    rescue
      remove c
      c.disconnect!
      raise
    end
end

# File activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb, line 650
def bulk_make_new_connections(num_new_conns_needed)
  num_new_conns_needed.times do
    # try_to_checkout_new_connection will not exceed pool's @size limit
    if new_conn = try_to_checkout_new_connection
      # make the new_conn available to the starving threads stuck @available Queue
      checkin(new_conn)
    end
  end
end

# File activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb, line 529
def checkin(conn)
  conn.lock.synchronize do
    synchronize do
      remove_connection_from_thread_cache conn

      conn._run_checkin_callbacks do
        conn.expire
      end

      @available.add conn
    end
  end
end

# File activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb, line 520
def checkout(checkout_timeout = @checkout_timeout)
  checkout_and_verify(acquire_connection(checkout_timeout))
end

# File activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb, line 856
def checkout_and_verify(c)
  c._run_checkout_callbacks do
    c.verify!
  end
  c
rescue
  remove c
  c.disconnect!
  raise
end

# File activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb, line 726
def checkout_for_exclusive_access(checkout_timeout)
  checkout(checkout_timeout)
rescue ConnectionTimeoutError
  # this block can't be easily moved into attempt_to_checkout_all_existing_connections's
  # rescue block, because doing so would put it outside of synchronize section, without
  # being in a critical section thread_report might become inaccurate
  msg = "could not obtain ownership of all database connections in #{checkout_timeout} seconds".dup

  thread_report = []
  @connections.each do |conn|
    unless conn.owner == Thread.current
      thread_report << "#{conn} is owned by #{conn.owner}"
    end
  end

  msg << " (#{thread_report.join(', ')})" if thread_report.any?

  raise ExclusiveConnectionTimeoutError, msg
end

# File activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb, line 851
def checkout_new_connection
  raise ConnectionNotEstablished unless @automatic_reconnect
  new_connection
end

# File activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb, line 478
def clear_reloadable_connections(raise_on_acquisition_timeout = true)
  with_exclusively_acquired_all_connections(raise_on_acquisition_timeout) do
    synchronize do
      @connections.each do |conn|
        if conn.in_use?
          conn.steal!
          checkin conn
        end
        conn.disconnect! if conn.requires_reloading?
      end
      @connections.delete_if(&:requires_reloading?)
      @available.clear
    end
  end
end

# File activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb, line 502
def clear_reloadable_connections!
  clear_reloadable_connections(false)
end

# File activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb, line 420
def connected?
  synchronize { @connections.any? }
end

# File activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb, line 379
def connection
  @thread_cached_conns[connection_cache_key(@lock_thread || Thread.current)] ||= checkout
end

# File activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb, line 665
def connection_cache_key(thread)
  thread
end

# File activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb, line 430
def disconnect(raise_on_acquisition_timeout = true)
  with_exclusively_acquired_all_connections(raise_on_acquisition_timeout) do
    synchronize do
      @connections.each do |conn|
        if conn.in_use?
          conn.steal!
          checkin conn
        end
        conn.disconnect!
      end
      @connections = []
      @available.clear
    end
  end
end

# File activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb, line 452
def disconnect!
  disconnect(false)
end

# File activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb, line 599
    def flush(minimum_idle = @idle_timeout)
      return if minimum_idle.nil?

      idle_connections = synchronize do
        @connections.select do |conn|
          !conn.in_use? && conn.seconds_idle >= minimum_idle
        end.each do |conn|
          conn.lease

          @available.delete conn
          @connections.delete conn
        end
      end

      idle_connections.each do |conn|
        conn.disconnect!
      end
    end

    # Disconnect all currently idle connections. Connections currently checked
    # out are unaffected.
    def flush!
      reap
      flush(-1)
    end

    def num_waiting_in_queue # :nodoc:
      @available.num_waiting
    end

    # Return connection pool's usage statistic
    # Example:
    #
    #    ActiveRecord::Base.connection_pool.stat # => { size: 15, connections: 1, busy: 1, dead: 0, idle: 0, waiting: 0, checkout_timeout: 5 }
    def stat
      synchronize do
        {
          size: size,
          connections: @connections.size,
          busy: @connections.count { |c| c.in_use? && c.owner.alive? },
          dead: @connections.count { |c| c.in_use? && !c.owner.alive? },
          idle: @connections.count { |c| !c.in_use? },
          waiting: num_waiting_in_queue,
          checkout_timeout: checkout_timeout
        }
      end
    end

    private
      #--
      # this is unfortunately not concurrent
      def bulk_make_new_connections(num_new_conns_needed)
        num_new_conns_needed.times do
          # try_to_checkout_new_connection will not exceed pool's @size limit
          if new_conn = try_to_checkout_new_connection
            # make the new_conn available to the starving threads stuck @available Queue
            checkin(new_conn)
          end
        end
      end

      #--
      # From the discussion on GitHub:
      #  https://github.com/rails/rails/pull/14938#commitcomment-6601951
      # This hook-in method allows for easier monkey-patching fixes needed by
      # JRuby users that use Fibers.
      def connection_cache_key(thread)
        thread
      end

      # Take control of all existing connections so a "group" action such as
      # reload/disconnect can be performed safely. It is no longer enough to
      # wrap it in +synchronize+ because some pool's actions are allowed
      # to be performed outside of the main +synchronize+ block.
      def with_exclusively_acquired_all_connections(raise_on_acquisition_timeout = true)
        with_new_connections_blocked do
          attempt_to_checkout_all_existing_connections(raise_on_acquisition_timeout)
          yield
        end
      end

      def attempt_to_checkout_all_existing_connections(raise_on_acquisition_timeout = true)
        collected_conns = synchronize do
          # account for our own connections
          @connections.select { |conn| conn.owner == Thread.current }
        end

        newly_checked_out = []
        timeout_time      = Time.now + (@checkout_timeout * 2)

        @available.with_a_bias_for(Thread.current) do
          loop do
            synchronize do
              return if collected_conns.size == @connections.size && @now_connecting == 0
              remaining_timeout = timeout_time - Time.now
              remaining_timeout = 0 if remaining_timeout < 0
              conn = checkout_for_exclusive_access(remaining_timeout)
              collected_conns   << conn
              newly_checked_out << conn
            end
          end
        end
      rescue ExclusiveConnectionTimeoutError
        # <tt>raise_on_acquisition_timeout == false</tt> means we are directed to ignore any
        # timeouts and are expected to just give up: we've obtained as many connections
        # as possible, note that in a case like that we don't return any of the
        # +newly_checked_out+ connections.

        if raise_on_acquisition_timeout
          release_newly_checked_out = true
          raise
        end
      rescue Exception # if something else went wrong
        # this can't be a "naked" rescue, because we have should return conns
        # even for non-StandardErrors
        release_newly_checked_out = true
        raise
      ensure
        if release_newly_checked_out && newly_checked_out
          # releasing only those conns that were checked out in this method, conns
          # checked outside this method (before it was called) are not for us to release
          newly_checked_out.each { |conn| checkin(conn) }
        end
      end

      #--
      # Must be called in a synchronize block.
      def checkout_for_exclusive_access(checkout_timeout)
        checkout(checkout_timeout)
      rescue ConnectionTimeoutError
        # this block can't be easily moved into attempt_to_checkout_all_existing_connections's
        # rescue block, because doing so would put it outside of synchronize section, without
        # being in a critical section thread_report might become inaccurate
        msg = "could not obtain ownership of all database connections in #{checkout_timeout} seconds".dup

        thread_report = []
        @connections.each do |conn|
          unless conn.owner == Thread.current
            thread_report << "#{conn} is owned by #{conn.owner}"
          end
        end

        msg << " (#{thread_report.join(', ')})" if thread_report.any?

        raise ExclusiveConnectionTimeoutError, msg
      end

      def with_new_connections_blocked
        synchronize do
          @threads_blocking_new_connections += 1
        end

        yield
      ensure
        num_new_conns_required = 0

        synchronize do
          @threads_blocking_new_connections -= 1

          if @threads_blocking_new_connections.zero?
            @available.clear

            num_new_conns_required = num_waiting_in_queue

            @connections.each do |conn|
              next if conn.in_use?

              @available.add conn
              num_new_conns_required -= 1
            end
          end
        end

        bulk_make_new_connections(num_new_conns_required) if num_new_conns_required > 0
      end

      # Acquire a connection by one of 1) immediately removing one
      # from the queue of available connections, 2) creating a new
      # connection if the pool is not at capacity, 3) waiting on the
      # queue for a connection to become available.
      #
      # Raises:
      # - ActiveRecord::ConnectionTimeoutError if a connection could not be acquired
      #
      #--
      # Implementation detail: the connection returned by +acquire_connection+
      # will already be "+connection.lease+ -ed" to the current thread.
      def acquire_connection(checkout_timeout)
        # NOTE: we rely on <tt>@available.poll</tt> and +try_to_checkout_new_connection+ to
        # +conn.lease+ the returned connection (and to do this in a +synchronized+
        # section). This is not the cleanest implementation, as ideally we would
        # <tt>synchronize { conn.lease }</tt> in this method, but by leaving it to <tt>@available.poll</tt>
        # and +try_to_checkout_new_connection+ we can piggyback on +synchronize+ sections
        # of the said methods and avoid an additional +synchronize+ overhead.
        if conn = @available.poll || try_to_checkout_new_connection
          conn
        else
          reap
          @available.poll(checkout_timeout)
        end
      end

      #--
      # if owner_thread param is omitted, this must be called in synchronize block
      def remove_connection_from_thread_cache(conn, owner_thread = conn.owner)
        @thread_cached_conns.delete_pair(connection_cache_key(owner_thread), conn)
      end
      alias_method :release, :remove_connection_from_thread_cache

      def new_connection
        Base.send(spec.adapter_method, spec.config).tap do |conn|
          conn.schema_cache = schema_cache.dup if schema_cache
        end
      end

      # If the pool is not at a <tt>@size</tt> limit, establish new connection. Connecting
      # to the DB is done outside main synchronized section.
      #--
      # Implementation constraint: a newly established connection returned by this
      # method must be in the +.leased+ state.
      def try_to_checkout_new_connection
        # first in synchronized section check if establishing new conns is allowed
        # and increment @now_connecting, to prevent overstepping this pool's @size
        # constraint
        do_checkout = synchronize do
          if @threads_blocking_new_connections.zero? && (@connections.size + @now_connecting) < @size
            @now_connecting += 1
          end
        end
        if do_checkout
          begin
            # if successfully incremented @now_connecting establish new connection
            # outside of synchronized section
            conn = checkout_new_connection
          ensure
            synchronize do
              if conn
                adopt_connection(conn)
                # returned conn needs to be already leased
                conn.lease
              end
              @now_connecting -= 1
            end
          end
        end
      end

      def adopt_connection(conn)
        conn.pool = self
        @connections << conn
      end

      def checkout_new_connection
        raise ConnectionNotEstablished unless @automatic_reconnect
        new_connection
      end

      def checkout_and_verify(c)
        c._run_checkout_callbacks do
          c.verify!
        end
        c
      rescue
        remove c
        c.disconnect!
        raise
      end
  end

  # ConnectionHandler is a collection of ConnectionPool objects. It is used
  # for keeping separate connection pools that connect to different databases.
  #
  # For example, suppose that you have 5 models, with the following hierarchy:
  #
  #   class Author < ActiveRecord::Base
  #   end
  #
  #   class BankAccount < ActiveRecord::Base
  #   end
  #
  #   class Book < ActiveRecord::Base
  #     establish_connection :library_db
  #   end
  #
  #   class ScaryBook < Book
  #   end
  #
  #   class GoodBook < Book
  #   end
  #
  # And a database.yml that looked like this:
  #
  #   development:
  #     database: my_application
  #     host: localhost
  #
  #   library_db:
  #     database: library
  #     host: some.library.org
  #
  # Your primary database in the development environment is "my_application"
  # but the Book model connects to a separate database called "library_db"
  # (this can even be a database on a different machine).
  #
  # Book, ScaryBook and GoodBook will all use the same connection pool to
  # "library_db" while Author, BankAccount, and any other models you create
  # will use the default connection pool to "my_application".
  #
  # The various connection pools are managed by a single instance of
  # ConnectionHandler accessible via ActiveRecord::Base.connection_handler.
  # All Active Record models use this handler to determine the connection pool that they
  # should use.
  #
  # The ConnectionHandler class is not coupled with the Active models, as it has no knowledge
  # about the model. The model needs to pass a specification name to the handler,
  # in order to look up the correct connection pool.
  class ConnectionHandler
    def self.unowned_pool_finalizer(pid_map) # :nodoc:
      lambda do |_|
        discard_unowned_pools(pid_map)
      end
    end

    def self.discard_unowned_pools(pid_map) # :nodoc:
      pid_map.each do |pid, pools|
        pools.values.compact.each(&:discard!) unless pid == Process.pid
      end
    end

    def initialize
      # These caches are keyed by spec.name (ConnectionSpecification#name).
      @owner_to_pool = Concurrent::Map.new(initial_capacity: 2) do |h, k|
        # Discard the parent's connection pools immediately; we have no need
        # of them
        ConnectionHandler.discard_unowned_pools(h)

        h[k] = Concurrent::Map.new(initial_capacity: 2)
      end

      # Backup finalizer: if the forked child never needed a pool, the above
      # early discard has not occurred
      ObjectSpace.define_finalizer self, ConnectionHandler.unowned_pool_finalizer(@owner_to_pool)
    end

    def connection_pool_list
      owner_to_pool.values.compact
    end
    alias :connection_pools :connection_pool_list

    def establish_connection(config)
      resolver = ConnectionSpecification::Resolver.new(Base.configurations)
      spec = resolver.spec(config)

      remove_connection(spec.name)

      message_bus = ActiveSupport::Notifications.instrumenter
      payload = {
        connection_id: object_id
      }
      if spec
        payload[:spec_name] = spec.name
        payload[:config] = spec.config
      end

      message_bus.instrument("!connection.active_record", payload) do
        owner_to_pool[spec.name] = ConnectionAdapters::ConnectionPool.new(spec)
      end

      owner_to_pool[spec.name]
    end

    # Returns true if there are any active connections among the connection
    # pools that the ConnectionHandler is managing.
    def active_connections?
      connection_pool_list.any?(&:active_connection?)
    end

    # Returns any connections in use by the current thread back to the pool,
    # and also returns connections to the pool cached by threads that are no
    # longer alive.
    def clear_active_connections!
      connection_pool_list.each(&:release_connection)
    end

    # Clears the cache which maps classes.
    #
    # See ConnectionPool#clear_reloadable_connections! for details.
    def clear_reloadable_connections!
      connection_pool_list.each(&:clear_reloadable_connections!)
    end

    def clear_all_connections!
      connection_pool_list.each(&:disconnect!)
    end

    # Disconnects all currently idle connections.
    #
    # See ConnectionPool#flush! for details.
    def flush_idle_connections!
      connection_pool_list.each(&:flush!)
    end

    # Locate the connection of the nearest super class. This can be an
    # active or defined connection: if it is the latter, it will be
    # opened and set as the active connection for the class it was defined
    # for (not necessarily the current class).
    def retrieve_connection(spec_name) #:nodoc:
      pool = retrieve_connection_pool(spec_name)
      raise ConnectionNotEstablished, "No connection pool with '#{spec_name}' found." unless pool
      pool.connection
    end

    # Returns true if a connection that's accessible to this class has
    # already been opened.
    def connected?(spec_name)
      conn = retrieve_connection_pool(spec_name)
      conn && conn.connected?
    end

    # Remove the connection for this class. This will close the active
    # connection and the defined connection (if they exist). The result
    # can be used as an argument for #establish_connection, for easily
    # re-establishing the connection.
    def remove_connection(spec_name)
      if pool = owner_to_pool.delete(spec_name)
        pool.automatic_reconnect = false
        pool.disconnect!
        pool.spec.config
      end
    end

    # Retrieving the connection pool happens a lot, so we cache it in @owner_to_pool.
    # This makes retrieving the connection pool O(1) once the process is warm.
    # When a connection is established or removed, we invalidate the cache.
    def retrieve_connection_pool(spec_name)
      owner_to_pool.fetch(spec_name) do
        # Check if a connection was previously established in an ancestor process,
        # which may have been forked.
        if ancestor_pool = pool_from_any_process_for(spec_name)
          # A connection was established in an ancestor process that must have
          # subsequently forked. We can't reuse the connection, but we can copy
          # the specification and establish a new connection with it.
          establish_connection(ancestor_pool.spec.to_hash).tap do |pool|
            pool.schema_cache = ancestor_pool.schema_cache if ancestor_pool.schema_cache
          end
        else
          owner_to_pool[spec_name] = nil
        end
      end
    end

    private

      def owner_to_pool
        @owner_to_pool[Process.pid]
      end

      def pool_from_any_process_for(spec_name)
        owner_to_pool = @owner_to_pool.values.reverse.find { |v| v[spec_name] }
        owner_to_pool && owner_to_pool[spec_name]
      end
  end
end

# File activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb, line 620
def flush!
  reap
  flush(-1)
end

# File activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb, line 366
def lock_thread=(lock_thread)
  if lock_thread
    @lock_thread = Thread.current
  else
    @lock_thread = nil
  end
end

# File activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb, line 808
def new_connection
  Base.send(spec.adapter_method, spec.config).tap do |conn|
    conn.schema_cache = schema_cache.dup if schema_cache
  end
end

# File activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb, line 577
def reap
  stale_connections = synchronize do
    @connections.select do |conn|
      conn.in_use? && !conn.owner.alive?
    end.each do |conn|
      conn.steal!
    end
  end

  stale_connections.each do |conn|
    if conn.active?
      conn.reset!
      checkin conn
    else
      remove conn
    end
  end
end

# File activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb, line 399
def release_connection(owner_thread = Thread.current)
  if conn = @thread_cached_conns.delete(connection_cache_key(owner_thread))
    checkin conn
  end
end

# File activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb, line 545
def remove(conn)
  needs_new_connection = false

  synchronize do
    remove_connection_from_thread_cache conn

    @connections.delete conn
    @available.delete conn

    # @available.any_waiting? => true means that prior to removing this
    # conn, the pool was at its max size (@connections.size == @size).
    # This would mean that any threads stuck waiting in the queue wouldn't
    # know they could checkout_new_connection, so let's do it for them.
    # Because condition-wait loop is encapsulated in the Queue class
    # (that in turn is oblivious to ConnectionPool implementation), threads
    # that are "stuck" there are helpless. They have no way of creating
    # new connections and are completely reliant on us feeding available
    # connections into the Queue.
    needs_new_connection = @available.any_waiting?
  end

  # This is intentionally done outside of the synchronized section as we
  # would like not to hold the main mutex while checking out new connections.
  # Thus there is some chance that needs_new_connection information is now
  # stale, we can live with that (bulk_make_new_connections will make
  # sure not to exceed the pool's @size limit).
  bulk_make_new_connections(1) if needs_new_connection
end

# File activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb, line 803
def remove_connection_from_thread_cache(conn, owner_thread = conn.owner)
  @thread_cached_conns.delete_pair(connection_cache_key(owner_thread), conn)
end

# File activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb, line 295
  def run
    return unless frequency && frequency > 0
    Thread.new(frequency, pool) { |t, p|
      loop do
        sleep t
        p.reap
        p.flush
      end
    }
  end
end

# File activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb, line 633
def stat
  synchronize do
    {
      size: size,
      connections: @connections.size,
      busy: @connections.count { |c| c.in_use? && c.owner.alive? },
      dead: @connections.count { |c| c.in_use? && !c.owner.alive? },
      idle: @connections.count { |c| !c.in_use? },
      waiting: num_waiting_in_queue,
      checkout_timeout: checkout_timeout
    }
  end
end

# File activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb, line 819
def try_to_checkout_new_connection
  # first in synchronized section check if establishing new conns is allowed
  # and increment @now_connecting, to prevent overstepping this pool's @size
  # constraint
  do_checkout = synchronize do
    if @threads_blocking_new_connections.zero? && (@connections.size + @now_connecting) < @size
      @now_connecting += 1
    end
  end
  if do_checkout
    begin
      # if successfully incremented @now_connecting establish new connection
      # outside of synchronized section
      conn = checkout_new_connection
    ensure
      synchronize do
        if conn
          adopt_connection(conn)
          # returned conn needs to be already leased
          conn.lease
        end
        @now_connecting -= 1
      end
    end
  end
end

# File activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb, line 409
def with_connection
  unless conn = @thread_cached_conns[connection_cache_key(Thread.current)]
    conn = connection
    fresh_connection = true
  end
  yield conn
ensure
  release_connection if fresh_connection
end

# File activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb, line 673
def with_exclusively_acquired_all_connections(raise_on_acquisition_timeout = true)
  with_new_connections_blocked do
    attempt_to_checkout_all_existing_connections(raise_on_acquisition_timeout)
    yield
  end
end

# File activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb, line 746
def with_new_connections_blocked
  synchronize do
    @threads_blocking_new_connections += 1
  end

  yield
ensure
  num_new_conns_required = 0

  synchronize do
    @threads_blocking_new_connections -= 1

    if @threads_blocking_new_connections.zero?
      @available.clear

      num_new_conns_required = num_waiting_in_queue

      @connections.each do |conn|
        next if conn.in_use?

        @available.add conn
        num_new_conns_required -= 1
      end
    end
  end

  bulk_make_new_connections(num_new_conns_required) if num_new_conns_required > 0
end