Class: HotTub::Pool

Inherits:
Object
  • Object
show all
Includes:
KnownClients, Reaper::Mixin
Defined in:
lib/hot_tub/pool.rb

Constant Summary

Constants included from KnownClients

KnownClients::KNOWN_CLIENTS

Instance Attribute Summary collapse

Attributes included from Reaper::Mixin

#reap_timeout, #reaper, #shutdown

Instance Method Summary collapse

Methods included from Reaper::Mixin

#kill_reaper, #spawn_reaper

Methods included from KnownClients

#clean_client, #close_client, #reap_client?

Constructor Details

#initialize(opts = {}, &client_block) ⇒ Pool

Thread-safe lazy connection pool

Example Net::HTTP

pool = HotTub::Pool.new(:size => 10) {
  uri = URI.parse("http://somewebservice.com")
  http = Net::HTTP.new(uri.host, uri.port)
  http.use_ssl = false
  http.start
  http
}
pool.run {|clnt| puts clnt.head('/').code }

Example Redis

# We don't want too many connections so we set our :max_size Under load our pool
# can grow to 30 connections. Once load dies down our pool can be reaped back down to 5
pool = HotTub::Pool.new(:size => 5, :max_size => 30, :reap_timeout => 60) { Redis.new }
pool.set('hot', 'stuff')
pool.get('hot')
# => 'stuff'

HotTub::Pool defaults never_block to true, which means if we run out of clients simply create a new client to continue operations. The pool will grow and extra clients will be reused until activity dies down. If you would like to block and possibly throw an exception rather than temporarily grow the set :size, set :never_block to false; wait_timeout defaults to 10 seconds.

Example with set pool size (will throw HotTub::Pool::Timeout exception)

pool = HotTub::Pool.new(:size => 1, :max_size => 1, :wait_timeout => 0.5) {
  uri = URI.parse("http://someslowwebservice.com")
  http = Net::HTTP.new(uri.host, uri.port)
  http.use_ssl = false
  http.start
  http
}
pool.run { |clnt| s clnt.head('/').code }

begin
  pool.run { |clnt| s clnt.head('/').code }
rescue HotTub::Pool::Timeout => e
  puts "Waited too long for a client: {e}"
end

OPTIONS

:name

A string representing the name of your pool used for logging.

:size

Default is 5. An integer that sets the size of the pool. Could be describe as minimum size the pool should grow to.

:max_size

Default is 0. An integer that represents the maximum number of connections allowed when :non_blocking is true. If set to 0, which is the default, there is no limit; connections will continue to open until load subsides long enough for reaping to occur.

:wait_timeout

Default is 10 seconds. An integer that represents the timeout when waiting for a client from the pool in seconds. After said time a HotTub::Pool::Timeout exception will be thrown

:close

Default is nil. Can be a symbol representing an method to call on a client to close the client or a lambda that accepts the client as a parameter that will close a client. The close option is performed on clients on reaping and shutdown after the client has been removed from the pool. When nil, as is the default, no action is performed.

:clean

Default is nil. Can be a symbol representing an method to call on a client to clean the client or a lambda that accepts the client as a parameter that will clean a client. When nil, as is the default, no action is performed.

:reap?

Default is nil. Can be a symbol representing an method to call on a client that returns a boolean marking a client for reaping, or a lambda that accepts the client as a parameter that returns a boolean boolean marking a client for reaping. When nil, as is the default, no action is performed.

:reaper

If set to false prevents, a HotTub::Reaper from initializing and all reaping will occur when the clients are returned to the pool, blocking the current thread.

:reap_timeout

Default is 600 seconds. An integer that represents the timeout for reaping the pool in seconds.

Raises:

  • (ArgumentError) 


91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
 
# File 'lib/hot_tub/pool.rb', line 91

def initialize(opts={},&client_block)
  raise ArgumentError, 'a block that initializes a new client is required' unless block_given?
  @name             = (opts[:name] || self.class.name)
  @size             = (opts[:size] || 5)            # in seconds
  @wait_timeout     = (opts[:wait_timeout] || 10)   # in seconds
  @reap_timeout     = (opts[:reap_timeout] || 600)  # the interval to reap connections in seconds
  @max_size         = (opts[:max_size] || 0)        # maximum size of pool when non-blocking, 0 means no limit

  @close_client     = opts[:close]                  # => lambda {|clnt| clnt.close} or :close
  @clean_client     = opts[:clean]                  # => lambda {|clnt| clnt.clean} or :clean
  @reap_client      = opts[:reap?]                  # => lambda {|clnt| clnt.reap?} or :reap? # should return boolean
  @client_block     = client_block

  @_pool            = []    # stores available clients
  @_pool.taint
  @_out             = []    # stores all checked out clients
  @_out.taint

  @mutex            = Mutex.new
  @cond             = ConditionVariable.new

  @shutdown         = false
  @blocking_reap    = (opts[:reaper] == false && !opts[:sessions])
  @reaper           = ((opts[:sessions] || (opts[:reaper] == false)) ? false : spawn_reaper)

  @never_block      = (@max_size == 0)

  at_exit {shutdown!} unless opts[:sessions]
end

Instance Attribute Details

#name ⇒ Object

Returns the value of attribute name.



6
7
8
 
# File 'lib/hot_tub/pool.rb', line 6

def name
  @name
end

Instance Method Details

#clean! ⇒ Object

Clean all clients currently checked into the pool. Its possible clients may be returned to the pool after cleaning



132
133
134
135
136
137
138
139
 
# File 'lib/hot_tub/pool.rb', line 132

def clean!
  HotTub.logger.info "[HotTub] Cleaning pool #{@name}!" if HotTub.logger
  @mutex.synchronize do
    @_pool.each do |clnt|
      clean_client(clnt)
    end
  end
end

#current_size ⇒ Object 



221
222
223
224
225
 
# File 'lib/hot_tub/pool.rb', line 221

def current_size
  @mutex.synchronize do
    _total_current_size
  end
end

#drain! ⇒ Object Also known as: close!

Drain the pool of all clients currently checked into the pool. After draining, wake all sleeping threads to allow repopulating the pool or if shutdown allow threads to quickly finish their work Its possible clients may be returned to the pool after cleaning



145
146
147
148
149
150
151
152
153
154
155
156
157
158
 
# File 'lib/hot_tub/pool.rb', line 145

def drain!
  HotTub.logger.info "[HotTub] Draining pool #{@name}!" if HotTub.logger
  @mutex.synchronize do
    begin
      while clnt = @_pool.pop
        close_client(clnt)
      end
    ensure
      @_out.clear
      @_pool.clear
      @cond.broadcast
    end
  end
end

#max_size=(max_size) ⇒ Object

We must reset our @never_block cache when we set max_size after initialization



229
230
231
232
 
# File 'lib/hot_tub/pool.rb', line 229

def max_size=max_size
  @never_block = (max_size == 0)
  @max_size = max_size
end

#reap! ⇒ Object

Remove and close extra clients Releases mutex each iteration because reaping is a low priority action



202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
 
# File 'lib/hot_tub/pool.rb', line 202

def reap!
  HotTub.logger.info "[HotTub] Reaping pool #{@name}!" if HotTub.log_trace?
  reaped = nil
  while !@shutdown
    @mutex.synchronize do
      if _reap?
        reaped = @_pool.shift
      else
        reaped = nil
      end
    end
    if reaped
      close_client(reaped)
    else
      break
    end
  end
end

#reset! ⇒ Object

Reset the pool. or if shutdown allow threads to quickly finish their work Clients from the previous pool will not return to pool.



164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
 
# File 'lib/hot_tub/pool.rb', line 164

def reset!
  HotTub.logger.info "[HotTub] Resetting pool #{@name}!" if HotTub.logger
  @mutex.synchronize do
    begin
      while clnt = @_pool.pop
        close_client(clnt)
      end
    ensure
      @_out.clear
      @_pool.clear
      @cond.broadcast
    end
  end
  nil
end

#run ⇒ Object

Preform an operations with a client/connection. Requires a block that receives the client.



123
124
125
126
127
128
 
# File 'lib/hot_tub/pool.rb', line 123

def run
  clnt = pop
  yield clnt
ensure
  push(clnt)
end

#shutdown! ⇒ Object

Kills the reaper and drains the pool.



181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
 
# File 'lib/hot_tub/pool.rb', line 181

def shutdown!
  HotTub.logger.info "[HotTub] Shutting down pool #{@name}!" if HotTub.logger
  @shutdown = true
  kill_reaper if @reaper
  @mutex.synchronize do
    begin
      while clnt = @_pool.pop
        close_client(clnt)
      end
    ensure
      @_out.clear
      @_pool.clear
      @cond.broadcast
    end
  end
  nil
end