Class: ActiveRecord::ConnectionAdapters::ConnectionPool
Relationships & Source Files | |
Namespace Children | |
Modules:
| |
Classes:
| |
Super Chains via Extension / Inclusion / Inheritance | |
Instance Chain:
self,
AbstractPool ,
QueryCache::ConnectionPoolConfiguration ,
MonitorMixin
|
|
Inherits: | Object |
Defined in: | activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb, activerecord/lib/active_record/connection_adapters/abstract/connection_pool/queue.rb, activerecord/lib/active_record/connection_adapters/abstract/connection_pool/reaper.rb |
Overview
Active Record Connection Pool
Connection pool base class for managing Active Record database connections.
Introduction
A connection pool synchronizes thread access to a limited number of database connections. The basic idea is that each thread checks out a database connection from the pool, uses that connection, and checks the connection back in. ConnectionPool
is completely thread-safe, and will ensure that a connection cannot be used by two threads at the same time, as long as ConnectionPool’s contract is correctly followed. It will also handle cases in which there are more threads than connections: if all connections have been checked out, and a thread tries to checkout a connection anyway, then ConnectionPool
will wait until some other thread has checked in a connection, or the #checkout_timeout has expired.
Obtaining (checking out) a connection
Connections can be obtained and used from a connection pool in several ways:
-
Simply use {ActiveRecord::Base.lease_connection}. When you’re done with the connection(s) and wish it to be returned to the pool, you call ActiveRecord::Base.connection_handler.clear_active_connections!. This is the default behavior for Active Record when used in conjunction with Action Pack’s request handling cycle.
-
Manually check out a connection from the pool with ActiveRecord::Base.connection_pool.checkout. You are responsible for returning this connection to the pool when finished by calling ActiveRecord::Base.connection_pool.checkin(connection).
-
Use {ActiveRecord::Base.connection_pool.with_connection(&block)}, which obtains a connection, yields it as the sole argument to the block, and returns it to the pool after the block completes.
Connections in the pool are actually AbstractAdapter
objects (or objects compatible with AbstractAdapter’s interface).
While a thread has a connection checked out from the pool using one of the above three methods, that connection will automatically be the one used by ::ActiveRecord
queries executing on that thread. It is not required to explicitly pass the checked out connection to Rails models or queries, for example.
Options
There are several connection-pooling-related options that you can add to your database connection configuration:
-
pool
: maximum number of connections the pool may manage (default 5). -
idle_timeout
: number of seconds that a connection will be kept unused in the pool before it is automatically disconnected (default 300 seconds). Set this to zero to keep connections forever. -
#checkout_timeout: number of seconds to wait for a connection to become available before giving up and raising a timeout error (default 5 seconds).
Class Method Summary
- .install_executor_hooks(executor = ActiveSupport::Executor)
-
.new(pool_config) ⇒ ConnectionPool
constructor
Creates a new
ConnectionPool
object.
Instance Attribute Summary
- #async_executor readonly
- #automatic_reconnect rw
- #checkout_timeout rw
-
#connected? ⇒ Boolean
readonly
Returns true if a connection has already been opened.
- #db_config readonly
- #pool_config readonly
- #reaper readonly
- #role readonly
- #schema_reflection rw
- #schema_reflection=(schema_reflection) rw
- #server_version readonly
- #shard readonly
- #size readonly
-
#active_connection
readonly
Internal use only
Alias for #active_connection?.
- #discarded? ⇒ Boolean readonly Internal use only
- #permanent_lease? ⇒ Boolean readonly Internal use only
Instance Method Summary
-
#active_connection? ⇒ Boolean
(also: #active_connection)
readonly
Returns true if there is an open connection being used for the current thread.
-
#checkin(conn)
Check-in a database connection back into the pool, indicating that you no longer need this connection.
-
#checkout(checkout_timeout = @checkout_timeout)
Check-out a database connection from the pool, indicating that you want to use it.
-
#clear_reloadable_connections(raise_on_acquisition_timeout = true)
Clears the cache which maps classes and re-connects connections that require reloading.
-
#clear_reloadable_connections!
Clears the cache which maps classes and re-connects connections that require reloading.
-
#connections
Returns an array containing the connections currently in the pool.
-
#disconnect(raise_on_acquisition_timeout = true)
Disconnects all connections in the pool, and clears the pool.
-
#disconnect!
Disconnects all connections in the pool, and clears the pool.
-
#flush(minimum_idle = @idle_timeout)
Disconnect all connections that have been idle for at least
minimum_idle
seconds. -
#flush!
Disconnect all currently idle connections.
-
#lease_connection
Retrieve the connection associated with the current thread, or call #checkout to obtain one if necessary.
-
#reap
Recover lost connections for the pool.
-
#release_connection(existing_lease = nil)
Signal that the thread is finished with the current connection.
-
#remove(conn)
Remove a connection from the connection pool.
- #schema_cache
-
#stat
Returns the connection pool’s usage statistic.
-
#with_connection(prevent_permanent_checkout: false)
Yields a connection from the connection pool to the block.
-
#acquire_connection(checkout_timeout)
private
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.
- #adopt_connection(conn) private
- #attempt_to_checkout_all_existing_connections(raise_on_acquisition_timeout = true) private
- #build_async_executor private
- #bulk_make_new_connections(num_new_conns_needed) private
- #checkout_and_verify(c) private
- #checkout_for_exclusive_access(checkout_timeout) private
- #checkout_new_connection private
- #connection_lease private
- #new_connection private
-
#release(conn, owner_thread = conn.owner)
private
Alias for #remove_connection_from_thread_cache.
- #remove_connection_from_thread_cache(conn, owner_thread = conn.owner) (also: #release) private
-
#try_to_checkout_new_connection
private
If the pool is not at a
@size
limit, establish new connection. -
#with_exclusively_acquired_all_connections(raise_on_acquisition_timeout = true)
private
Take control of all existing connections so a “group” action such as reload/disconnect can be performed safely.
- #with_new_connections_blocked private
- #connection_class Internal use only
-
#discard!
Internal use only
Discards all connections in the pool (even if they’re currently leased!), along with the pool itself.
- #inspect Internal use only
- #internal_metadata Internal use only
- #migration_context Internal use only
- #migrations_paths Internal use only
- #num_waiting_in_queue Internal use only
- #pin_connection!(lock_thread) Internal use only
- #schedule_query(future_result) Internal use only
- #schema_migration Internal use only
- #unpin_connection! Internal use only
QueryCache::ConnectionPoolConfiguration
- Included
#checkout_and_verify, #clear_query_cache, #dirties_query_cache, | |
#disable_query_cache | Disable the query cache within the block. |
#disable_query_cache!, #enable_query_cache, #enable_query_cache!, #initialize, #query_cache, #query_cache_enabled |
Constructor Details
.new(pool_config) ⇒ ConnectionPool
Creates a new ConnectionPool
object. #pool_config is a PoolConfig
object which describes database connection information (e.g. adapter, host name, username, password, etc), as well as the maximum size for this ConnectionPool
.
The default ConnectionPool
maximum size is 5.
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 226
def initialize(pool_config) super() @pool_config = pool_config @db_config = pool_config.db_config @role = pool_config.role @shard = pool_config.shard @checkout_timeout = db_config.checkout_timeout @idle_timeout = db_config.idle_timeout @size = db_config.pool # This variable tracks the cache of threads mapped to reserved connections, with the # sole purpose of speeding up the connection method. It is not the authoritative # registry of which thread owns which connection. Connection ownership is tracked by # the connection.owner attr on each connection instance. # The invariant works like this: if there is mapping of <tt>thread => conn</tt>, # then that thread does indeed own that conn. However, an absence of such # mapping does not mean that the thread doesn't own the said connection. In # that case conn.owner attr should be consulted. # Access and modification of <tt>@leases</tt> does not require # synchronization. @leases = LeaseRegistry.new @connections = [] @automatic_reconnect = true # Connection pool allows for concurrent (outside the main synchronize section) # establishment of new connections. This variable tracks the number of threads # currently in the process of independently establishing connections to the DB. @now_connecting = 0 @threads_blocking_new_connections = 0 @available = ConnectionLeasingQueue.new self @pinned_connection = nil @pinned_connections_depth = 0 @async_executor = build_async_executor @schema_cache = nil @reaper = Reaper.new(self, db_config.reaping_frequency) @reaper.run end
Class Method Details
.install_executor_hooks(executor = ActiveSupport::Executor)
[ GitHub ]# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 206
def install_executor_hooks(executor = ActiveSupport::Executor) executor.register_hook(ExecutorHooks) end
Instance Attribute Details
#active_connection (readonly)
Alias for #active_connection?.
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 373
alias_method :active_connection, :active_connection? # :nodoc:
#async_executor (readonly)
[ GitHub ]#automatic_reconnect (rw)
[ GitHub ]# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 215
attr_accessor :automatic_reconnect, :checkout_timeout
#checkout_timeout (rw)
[ GitHub ]# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 215
attr_accessor :automatic_reconnect, :checkout_timeout
#connected? ⇒ Boolean
(readonly)
Returns true if a connection has already been opened.
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 421
def connected? synchronize { @connections.any?(&:connected?) } end
#db_config (readonly)
[ GitHub ]# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 216
attr_reader :db_config, :size, :reaper, :pool_config, :async_executor, :role, :shard
#discarded? ⇒ Boolean
(readonly)
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 488
def discarded? # :nodoc: @connections.nil? end
#permanent_lease? ⇒ Boolean
(readonly)
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 315
def permanent_lease? # :nodoc: connection_lease.sticky.nil? end
#pool_config (readonly)
[ GitHub ]#reaper (readonly)
[ GitHub ]# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 216
attr_reader :db_config, :size, :reaper, :pool_config, :async_executor, :role, :shard
#role (readonly)
[ GitHub ]# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 216
attr_reader :db_config, :size, :reaper, :pool_config, :async_executor, :role, :shard
#schema_reflection (rw)
[ GitHub ]# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 218
delegate :schema_reflection, :server_version, to: :pool_config
#schema_reflection=(schema_reflection) (rw)
[ GitHub ]# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 283
def schema_reflection=(schema_reflection) pool_config.schema_reflection = schema_reflection @schema_cache = nil end
#server_version (readonly)
[ GitHub ]# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 218
delegate :schema_reflection, :server_version, to: :pool_config
#shard (readonly)
[ GitHub ]# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 216
attr_reader :db_config, :size, :reaper, :pool_config, :async_executor, :role, :shard
#size (readonly)
[ GitHub ]# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 216
attr_reader :db_config, :size, :reaper, :pool_config, :async_executor, :role, :shard
Instance Method Details
#acquire_connection(checkout_timeout) (private)
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
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 843
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 # Retry after reaping, which may return an available connection, # remove an inactive connection, or both if conn = @available.poll || try_to_checkout_new_connection conn else @available.poll(checkout_timeout) end end rescue ConnectionTimeoutError => ex raise ex.set_pool(self) end
#active_connection? ⇒ Boolean
(readonly)
Also known as: #active_connection
Returns true if there is an open connection being used for the current thread.
This method only works for connections that have been obtained through #lease_connection or #with_connection methods. Connections obtained through #checkout will not be detected by #active_connection?
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 370
def active_connection? connection_lease.connection end
#adopt_connection(conn) (private)
[ GitHub ]# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 913
def adopt_connection(conn) conn.pool = self @connections << conn # We just created the first connection, it's time to load the schema # cache if that wasn't eagerly done before if @schema_cache.nil? && ActiveRecord.lazily_load_schema_cache schema_cache.load! end end
#attempt_to_checkout_all_existing_connections(raise_on_acquisition_timeout = true) (private)
[ GitHub ]# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 734
def attempt_to_checkout_all_existing_connections(raise_on_acquisition_timeout = true) collected_conns = synchronize do reap # No need to wait for dead owners # account for our own connections @connections.select { |conn| conn.owner == ActiveSupport::IsolatedExecutionState.context } end newly_checked_out = [] timeout_time = Process.clock_gettime(Process::CLOCK_MONOTONIC) + (@checkout_timeout * 2) @available.with_a_bias_for(ActiveSupport::IsolatedExecutionState.context) do loop do synchronize do return if collected_conns.size == @connections.size && @now_connecting == 0 remaining_timeout = timeout_time - Process.clock_gettime(Process::CLOCK_MONOTONIC) 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
#build_async_executor (private)
[ GitHub ]# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 695
def build_async_executor case ActiveRecord.async_query_executor when :multi_thread_pool if @db_config.max_threads > 0 Concurrent::ThreadPoolExecutor.new( min_threads: @db_config.min_threads, max_threads: @db_config.max_threads, max_queue: @db_config.max_queue, fallback_policy: :caller_runs ) end when :global_thread_pool ActiveRecord.global_thread_pool_async_query_executor end end
#bulk_make_new_connections(num_new_conns_needed) (private)
[ GitHub ]# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 713
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
#checkin(conn)
Check-in a database connection back into the pool, indicating that you no longer need this connection.
conn
: an AbstractAdapter
object, which was obtained by earlier by calling #checkout on this pool.
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 564
def checkin(conn) return if @pinned_connection.equal?(conn) conn.lock.synchronize do synchronize do connection_lease.clear(conn) conn._run_checkin_callbacks do conn.expire end @available.add conn end end end
#checkout(checkout_timeout = @checkout_timeout)
Check-out a database connection from the pool, indicating that you want to use it. You should call #checkin when you no longer need this.
This is done by either returning and leasing existing connection, or by creating a new connection and leasing it.
If all connections are leased and the pool is at capacity (meaning the number of currently leased connections is greater than or equal to the size limit set), an ::ActiveRecord::ConnectionTimeoutError
exception will be raised.
Returns: an AbstractAdapter
object.
Raises:
-
::ActiveRecord::ConnectionTimeoutError
no connection can be obtained from the pool.
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 541
def checkout(checkout_timeout = @checkout_timeout) if @pinned_connection @pinned_connection.lock.synchronize do synchronize do @pinned_connection.verify! # Any leased connection must be in @connections otherwise # some methods like #connected? won't behave correctly unless @connections.include?(@pinned_connection) @connections << @pinned_connection end end end @pinned_connection else checkout_and_verify(acquire_connection(checkout_timeout)) end end
#checkout_and_verify(c) (private)
[ GitHub ]# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 929
def checkout_and_verify(c) c._run_checkout_callbacks do c.clean! end c rescue Exception remove c c.disconnect! raise end
#checkout_for_exclusive_access(checkout_timeout) (private)
[ GitHub ]# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 783
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" thread_report = [] @connections.each do |conn| unless conn.owner == ActiveSupport::IsolatedExecutionState.context thread_report << "#{conn} is owned by #{conn.owner}" end end msg << " (#{thread_report.join(', ')})" if thread_report.any? raise ExclusiveConnectionTimeoutError.new(msg, connection_pool: self) end
#checkout_new_connection (private)
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 924
def checkout_new_connection raise ConnectionNotEstablished unless @automatic_reconnect new_connection end
#clear_reloadable_connections(raise_on_acquisition_timeout = true)
Clears the cache which maps classes and re-connects connections that require reloading.
Raises:
-
::ActiveRecord::ExclusiveConnectionTimeoutError
if unable to gain ownership of all connections in the pool within a timeout interval (default duration isspec.db_config.checkout_timeout * 2
seconds).
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 499
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
#clear_reloadable_connections!
Clears the cache which maps classes and re-connects connections that require reloading.
The pool first tries to gain ownership of all connections. If unable to do so within a timeout interval (default duration is spec.db_config.checkout_timeout * 2
seconds), then the pool forcefully clears the cache and reloads connections without any regard for other connection owning threads.
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 523
def clear_reloadable_connections! clear_reloadable_connections(false) end
#connection_class
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 361
def connection_class # :nodoc: pool_config.connection_class end
#connection_lease (private)
[ GitHub ]# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 691
def connection_lease @leases[ActiveSupport::IsolatedExecutionState.context] end
#connections
Returns an array containing the connections currently in the pool. Access to the array does not require synchronization on the pool because the array is newly created and not retained by the pool.
However; this method bypasses the ConnectionPool’s thread-safe connection access pattern. A returned connection may be owned by another thread, unowned, or by happen-stance owned by the calling thread.
Calling methods on a connection without ownership is subject to the thread-safety guarantees of the underlying method. Many of the methods on connection adapter classes are inherently multi-thread unsafe.
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 436
def connections synchronize { @connections.dup } end
#discard!
Discards all connections in the pool (even if they’re currently leased!), along with the pool itself. Any further interaction with the pool (except #spec
and #schema_cache) is undefined.
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 478
def discard! # :nodoc: synchronize do return if self.discarded? @connections.each do |conn| conn.discard! end @connections = @available = @leases = nil end end
#disconnect(raise_on_acquisition_timeout = true)
Disconnects all connections in the pool, and clears the pool.
Raises:
-
::ActiveRecord::ExclusiveConnectionTimeoutError
if unable to gain ownership of all connections in the pool within a timeout interval (default duration isspec.db_config.checkout_timeout * 2
seconds).
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 446
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 = [] @leases.clear @available.clear end end end
#disconnect!
Disconnects all connections in the pool, and clears the pool.
The pool first tries to gain ownership of all connections. If unable to do so within a timeout interval (default duration is spec.db_config.checkout_timeout * 2
seconds), then the pool is forcefully disconnected without any regard for other connection owning threads.
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 469
def disconnect! disconnect(false) end
#flush(minimum_idle = @idle_timeout)
Disconnect all connections that have been idle for at least minimum_idle
seconds. Connections currently checked out, or that were checked in less than minimum_idle
seconds ago, are unaffected.
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 637
def flush(minimum_idle = @idle_timeout) return if minimum_idle.nil? idle_connections = synchronize do return if self.discarded? @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
#flush!
Disconnect all currently idle connections. Connections currently checked out are unaffected.
#inspect
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 272
def inspect # :nodoc: name_field = " name=#{db_config.name.inspect}" unless db_config.name == "primary" shard_field = " shard=#{@shard.inspect}" unless @shard == :default "#<#{self.class.name} env_name=#{db_config.env_name.inspect}#{name_field} role=#{role.inspect}#{shard_field}>" end
#internal_metadata
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 300
def # :nodoc: InternalMetadata.new(self) end
#lease_connection
Retrieve the connection associated with the current thread, or call #checkout to obtain one if necessary.
#lease_connection
can be called any number of times; the connection is held in a cache keyed by a thread.
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 309
def lease_connection lease = connection_lease lease.sticky = true lease.connection ||= checkout end
#migration_context
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 288
def migration_context # :nodoc: MigrationContext.new(migrations_paths, schema_migration, ) end
#migrations_paths
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 292
def migrations_paths # :nodoc: db_config.migrations_paths || Migrator.migrations_paths end
#new_connection (private)
[ GitHub ]# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 873
def new_connection connection = db_config.new_connection connection.pool = self connection rescue ConnectionNotEstablished => ex raise ex.set_pool(self) end
#num_waiting_in_queue
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 664
def num_waiting_in_queue # :nodoc: @available.num_waiting end
#pin_connection!(lock_thread)
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 319
def pin_connection!(lock_thread) # :nodoc: @pinned_connection ||= (connection_lease&.connection || checkout) @pinned_connections_depth += 1 # Any leased connection must be in @connections otherwise # some methods like #connected? won't behave correctly unless @connections.include?(@pinned_connection) @connections << @pinned_connection end @pinned_connection.lock_thread = ActiveSupport::IsolatedExecutionState.context if lock_thread @pinned_connection.verify! # eagerly validate the connection @pinned_connection.begin_transaction joinable: false, _lazy: false end
#reap
Recover lost connections for the pool. A lost connection can occur if a programmer forgets to checkin a connection at the end of a thread or a thread dies unexpectedly.
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 614
def reap stale_connections = synchronize do return if self.discarded? @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
#release(conn, owner_thread = conn.owner) (private)
Alias for #remove_connection_from_thread_cache.
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 871
alias_method :release, :remove_connection_from_thread_cache
#release_connection(existing_lease = nil)
Signal that the thread is finished with the current connection. #release_connection
releases the connection-thread association and returns the connection to the pool.
This method only works for connections that have been obtained through #lease_connection or #with_connection methods, connections obtained through #checkout will not be automatically released.
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 382
def release_connection(existing_lease = nil) if conn = connection_lease.release checkin conn return true end false end
#remove(conn)
Remove a connection from the connection pool. The connection will remain open and active but will no longer be managed by this pool.
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 582
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
#remove_connection_from_thread_cache(conn, owner_thread = conn.owner) (private) Also known as: #release
[ GitHub ]# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 868
def remove_connection_from_thread_cache(conn, owner_thread = conn.owner) @leases[owner_thread].clear(conn) end
#schedule_query(future_result)
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 685
def schedule_query(future_result) # :nodoc: @async_executor.post { future_result.execute_or_skip } Thread.pass end
#schema_cache
[ GitHub ]# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 279
def schema_cache @schema_cache ||= BoundSchemaReflection.new(schema_reflection, self) end
#schema_migration
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 296
def schema_migration # :nodoc: SchemaMigration.new(self) end
#stat
Returns the connection pool’s usage statistic.
ActiveRecord::Base.connection_pool.stat # => { size: 15, connections: 1, busy: 1, dead: 0, idle: 0, waiting: 0, checkout_timeout: 5 }
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 671
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
#try_to_checkout_new_connection (private)
If the pool is not at a @size
limit, establish new connection. Connecting to the DB is done outside main synchronized section.
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 886
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
#unpin_connection!
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 334
def unpin_connection! # :nodoc: raise "There isn't a pinned connection #{object_id}" unless @pinned_connection clean = true @pinned_connection.lock.synchronize do @pinned_connections_depth -= 1 connection = @pinned_connection @pinned_connection = nil if @pinned_connections_depth.zero? if connection.transaction_open? connection.rollback_transaction else # Something committed or rolled back the transaction clean = false connection.reset! end if @pinned_connection.nil? connection.steal! connection.lock_thread = nil checkin(connection) end end clean end
#with_connection(prevent_permanent_checkout: false)
Yields a connection from the connection pool to the block. If no connection is already checked out by the current thread, a connection will be checked out from the pool, yielded to the block, and then returned to the pool when the block is finished. If a connection has already been checked out on the current thread, such as via #lease_connection or #with_connection
, that existing connection will be the one yielded and it will not be returned to the pool automatically at the end of the block; it is expected that such an existing connection will be properly returned to the pool by the code that checked it out.
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 399
def with_connection(prevent_permanent_checkout: false) lease = connection_lease sticky_was = lease.sticky lease.sticky = false if prevent_permanent_checkout if lease.connection begin yield lease.connection ensure lease.sticky = sticky_was if prevent_permanent_checkout && !sticky_was end else begin yield lease.connection = checkout ensure lease.sticky = sticky_was if prevent_permanent_checkout && !sticky_was release_connection(lease) unless lease.sticky end end end
#with_exclusively_acquired_all_connections(raise_on_acquisition_timeout = true) (private)
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.
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 727
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
#with_new_connections_blocked (private)
[ GitHub ]# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb', line 803
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