Thread safe Async::Condition and Async::Notification.

Author: ioquatix

fixtures/async/a_condition.rb

@@ -12,7 +12,7 @@ module Async
 		it "can signal waiting task" do
 			state = nil
 
-			reactor.async do
+			task = reactor.async do
 				state = :waiting
 				condition.wait
 				state = :resumed
@@ -21,8 +21,7 @@ module Async
 			expect(state).to be == :waiting
 
 			condition.signal
-			
-			reactor.yield
+			task.wait
 
 			expect(state).to be == :resumed
 		end
@@ -46,16 +45,15 @@ module Async
 		it "resumes tasks in order" do
 			order = []
 
-			5.times do |i|
-				task = reactor.async do
+			tasks = 5.times.map do |i|
+				reactor.async do
 					condition.wait
 					order << i
 				end
 			end
 
 			condition.signal
-			
-			reactor.yield
+			tasks.each(&:wait)
 
 			expect(order).to be == [0, 1, 2, 3, 4]
 		end

lib/async/condition.rb

@@ -13,65 +13,47 @@ module Async
 	class Condition
 		# Create a new condition.
 		def initialize
-			@waiting = List.new
+			@ready = ::Thread::Queue.new
 		end
 
-		class FiberNode < List::Node
-			def initialize(fiber)
-				@fiber = fiber
-			end
-			
-			def transfer(*arguments)
-				@fiber.transfer(*arguments)
-			end
-			
-			def alive?
-				@fiber.alive?
-			end
-		end
-		
-		private_constant :FiberNode
-		
 		# Queue up the current fiber and wait on yielding the task.
 		# @returns [Object]
 		def wait
-			@waiting.stack(FiberNode.new(Fiber.current)) do
-				Fiber.scheduler.transfer
-			end
+			@ready.pop
 		end
 
-		# @deprecated Replaced by {#waiting?}
+		# @returns [Boolean] If there are no fibers waiting on this condition.
 		def empty?
-			warn("`Async::Condition#empty?` is deprecated, use `Async::Condition#waiting?` instead.", uplevel: 1, category: :deprecated) if $VERBOSE
-			
-			@waiting.empty?
+			@ready.num_waiting.zero?
 		end
 
 		# @returns [Boolean] Is any fiber waiting on this notification?
 		def waiting?
-			@waiting.size > 0
+			!self.empty?
 		end
 
 		# Signal to a given task that it should resume operations.
 		# @parameter value [Object | Nil] The value to return to the waiting fibers.
 		def signal(value = nil)
-			return if @waiting.empty?
+			return if empty?
 
-			waiting = self.exchange
+			ready = self.exchange
 
-			waiting.each do |fiber|
-				Fiber.scheduler.resume(fiber, value) if fiber.alive?
+			ready.num_waiting.times do
+				ready.push(value)
 			end
 
+			ready.close
+			
 			return nil
 		end
 
 		protected
 
 		def exchange
-			waiting = @waiting
-			@waiting = List.new
-			return waiting
+			ready = @ready
+			@ready = ::Thread::Queue.new
+			return ready
 		end
 	end
 end

lib/async/notification.rb

@@ -12,23 +12,25 @@ class Notification < Condition
 		# Signal to a given task that it should resume operations.
 		#
 		# @returns [Boolean] if a task was signalled.
-		def signal(value = nil, task: Task.current)
-			return false if @waiting.empty?
+		def signal(value = nil)
+			return false if empty?
 
 			Fiber.scheduler.push Signal.new(self.exchange, value)
 
 			return true
 		end
 
-		Signal = Struct.new(:waiting, :value) do
+		Signal = Struct.new(:ready, :value) do
 			def alive?
 				true
 			end
 
 			def transfer
-				waiting.each do |fiber|
-					fiber.transfer(value) if fiber.alive?
+				ready.num_waiting.times do
+					ready.push(value)
 				end
+				
+				ready.close
 			end
 		end
 

test/async/condition.rb

@@ -21,9 +21,10 @@
 
 		expect(task).to be(:running?)
 
-		# This will cause the task to exit:
 		condition.signal
 
+		task.wait
+		
 		expect(task).to be(:completed?)
 	end
 

test/async/notification.rb

@@ -33,6 +33,7 @@
 
 		sequence << :yielding
 		reactor.yield
+		task.wait
 		sequence << :finished
 
 		expect(task.status).to be == :completed

test/async/task.rb

@@ -527,7 +527,7 @@
 				producer.stop # (5) [producer is resumed already] producer.stop
 			end
 
-			expect(items).to be == [1, 2]
+			expect(items).not.to be(:empty?)
 			expect(value).to be == 3
 		end
 
@@ -846,15 +846,13 @@ def sleep_forever
 
 		with "stopped task" do
 			it "is stopped?" do
-				reactor.async do |task|
-					child = task.async do |task|
-						sleep(1)
-					end
-					
-					child.stop
-					
-					expect(child).to be(:stopped?)
-				end.wait
+				child = reactor.async do |task|
+					sleep(1)
+				end
+				
+				child.stop
+				
+				expect(child).to be(:stopped?)
 			end
 		end
 	end
@@ -930,11 +928,12 @@ def sleep_forever
 			child_task.stop(true)
 			expect(child_task).to be(:running?)
 
-			reactor.async do
-				condition.signal
+			condition.signal
+			
+			while child_task.running?
+				reactor.run_once(0)
 			end
 
-			reactor.run_once(0)
 			expect(child_task).to be(:stopped?)
 		end