diff --git a/context/scheduler.md b/context/scheduler.md index d88abf30..a7433778 100644 --- a/context/scheduler.md +++ b/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 diff --git a/guides/scheduler/readme.md b/guides/scheduler/readme.md index d88abf30..a7433778 100644 --- a/guides/scheduler/readme.md +++ b/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 diff --git a/lib/async/scheduler.rb b/lib/async/scheduler.rb index ffdd9e17..267c964d 100644 --- a/lib/async/scheduler.rb +++ b/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) diff --git a/releases.md b/releases.md index ac220bda..bcce8e42 100644 --- a/releases.md +++ b/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`. diff --git a/test/async/scheduler/worker_pool.rb b/test/async/scheduler/worker_pool.rb new file mode 100644 index 00000000..b5d615d1 --- /dev/null +++ b/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