Merge pull request #168 from rage-rb/fiber-await-channels

Use different channels in `Fiber.await`

Author: rsamoilov

lib/rage/fiber.rb

@@ -118,6 +118,14 @@ def __block_channel(force = false)
     "block:#{object_id}:#{@__block_channel_i}"
   end
 
+  # @private
+  def __await_channel(force = false)
+    @__fiber_channel_i ||= 0
+    @__fiber_channel_i += 1 if force
+
+    "await:#{object_id}:#{@__fiber_channel_i}"
+  end
+
   # @private
   attr_accessor :__awaited_fileno
 
@@ -148,6 +156,7 @@ class << self
   # @note This method should only be used when multiple fibers have to be processed in parallel. There's no need to use `Fiber.await` for single IO calls.
   def self.await(fibers)
     f, fibers = Fiber.current, Array(fibers)
+    await_channel = f.__await_channel(true)
 
     # check which fibers are alive (i.e. have yielded) and which have errored out
     i, err, num_wait_for = 0, nil, 0
@@ -169,7 +178,7 @@ def self.await(fibers)
     end
 
     # wait on async fibers; resume right away if one of the fibers errors out
-    Iodine.subscribe("await:#{f.object_id}") do |_, err|
+    Iodine.subscribe(await_channel) do |_, err|
       if err == AWAIT_ERROR_MESSAGE
         f.resume
       else
@@ -179,7 +188,7 @@ def self.await(fibers)
     end
 
     Fiber.defer(-1)
-    Iodine.defer { Iodine.unsubscribe("await:#{f.object_id}") }
+    Iodine.defer { Iodine.unsubscribe(await_channel) }
 
     # if num_wait_for is not 0 means we exited prematurely because of an error
     if num_wait_for > 0

lib/rage/fiber_scheduler.rb

@@ -130,10 +130,10 @@ def fiber(&block)
         Thread.current[:rage_logger] = logger
         Fiber.current.__set_result(block.call)
         # send a message for `Fiber.await` to work
-        Iodine.publish("await:#{parent.object_id}", "", Iodine::PubSub::PROCESS) if parent.alive?
+        Iodine.publish(parent.__await_channel, "", Iodine::PubSub::PROCESS) if parent.alive?
       rescue Exception => e
         Fiber.current.__set_err(e)
-        Iodine.publish("await:#{parent.object_id}", Fiber::AWAIT_ERROR_MESSAGE, Iodine::PubSub::PROCESS) if parent.alive?
+        Iodine.publish(parent.__await_channel, Fiber::AWAIT_ERROR_MESSAGE, Iodine::PubSub::PROCESS) if parent.alive?
       end
     end
 

spec/fiber_spec.rb

@@ -183,4 +183,14 @@
       -> { expect(e).to be_a(StandardError) }
     end
   end
+
+  it "correctly processes several awaits in a row" do
+    within_reactor do
+      Fiber.await(Fiber.schedule { sleep 0.1 })
+      Fiber.await(Fiber.schedule { sleep 0.2 })
+      Fiber.await(Fiber.schedule { sleep 0.3 })
+
+      -> {}
+    end
+  end
 end