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
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
|
# frozen_string_literal: true
require "cases/helper"
module ActiveRecord
module ConnectionAdapters
class ReaperTest < ActiveRecord::TestCase
class FakePool
attr_reader :reaped
attr_reader :flushed
def initialize(discarded: false)
@reaped = false
@discarded = discarded
end
def reap
@reaped = true
end
def flush
@flushed = true
end
def discard!
@discarded = true
end
def discarded?
@discarded
end
end
# A reaper with nil time should never reap connections
def test_nil_time
fp = FakePool.new
assert_not fp.reaped
reaper = ConnectionPool::Reaper.new(fp, nil)
reaper.run
assert_not fp.reaped
ensure
fp.discard!
end
def test_some_time
fp = FakePool.new
assert_not fp.reaped
reaper = ConnectionPool::Reaper.new(fp, 0.0001)
reaper.run
until fp.flushed
Thread.pass
end
assert fp.reaped
assert fp.flushed
ensure
fp.discard!
end
def test_pool_has_reaper
config = ActiveRecord::Base.configurations.configs_for(env_name: "arunit", name: "primary")
pool_config = PoolConfig.new(ActiveRecord::Base, config, :writing, :default)
pool = ConnectionPool.new(pool_config)
assert pool.reaper
ensure
pool.discard!
end
def test_reaping_frequency_configuration
pool_config = duplicated_pool_config(reaping_frequency: "10.01")
pool = ConnectionPool.new(pool_config)
assert_equal 10.01, pool.reaper.frequency
ensure
pool.discard!
end
def test_connection_pool_starts_reaper
pool_config = duplicated_pool_config(reaping_frequency: "0.0001")
pool = ConnectionPool.new(pool_config)
conn, child = new_conn_in_thread(pool)
assert_predicate conn, :in_use?
child.terminate
wait_for_conn_idle(conn)
assert_not_predicate conn, :in_use?
ensure
pool.discard!
end
def test_reaper_works_after_pool_discard
pool_config = duplicated_pool_config(reaping_frequency: "0.0001")
2.times do
pool = ConnectionPool.new(pool_config)
conn, child = new_conn_in_thread(pool)
assert_predicate conn, :in_use?
child.terminate
wait_for_conn_idle(conn)
assert_not_predicate conn, :in_use?
pool.discard!
end
end
# This doesn't test the reaper directly, but we want to test the action
# it would take on a discarded pool
def test_reap_flush_on_discarded_pool
pool_config = duplicated_pool_config
pool = ConnectionPool.new(pool_config)
pool.discard!
assert_nothing_raised do
pool.reap
pool.flush
end
end
if Process.respond_to?(:fork)
def test_connection_pool_starts_reaper_in_fork
pool_config = duplicated_pool_config(reaping_frequency: "0.0001")
pool = ConnectionPool.new(pool_config)
pool.checkout
# We currently have a bug somewhere which leads for this test case to be deadlocked
# and timeout after 30 minutes on the CI. Until that bug is fixed, this test is made
# to timeout after a short period of time to reduce the damage.
reader, writer = IO.pipe
pid = fork do
reader.close
pool = ConnectionPool.new(pool_config)
conn, child = new_conn_in_thread(pool)
child.terminate
wait_for_conn_idle(conn)
writer.close
if conn.in_use?
exit!(1)
else
exit!(0)
end
end
writer.close
completed = reader.wait_readable(20)
reader.close
unless completed
Process.kill("ABRT", pid)
end
_, status = Process.wait2(pid)
assert_predicate status, :success?
ensure
pool.discard!
end
end
def test_reaper_does_not_reap_discarded_connection_pools
discarded_pool = FakePool.new(discarded: true)
pool = FakePool.new
frequency = 0.001
ConnectionPool::Reaper.new(discarded_pool, frequency).run
ConnectionPool::Reaper.new(pool, frequency).run
until pool.flushed
Thread.pass
end
assert_not discarded_pool.reaped
assert pool.reaped
ensure
pool.discard!
end
private
def duplicated_pool_config(merge_config_options = {})
old_config = ActiveRecord::Base.connection_pool.db_config.configuration_hash.merge(merge_config_options)
db_config = ActiveRecord::DatabaseConfigurations::HashConfig.new("arunit", "primary", old_config.dup)
PoolConfig.new(ActiveRecord::Base, db_config, :writing, :default)
end
def new_conn_in_thread(pool)
event = Concurrent::Event.new
conn = nil
child = Thread.new do
conn = pool.checkout
conn.query("SELECT 1") # ensure connected
event.set
Thread.stop
end
event.wait
[conn, child]
end
def wait_for_conn_idle(conn, timeout = 5)
start = Process.clock_gettime(Process::CLOCK_MONOTONIC)
while conn.in_use? && Process.clock_gettime(Process::CLOCK_MONOTONIC) - start < timeout
Thread.pass
end
end
end
end
end
|
