Class: ActiveRecord::ConnectionAdapters::ConnectionPool

Inherits:
Object
  • Object
show all
Includes:
AbstractPool, QueryCache::ConnectionPoolConfiguration, MonitorMixin
Defined in:
lib/active_record/connection_adapters/abstract/connection_pool.rb,
lib/active_record/connection_adapters/abstract/connection_pool/queue.rb,
lib/active_record/connection_adapters/abstract/connection_pool/reaper.rb

Overview

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.

Obtaining (checking out) a connection

Connections can be obtained and used from a connection pool in several ways:

  1. Simply use ActiveRecord::Base.connection as with Active Record 2.1 and earlier (pre-connection-pooling). Eventually, when you're done with the connection(s) and wish it to be returned to the pool, you call ActiveRecord::Base.clear_active_connections!. This will be the default behavior for Active Record when used in conjunction with Action Pack's request handling cycle.

  2. 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).

  3. 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).

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).

– Synchronization policy:

  • all public methods can be called outside synchronize

  • access to these instance variables needs to be in synchronize:

    • @connections

    • @now_connecting

  • private methods that require being called in a synchronize blocks are now explicitly documented

Defined Under Namespace

Modules: BiasableQueue Classes: ConnectionLeasingQueue, Queue, Reaper

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from AbstractPool

#get_schema_cache, #lazily_set_schema_cache, #set_schema_cache

Methods included from QueryCache::ConnectionPoolConfiguration

#disable_query_cache!, #enable_query_cache!, #query_cache_enabled

Constructor Details

#initialize(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.



120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 120

def initialize(pool_config)
  super()

  @pool_config = pool_config
  @db_config = pool_config.db_config
  @connection_class = pool_config.connection_class
  @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>@thread_cached_conns</tt> does not require
  # synchronization.
  @thread_cached_conns = Concurrent::Map.new(initial_capacity: @size)

  @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

  @lock_thread = false

  @async_executor = build_async_executor

  lazily_set_schema_cache

  @reaper = Reaper.new(self, db_config.reaping_frequency)
  @reaper.run
end

Instance Attribute Details

#async_executor ⇒ Object (readonly)

Returns the value of attribute async_executor.



108
109
110
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 108

def async_executor
  @async_executor
end

#automatic_reconnect ⇒ Object

Returns the value of attribute automatic_reconnect.



107
108
109
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 107

def automatic_reconnect
  @automatic_reconnect
end

#checkout_timeout ⇒ Object

Returns the value of attribute checkout_timeout.



107
108
109
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 107

def checkout_timeout
  @checkout_timeout
end

#connection_class ⇒ Object (readonly) Also known as: connection_klass

Returns the value of attribute connection_class.



108
109
110
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 108

def connection_class
  @connection_class
end

#db_config ⇒ Object (readonly)

Returns the value of attribute db_config.



108
109
110
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 108

def db_config
  @db_config
end

#pool_config ⇒ Object (readonly)

Returns the value of attribute pool_config.



108
109
110
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 108

def pool_config
  @pool_config
end

#reaper ⇒ Object (readonly)

Returns the value of attribute reaper.



108
109
110
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 108

def reaper
  @reaper
end

#role ⇒ Object (readonly)

Returns the value of attribute role.



108
109
110
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 108

def role
  @role
end

#shard ⇒ Object (readonly)

Returns the value of attribute shard.



108
109
110
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 108

def shard
  @shard
end

#size ⇒ Object (readonly)

Returns the value of attribute size.



108
109
110
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 108

def size
  @size
end

Instance Method Details

#active_connection? ⇒ Boolean

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 #connection or #with_connection methods. Connections obtained through #checkout will not be detected by #active_connection?

Returns:

  • (Boolean) 


189
190
191
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 189

def active_connection?
  @thread_cached_conns[connection_cache_key(current_thread)]
end

#checkin(conn) ⇒ Object

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.



349
350
351
352
353
354
355
356
357
358
359
360
361
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 349

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

#checkout(checkout_timeout = @checkout_timeout) ⇒ Object

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.



340
341
342
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 340

def checkout(checkout_timeout = @checkout_timeout)
  checkout_and_verify(acquire_connection(checkout_timeout))
end

#clear_reloadable_connections(raise_on_acquisition_timeout = true) ⇒ Object

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 is spec.db_config.checkout_timeout * 2 seconds).



298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 298

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! ⇒ Object

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.



322
323
324
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 322

def clear_reloadable_connections!
  clear_reloadable_connections(false)
end

#connected? ⇒ Boolean

Returns true if a connection has already been opened.

Returns:

  • (Boolean) 


221
222
223
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 221

def connected?
  synchronize { @connections.any? }
end

#connection ⇒ Object

Retrieve the connection associated with the current thread, or call #checkout to obtain one if necessary.

#connection can be called any number of times; the connection is held in a cache keyed by a thread.



180
181
182
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 180

def connection
  @thread_cached_conns[connection_cache_key(current_thread)] ||= checkout
end

#connections ⇒ Object

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.



236
237
238
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 236

def connections
  synchronize { @connections.dup }
end

#discard! ⇒ Object

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.

See AbstractAdapter#discard!



277
278
279
280
281
282
283
284
285
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 277

def discard! # :nodoc:
  synchronize do
    return if self.discarded?
    @connections.each do |conn|
      conn.discard!
    end
    @connections = @available = @thread_cached_conns = nil
  end
end

#discarded? ⇒ Boolean

:nodoc:

Returns:

  • (Boolean) 


287
288
289
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 287

def discarded? # :nodoc:
  @connections.nil?
end

#disconnect(raise_on_acquisition_timeout = true) ⇒ Object

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 is spec.db_config.checkout_timeout * 2 seconds).



246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 246

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

#disconnect! ⇒ Object

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.



268
269
270
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 268

def disconnect!
  disconnect(false)
end

#flush(minimum_idle = @idle_timeout) ⇒ Object

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.



420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 420

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! ⇒ Object

Disconnect all currently idle connections. Connections currently checked out are unaffected.



442
443
444
445
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 442

def flush!
  reap
  flush(-1)
end

#lock_thread=(lock_thread) ⇒ Object 



167
168
169
170
171
172
173
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 167

def lock_thread=(lock_thread)
  if lock_thread
    @lock_thread = Thread.current
  else
    @lock_thread = nil
  end
end

#num_waiting_in_queue ⇒ Object

:nodoc:



447
448
449
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 447

def num_waiting_in_queue # :nodoc:
  @available.num_waiting
end

#reap ⇒ Object

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.



397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 397

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_connection(owner_thread = Thread.current) ⇒ Object

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 #connection or #with_connection methods, connections obtained through #checkout will not be automatically released.



200
201
202
203
204
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 200

def release_connection(owner_thread = Thread.current)
  if conn = @thread_cached_conns.delete(connection_cache_key(owner_thread))
    checkin conn
  end
end

#remove(conn) ⇒ Object

Remove a connection from the connection pool. The connection will remain open and active but will no longer be managed by this pool.



365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 365

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

#schedule_query(future_result) ⇒ Object

:nodoc:



469
470
471
472
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 469

def schedule_query(future_result) # :nodoc:
  @async_executor.post { future_result.execute_or_skip }
  Thread.pass
end

#stat ⇒ Object

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 }



455
456
457
458
459
460
461
462
463
464
465
466
467
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 455

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

#with_connection ⇒ Object

If a connection obtained through #connection or #with_connection methods already exists yield it to the block. If no such connection exists checkout a connection, yield it to the block, and checkin the connection when finished.



210
211
212
213
214
215
216
217
218
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 210

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