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
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
|
require_relative 'base_shared'
module Concurrent::Channel::Buffer
RSpec.describe Unbuffered, edge: true, notravis: true do
subject { described_class.new }
it_behaves_like :channel_buffer
specify do
expect(subject).to be_blocking
end
specify do
expect(subject.capacity).to eq 1
end
context '#size' do
it 'is 0 when first created' do
expect(subject.size).to eq 0
end
it 'is 1 when a putting thread is waiting' do
t = in_thread { subject.put(:foo) }
t.join(0.1)
expect(subject.size).to eq 1
t.kill # cleanup
end
it 'is 0 when there are taking threads but no putting threads' do
t = in_thread { subject.take }
t.join(0.1)
expect(subject.size).to eq 0
t.kill # cleanup
end
end
context '#empty?' do
it 'is true when there are no putting threads' do
expect(subject).to be_empty
end
it 'is false when there are waiting putting threads' do
t = in_thread { subject.put(:foo) }
t.join(0.1)
expect(subject).to_not be_empty
t.kill # cleanup
end
end
context '#full?' do
it 'is false when there are no putting threads' do
expect(subject).to_not be_full
end
it 'is false when there are waiting putting threads' do
t = in_thread { subject.put(:foo) }
t.join(0.1)
expect(subject).to be_full
t.kill # cleanup
end
end
context '#put' do
it 'does not enqueue the item when closed' do
subject.close
subject.put(:foo)
expect(subject).to be_empty
end
it 'returns false when closed' do
subject.close
expect(subject.put(:foo)).to be false
end
it 'blocks until a thread is ready to take' do
subject # initialize on this thread
bucket = Concurrent::AtomicReference.new(nil)
t = in_thread do
subject.put(42)
bucket.value = 42
end
t.join(0.1)
before = bucket.value
subject.take
t.join(0.1)
after = bucket.value
expect(before).to be nil
expect(after).to eq 42
expect(t.status).to be false
end
it 'delivers when closed after put starts' do
t = in_thread do
subject.put(:foo)
end
t.join(0.1)
subject.close
item = subject.take
t.kill #clean up
expect(item).to eq :foo
end
end
context '#offer' do
it 'returns false immediately when a put in in progress' do
subject # initialize on this thread
t = in_thread do
subject.put(:foo) # block the thread
end
t.join(0.1)
ok = subject.offer(:bar)
subject.poll # release the blocked thread
expect(ok).to be false
end
it 'gives the item to a waiting taker and returns true' do
subject # initialize on this thread
bucket = Concurrent::AtomicReference.new(nil)
t = in_thread do
bucket.value = subject.take
end
t.join(0.1)
before = bucket.value
ok = subject.offer(42)
t.join(0.1)
after = bucket.value
expect(ok).to be true
expect(before).to be nil
expect(after).to eq 42
end
end
context '#take' do
it 'returns false immediately when a put in in progress' do
subject # initialize on this thread
t = in_thread do
subject.put(:foo) # block the thread
end
t.join(0.1)
ok = subject.offer(:bar)
subject.poll # release the blocked thread
expect(ok).to be false
end
it 'gives the item to a waiting taker and returns true' do
subject # initialize on this thread
bucket = Concurrent::AtomicReference.new(nil)
t = in_thread do
bucket.value = subject.take
end
t.join(0.1)
before = bucket.value
ok = subject.offer(42)
t.join(0.1)
after = bucket.value
expect(ok).to be true
expect(before).to be nil
expect(after).to eq 42
end
end
context '#next' do
it 'blocks when no putting and returns <item>, true when one arrives' do
subject # initialize on this thread
bucket = Concurrent::AtomicReference.new([])
t = in_thread do
bucket.value = subject.next
end
t.join(0.1)
before = bucket.value
subject.put(42)
t.join(0.1)
after = bucket.value
expect(before).to eq []
expect(after.first).to eq 42
expect(after.last).to be true
expect(t.status).to be false
end
it 'returns <item>, true when there are multiple putting' do
subject # initialize on this thread
threads = 2.times.collect do
in_thread do
subject.put(42)
end
end
threads.each {|t| t.join(0.1)}
item, more = subject.next
subject.poll # clear the channel
expect(item).to eq 42
expect(more).to be true
end
it 'returns <item>, true when closed and last item' do
t = in_thread do
subject.put(:foo)
end
t.join(0.1)
subject.close
item, more = subject.next
t.kill #clean up
expect(item).to eq :foo
expect(more).to be true
end
it 'returns Concurrent::NULL, false when closed and no items remain' do
t = in_thread do
subject.put(:foo)
end
subject.close
subject.next
item, more = subject.next
t.kill #clean up
expect(item).to eq Concurrent::NULL
expect(more).to be false
end
end
end
end
|
