e782fcef4f
* 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
95 lines
2.1 KiB
Ruby
95 lines
2.1 KiB
Ruby
# frozen_string_literal: true
|
|
|
|
class ConnectionPool::SharedTimedStack
|
|
def initialize(max = 0, &block)
|
|
@create_block = block
|
|
@max = max
|
|
@created = 0
|
|
@queue = []
|
|
@tagged_queue = Hash.new { |hash, key| hash[key] = [] }
|
|
@mutex = Mutex.new
|
|
@resource = ConditionVariable.new
|
|
end
|
|
|
|
def push(connection)
|
|
@mutex.synchronize do
|
|
store_connection(connection)
|
|
@resource.broadcast
|
|
end
|
|
end
|
|
|
|
alias << push
|
|
|
|
def pop(preferred_tag, timeout = 5.0)
|
|
deadline = current_time + timeout
|
|
|
|
@mutex.synchronize do
|
|
loop do
|
|
return fetch_preferred_connection(preferred_tag) unless @tagged_queue[preferred_tag].empty?
|
|
|
|
connection = try_create(preferred_tag)
|
|
return connection if connection
|
|
|
|
to_wait = deadline - current_time
|
|
raise Timeout::Error, "Waited #{timeout} sec" if to_wait <= 0
|
|
|
|
@resource.wait(@mutex, to_wait)
|
|
end
|
|
end
|
|
end
|
|
|
|
def empty?
|
|
size.zero?
|
|
end
|
|
|
|
def size
|
|
@mutex.synchronize do
|
|
@queue.size
|
|
end
|
|
end
|
|
|
|
def flush
|
|
@mutex.synchronize do
|
|
@queue.delete_if do |connection|
|
|
delete = !connection.in_use && (connection.dead || connection.seconds_idle >= RequestPool::MAX_IDLE_TIME)
|
|
|
|
if delete
|
|
@tagged_queue[connection.site].delete(connection)
|
|
connection.close
|
|
@created -= 1
|
|
end
|
|
|
|
delete
|
|
end
|
|
end
|
|
end
|
|
|
|
private
|
|
|
|
def try_create(preferred_tag)
|
|
if @created == @max && !@queue.empty?
|
|
throw_away_connection = @queue.pop
|
|
@tagged_queue[throw_away_connection.site].delete(throw_away_connection)
|
|
@create_block.call(preferred_tag)
|
|
elsif @created != @max
|
|
connection = @create_block.call(preferred_tag)
|
|
@created += 1
|
|
connection
|
|
end
|
|
end
|
|
|
|
def fetch_preferred_connection(preferred_tag)
|
|
connection = @tagged_queue[preferred_tag].pop
|
|
@queue.delete(connection)
|
|
connection
|
|
end
|
|
|
|
def current_time
|
|
Process.clock_gettime(Process::CLOCK_MONOTONIC)
|
|
end
|
|
|
|
def store_connection(connection)
|
|
@tagged_queue[connection.site].push(connection)
|
|
@queue.push(connection)
|
|
end
|
|
end
|