Class: Farce::Abstract::Scheduler Abstract

Inherits:
Executor
  • Object
show all
Includes:
Internal::Inspect
Defined in:
lib/farce/abstract/scheduler.rb

Overview

This class is abstract.

Use Scheduler, Unshared::Scheduler, or Local::Scheduler.

A task scheduler for one worker or an installed fiber scheduler.

Use Scheduler.create to start a worker, or install an instance on the current thread with Fiber.set_scheduler. #close rejects new submissions and drains queued tasks. A scheduler can only be launched or installed once.

Direct Known Subclasses

Local::Scheduler, Scheduler, Unshared::Scheduler

Class Method Summary collapse

Instance Method Summary collapse

Methods inherited from Executor

#close_all, #execute, #ractor_shareable?

Methods included from Internal::Noncopyable

#duplicable?

Constructor Details

#initialize(capacity: 1024, backend: Farce.config.io_backend) ⇒ Scheduler #initialize(capacity: 1024, &block) ⇒ Scheduler

Creates a scheduler without starting a worker. Use this instead of create when you want to launch it later with #launch, or install it on the current thread with Fiber.set_scheduler(scheduler).

Overloads:

  • #initialize(capacity: 1024, backend: Farce.config.io_backend) ⇒ Scheduler

    Uses the configured fiber scheduler when launched or installed.

    Parameters:

    • capacity (Integer, nil) (defaults to: 1024) —

      Pending-task capacity, nil creates an unbounded queue.

    • backend (Symbol) (defaults to: Farce.config.io_backend) —

      IO backend for the built-in scheduler, defaults to Farce configuration.

  • #initialize(capacity: 1024, &block) ⇒ Scheduler

    Uses another library's fiber scheduler while retaining Farce's ability to accept tasks from other threads and Ractors.

    Supply a block that constructs and returns that scheduler. The block runs once when this handle is launched or installed, in the thread that will run the tasks, rather than during new. Scheduler requires a Ractor-shareable block. Unshared::Scheduler may capture local state. Farce installs its result. Submit tasks separately with #schedule. Configure the supplied scheduler in the block. Farce's backend option is ignored for this overload.

    Examples:

    Choose Async, then launch the worker later

    require "async"
require "farce"

scheduler = Farce::Scheduler.new { Async::Scheduler.new }
worker = scheduler.launch_thread
scheduler.schedule { puts "Running on Async" }

    Parameters:

    • capacity (Integer, nil) (defaults to: 1024) —

      Pending-task capacity, nil creates an unbounded queue.

    Yield Returns:

    • (Object) —

      The fiber scheduler to install when launched or installed.



156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
 
# File 'lib/farce/abstract/scheduler.rb', line 156

def initialize(capacity: 1024, backend: Internal::FROZEN_CONFIG.io_backend, queue: nil, external: false,
               pool_worker: nil, queue_owner: true, &constructor)
  constructor ||= Internal::FROZEN_CONFIG.fiber_scheduler_constructor
  @capacity    = capacity ? Integer(capacity) : nil
  @backend     = backend
  @constructor = prepare_constructor(constructor)
  @external    = external || !constructor.nil?
  @owner       = Internal::Atom.new(nil)
  @state       = Internal::Atom.new(:initialized)
  @error       = Internal::Atom.new
  @queue       = queue || new_queue(capacity: @capacity)
  @pool_worker = pool_worker
  @queue_owner = queue_owner
  super()
end

Class Method Details

.create(executor = nil, name: nil, priority: nil, capacity: 1024, backend: Farce.config.io_backend) ⇒ Scheduler .create(executor = nil, name: nil, priority: nil, capacity: 1024, &block) ⇒ Scheduler

Creates a new scheduler and starts it in a background ractor, thread, or custom executor.

Overloads:

  • .create(executor = nil, name: nil, priority: nil, capacity: 1024, backend: Farce.config.io_backend) ⇒ Scheduler

    Runs tasks on the configured fiber scheduler. The backend option selects the IO driver when using Farce's built-in scheduler.

    Examples:

    scheduler = Farce::Scheduler.create
scheduler.schedule { puts "Running on the built-in scheduler" }

    Parameters:

    • executor (Class, nil) (defaults to: nil) —

      Ractor or Thread, nil selects the runtime default.

    • name (String, nil) (defaults to: nil) —

      Optional worker name.

    • priority (Integer, nil) (defaults to: nil) —

      Optional worker thread priority.

    • capacity (Integer, nil) (defaults to: 1024) —

      Pending-task capacity, nil creates an unbounded queue.

    • backend (Symbol) (defaults to: Farce.config.io_backend) —

      IO backend for the built-in scheduler, defaults to Farce configuration.

    Returns:

  • .create(executor = nil, name: nil, priority: nil, capacity: 1024, &block) ⇒ Scheduler

    Runs tasks on a fiber scheduler supplied by your block. Use this to send work through Farce to another library's event loop, such as Async.

    Construct and return the scheduler inside the block: Farce calls it once in the worker, where that event loop will run, and installs the result. The block configures the scheduler. Submit actual tasks with #schedule. Configure any IO backend on the returned scheduler itself. Farce's backend option is ignored when a block is supplied.

    Scheduler requires a Ractor-shareable block. Require the library before calling create and avoid capturing a scheduler created on another thread or Ractor.

    Examples:

    Run tasks on Async in a worker thread

    require "async"
require "farce"

# Using Thread as an executor as at the moment Async cannot run on a non-main Ractor
scheduler = Farce::Scheduler.create(Thread) { Async::Scheduler.new }
scheduler.schedule { puts "Running on Async" }

    Run tasks on CarbonFiber in a worker ractor

    require "carbon_fiber"
require "farce"

scheduler = Farce::Scheduler.create(Ractor) { CarbonFiber::Scheduler.new }
scheduler.schedule { puts "Running on CarbonFiber" }

    Parameters:

    • executor (Class, nil) (defaults to: nil) —

      Ractor or Thread. nil selects the runtime default. Choose an executor supported by the supplied scheduler and its dependencies.

    • name (String, nil) (defaults to: nil) —

      Optional worker name.

    • priority (Integer, nil) (defaults to: nil) —

      Optional worker thread priority.

    • capacity (Integer, nil) (defaults to: 1024) —

      Pending-task capacity, nil creates an unbounded queue.

    Yield Returns:

    • (Object) —

      The fiber scheduler to install in the worker.

    Returns:

See Also:



120
121
122
 
# File 'lib/farce/abstract/scheduler.rb', line 120

def self.create(executor = nil, name: nil, priority: nil, **, &)
  new(**, &).tap { it.launch(executor, name:, priority:) }
end

.current ⇒ Scheduler?

Returns a Farce scheduler for the current thread's installed fiber scheduler. Returns nil when no fiber scheduler is installed.

Creates and installs a wrapper when necessary, starting its dispatcher on that scheduler.

Additional keyword arguments are forwarded to #initialize when wrapping.

Examples:

Create a scheduler for Async, and send it to a different Ractor

require "async"
require "farce"

# No fiber scheduler set here
Farce::Scheduler.current # => nil

Async do
  # A scheduler that will hand tasks to Async::Scheduler
  scheduler = Farce::Scheduler.current # => #<Farce::Scheduler>

  # This looks like doing the same as nested Async { ... } calls or
  # Fiber.schedule { ... }, …
  scheduler.schedule { puts "Scheduled from the main Ractor" }

  # … but it can be called from other threads and ractors
  Ractor.new(scheduler) do |scheduler|
    scheduler.schedule { puts "Scheduled from a background Ractor" }
  end
end

Full circle

scheduler = Farce::Scheduler.new
Farce::Scheduler.current # => nil
Fiber.set_scheduler(scheduler)
Farce::Scheduler.current == scheduler # => true

Returns:



55
56
57
58
59
60
61
62
63
64
 
# File 'lib/farce/abstract/scheduler.rb', line 55

def self.current(**)
  return unless Fiber.respond_to?(:scheduler)
  return unless scheduler = Fiber.scheduler
  register = Internal::Storage.store_if_absent(self) { Internal::Storage.new }
  register.store_if_absent(scheduler) do
    wrapper = new(**, external: true)
    wrapper.__send__(:set_scheduler!, scheduler:)
    wrapper
  end
end

Instance Method Details

#alive? ⇒ Boolean

Returns Whether the dispatcher is in its :running state.

Returns:

  • (Boolean) —

    Whether the dispatcher is in its :running state.

See Also:



392
 
# File 'lib/farce/abstract/scheduler.rb', line 392

def alive? = @state.value == :running

#capacity ⇒ BasicObject

Pending-task capacity. Nil means the queue is unbounded.



173
 
# File 'lib/farce/abstract/scheduler.rb', line 173

def capacity = @queue.capacity

#capacity=(value) ⇒ BasicObject

Resize the pending queue without discarding tasks. JRuby and TruffleRuby cannot switch between bounded and unbounded queues.



177
178
179
 
# File 'lib/farce/abstract/scheduler.rb', line 177

def capacity=(value)
  @queue.capacity = value.nil? ? nil : Integer(value)
end

#close ⇒ Scheduler

Requests shutdown and rejects further submissions. Repeated calls are harmless. The dispatcher drains queued tasks before closing the queue. This method does not wait for dispatch or running tasks to finish.

This wakes a dispatcher blocked on an empty queue. It does not directly close the underlying fiber scheduler.

Returns:



361
362
363
364
365
366
367
368
369
370
 
# File 'lib/farce/abstract/scheduler.rb', line 361

def close
  until closed?
    state = @state.value
    @state.compare_and_set(state, :closing) unless CLOSED_STATES.include?(state)
  end
  # if the scheduler never acquired an owner, #schedule might be blocked
  @owner.compare_and_set(nil, Ractor.current)
  @queue_owner ? @queue.seal : wake_dispatcher
  self
end

#closed? ⇒ Boolean

Returns Whether new submissions are rejected, including while closing or after a dispatch error. Does not imply all tasks have finished.

Returns:

  • (Boolean) —

    Whether new submissions are rejected, including while closing or after a dispatch error. Does not imply all tasks have finished.

See Also:



388
 
# File 'lib/farce/abstract/scheduler.rb', line 388

def closed? = CLOSED_STATES.include?(@state.value)

#error ⇒ Exception?

Returns The error encountered by the scheduler, if any.

Returns:

  • (Exception, nil) —

    The error encountered by the scheduler, if any.



182
183
184
185
 
# File 'lib/farce/abstract/scheduler.rb', line 182

def error
  value = @error.value
  value.is_a?(Envelope) ? value.value : value
end

#external_scheduler ⇒ Object?

Retrieves the constructed external scheduler from its owning Ractor.

Returns:

  • (Object, nil) —

    The external scheduler, or nil before setup or when using the built-in scheduler.

Raises:



299
300
301
302
303
 
# File 'lib/farce/abstract/scheduler.rb', line 299

def external_scheduler
  return unless wraps_external?
  return Internal::Storage[self] if owner.nil? || owner == Ractor.current
  raise Ractor::IsolationError, "external_scheduler is owned by a different Ractor"
end

#launch(executor = nil, name: nil, priority: nil) ⇒ Ractor, ...

Starts dispatching tasks in a new worker and waits for setup to leave its launching/setup states before returning. Returns the created worker instance.

Examples:

Using the return value to assign a ThreadGroup

scheduler = Farce::Scheduler.new
worker    = scheduler.launch_thread
group     = ThreadGroup.new

group.add(worker)
group.enclose

Using a custom worker

class MyWorker
  def initialize(...)
    # flip a coin whether we're using Ractor or Thread
    @worker = [Ractor, Thread].sample.new(...)
  end

  def join = @worker.join
end

scheduler = Farce::Scheduler.new
scheduler.launch(MyWorker)

Parameters:

  • executor (Class, nil) (defaults to: nil) —

    Ractor or Thread. Defaults to Ractor when native Ractors are available, otherwise Thread.

  • name (String, nil) (defaults to: nil) —

    Optional worker name.

  • priority (Integer, nil) (defaults to: nil) —

    Optional worker thread priority.

Returns:

  • (Ractor, Thread, Object) —

    The worker.

Raises:

  • (RuntimeError) —

    If this handle has already been launched, installed, or closed.



231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
 
# File 'lib/farce/abstract/scheduler.rb', line 231

def launch(executor = nil, name: nil, priority: nil)
  raise "Scheduler may not be reused" unless @state.compare_and_set(:initialized, :launching)

  executor ||= Internal.native_ractors? ? Ractor : Thread
  priority &&= Integer(priority)
  name     &&= -String(name)
  callback   = launch_callback

  return executor.new(self, priority, name, &callback) unless executor <= Ractor
  # Requiring an autoloaded implementation from a new Ractor can deadlock
  # while its launcher waits for setup to finish.
  Internal.const_get(:FiberScheduler, false)
  Internal.const_get(:Storage, false)
  Internal.const_get(:ThreadPool, false)
  executor.new(self, priority, name, name:, &callback)
rescue Exception => e # rubocop:disable Lint/RescueException
  @state.value = :error
  @error.value = Envelope.new(e)
  raise
ensure
  # if we return the thread too early, someone could kill it while another threads blocks on a schedule call
  if System.windows? && executor <= Ractor
    sleep 0.001 while @state.value == :launching
    sleep 0.001 while @state.value == :setup
  else
    @state.wait_until_changed(:launching)
    @state.wait_until_changed(:setup)
  end
end

#launch_ractor(name: nil, priority: nil) ⇒ Ractor

Launches a Ractor worker. Accepts the keyword arguments of #launch.

Parameters:

  • name (String, nil) (defaults to: nil) —

    Optional worker name.

  • priority (Integer, nil) (defaults to: nil) —

    Optional worker thread priority.

Returns:



192
 
# File 'lib/farce/abstract/scheduler.rb', line 192

def launch_ractor(...) = launch(Ractor, ...)

#launch_thread(name: nil, priority: nil) ⇒ Thread

Launches a Thread worker. Accepts the keyword arguments of #launch.

Parameters:

  • name (String, nil) (defaults to: nil) —

    Optional worker name.

  • priority (Integer, nil) (defaults to: nil) —

    Optional worker thread priority.

Returns:

  • (Thread) —

    The worker.



199
 
# File 'lib/farce/abstract/scheduler.rb', line 199

def launch_thread(...) = launch(Thread, ...)

#local?(wait: true) ⇒ Boolean

Returns Whether the executor is local for the current Ractor.

Parameters:

  • wait (Boolean) (defaults to: true) —

    Whether to wait for the owner to be non-nil before checking if the executor is local.

Returns:

  • (Boolean) —

    Whether the executor is local for the current Ractor.



349
350
351
352
 
# File 'lib/farce/abstract/scheduler.rb', line 349

def local?(wait: true)
  owner = wait ? @owner.wait_until_non_nil : @owner.value
  owner == Ractor.current
end

#owner ⇒ Ractor?

Returns The owning Ractor, or nil before ownership is assigned.

Returns:

  • (Ractor, nil) —

    The owning Ractor, or nil before ownership is assigned.



262
 
# File 'lib/farce/abstract/scheduler.rb', line 262

def owner = @owner.value

#owner=(value) ⇒ BasicObject

Assigns ownership during setup. An existing owner cannot be changed. This is used by schedule to determine if a task is scheduled locally or remotely.

If #schedule is called without setting auto_local to false, it will block until the owner can be determined. Setting the owner explicitly avoids blocking on ownership determination.

Examples:

scheduler = Farce::Scheduler.new

# Scheduler hasn't been launched, so it could block.
scheduler.owner = Ractor.current

# Now this doesn't block
scheduler.schedule { some_task }

# Can use #launch_thread, but #launch_ractor would fail now that the owner is set.
scheduler.launch_thread

Parameters:

Raises:

  • (ArgumentError) —

    If value is not a Ractor/nil or changes an assigned owner.



284
285
286
287
288
289
 
# File 'lib/farce/abstract/scheduler.rb', line 284

def owner=(value)
  raise ArgumentError, "owner must be a Ractor" unless value.nil? || value.is_a?(Ractor)
  return if @owner.compare_and_set(nil, value)
  return if owner == value
  raise ArgumentError, "owner has already been set"
end

#schedule(*args, mode: :smart, auto_local: true, timeout: nil) {|*args| ... } ⇒ self?

Enqueues a task, waiting for queue space if necessary. Returns after submission. It does not wait for task completion or return the task's value.

With auto_local enabled, waits for ownership to be assigned and uses :local mode whenever the caller and owner are in the same Ractor, including different threads in that Ractor.

Local tasks retain their block and arguments.

Otherwise, the block must be convertible to a Ractor-shareable proc. Pass task data as arguments so the selected transfer mode can be applied to it.

A timeout limits the combined wait for ownership and queue space. Expiry returns nil without enqueuing the task. Accepted tasks have no execution timeout.

Valid modes are:

  • :smart - Applies smart rules or a dedicated proxy. Raises for unsupported values. This is the default for most mode-backed classes.
  • :smart_raise - Applies smart rules or a dedicated proxy, with a raising fallback.
  • :smart_copy - Applies smart rules or a dedicated proxy, then falls back to copying or Marshal. This is the default for Port.
  • :smart_move - Applies smart rules or a dedicated proxy, then falls back to moving.
  • :copy - The value will be copied between Ractors.
  • :marshal - The value will be serialized immediately and reconstructed once per Ractor with Ruby's Marshal protocol. Contents must support Marshal. Nested shareable values may be reconstructed too.
  • :make_shareable - The value will be made Ractor-shareable using Ractor.make_shareable.
  • :move - The value will be moved between Ractors. This saves memory compared to copying, and supports values that can't be copied but moved (like IO objects). However, the value will no longer be accessible on the Ractor that pushed it.
  • :mutable - A Mutable instance will be created for the value. This isn't done recursively and thus will fail for nested unshareable values.
  • :local - The value will be kept local to the Ractor that pushed it. Another ractor trying to receive it will get an error. Useful for usage contained within a single Ractor.
  • :proxy - The value will be wrapped in a Farce::Proxy that executes calls in the original Ractor.
  • :raise - An error will be raised if the value is not Ractor-shareable. Useful for enforcing shareability.
  • :dedup - The value will be deduplicated using Farce.dedup, then made Ractor-shareable. This may update and freeze the original. Already-shareable values pass through unchanged.
  • :shareable_copy - The value will be copied and the copy will be made Ractor-shareable.

Parameters:

  • args (Array<Object>) —

    Positional arguments passed to the task block.

  • mode (Symbol) (defaults to: :smart) —

    Argument transfer mode from MODES, including :copy, :marshal, and :local. See ModeManager#wrap.

  • auto_local (Boolean) (defaults to: true) —

    Whether to override mode with :local in the owning Ractor.

  • timeout (Numeric, nil) (defaults to: nil) —

    Maximum scheduling wait in seconds. Nil waits indefinitely, zero polls.

Yields:

  • (*args) —

    The task to execute.

Returns:

  • (self, nil) —

    self when submitted, or nil when the scheduling timeout expires.

Raises:

See Also:



333
 
# File 'lib/farce/abstract/scheduler.rb', line 333

def schedule(...) = enqueue_task(false, ...)

#state ⇒ Symbol

Returns The current state of the scheduler. One of:

  • :initialized: The scheduler has been created but not yet launched.
  • :launching: The scheduler is in the process of being launched on a background worker.
  • :setup: The scheduler is in the process of being setting up a fiber scheduler.
  • :running: The scheduler is actively dispatching tasks.
  • :closing: The scheduler has been requested to shut down and is draining its queue.
  • :closed: The scheduler has finished shutting down.
  • :error: The scheduler encountered an error during dispatch.

Returns:

  • (Symbol) —

    The current state of the scheduler. One of:

    • :initialized: The scheduler has been created but not yet launched.
    • :launching: The scheduler is in the process of being launched on a background worker.
    • :setup: The scheduler is in the process of being setting up a fiber scheduler.
    • :running: The scheduler is actively dispatching tasks.
    • :closing: The scheduler has been requested to shut down and is draining its queue.
    • :closed: The scheduler has finished shutting down.
    • :error: The scheduler encountered an error during dispatch.


381
 
# File 'lib/farce/abstract/scheduler.rb', line 381

def state = @state.value

#try_schedule(*args, mode: :smart, auto_local: true) ⇒ self?

Enqueues a task without waiting for queue space. Arguments and transfer modes follow #schedule. With auto_local enabled, it still waits for ownership to be assigned.

Returns self when submitted, or nil when the queue is full.

Parameters:

  • args (Array) —

    Arguments to pass to the scheduled block. May be handled according to mode.

  • mode (Symbol) (defaults to: :smart) —

    The mode in which to schedule the task. Defaults to :smart. May be ignored by some executors.

  • auto_local (Boolean) (defaults to: true) —

    Whether to automatically use a local mode when possible. Defaults to true. May be ignored by some executors.

Returns:

  • (self, nil) —

    self when submitted, or nil when the queue is full.



344
345
346
 
# File 'lib/farce/abstract/scheduler.rb', line 344

def try_schedule(*, mode: :smart, auto_local: true, &)
  enqueue_task(true, *, mode:, auto_local:, &)
end

#wraps_external? ⇒ Boolean

Returns Whether a fiber scheduler constructor was supplied.

Returns:

  • (Boolean) —

    Whether a fiber scheduler constructor was supplied.



292
 
# File 'lib/farce/abstract/scheduler.rb', line 292

def wraps_external? = @external