Skip to content

Commit 3adcaa0

Browse files
committed
implemented remove_probe
1 parent 46531d7 commit 3adcaa0

File tree

2 files changed

+32
-7
lines changed

2 files changed

+32
-7
lines changed

lib/concurrent/channel/unbuffered_channel.rb

Lines changed: 9 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,11 @@ def initialize
55
@mutex = Mutex.new
66
@condition = Condition.new
77

8-
@wait_set = []
8+
@probe_set = []
9+
end
10+
11+
def probe_set_size
12+
@mutex.synchronize { @probe_set.size }
913
end
1014

1115
def push(value)
@@ -21,19 +25,20 @@ def pop
2125

2226
def select(probe)
2327
@mutex.synchronize do
24-
@wait_set << probe
28+
@probe_set << probe
2529
@condition.signal
2630
end
2731
end
2832

2933
def remove_probe(probe)
34+
@mutex.synchronize { @probe_set.delete(probe) }
3035
end
3136

3237
private
3338
def first_waiting_probe
3439
@mutex.synchronize do
35-
@condition.wait(@mutex) while @wait_set.empty?
36-
@wait_set.shift
40+
@condition.wait(@mutex) while @probe_set.empty?
41+
@probe_set.shift
3742
end
3843
end
3944

spec/concurrent/channel/unbuffered_channel_spec.rb

Lines changed: 23 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,8 @@ module Concurrent
44

55
describe UnbufferedChannel do
66

7-
let!(:channel) { subject } # let is not thread safe, let! creates the object before ensuring uniqueness
7+
let(:channel) { subject }
8+
let(:probe) { Probe.new }
89

910
context 'with one thread' do
1011

@@ -58,8 +59,6 @@ module Concurrent
5859

5960
describe 'select' do
6061

61-
let(:probe) { Probe.new }
62-
6362
it 'does not block' do
6463
t = Thread.new { channel.select(probe) }
6564

@@ -97,5 +96,26 @@ module Concurrent
9796
end
9897

9998
end
99+
100+
describe 'probe set' do
101+
102+
it 'has size zero after creation' do
103+
channel.probe_set_size.should eq 0
104+
end
105+
106+
it 'increases size after a select' do
107+
channel.select(probe)
108+
channel.probe_set_size.should eq 1
109+
end
110+
111+
it 'decreases size after a removal' do
112+
channel.select(probe)
113+
channel.remove_probe(probe)
114+
channel.probe_set_size.should eq 0
115+
end
116+
117+
end
118+
119+
100120
end
101121
end

0 commit comments

Comments
 (0)