1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
|
# frozen_string_literal: true
require_relative './connection_pool/shared_connection_pool'
class RequestPool
def self.current
@current ||= RequestPool.new
end
class Reaper
attr_reader :pool, :frequency
def initialize(pool, frequency)
@pool = pool
@frequency = frequency
end
def run
return unless frequency&.positive?
Thread.new(frequency, pool) do |t, p|
loop do
sleep t
p.flush
end
end
end
end
MAX_IDLE_TIME = 30
WAIT_TIMEOUT = 5
MAX_POOL_SIZE = ENV.fetch('MAX_REQUEST_POOL_SIZE', 512).to_i
class Connection
attr_reader :site, :last_used_at, :created_at, :in_use, :dead, :fresh
def initialize(site)
@site = site
@http_client = http_client
@last_used_at = nil
@created_at = current_time
@dead = false
@fresh = true
end
def use
@last_used_at = current_time
@in_use = true
retries = 0
begin
yield @http_client
rescue HTTP::ConnectionError
# It's possible the connection was closed, so let's
# try re-opening it once
close
if @fresh || retries.positive?
raise
else
@http_client = http_client
retries += 1
retry
end
rescue StandardError
# If this connection raises errors of any kind, it's
# better if it gets reaped as soon as possible
close
@dead = true
raise
end
ensure
@fresh = false
@in_use = false
end
def seconds_idle
current_time - (@last_used_at || @created_at)
end
def close
@http_client.close
end
private
def http_client
Request.http_client.persistent(@site, timeout: MAX_IDLE_TIME)
end
def current_time
Process.clock_gettime(Process::CLOCK_MONOTONIC)
end
end
def initialize
@pool = ConnectionPool::SharedConnectionPool.new(size: MAX_POOL_SIZE, timeout: WAIT_TIMEOUT) { |site| Connection.new(site) }
@reaper = Reaper.new(self, 30)
@reaper.run
end
def with(site, &block)
@pool.with(site) do |connection|
ActiveSupport::Notifications.instrument('with.request_pool', miss: connection.fresh, host: connection.site) do
connection.use(&block)
end
end
end
delegate :size, :flush, to: :@pool
end
|
