Add Promise::Channel

pitr-ch · pitr-ch · commit 0f33d0914be2 · 2016-12-27T01:23:27.000+01:00
diff --git a/lib/concurrent/edge/old_channel_integration.rb b/lib/concurrent/edge/old_channel_integration.rb
@@ -0,0 +1,54 @@
+module Concurrent
+  module Promises
+    module FactoryMethods
+
+      # @!visibility private
+
+      module OldChannelIntegration
+
+        # @!visibility private
+
+        # only proof of concept
+        # @return [Future]
+        def select(*channels)
+          # TODO (pitr-ch 26-Mar-2016): re-do, has to be non-blocking
+          future do
+            # noinspection RubyArgCount
+            Channel.select do |s|
+              channels.each do |ch|
+                s.take(ch) { |value| [value, ch] }
+              end
+            end
+          end
+        end
+      end
+
+      include OldChannelIntegration
+    end
+
+    class Future < AbstractEventFuture
+
+      # @!visibility private
+
+      module OldChannelIntegration
+
+        # @!visibility private
+
+        # Zips with selected value form the suplied channels
+        # @return [Future]
+        def then_select(*channels)
+          future = Concurrent::Promises.select(*channels)
+          ZipFuturesPromise.new_blocked_by2(self, future, @DefaultExecutor).future
+        end
+
+        # @note may block
+        # @note only proof of concept
+        def then_put(channel)
+          on_fulfillment_using(:io, channel) { |value, channel| channel.put value }
+        end
+      end
+
+      include OldChannelIntegration
+    end
+  end
+end
diff --git a/lib/concurrent/edge/promises.rb b/lib/concurrent/edge/promises.rb
@@ -1951,57 +1951,128 @@ def then_ask(actor)
       include ActorIntegration
     end
 
-    ### Experimental features follow
+    class Channel < Concurrent::Synchronization::Object
+      safe_initialization!
 
-    module FactoryMethods
+      # Default size of the Channel, makes it accept unlimited number of messages.
+      UNLIMITED = Object.new
+      UNLIMITED.singleton_class.class_eval do
+        include Comparable
 
-      # @!visibility private
+        def <=>(other)
+          1
+        end
 
-      module ChannelIntegration
+        def to_s
+          'unlimited'
+        end
+      end
+
+      # A channel to pass messages between promises. The size is limited to support back pressure.
+      # @param [Integer, UNLIMITED] size the maximum number of messages stored in the channel.
+      def initialize(size = UNLIMITED)
+        super()
+        @Size        = size
+        # TODO (pitr-ch 26-Dec-2016): replace with lock-free implementation
+        @Mutex       = Mutex.new
+        @Probes      = []
+        @Messages    = []
+        @PendingPush = []
+      end
 
-        # @!visibility private
 
-        # only proof of concept
-        # @return [Future]
-        def select(*channels)
-          # TODO (pitr-ch 26-Mar-2016): re-do, has to be non-blocking
-          future do
-            # noinspection RubyArgCount
-            Channel.select do |s|
-              channels.each do |ch|
-                s.take(ch) { |value| [value, ch] }
+      # Returns future which will fulfill when the message is added to the channel. Its value is the message.
+      # @param [Object] message
+      # @return [Future]
+      def push(message)
+        @Mutex.synchronize do
+          while true
+            if @Probes.empty?
+              if @Size > @Messages.size
+                @Messages.push message
+                return Promises.fulfilled_future message
+              else
+                pushed = Promises.resolvable_future
+                @PendingPush.push [message, pushed]
+                return pushed.with_hidden_resolvable
+              end
+            else
+              probe = @Probes.shift
+              if probe.fulfill [self, message], false
+                return Promises.fulfilled_future(message)
               end
             end
           end
         end
       end
 
-      include ChannelIntegration
+      # Returns a future witch will become fulfilled with a value from the channel when one is available.
+      # @param [ResolvableFuture] probe the future which will be fulfilled with a channel value
+      # @return [Future] the probe, its value will be the message when available.
+      def pop(probe = Concurrent::Promises.resolvable_future)
+        # TODO (pitr-ch 26-Dec-2016): improve performance
+        pop_for_select(probe).then(&:last)
+      end
+
+      # @!visibility private
+      def pop_for_select(probe = Concurrent::Promises.resolvable_future)
+        @Mutex.synchronize do
+          if @Messages.empty?
+            @Probes.push probe
+          else
+            message = @Messages.shift
+            probe.fulfill [self, message]
+
+            unless @PendingPush.empty?
+              message, pushed = @PendingPush.shift
+              @Messages.push message
+              pushed.fulfill message
+            end
+          end
+        end
+        probe
+      end
+
+      # @return [String] Short string representation.
+      def to_s
+        format '<#%s:0x%x size:%s>', self.class, object_id << 1, @Size
+      end
+
+      alias_method :inspect, :to_s
     end
 
     class Future < AbstractEventFuture
+      module NewChannelIntegration
 
-      # @!visibility private
+        # @param [Channel] channel to push to.
+        # @return [Future] a future which is fulfilled after the message is pushed to the channel.
+        #   May take a moment if the channel is full.
+        def then_push_channel(channel)
+          self.then { |value| channel.push value }.flat_future
+        end
+
+        # TODO (pitr-ch 26-Dec-2016): does it make sense to have rescue an chain variants as well, check other integrations as well
+      end
 
-      module ChannelIntegration
+      include NewChannelIntegration
+    end
 
-        # @!visibility private
+    module FactoryMethods
 
-        # Zips with selected value form the suplied channels
-        # @return [Future]
-        def then_select(*channels)
-          future = Concurrent::Promises.select(*channels)
-          ZipFuturesPromise.new_blocked_by2(self, future, @DefaultExecutor).future
-        end
+      module NewChannelIntegration
 
-        # @note may block
-        # @note only proof of concept
-        def then_put(channel)
-          on_fulfillment_using(:io, channel) { |value, channel| channel.put value }
+        # Selects a channel which is ready to be read from.
+        # @param [Channel] channels
+        # @return [Future] a future which is fulfilled with pair [channel, message] when one of the channels is
+        #   available for reading
+        def select_channel(*channels)
+          probe = Promises.resolvable_future
+          channels.each { |ch| ch.pop_for_select probe }
+          probe
         end
       end
 
-      include ChannelIntegration
+      include NewChannelIntegration
     end
 
   end
diff --git a/spec/concurrent/edge/promises_spec.rb b/spec/concurrent/edge/promises_spec.rb
@@ -426,19 +426,17 @@ def behaves_as_delay(delay, value)
     end
 
     it 'with channel' do
-      ch1 = Concurrent::Channel.new
-      ch2 = Concurrent::Channel.new
+      ch1 = Concurrent::Promises::Channel.new
+      ch2 = Concurrent::Promises::Channel.new
 
-      result = Concurrent::Promises.select(ch1, ch2)
-      ch1.put 1
-      expect(result.value!).to eq [1, ch1]
+      result = Concurrent::Promises.select_channel(ch1, ch2)
+      ch1.push 1
+      expect(result.value!).to eq [ch1, 1]
 
 
-      future { 1+1 }.
-          then_put(ch1)
-      result = future { '%02d' }.
-          then_select(ch1, ch2).
-          then { |format, (value, channel)| format format, value }
+      future { 1+1 }.then_push_channel(ch1)
+      result = (Concurrent::Promises.future { '%02d' } & Concurrent::Promises.select_channel(ch1, ch2)).
+          then { |format, (channel, value)| format format, value }
       expect(result.value!).to eq '02'
     end
   end
@@ -487,7 +485,7 @@ def behaves_as_delay(delay, value)
 
   describe 'Throttling' do
     specify do
-      limit = 4
+      limit    = 4
       throttle = Concurrent::Throttle.new limit
       counter  = Concurrent::AtomicFixnum.new
       testing  = -> *args do
@@ -523,4 +521,75 @@ def behaves_as_delay(delay, value)
           end).value!.all? { |v| v <= limit }).to be_truthy
     end
   end
+
+  describe 'Promises::Channel' do
+    specify do
+      channel = Concurrent::Promises::Channel.new 1
+
+      pushed1 = channel.push 1
+      expect(pushed1.resolved?).to be_truthy
+      expect(pushed1.value!).to eq 1
+
+      pushed2 = channel.push 2
+      expect(pushed2.resolved?).to be_falsey
+
+      popped = channel.pop
+      expect(pushed1.value!).to eq 1
+      expect(pushed2.resolved?).to be_truthy
+      expect(pushed2.value!).to eq 2
+      expect(popped.value!).to eq 1
+
+      popped = channel.pop
+      expect(popped.value!).to eq 2
+
+      popped = channel.pop
+      expect(popped.resolved?).to be_falsey
+
+      pushed3 = channel.push 3
+      expect(popped.value!).to eq 3
+      expect(pushed3.resolved?).to be_truthy
+      expect(pushed3.value!).to eq 3
+    end
+
+    specify do
+      ch1 = Concurrent::Promises::Channel.new
+      ch2 = Concurrent::Promises::Channel.new
+      ch3 = Concurrent::Promises::Channel.new
+
+      add = -> do
+        (ch1.pop & ch2.pop).then do |a, b|
+          if a == :done && b == :done
+            :done
+          else
+            ch3.push a + b
+            add.call
+          end
+        end
+      end
+
+      ch1.push 1
+      ch2.push 2
+      ch1.push 'a'
+      ch2.push 'b'
+      ch1.push nil
+      ch2.push true
+
+      result = Concurrent::Promises.future(&add).run.result
+      expect(result[0..1]).to eq [false, nil]
+      expect(result[2]).to be_a_kind_of(NoMethodError)
+      expect(ch3.pop.value!).to eq 3
+      expect(ch3.pop.value!).to eq 'ab'
+
+      ch1.push 1
+      ch2.push 2
+      ch1.push 'a'
+      ch2.push 'b'
+      ch1.push :done
+      ch2.push :done
+
+      expect(Concurrent::Promises.future(&add).run.result).to eq [true, :done, nil]
+      expect(ch3.pop.value!).to eq 3
+      expect(ch3.pop.value!).to eq 'ab'
+    end
+  end
 end
