Class: Farce::Unshared::Scheduler

Inherits:
Abstract::Scheduler show all
Includes:
Farce::Unshareable, Internal::LocalExecutorMethods
Defined in:
lib/farce/unshared/scheduler.rb

Overview

A task scheduler that runs work within its creating Ractor.

Use Scheduler.create to start a background thread, or install an instance with Fiber.set_scheduler on the current thread. Tasks retain their blocks and argument references. The scheduler cannot be copied or moved to another Ractor.

auto_local: true ignores mode. With auto_local: false, only :local and smart modes are accepted. All accepted modes keep tasks local.

Examples:

Run tasks with captured local state

results   = Farce::Unshared::Queue.new
scheduler = Farce::Unshared::Scheduler.create
scheduler.schedule { results.push(:done) }
results.pop # => :done
scheduler.close

Class Method Summary collapse

Instance Method Summary collapse

Methods included from Farce::Unshareable

#ractor_shareable?

Methods inherited from Abstract::Scheduler

#alive?, #capacity, #capacity=, #close, #closed?, current, #error, #external_scheduler, #launch_thread, #local?, #owner, #state, #try_schedule, #wraps_external?

Methods inherited from Abstract::Executor

#close, #close_all, #closed?, #error, #execute, #local?, #ractor_shareable?, #state, #try_schedule

Methods included from Internal::Noncopyable

#duplicable?

Constructor Details

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

Create a scheduler without starting a worker. A constructor block builds the fiber scheduler when launched or installed. It may capture objects from this Ractor.

Parameters:

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

    Pending-task capacity, or nil for an unbounded queue.

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

    IO backend for the built-in scheduler.

Yield Returns:

  • (Object) —

    Fiber scheduler to install in the worker.



50
51
52
53
 
# File 'lib/farce/unshared/scheduler.rb', line 50

def initialize(...)
  super
  self.owner = Ractor.current
end

Class Method Details

.create(executor = Thread, name: nil, priority: nil) ⇒ Scheduler

Start a worker thread in this Ractor. Returns the running scheduler. Additional options and the block are forwarded to #initialize.

Parameters:

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

    Thread or a custom executor that stays in this Ractor.

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

    Optional worker name.

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

    Optional worker thread priority.

Yield Returns:

  • (Object) —

    Fiber scheduler to install in the worker.

Returns:



40
 
# File 'lib/farce/unshared/scheduler.rb', line 40

def self.create(executor = Thread, name: nil, priority: nil, **, &) = super

Instance Method Details

#launch(executor = Thread, name: nil, priority: nil) ⇒ Thread, Object

Start dispatching tasks in a new thread or a custom executor in this Ractor. Rejects Ractor executors before changing the scheduler's state.

Parameters:

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

    Thread or a custom executor that stays in this Ractor.

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

    Optional worker name.

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

    Optional worker thread priority.

Returns:

  • (Thread, Object) —

    The worker.

Raises:

  • (ArgumentError) —

    If a Ractor executor is supplied.



62
63
64
65
 
# File 'lib/farce/unshared/scheduler.rb', line 62

def launch(executor = Thread, name: nil, priority: nil)
  raise ArgumentError, "Unshared schedulers cannot launch Ractor workers" if executor && executor <= Ractor
  super(executor || Thread, name:, priority:)
end

#launch_ractor ⇒ BasicObject

Raises:

  • (ArgumentError) —

    Unshared schedulers cannot launch Ractor workers.



68
 
# File 'lib/farce/unshared/scheduler.rb', line 68

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

#owner=(value) ⇒ BasicObject

Parameters:

  • value (Ractor) —

    The creating Ractor.

Raises:

  • (ArgumentError) —

    If another Ractor is supplied or the assigned owner changes.



72
73
74
75
 
# File 'lib/farce/unshared/scheduler.rb', line 72

def owner=(value)
  raise ArgumentError, "owner must be the current Ractor" unless value == Ractor.current
  super
end

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

Enqueue a task while retaining its block and argument references. mode is ignored when auto_local is true. Otherwise, only local and smart modes are accepted.

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.

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

    Maximum seconds to wait for ownership and queue space. Nil waits indefinitely. Zero only submits if no waiting is needed. Must be finite and non-negative. An expired attempt returns nil and does not enqueue the task. This does not limit task execution time. Executors without a queue may ignore it.

Returns:

  • (self, nil) —

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



29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
 
# File 'lib/farce/unshared/scheduler.rb', line 29

class Scheduler < Abstract::Scheduler
  include Unshareable
  include Internal::LocalExecutorMethods

  # Start a worker thread in this Ractor. Returns the running scheduler.
  # Additional options and the block are forwarded to {#initialize}.
  # @param executor [Class, nil] Thread or a custom executor that stays in this Ractor.
  # @param name [String, nil] Optional worker name.
  # @param priority [Integer, nil] Optional worker thread priority.
  # @yieldreturn [Object] Fiber scheduler to install in the worker.
  # @return [Scheduler]
  def self.create(executor = Thread, name: nil, priority: nil, **, &) = super

  # Create a scheduler without starting a worker.
  # A constructor block builds the fiber scheduler when launched or installed.
  # It may capture objects from this Ractor.
  #
  # @overload initialize(capacity: 1024, backend: Farce.config.io_backend, &constructor)
  #   @param capacity [Integer, nil] Pending-task capacity, or nil for an unbounded queue.
  #   @param backend [Symbol] IO backend for the built-in scheduler.
  #   @yieldreturn [Object] Fiber scheduler to install in the worker.
  def initialize(...)
    super
    self.owner = Ractor.current
  end

  # Start dispatching tasks in a new thread or a custom executor in this Ractor.
  # Rejects Ractor executors before changing the scheduler's state.
  # @param executor [Class, nil] Thread or a custom executor that stays in this Ractor.
  # @param name [String, nil] Optional worker name.
  # @param priority [Integer, nil] Optional worker thread priority.
  # @return [Thread, Object] The worker.
  # @raise [ArgumentError] If a Ractor executor is supplied.
  def launch(executor = Thread, name: nil, priority: nil)
    raise ArgumentError, "Unshared schedulers cannot launch Ractor workers" if executor && executor <= Ractor
    super(executor || Thread, name:, priority:)
  end

  # @raise [ArgumentError] Unshared schedulers cannot launch Ractor workers.
  def launch_ractor(...) = launch(Ractor, ...)

  # @param value [Ractor] The creating Ractor.
  # @raise [ArgumentError] If another Ractor is supplied or the assigned owner changes.
  def owner=(value)
    raise ArgumentError, "owner must be the current Ractor" unless value == Ractor.current
    super
  end

  private

  def prepare_constructor(constructor) = constructor
  def new_queue(capacity:)             = Internal::UnsharedQueue.new(capacity:)
  def launch_callback                  = ->(_scheduler, priority, name) { launch!(name, priority) }

  def prepare_task(args, block, mode, auto_local, **)
    validate_local_mode(mode, auto_local)
    Internal::LocalScheduledTask.new(args, block)
  end
end