Class: ActiveRecord::ConnectionAdapters::ConnectionPool

Inherits:
Object
  • Object
show all
Defined in:
lib/activerecord_threadsafe_fix_3_0.rb,
lib/activerecord_threadsafe_fix_3_1.rb

Overview

Monkey Patch time!

Instance Attribute Summary collapse

Class Method 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.



78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
 
# File 'lib/activerecord_threadsafe_fix_3_0.rb', line 78

def initialize(spec)
  @spec = spec

  # The cache of reserved connections mapped to threads
  @reserved_connections = {}

  # The mutex used to synchronize pool access
  @lock = Monitor.new
  # @queue = @lock.new_cond
  @queue = Queue.new(@lock)
  

  # default 5 second timeout unless on ruby 1.9
  @timeout = spec.config[:wait_timeout] || 5

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

  @connections = []
end

Instance Attribute Details

#connections ⇒ Object (readonly)

Returns the value of attribute connections.



70
71
72
 
# File 'lib/activerecord_threadsafe_fix_3_0.rb', line 70

def connections
  @connections
end

#spec ⇒ Object (readonly)

Returns the value of attribute spec.



70
71
72
 
# File 'lib/activerecord_threadsafe_fix_3_0.rb', line 70

def spec
  @spec
end

Class Method Details

.make_synchronized!(*syms) ⇒ Object 



23
24
25
26
27
28
29
30
31
32
33
 
# File 'lib/activerecord_threadsafe_fix_3_1.rb', line 23

def make_synchronized!(*syms)
  syms.each do |sym|
    orig = "__orig_#{sym}__".to_sym
    alias_method orig, sym
    define_method(sym) do |*args, &block|
      synchronize do
        send(orig, *args, &block)
      end
    end
  end
end

Instance Method Details

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



227
228
229
230
231
232
233
 
# File 'lib/activerecord_threadsafe_fix_3_0.rb', line 227

def checkin(conn)
  synchronize do
    conn.send(:_run_checkin_callbacks) do
      @queue.add conn
    end
  end
end

#checkout ⇒ 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 an existing connection, or by creating a new connection. If the maximum number of connections for this pool has already been reached, but the pool is empty (i.e. they’re all being used), then this method will wait until a thread has checked in a connection. The wait time is bounded however: if no connection can be checked out within the timeout specified for this pool, then a ConnectionTimeoutError exception will be raised.

Returns: an AbstractAdapter object.

Raises:

  • ConnectionTimeoutError: no connection can be obtained from the pool within the timeout period.



211
212
213
214
215
216
217
218
219
220
 
# File 'lib/activerecord_threadsafe_fix_3_0.rb', line 211

def checkout
  if conn = @queue.poll
    checkout_and_verify(conn)
  elsif @connections.size < @size
    checkout_new_connection
  else
    # clear_stale_cached_connections!
    checkout_and_verify(@queue.poll(@timeout))
  end
end

#clear_reloadable_connections! ⇒ Object

Clears the cache which maps classes



155
156
157
158
159
160
161
162
163
164
165
166
167
168
 
# File 'lib/activerecord_threadsafe_fix_3_0.rb', line 155

def clear_reloadable_connections!
  synchronize do
    @reserved_connections.each do |name, conn|
      checkin conn
    end
    @reserved_connections = {}
    @connections.each do |conn|
      conn.disconnect! if conn.requires_reloading?
    end
    @connections.delete_if do |conn|
      conn.requires_reloading?
    end
  end
end

#clear_stale_cached_connections! ⇒ Object

Return any checked-out connections back to the pool by threads that are no longer alive.



183
184
185
186
187
188
189
190
191
192
193
 
# File 'lib/activerecord_threadsafe_fix_3_0.rb', line 183

def clear_stale_cached_connections!
  synchronize do
    keys = @reserved_connections.keys - Thread.list.find_all { |t|
      t.alive?
    }.map { |thread| thread.object_id }
    keys.each do |key|
      checkin @reserved_connections[key]
      @reserved_connections.delete(key)
    end
  end
end

#connected? ⇒ Boolean

Returns true if a connection has already been opened.

Returns:

  • (Boolean) 


134
135
136
137
138
 
# File 'lib/activerecord_threadsafe_fix_3_0.rb', line 134

def connected?
  synchronize do
    !@connections.empty?
  end
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 hash keyed by the thread id.



104
105
106
107
108
 
# File 'lib/activerecord_threadsafe_fix_3_0.rb', line 104

def connection
  synchronize do
    @reserved_connections[current_connection_id] ||= checkout
  end
end

#disconnect! ⇒ Object

Disconnects all connections in the pool, and clears the pool.



141
142
143
144
145
146
147
148
149
150
151
152
 
# File 'lib/activerecord_threadsafe_fix_3_0.rb', line 141

def disconnect!
  synchronize do
    @reserved_connections.each do |name,conn|
      checkin conn
    end
    @reserved_connections = {}
    @connections.each do |conn|
      conn.disconnect!
    end
    @connections = []
  end
end

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

Returns:



20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
 
# File 'lib/activerecord_threadsafe_fix_3_1.rb', line 20

def initialize(spec)
  @spec = spec

  # The cache of reserved connections mapped to threads
  @reserved_connections = {}

  # The mutex used to synchronize pool access
  @lock = Monitor.new
  # @queue = @lock.new_cond
  @queue = Queue.new(@lock)
  

  # default 5 second timeout unless on ruby 1.9
  @timeout = spec.config[:wait_timeout] || 5

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

  @connections = []
end

#release_connection(with_id = current_connection_id) ⇒ 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.



113
114
115
116
117
118
 
# File 'lib/activerecord_threadsafe_fix_3_0.rb', line 113

def release_connection(with_id = current_connection_id)
  synchronize do
    conn = @reserved_connections.delete(with_id)
    checkin conn if conn
  end
end

#verify_active_connections! ⇒ Object

Verify active connections and remove and disconnect connections associated with stale threads.



172
173
174
175
176
177
178
179
 
# File 'lib/activerecord_threadsafe_fix_3_0.rb', line 172

def verify_active_connections! #:nodoc:
  synchronize do
    clear_stale_cached_connections!
    @connections.each do |connection|
      connection.verify!
    end
  end
end

#with_connection ⇒ Object

If a connection already exists yield it to the block. If no connection exists checkout a connection, yield it to the block, and checkin the connection when finished.



123
124
125
126
127
128
129
130
131
 
# File 'lib/activerecord_threadsafe_fix_3_0.rb', line 123

def with_connection
  connection_id = current_connection_id
  fresh_connection = synchronize do
    true unless @reserved_connections[connection_id]
  end
  yield connection
ensure
  release_connection(connection_id) if fresh_connection
end