Add Async::Scheduler.enable_worker_pool and .worker_pool=

context/scheduler.md

@@ -88,6 +88,34 @@ Fiber.schedule do
 end
 ```
 
+### Worker Pool
+
+Ruby 3.4 introduced the `blocking_operation_wait` fiber scheduler hook, which allows the scheduler to execute blocking operations (e.g. `rb_nogvl`) on a worker pool, rather than blocking the event loop. The worker pool is disabled by default. You can enable it using the `ASYNC_SCHEDULER_WORKER_POOL=true` environment variable, which is read when `async` is loaded, or at runtime using {ruby Async::Scheduler.enable_worker_pool}:
+
+```ruby
+require "async"
+
+Async::Scheduler.enable_worker_pool
+
+Sync do
+	# Blocking operations are executed by the worker pool.
+end
+```
+
+The setting applies to schedulers created afterwards without an explicit `worker_pool:` argument, including those created by `Sync`, `Async` and {ruby Async::Reactor}. Each scheduler creates its own worker pool and closes it when the scheduler is closed. To customize the worker pool, set {ruby Async::Scheduler.worker_pool} to a factory which returns a new worker pool for each scheduler, or `nil` to disable it:
+
+```ruby
+Async::Scheduler.worker_pool = ->{IO::Event::WorkerPool.new(maximum_worker_count: 4)}
+```
+
+An explicit `worker_pool:` argument takes precedence over these settings:
+
+```ruby
+scheduler = Async::Scheduler.new(worker_pool: true)
+```
+
+The worker pool has overhead, which can be significant compared to the work being offloaded, so you should benchmark your application with and without it.
+
 ## Design
 
 ### Optimistic vs Pessimistic Scheduling

guides/scheduler/readme.md

@@ -88,6 +88,34 @@ Fiber.schedule do
 end
 ```
 
+### Worker Pool
+
+Ruby 3.4 introduced the `blocking_operation_wait` fiber scheduler hook, which allows the scheduler to execute blocking operations (e.g. `rb_nogvl`) on a worker pool, rather than blocking the event loop. The worker pool is disabled by default. You can enable it using the `ASYNC_SCHEDULER_WORKER_POOL=true` environment variable, which is read when `async` is loaded, or at runtime using {ruby Async::Scheduler.enable_worker_pool}:
+
+```ruby
+require "async"
+
+Async::Scheduler.enable_worker_pool
+
+Sync do
+	# Blocking operations are executed by the worker pool.
+end
+```
+
+The setting applies to schedulers created afterwards without an explicit `worker_pool:` argument, including those created by `Sync`, `Async` and {ruby Async::Reactor}. Each scheduler creates its own worker pool and closes it when the scheduler is closed. To customize the worker pool, set {ruby Async::Scheduler.worker_pool} to a factory which returns a new worker pool for each scheduler, or `nil` to disable it:
+
+```ruby
+Async::Scheduler.worker_pool = ->{IO::Event::WorkerPool.new(maximum_worker_count: 4)}
+```
+
+An explicit `worker_pool:` argument takes precedence over these settings:
+
+```ruby
+scheduler = Async::Scheduler.new(worker_pool: true)
+```
+
+The worker pool has overhead, which can be significant compared to the work being offloaded, so you should benchmark your application with and without it.
+
 ## Design
 
 ### Optimistic vs Pessimistic Scheduling

lib/async/scheduler.rb

@@ -28,6 +28,7 @@ module Async
 
 	# Handles scheduling of fibers. Implements the fiber scheduler interface.
 	class Scheduler < Node
+		# Whether the `ASYNC_SCHEDULER_WORKER_POOL` environment variable enabled the worker pool when `async/scheduler` was loaded. If so, {Scheduler.enable_worker_pool} is invoked.
 		WORKER_POOL = ENV.fetch("ASYNC_SCHEDULER_WORKER_POOL", nil).then do |value|
 			value == "true" ? true : nil
 		end
@@ -70,12 +71,57 @@ def blocking_operation_wait(work)
 			WorkerPool = nil
 		end
 
+		@@worker_pool = nil
+
+		# The factory used to create a worker pool for each scheduler that is created without an explicit `worker_pool:` argument, including those created by {Kernel::Sync}, {Kernel::Async} and {Reactor}.
+		#
+		# @public Since *Async v2.47*.
+		# @returns [Proc | Method | Nil] A callable which returns a new worker pool, or `nil` if the worker pool is disabled.
+		def self.worker_pool
+			@@worker_pool
+		end
+
+		# Set the factory used to create a worker pool for each scheduler that is created without an explicit `worker_pool:` argument. Changes only affect schedulers created afterwards.
+		#
+		# A scheduler closes its worker pool when it is closed, so the factory must return a new worker pool each time it is called, e.g. `->{IO::Event::WorkerPool.new(maximum_worker_count: 4)}`.
+		#
+		# @public Since *Async v2.47*.
+		# @parameter factory [Proc | Method | Nil] A callable which returns a new worker pool (or `nil`), or `nil` to disable the worker pool.
+		# @raises [ArgumentError] If the factory is not callable, e.g. a worker pool instance.
+		def self.worker_pool=(factory)
+			if WorkerPool and factory.is_a?(WorkerPool)
+				raise ArgumentError, "Worker pools can't be shared between schedulers, use a callable that returns a new worker pool instead!"
+			elsif factory.nil? or factory.respond_to?(:call)
+				@@worker_pool = factory
+			else
+				raise ArgumentError, "Worker pool factory must be callable or nil: #{factory.inspect}"
+			end
+		end
+
+		# Enable the default worker pool for each scheduler that is created without an explicit `worker_pool:` argument. Does nothing if `io-event` was built without worker pool support.
+		#
+		# @public Since *Async v2.47*.
+		def self.enable_worker_pool
+			self.worker_pool = WorkerPool&.method(:new)
+		end
+
+		# Create a new worker pool using {worker_pool}.
+		#
+		# @public Since *Async v2.47*.
+		# @returns [IO::Event::WorkerPool | Nil] A new worker pool, or `nil` if the worker pool is disabled.
+		def self.make_worker_pool
+			@@worker_pool&.call
+		end
+
+		enable_worker_pool if WORKER_POOL
+
 		# Create a new scheduler.
 		#
 		# @public Since *Async v1*.
 		# @parameter parent [Node | Nil] The parent node to use for task hierarchy.
 		# @parameter selector [IO::Event::Selector] The selector to use for event handling.
-		def initialize(parent = nil, selector: nil, profiler: Profiler&.default, worker_pool: WORKER_POOL)
+		# @parameter worker_pool [IO::Event::WorkerPool | Boolean | Nil] The worker pool to use for blocking operations, `true` to create a default worker pool, or `nil` to disable it. Defaults to a new worker pool created by {Scheduler.worker_pool}.
+		def initialize(parent = nil, selector: nil, profiler: Profiler&.default, worker_pool: Scheduler.make_worker_pool)
 			super(parent)
 
 			@selector = selector || ::IO::Event::Selector.new(Fiber.current)

releases.md

@@ -1,5 +1,31 @@
 # Releases
 
+## Unreleased
+
+  - Add `Async::Scheduler.enable_worker_pool` and `Async::Scheduler.worker_pool=` to enable the worker pool at runtime, for schedulers created without an explicit `worker_pool:` argument, including those created by `Sync`, `Async` and `Async::Reactor.new`. The `ASYNC_SCHEDULER_WORKER_POOL` environment variable is still supported.
+
+### Configuring the Worker Pool at Runtime
+
+The worker pool could previously only be enabled for schedulers you don't create yourself by setting `ASYNC_SCHEDULER_WORKER_POOL=true` before `async` was loaded. You can now enable it from application code:
+
+``` ruby
+require "async"
+
+Async::Scheduler.enable_worker_pool
+
+Sync do
+	# Blocking operations (e.g. `rb_nogvl`) are executed by the worker pool.
+end
+```
+
+Each scheduler creates its own worker pool, and closes it when the scheduler is closed. `enable_worker_pool` (like the environment variable) uses the `IO::Event::WorkerPool` defaults, which is a single worker. Previously, a custom worker pool such as one with a higher `maximum_worker_count` required creating the scheduler yourself with an explicit `worker_pool:` argument. Now you can set a factory which returns a new worker pool for each scheduler:
+
+``` ruby
+Async::Scheduler.worker_pool = ->{IO::Event::WorkerPool.new(maximum_worker_count: 4)}
+```
+
+Set `Async::Scheduler.worker_pool = nil` to disable the worker pool again. An explicit `worker_pool:` argument to `Async::Scheduler.new` takes precedence over these settings. Changes only affect schedulers created afterwards.
+
 ## v2.46.0
 
   - Fixed `Thread#kill` raising a `TypeError` while shutting down a thread with an active `Async::Reactor`.

test/async/scheduler/worker_pool.rb

@@ -0,0 +1,186 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2026, by Samuel Williams.
+
+require "async"
+
+describe Async::Scheduler do
+	def around
+		worker_pool = Async::Scheduler.worker_pool
+
+		super
+	ensure
+		Async::Scheduler.worker_pool = worker_pool
+	end
+
+	# Run a blocking operation in a new reactor, and return the reactor's worker pool.
+	def sync_worker_pool
+		Sync do
+			worker_pool = Fiber.scheduler.instance_variable_get(:@worker_pool)
+
+			if worker_pool
+				expect(IO::Event::WorkerPool.busy(duration: 0.001)).to have_keys(result: be == :completed)
+			end
+
+			worker_pool
+		end
+	end
+
+	with ".worker_pool" do
+		it "defaults to the environment variable" do
+			if ENV["ASYNC_SCHEDULER_WORKER_POOL"] == "true"
+				expect(Async::Scheduler::WORKER_POOL).to be == true
+				expect(Async::Scheduler.worker_pool).to be == Async::Scheduler::WorkerPool&.method(:new)
+			else
+				expect(Async::Scheduler::WORKER_POOL).to be_nil
+				expect(Async::Scheduler.worker_pool).to be_nil
+			end
+		end
+
+		it "can be set to a factory" do
+			factory = proc{nil}
+			Async::Scheduler.worker_pool = factory
+
+			expect(Async::Scheduler.worker_pool).to be_equal(factory)
+		end
+
+		it "can be disabled" do
+			Async::Scheduler.worker_pool = proc{nil}
+			Async::Scheduler.worker_pool = nil
+
+			expect(Async::Scheduler.worker_pool).to be_nil
+		end
+
+		it "is shared with subclasses" do
+			factory = proc{nil}
+			Async::Reactor.worker_pool = factory
+
+			expect(Async::Scheduler.worker_pool).to be_equal(factory)
+		end
+
+		it "can't be a worker pool instance" do
+			skip_unless_constant_defined(:WorkerPool, IO::Event)
+
+			worker_pool = IO::Event::WorkerPool.new
+
+			expect do
+				Async::Scheduler.worker_pool = worker_pool
+			end.to raise_exception(ArgumentError, message: be =~ /can't be shared/)
+		ensure
+			worker_pool&.close
+		end
+
+		it "can't be a boolean" do
+			expect do
+				Async::Scheduler.worker_pool = true
+			end.to raise_exception(ArgumentError, message: be =~ /must be callable/)
+		end
+	end
+
+	with ".enable_worker_pool" do
+		it "sets the default factory" do
+			Async::Scheduler.worker_pool = nil
+			Async::Scheduler.enable_worker_pool
+
+			expect(Async::Scheduler.worker_pool).to be == Async::Scheduler::WorkerPool&.method(:new)
+		end
+	end
+
+	with "worker pool" do
+		def before
+			skip_unless_constant_defined(:WorkerPool, IO::Event)
+
+			super
+		end
+
+		it "is enabled for schedulers created by Sync" do
+			Async::Scheduler.enable_worker_pool
+
+			Sync do
+				expect(Fiber.scheduler).to respond_to(:blocking_operation_wait)
+			end
+
+			worker_pool = sync_worker_pool
+			expect(worker_pool).to be_a(IO::Event::WorkerPool)
+			expect(worker_pool.statistics).to have_keys(
+				call_count: be > 0,
+				shutdown: be == true,
+			)
+		end
+
+		it "is disabled for schedulers created by Sync" do
+			Async::Scheduler.worker_pool = nil
+
+			Sync do
+				expect(Fiber.scheduler).not.to respond_to(:blocking_operation_wait)
+			end
+
+			expect(sync_worker_pool).to be_nil
+		end
+
+		it "creates a new worker pool for each scheduler" do
+			Async::Scheduler.enable_worker_pool
+
+			first = sync_worker_pool
+			second = sync_worker_pool
+
+			expect(first).to be_a(IO::Event::WorkerPool)
+			expect(second).to be_a(IO::Event::WorkerPool)
+			expect(second).not.to be_equal(first)
+
+			expect(first.statistics).to have_keys(call_count: be == 1)
+			expect(second.statistics).to have_keys(call_count: be == 1)
+		end
+
+		it "can use a callable to create each worker pool" do
+			calls = 0
+
+			Async::Scheduler.worker_pool = proc do
+				calls += 1
+				IO::Event::WorkerPool.new(maximum_worker_count: 2)
+			end
+
+			first = sync_worker_pool
+			second = sync_worker_pool
+
+			expect(calls).to be == 2
+			expect(second).not.to be_equal(first)
+			expect(first.statistics).to have_keys(maximum_worker_count: be == 2, call_count: be == 1)
+			expect(second.statistics).to have_keys(maximum_worker_count: be == 2, call_count: be == 1)
+		end
+
+		with "explicit worker_pool:" do
+			it "can disable the worker pool" do
+				Async::Scheduler.enable_worker_pool
+
+				scheduler = Async::Scheduler.new(worker_pool: nil)
+
+				expect(scheduler).not.to respond_to(:blocking_operation_wait)
+			ensure
+				scheduler&.close
+			end
+
+			it "can enable the worker pool" do
+				Async::Scheduler.worker_pool = nil
+
+				scheduler = Async::Scheduler.new(worker_pool: true)
+
+				expect(scheduler).to respond_to(:blocking_operation_wait)
+			ensure
+				scheduler&.close
+			end
+
+			it "can use a specific worker pool" do
+				Async::Scheduler.worker_pool = proc{raise "Should not be called!"}
+
+				worker_pool = IO::Event::WorkerPool.new
+				scheduler = Async::Scheduler.new(worker_pool: worker_pool)
+
+				expect(scheduler.instance_variable_get(:@worker_pool)).to be_equal(worker_pool)
+			ensure
+				scheduler&.close
+			end
+		end
+	end
+end