0d9ffe56fb
* Add request pool to improve delivery performance Fix #7909 * Ensure connection is closed when exception interrupts execution * Remove Timeout#timeout from socket connection * Fix infinite retrial loop on HTTP::ConnectionError * Close sockets on failure, reduce idle time to 90 seconds * Add MAX_REQUEST_POOL_SIZE option to limit concurrent connections to the same server * Use a shared pool size, 512 by default, to stay below open file limit * Add some tests * Add more tests * Reduce MAX_IDLE_TIME from 90 to 30 seconds, reap every 30 seconds * Use a shared pool that returns preferred connection but re-purposes other ones when needed * Fix wrong connection being returned on subsequent calls within the same thread * Reduce mutex calls on flushes from 2 to 1 and add test for reaping
63 Zeilen
1,5 KiB
Ruby
63 Zeilen
1,5 KiB
Ruby
# frozen_string_literal: true
|
|
|
|
require 'connection_pool'
|
|
require_relative './shared_timed_stack'
|
|
|
|
class ConnectionPool::SharedConnectionPool < ConnectionPool
|
|
def initialize(options = {}, &block)
|
|
super(options, &block)
|
|
|
|
@available = ConnectionPool::SharedTimedStack.new(@size, &block)
|
|
end
|
|
|
|
delegate :size, :flush, to: :@available
|
|
|
|
def with(preferred_tag, options = {})
|
|
Thread.handle_interrupt(Exception => :never) do
|
|
conn = checkout(preferred_tag, options)
|
|
|
|
begin
|
|
Thread.handle_interrupt(Exception => :immediate) do
|
|
yield conn
|
|
end
|
|
ensure
|
|
checkin(preferred_tag)
|
|
end
|
|
end
|
|
end
|
|
|
|
def checkout(preferred_tag, options = {})
|
|
if ::Thread.current[key(preferred_tag)]
|
|
::Thread.current[key_count(preferred_tag)] += 1
|
|
::Thread.current[key(preferred_tag)]
|
|
else
|
|
::Thread.current[key_count(preferred_tag)] = 1
|
|
::Thread.current[key(preferred_tag)] = @available.pop(preferred_tag, options[:timeout] || @timeout)
|
|
end
|
|
end
|
|
|
|
def checkin(preferred_tag)
|
|
if ::Thread.current[key(preferred_tag)]
|
|
if ::Thread.current[key_count(preferred_tag)] == 1
|
|
@available.push(::Thread.current[key(preferred_tag)])
|
|
::Thread.current[key(preferred_tag)] = nil
|
|
else
|
|
::Thread.current[key_count(preferred_tag)] -= 1
|
|
end
|
|
else
|
|
raise ConnectionPool::Error, 'no connections are checked out'
|
|
end
|
|
|
|
nil
|
|
end
|
|
|
|
private
|
|
|
|
def key(tag)
|
|
:"#{@key}-#{tag}"
|
|
end
|
|
|
|
def key_count(tag)
|
|
:"#{@key_count}-#{tag}"
|
|
end
|
|
end
|