blob: 27a42c82c1f56f98e684370a6403bce16d4a3d96 (
plain)
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
|
# Copyright 2015 gRPC authors.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
require 'grpc'
Thread.abort_on_exception = true
describe GRPC::Pool do
Pool = GRPC::Pool
describe '#new' do
it 'raises if a non-positive size is used' do
expect { Pool.new(0) }.to raise_error
expect { Pool.new(-1) }.to raise_error
expect { Pool.new(Object.new) }.to raise_error
end
it 'is constructed OK with a positive size' do
expect { Pool.new(1) }.not_to raise_error
end
end
describe '#ready_for_work?' do
it 'before start it is not ready' do
p = Pool.new(1)
expect(p.ready_for_work?).to be(false)
end
it 'it stops being ready after all workers are busy' do
p = Pool.new(5)
p.start
wait_mu = Mutex.new
wait_cv = ConditionVariable.new
wait = true
job = proc do
wait_mu.synchronize do
wait_cv.wait(wait_mu) while wait
end
end
5.times do
expect(p.ready_for_work?).to be(true)
p.schedule(&job)
end
expect(p.ready_for_work?).to be(false)
wait_mu.synchronize do
wait = false
wait_cv.broadcast
end
end
end
describe '#schedule' do
it 'return if the pool is already stopped' do
p = Pool.new(1)
p.stop
job = proc {}
expect { p.schedule(&job) }.to_not raise_error
end
it 'adds jobs that get run by the pool' do
p = Pool.new(1)
p.start
o, q = Object.new, Queue.new
job = proc { q.push(o) }
p.schedule(&job)
expect(q.pop).to be(o)
p.stop
end
end
describe '#stop' do
it 'works when there are no scheduled tasks' do
p = Pool.new(1)
expect { p.stop }.not_to raise_error
end
it 'stops jobs when there are long running jobs' do
p = Pool.new(1)
p.start
wait_forever_mu = Mutex.new
wait_forever_cv = ConditionVariable.new
wait_forever = true
job_running = Queue.new
job = proc do
job_running.push(Object.new)
wait_forever_mu.synchronize do
wait_forever_cv.wait while wait_forever
end
end
p.schedule(&job)
job_running.pop
expect { p.stop }.not_to raise_error
end
end
describe '#start' do
it 'runs jobs as they are scheduled' do
p = Pool.new(5)
o, q = Object.new, Queue.new
p.start
n = 5 # arbitrary
n.times do
p.schedule(o, &q.method(:push))
expect(q.pop).to be(o)
end
p.stop
end
end
end
|
