Class: ConnectionPool

Inherits:
Object
  • Object
show all
Includes:
ForkTracker
Defined in:
lib/connection_pool.rb,
lib/connection_pool.rb,
lib/connection_pool/version.rb,
lib/connection_pool/wrapper.rb

Overview

Generic connection pool class for sharing a limited number of objects or network connections among many threads. Note: pool elements are lazily created.

Example usage with block (faster):

@pool = ConnectionPool.new { Redis.new }
@pool.with do |redis|
  redis.lpop('my-list') if redis.llen('my-list') > 0
end

Using optional timeout override (for that single invocation)

@pool.with(timeout: 2.0) do |redis|
  redis.lpop('my-list') if redis.llen('my-list') > 0
end

Example usage replacing an existing connection (slower):

$redis = ConnectionPool.wrap { Redis.new }

def do_work
  $redis.lpop('my-list') if $redis.llen('my-list') > 0
end

Accepts the following options:

  • :size - number of connections to pool, defaults to 5

  • :timeout - amount of time to wait for a connection if none currently available, defaults to 5 seconds

  • :auto_reload_after_fork - automatically drop all connections after fork, defaults to true

Defined Under Namespace

Modules: ForkTracker Classes: Error, PoolShuttingDownError, TimedStack, TimeoutError, Wrapper

Constant Summary collapse

DEFAULTS = 
{size: 5, timeout: 5, auto_reload_after_fork: true}
VERSION = 
"2.4.1"

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Methods included from ForkTracker

#_fork

Constructor Details

#initialize(options = {}, &block) ⇒ ConnectionPool

Returns a new instance of ConnectionPool.

Raises:

  • (ArgumentError) 


90
91
92
93
94
95
96
97
98
99
100
101
102
103
 
# File 'lib/connection_pool.rb', line 90

def initialize(options = {}, &block)
  raise ArgumentError, "Connection pool requires a block" unless block

  options = DEFAULTS.merge(options)

  @size = Integer(options.fetch(:size))
  @timeout = options.fetch(:timeout)
  @auto_reload_after_fork = options.fetch(:auto_reload_after_fork)

  @available = TimedStack.new(@size, &block)
  @key = :"pool-#{@available.object_id}"
  @key_count = :"pool-#{@available.object_id}-count"
  INSTANCES[self] = self if INSTANCES
end

Instance Attribute Details

#auto_reload_after_fork ⇒ Object (readonly)

Automatically drop all connections after fork



166
167
168
 
# File 'lib/connection_pool.rb', line 166

def auto_reload_after_fork
  @auto_reload_after_fork
end

#size ⇒ Object (readonly)

Size of this connection pool



164
165
166
 
# File 'lib/connection_pool.rb', line 164

def size
  @size
end

Class Method Details

.after_fork ⇒ Object 



52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
 
# File 'lib/connection_pool.rb', line 52

def self.after_fork
  INSTANCES.values.each do |pool|
    next unless pool.auto_reload_after_fork

    # We're on after fork, so we know all other threads are dead.
    # All we need to do is to ensure the main thread doesn't have a
    # checked out connection
    pool.checkin(force: true)
    pool.reload do |connection|
      # Unfortunately we don't know what method to call to close the connection,
      # so we try the most common one.
      connection.close if connection.respond_to?(:close)
    end
  end
  nil
end

.wrap(options, &block) ⇒ Object 



44
45
46
 
# File 'lib/connection_pool.rb', line 44

def self.wrap(options, &block)
  Wrapper.new(options, &block)
end

Instance Method Details

#available ⇒ Object

Number of pool entries available for checkout at this instant.



169
170
171
 
# File 'lib/connection_pool.rb', line 169

def available
  @available.length
end

#checkin(force: false) ⇒ Object 



129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
 
# File 'lib/connection_pool.rb', line 129

def checkin(force: false)
  if ::Thread.current[@key]
    if ::Thread.current[@key_count] == 1 || force
      @available.push(::Thread.current[@key])
      ::Thread.current[@key] = nil
      ::Thread.current[@key_count] = nil
    else
      ::Thread.current[@key_count] -= 1
    end
  elsif !force
    raise ConnectionPool::Error, "no connections are checked out"
  end

  nil
end

#checkout(options = {}) ⇒ Object 



119
120
121
122
123
124
125
126
127
 
# File 'lib/connection_pool.rb', line 119

def checkout(options = {})
  if ::Thread.current[@key]
    ::Thread.current[@key_count] += 1
    ::Thread.current[@key]
  else
    ::Thread.current[@key_count] = 1
    ::Thread.current[@key] = @available.pop(options[:timeout] || @timeout)
  end
end

#reload(&block) ⇒ Object

Reloads the ConnectionPool by passing each connection to block and then removing it the pool. Subsequent checkouts will create new connections as needed.



159
160
161
 
# File 'lib/connection_pool.rb', line 159

def reload(&block)
  @available.shutdown(reload: true, &block)
end

#shutdown(&block) ⇒ Object

Shuts down the ConnectionPool by passing each connection to block and then removing it from the pool. Attempting to checkout a connection after shutdown will raise ConnectionPool::PoolShuttingDownError.



150
151
152
 
# File 'lib/connection_pool.rb', line 150

def shutdown(&block)
  @available.shutdown(&block)
end

#with(options = {}) ⇒ Object Also known as: then 



105
106
107
108
109
110
111
112
113
114
115
116
 
# File 'lib/connection_pool.rb', line 105

def with(options = {})
  Thread.handle_interrupt(Exception => :never) do
    conn = checkout(options)
    begin
      Thread.handle_interrupt(Exception => :immediate) do
        yield conn
      end
    ensure
      checkin
    end
  end
end