Introduce Controller#wait_until_free for race-free clean-up.

Author: ioquatix

async-pool.gemspec

@@ -25,5 +25,5 @@ Gem::Specification.new do |spec|
 
 	spec.required_ruby_version = ">= 3.2"
 
-	spec.add_dependency "async", ">= 1.25"
+	spec.add_dependency "async", ">= 2.0"
 end

benchmark/acquire.rb

@@ -0,0 +1,21 @@
+
+require "sus/fixtures/benchmark"
+require "sus/fixtures/async/scheduler_context"
+
+require "async/pool"
+require "async/pool/resource"
+
+include Sus::Fixtures::Benchmark
+include Sus::Fixtures::Async::SchedulerContext
+
+describe Async::Pool::Controller do
+	let(:pool) {subject.new(Async::Pool::Resource)}
+	measure Async::Pool::Controller do |repeats|
+		pool = self.pool
+		
+		repeats.times do
+			pool.acquire do |resource|
+			end
+		end
+	end
+end

gems.rb

@@ -23,6 +23,7 @@
 
 	gem "traces"
 	gem "sus-fixtures-async"
+	gem "sus-fixtures-benchmark"
 
 	gem "bake-test"
 	gem "bake-test-external"

lib/async/pool/controller.rb

@@ -9,9 +9,10 @@
 require "console/logger"
 
 require "async"
-require "async/notification"
 require "async/semaphore"
 
+require "thread"
+
 module Async
 	module Pool
 		# A resource pool controller.
@@ -47,7 +48,8 @@ def initialize(constructor, limit: nil, concurrency: (limit || 1), policy: nil,
 				@available = []
 
 				# Used to signal when a resource has been released:
-				@notification = Async::Notification.new
+				@mutex = Thread::Mutex.new
+				@condition = Thread::ConditionVariable.new
 			end
 
 			# @attribute [Proc] The constructor used to create new resources.
@@ -124,8 +126,23 @@ def available?
 			end
 
 			# Wait until a pool resource has been freed.
+			# @deprecated Use {wait_until_free} instead.
 			def wait
-				@notification.wait
+				@mutex.synchronize do
+					@condition.wait(@mutex)
+				end
+			end
+			
+			# Wait until the pool is not busy (no resources in use).
+			def wait_until_free
+				@mutex.synchronize do
+					if busy?
+						yield self if block_given?
+						
+						# Wait until the pool is not busy:
+						@condition.wait(@mutex) while busy?
+					end
+				end
 			end
 
 			# Whether the pool is empty.
@@ -221,7 +238,7 @@ def retire(resource)
 
 				resource.close
 
-				@notification.signal
+				@mutex.synchronize {@condition.broadcast}
 
 				return true
 			end
@@ -285,19 +302,17 @@ def reuse(resource)
 
 				@resources[resource] = usage - 1
 
-				@notification.signal
+				@mutex.synchronize {@condition.broadcast}
 
 				return true
 			end
 
 			def wait_for_resource
 				# If we fail to create a resource (below), we will end up waiting for one to become resources.
 				until resource = available_resource
-					@notification.wait
+					@mutex.synchronize {@condition.wait(@mutex)}
 				end
-				
 				# Be careful not to context switch or fail here.
-				
 				return resource
 			end
 
@@ -345,10 +360,8 @@ def acquire_existing_resource
 							return resource
 						end
 					end
-					
-					@notification.wait
+					@mutex.synchronize {@condition.wait(@mutex)}
 				end
-				
 				# Only when the pool has been completely drained, return nil:
 				return nil
 			end

test/async/pool/controller.rb

@@ -407,10 +407,28 @@ def failures(repeats: 500, time_scale: 0.001, &block)
 			expect(pool).not.to be(:busy?)
 		end
 	end
-end
+	
+	with "#wait_until_free" do
+		it "waits until the pool is not busy and yields if waited" do
+			pool = subject.new(Async::Pool::Resource, limit: 1)
+			resource = pool.acquire
+			waited = false
+			finished = false
 
-describe Async::Pool::Controller do
-	let(:pool) {subject.new(Async::Pool::Resource)}
+			Async do
+				Async::Task.current.sleep(0.01)
+				pool.release(resource)
+			end
+
+			pool.wait_until_free do
+				waited = true
+			end
+
+			finished = true
+			expect(waited).to be == true
+			expect(finished).to be == true
+		end
+	end
 
 	with "#close" do
 		it "closes all resources when going out of scope" do