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
|
require "spec/helper/all"
DELAY = 0.25
QUERY = "select sleep(#{DELAY})"
describe EventMachine::Synchrony::ConnectionPool do
xit "should queue requests if pool size is exceeded" do
EventMachine.run do
db = EventMachine::Synchrony::ConnectionPool.new(size: 1) do
Mysql2::EM::Client.new
end
Fiber.new {
start = now
multi = EventMachine::Synchrony::Multi.new
multi.add :a, db.aquery(QUERY)
multi.add :b, db.aquery(QUERY)
res = multi.perform
(now - start.to_f).should be_within(DELAY * 2 * 0.15).of(DELAY * 2)
res.responses[:callback].size.should == 2
res.responses[:errback].size.should == 0
EventMachine.stop
}.resume
end
end
xit "should execute multiple async pool requests within same fiber" do
EventMachine.run do
db = EventMachine::Synchrony::ConnectionPool.new(size: 2) do
Mysql2::EM::Client.new
end
Fiber.new {
start = now
multi = EventMachine::Synchrony::Multi.new
multi.add :a, db.aquery(QUERY)
multi.add :b, db.aquery(QUERY)
res = multi.perform
(now - start.to_f).should be_within(DELAY * 0.15).of(DELAY)
res.responses[:callback].size.should == 2
res.responses[:errback].size.should == 0
EventMachine.stop
}.resume
end
end
xit "should share connection pool between different fibers" do
EventMachine.run do
db = EventMachine::Synchrony::ConnectionPool.new(size: 2) do
Mysql2::EM::Client.new
end
Fiber.new {
start = now
results = []
fiber = Fiber.current
2.times do |n|
Fiber.new {
results.push db.query(QUERY)
fiber.transfer if results.size == 2
}.resume
end
# wait for workers
Fiber.yield
(now - start.to_f).should be_within(DELAY * 0.15).of(DELAY)
results.size.should == 2
EventMachine.stop
}.resume
end
end
xit "should share connection pool between different fibers & async requests" do
EventMachine.run do
db = EventMachine::Synchrony::ConnectionPool.new(size: 5) do
Mysql2::EM::Client.new
end
Fiber.new {
start = now
results = []
fiber = Fiber.current
2.times do |n|
Fiber.new {
multi = EventMachine::Synchrony::Multi.new
multi.add :a, db.aquery(QUERY)
multi.add :b, db.aquery(QUERY)
results.push multi.perform
fiber.transfer if results.size == 3
}.resume
end
Fiber.new {
# execute a synchronous requests
results.push db.query(QUERY)
fiber.transfer if results.size == 3
}.resume
# wait for workers
Fiber.yield
(now - start.to_f).should be_within(DELAY * 0.15).of(DELAY)
results.size.should == 3
EventMachine.stop
}.resume
end
end
describe '#pool_status' do
it 'should return right initial size' do
(1..10).each do |count|
pool = EventMachine::Synchrony::ConnectionPool.new(size: count) { }
status = pool.pool_status
expect(status).to include available: count
expect(status).to include reserved: 0
expect(status).to include pending: 0
end
end
it 'should return up-to-date statusrmation' do
sleep = 0.5
count = 5
EM.run do
pool = EM::Synchrony::ConnectionPool.new(size: count) do
-> { EM::Synchrony.sleep(sleep) }
end
(1..count).each do |used|
Fiber.new { pool.call }.resume
status = pool.pool_status
expect(status).to include available: count - used
expect(status).to include reserved: used
expect(status).to include pending: 0
end
(1..count).each do |used|
Fiber.new { pool.call }.resume
status = pool.pool_status
expect(status).to include available: 0
expect(status).to include reserved: count
expect(status).to include pending: used
end
Fiber.new {
EM::Synchrony.sleep(sleep + 0.1)
expect(pool.pool_status).to include pending: 0
EM::Synchrony.sleep(sleep + 0.1)
status = pool.pool_status
expect(status).to include available: count
expect(status).to include reserved: 0
expect(status).to include pending: 0
EM.stop
}.resume
end
end
end
end
|
