Better Channel close behavior.

Author: jdantonio

examples/a-tour-of-go-channels/default-selection.rb

@@ -12,7 +12,7 @@
 
 loop do
   Channel.select do |s|
-    s.take(tick) { print "tick.\n" }
+    s.take(tick) { |t| print "tick.\n" if t }
     s.take(boom) do
       print "BOOM!\n"
       exit

examples/go-by-example-channels/rate-limiting.rb

@@ -17,8 +17,7 @@
 
 limiter = Channel.ticker(0.2)
 requests.each do |req|
-  ~limiter
-  print "request #{req} #{Channel::Tick.new}\n"
+  print "request #{req} #{Channel::Tick.new}\n" if ~limiter
 end
 print "\n"
 
@@ -27,8 +26,9 @@
   bursty_limiter << Channel::Tick.new
 end
 
+ticker = Channel.ticker(0.2)
 Channel.go do
-  Channel.ticker(0.2).each do |t|
+  ticker.each do |t|
     bursty_limiter << t
   end
 end
@@ -44,6 +44,9 @@
   print "request #{req} #{Channel::Tick.new}\n"
 end
 
+limiter.close
+ticker.close
+
 __END__
 request 1 2012-10-19 00:38:18.687438 +0000 UTC
 request 2 2012-10-19 00:38:18.887471 +0000 UTC

examples/go-by-example-channels/ticker.rb

@@ -10,7 +10,7 @@
 ticker = Channel.ticker(0.5)
 Channel.go do
   ticker.each do |tick|
-    print "Tick at #{tick}\n"
+    print "Tick at #{tick}\n" if tick
   end
 end
 

examples/go-by-example-channels/timers.rb

@@ -9,13 +9,11 @@
 
 timer1 = Channel.timer(2)
 
-~timer1
-puts 'Timer 1 expired'
+puts 'Timer 1 expired' if ~timer1
 
 timer2 = Channel.timer(1)
 Channel.go do
-  ~timer2
-  print "Timer 2 expired\n"
+  print "Timer 2 expired\n" if ~timer2
 end
 
 stop2 = timer2.stop

lib/concurrent/channel.rb

@@ -12,7 +12,7 @@ class Channel
     include Enumerable
 
     # NOTE: Move to global IO pool once stable
-    GOROUTINES = Concurrent::CachedThreadPool.new
+    GOROUTINES = Concurrent::CachedThreadPool.new(auto_terminate: true)
     private_constant :GOROUTINES
 
     BUFFER_TYPES = {

lib/concurrent/channel/buffer/base.rb

@@ -1,4 +1,5 @@
 require 'concurrent/synchronization/lockable_object'
+require 'concurrent/utility/at_exit'
 
 module Concurrent
   class Channel
@@ -32,6 +33,7 @@ def initialize(*args)
             @capacity = 0
             @buffer = nil
             ns_initialize(*args)
+            AtExit.add(self) { terminate_at_exit }
           end
         end
 
@@ -232,6 +234,11 @@ def ns_full?
         def ns_closed?
           @closed
         end
+
+        # @!visibility private
+        def terminate_at_exit
+          synchronize { self.closed = true }
+        end
       end
     end
   end

lib/concurrent/channel/buffer/ticker.rb

@@ -19,12 +19,14 @@ def ns_initialize(interval)
 
         def do_poll
           synchronize do
-            if !ns_closed? && (now = Concurrent.monotonic_time) >= @next_tick
+            if ns_closed?
+              return Concurrent::NULL, false
+            elsif (now = Concurrent.monotonic_time) >= @next_tick
               tick = Concurrent::Channel::Tick.new(@next_tick)
               @next_tick = now + @interval
-              return tick
+              return tick, true
             else
-              return Concurrent::NULL
+              return nil, true
             end
           end
         end

lib/concurrent/channel/buffer/timer.rb

@@ -18,26 +18,28 @@ def offer(item)
         end
 
         def take
-          # a Go timer will block forever if stopped
           loop do
-            tick = do_poll
-            return tick if tick != Concurrent::NULL
-            Thread.pass
+            tick, _ = do_poll
+            if tick
+              return tick
+            else
+              Thread.pass
+            end
           end
         end
 
         def next
-          # a Go timer will block forever if stopped
-          # it will always return `true` for more
           loop do
-            tick = do_poll
-            return tick, true if tick != Concurrent::NULL
+            tick, more = do_poll
+            return tick, more if tick
             Thread.pass
           end
         end
 
         def poll
-          do_poll
+          tick, _ = do_poll
+          tick = Concurrent::NULL unless tick
+          tick
         end
 
         private
@@ -61,12 +63,14 @@ def ns_full?
 
         def do_poll
           synchronize do
-            if !ns_closed? && Concurrent.monotonic_time >= @tick
+            if ns_closed?
+              return Concurrent::NULL, false
+            elsif Concurrent.monotonic_time >= @tick
               # only one listener gets notified
               self.closed = true
-              return Concurrent::Channel::Tick.new(@tick)
+              return Concurrent::Channel::Tick.new(@tick), false
             else
-              return Concurrent::NULL
+              return nil, true
             end
           end
         end

spec/concurrent/channel/buffer/ticker_spec.rb

@@ -16,6 +16,13 @@ module Concurrent::Channel::Buffer
         expected.times { actual += 1 if subject.take.is_a? Concurrent::Channel::Tick }
         expect(actual).to eq expected
       end
+
+      it 'returns Concurrent::NULL when closed after trigger' do
+        subject.take
+        subject.close
+        expect(subject).to be_closed
+        expect(subject.take).to eq Concurrent::NULL
+      end
     end
 
     context '#poll' do
@@ -37,6 +44,17 @@ module Concurrent::Channel::Buffer
         expected.times { actual += 1 if subject.next.first.is_a? Concurrent::Channel::Tick }
         expect(actual).to eq expected
       end
+
+      it 'returns true for more while open' do
+        _, more = subject.next
+        expect(more).to be true
+      end
+
+      it 'returns false for more once closed' do
+        subject.close
+        _, more = subject.next
+        expect(more).to be false
+      end
     end
   end
 end

spec/concurrent/channel/buffer/timer_spec.rb

@@ -33,6 +33,11 @@ module Concurrent::Channel::Buffer
         end
         expect(subject).to be_closed
       end
+
+    it 'returns false for more' do
+      _, more = subject.next
+      expect(more).to be false
+    end
     end
   end
 end

spec/concurrent/channel/buffer/timing_buffer_shared.rb

@@ -69,15 +69,9 @@
       expect(actual - start).to be >= 0.1
     end
 
-    it 'blocks forever when closed' do
+    it 'returns Concurrent::NULL when closed' do
       subject.close
-      t = Thread.new do
-        subject.take
-      end
-      actual = t.join(1)
-      t.kill # clean up
-
-      expect(actual).to be_falsey
+      expect(subject.take).to eq Concurrent::NULL
     end
   end
 
@@ -129,20 +123,9 @@
       expect(value).to be_a Concurrent::Channel::Tick
     end
 
-    it 'returns true for more' do
-      _, more = subject.next
-      expect(more).to be true
-    end
-
-    it 'blocks forever when closed' do
+    it 'returns Concurrent::NULL, false when closed' do
       subject.close
-      t = Thread.new do
-        subject.next
-      end
-      actual = t.join(1)
-      t.kill # clean up
-
-      expect(actual).to be_falsey
+      expect(subject.take).to eq Concurrent::NULL
     end
 
     it 'triggers after the specified time interval' do

spec/concurrent/channel/integration_spec.rb

@@ -1,4 +1,4 @@
-describe 'channel integration tests', notravis: true do
+describe 'channel integration tests' do
 
   let!(:examples_root) { File.expand_path(File.join(File.dirname(__FILE__), '../../../examples')) }
 
@@ -65,7 +65,7 @@
       expect(result).to eq expected
     end
 
-    specify 'default-selection.rb' do
+    specify 'default-selection.rb', notravis: true do
 expected = <<-STDOUT
     .
     .

spec/concurrent/channel_spec.rb

@@ -557,7 +557,7 @@ module Concurrent
       end
     end
 
-    context 'goroutines' do
+    context 'goroutines', notravis: true do
 
       let(:default_executor) { Channel.const_get(:GOROUTINES) }
 
@@ -608,7 +608,7 @@ module Concurrent
             actual < expected
           end
 
-          latch.wait(3)
+          latch.wait(10)
           expect(actual).to eq expected
         end
       end