Class: ActiveRecord::ConnectionAdapters::ConnectionPool

Inherits:
Object
  • Object
show all
Includes:
MonitorMixin
Defined in:
lib/active_record/connection_adapters/abstract/connection_pool.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: number indicating size of connection pool (default 5)

  • checkout_timeout: number of seconds to block and wait for a connection before giving up and raising a timeout error (default 5 seconds).

  • reaping_frequency: frequency in seconds to periodically run the Reaper, which attempts to find and recover connections from dead threads, which can occur if a programmer forgets to close a connection at the end of a thread or a thread dies unexpectedly. Regardless of this setting, the Reaper will be invoked before every blocking wait. (Default nil, which means don’t schedule the Reaper).

– Synchronization policy:

  • all public methods can be called outside synchronize

  • access to these i-vars 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

Constructor Details

#initialize(spec) ⇒ ConnectionPool

Creates a new ConnectionPool object. spec is a ConnectionSpecification 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.



320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 320

def initialize(spec)
  super()

  @spec = spec

  @checkout_timeout = (spec.config[:checkout_timeout] && spec.config[:checkout_timeout].to_f) || 5
  @reaper = Reaper.new(self, (spec.config[:reaping_frequency] && spec.config[:reaping_frequency].to_f))
  @reaper.run

  # default max pool size to 5
  @size = (spec.config[:pool] && spec.config[:pool].to_i) || 5

  # The cache of threads mapped to reserved connections, the sole purpose
  # of the cache is to speed-up +connection+ method, it is not the authoritative
  # registry of which thread owns which connection, that is tracked by
  # +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 a 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 +@thread_cached_conns+ 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

  # A boolean toggle that allows/disallows new connections.
  @new_cons_enabled = true

  @available = ConnectionLeasingQueue.new self
end

Instance Attribute Details

#automatic_reconnect ⇒ Object

Returns the value of attribute automatic_reconnect.



311
312
313
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 311

def automatic_reconnect
  @automatic_reconnect
end

#checkout_timeout ⇒ Object

Returns the value of attribute checkout_timeout.



311
312
313
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 311

def checkout_timeout
  @checkout_timeout
end

#connections ⇒ Object (readonly)

Returns the value of attribute connections.



312
313
314
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 312

def connections
  @connections
end

#reaper ⇒ Object (readonly)

Returns the value of attribute reaper.



312
313
314
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 312

def reaper
  @reaper
end

#schema_cache ⇒ Object

Returns the value of attribute schema_cache.



311
312
313
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 311

def schema_cache
  @schema_cache
end

#size ⇒ Object (readonly)

Returns the value of attribute size.



312
313
314
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 312

def size
  @size
end

#spec ⇒ Object (readonly)

Returns the value of attribute spec.



312
313
314
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 312

def spec
  @spec
end

Instance Method Details

#active_connection? ⇒ Boolean

Is there an open connection that is 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) 


372
373
374
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 372

def active_connection?
  @thread_cached_conns[connection_cache_key(Thread.current)]
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.



509
510
511
512
513
514
515
516
517
518
519
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 509

def checkin(conn)
  synchronize do
    remove_connection_from_thread_cache conn

    conn._run_checkin_callbacks do
      conn.expire
    end

    @available.add conn
  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.



500
501
502
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 500

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.config[:checkout_timeout] * 2 seconds).



444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 444

def clear_reloadable_connections(raise_on_acquisition_timeout = true)
  num_new_conns_required = 0

  with_exclusively_acquired_all_connections(raise_on_acquisition_timeout) do
    synchronize do
      @connections.each do |conn|
        checkin conn
        conn.disconnect! if conn.requires_reloading?
      end
      @connections.delete_if(&:requires_reloading?)

      @available.clear

      if @connections.size < @size
        # because of the pruning done by this method, we might be running
        # low on connections, while threads stuck in queue are helpless
        # (not being able to establish new connections for themselves),
        # see also more detailed explanation in +remove+
        num_new_conns_required = num_waiting_in_queue - @connections.size
      end

      @connections.each do |conn|
        @available.add conn
      end
    end
  end

  bulk_make_new_connections(num_new_conns_required) if num_new_conns_required > 0
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.config[:checkout_timeout] * 2 seconds), the pool forcefully clears the cache and reloads connections without any regard for other connection owning threads.



482
483
484
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 482

def clear_reloadable_connections!
  clear_reloadable_connections(false)
end

#connected? ⇒ Boolean

Returns true if a connection has already been opened.

Returns:

  • (Boolean) 


404
405
406
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 404

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.



363
364
365
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 363

def connection
  @thread_cached_conns[connection_cache_key(Thread.current)] ||= checkout
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.config[:checkout_timeout] * 2 seconds).



414
415
416
417
418
419
420
421
422
423
424
425
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 414

def disconnect(raise_on_acquisition_timeout = true)
  with_exclusively_acquired_all_connections(raise_on_acquisition_timeout) do
    synchronize do
      @connections.each do |conn|
        checkin conn
        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.config[:checkout_timeout] * 2 seconds), the pool is forcefully disconnected without any regard for other connection owning threads.



433
434
435
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 433

def disconnect!
  disconnect(false)
end

#num_waiting_in_queue ⇒ Object

:nodoc:



574
575
576
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 574

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.



555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 555

def reap
  stale_connections = synchronize do
    @connections.select do |conn|
      conn.in_use? && !conn.owner.alive?
    end
  end

  stale_connections.each do |conn|
    synchronize do
      if conn.active?
        conn.reset!
        checkin conn
      else
        remove conn
      end
    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.



383
384
385
386
387
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 383

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.



523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 523

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

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



393
394
395
396
397
398
399
400
401
 
# File 'lib/active_record/connection_adapters/abstract/connection_pool.rb', line 393

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