Class: Farce::Local::Scheduler

Inherits:
Abstract::Scheduler show all
Includes:
Scoped, Shareable::Unfreezable, Internal::LocalExecutorMethods
Defined in:
lib/farce/local/scheduler.rb

Overview

A shareable handle with an independent task scheduler in each scope.

new leaves each scheduler unstarted. Launch it with Abstract::Scheduler#launch_thread, or install it with Fiber.set_scheduler. Scheduler.create starts a worker thread in each scope when that scope is first used. The default scope is :ractor.

Tasks retain their blocks and argument references. Abstract::Executor#execute runs directly in the calling fiber. A fiber scheduler constructor must be convertible to a shareable proc for use in each Ractor. #close closes the current scope. #close_all closes all scopes and prevents new schedulers from starting. Scheduler.current returns an Unshared::Scheduler.

Examples:

Start one worker in each Ractor

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

Instance Attribute Summary

Attributes included from Scoped

#scope

Class Method Summary collapse

Instance Method Summary collapse

Methods included from Shareable

#ractor_shareable?

Methods inherited from Abstract::Scheduler

#launch_ractor, #launch_thread, #local?

Methods inherited from Abstract::Executor

#execute, #local?, #ractor_shareable?

Methods included from Internal::Noncopyable

#duplicable?

Constructor Details

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

Creates a handle whose scoped schedulers remain unstarted. The constructor block builds the fiber scheduler when a scope is launched or installed. Valid scopes are:

  • :ractor - The value is shared by all threads within the same Ractor.
  • :thread_group - The value is shared by all threads within the same thread group.
  • :thread - The value is shared by all fibers within the same thread.
  • :fiber_storage - The value is shared by all fibers using the same storage. Unless explicitly specified, storage is inherited from the parent fiber for blocking fibers, but not for non-blocking fibers (such as those created by #schedule).
  • :fiber - Each fiber has its own independent value.

Parameters:

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

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

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

    IO backend for the built-in scheduler.

  • scope (Symbol) (defaults to: :ractor) —

    The scheduler scope. Defaults to :ractor.

Yield Returns:

  • (Object) —

    Fiber scheduler to install in each scope.



56
57
58
59
60
61
62
63
64
65
66
67
 
# File 'lib/farce/local/scheduler.rb', line 56

def initialize(capacity: 1024, backend: Internal::FROZEN_CONFIG.io_backend, scope: :ractor,
               worker: nil, &constructor)
  @constructor        = prepare_constructor(constructor || Internal::FROZEN_CONFIG.fiber_scheduler_constructor)
  @capacity           = capacity && Integer(capacity)
  @backend            = backend
  @worker             = worker
  @schedulers         = Strict::Map.new
  @closed_all         = Internal::Flag.new(false)
  @initialization_key = Object.new.freeze
  @registration_key   = Object.new.freeze
  super(scope:, capacity: capacity && Integer(capacity), backend:)
end

Class Method Details

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

Starts a worker in this scope and retains the launch settings for other scopes. Additional options and the block are forwarded to #initialize.

Parameters:

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

    Thread or a custom executor that stays in each 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 each worker.

Returns:

Raises:

  • (ArgumentError) 


42
43
44
45
46
 
# File 'lib/farce/local/scheduler.rb', line 42

def self.create(executor = Thread, name: nil, priority: nil, **, &)
  raise ArgumentError, "Local schedulers cannot launch Ractor workers" if executor && executor <= Ractor
  worker = [executor || Thread, name && -String(name), priority && Integer(priority)].freeze
  new(**, worker:, &)
end

.current ⇒ Unshared::Scheduler?

Returns the current thread's unshared scheduler, or nil when none is installed.

Returns:



33
 
# File 'lib/farce/local/scheduler.rb', line 33

def self.current(**) = Unshared::Scheduler.current(**)

Instance Method Details

#alive? ⇒ Boolean

Returns Whether the current scope is dispatching tasks.

Returns:

  • (Boolean) —

    Whether the current scope is dispatching tasks.



133
 
# File 'lib/farce/local/scheduler.rb', line 133

def alive? = current_scheduler&.alive? || false

#capacity ⇒ BasicObject

Pending-task capacity in this scope. New scopes retain the constructor value.



70
 
# File 'lib/farce/local/scheduler.rb', line 70

def capacity = (scheduler = current_scheduler) ? scheduler.capacity : @capacity

#capacity=(value) ⇒ BasicObject

Resize the queue in this scope without starting a worker.

Raises:



73
74
75
76
 
# File 'lib/farce/local/scheduler.rb', line 73

def capacity=(value)
  raise SchedulerClosedError, "cannot configure a closed scheduler" if @closed_all.value
  scoped_value.capacity = value
end

#close ⇒ self

Closes the current scope and rejects new submissions there.

Returns:

  • (self) 


104
105
106
107
 
# File 'lib/farce/local/scheduler.rb', line 104

def close
  current_scheduler&.close
  self
end

#close_all ⇒ self

Requests shutdown in every scope and Ractor and rejects future scopes. Does not wait for queued or running tasks to finish.

Returns:

  • (self) 


112
113
114
115
116
117
118
119
120
121
 
# File 'lib/farce/local/scheduler.rb', line 112

def close_all
  return self unless @closed_all.compare_and_set(false, true)
  @schedulers.each_pair do |scheduler, queue|
    queue.push([scheduler, :close].freeze, clock: 0)
  rescue ClosedQueueError
    # The owning Ractor may already have shut down its supervisor.
    nil
  end
  close
end

#closed? ⇒ Boolean

Returns Whether the current scope rejects new tasks.

Returns:

  • (Boolean) —

    Whether the current scope rejects new tasks.



130
 
# File 'lib/farce/local/scheduler.rb', line 130

def closed? = @closed_all.value || scoped_value.closed?

#error ⇒ Exception?

Returns The current scope's scheduler error.

Returns:

  • (Exception, nil) —

    The current scope's scheduler error.



127
 
# File 'lib/farce/local/scheduler.rb', line 127

def error = current_scheduler&.error

#external_scheduler ⇒ Object?

Returns The fiber scheduler in the current scope, when supplied by a constructor.

Returns:

  • (Object, nil) —

    The fiber scheduler in the current scope, when supplied by a constructor.



148
 
# File 'lib/farce/local/scheduler.rb', line 148

def external_scheduler = current_scheduler&.external_scheduler

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

Starts a worker for the current scope. Other scopes retain their creation behavior.

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.



96
97
98
99
100
 
# File 'lib/farce/local/scheduler.rb', line 96

def launch(executor = Thread, name: nil, priority: nil)
  raise ArgumentError, "Local schedulers cannot launch Ractor workers" if executor && executor <= Ractor
  raise "Scheduler may not be reused" if @closed_all.value
  active_scoped_value.launch(executor, name:, priority:)
end

#owner ⇒ Ractor

Returns The current scope's owning Ractor.

Returns:

  • (Ractor) —

    The current scope's owning Ractor.



136
 
# File 'lib/farce/local/scheduler.rb', line 136

def owner = Ractor.current

#owner=(value) ⇒ BasicObject

Parameters:

  • value (Ractor) —

    The current Ractor.

Raises:

  • (ArgumentError) —

    If another owner is supplied.



140
141
142
 
# File 'lib/farce/local/scheduler.rb', line 140

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

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

Enqueue a task in the current scope, preserving its block and arguments. mode is ignored when auto_local is true. Otherwise, only local and smart modes are accepted.

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

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.



83
 
# File 'lib/farce/local/scheduler.rb', line 83

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

#state ⇒ Symbol

Returns The current scope's scheduler state.

Returns:

  • (Symbol) —

    The current scope's scheduler state.



124
 
# File 'lib/farce/local/scheduler.rb', line 124

def state = current_scheduler&.state || :closed

#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.



86
87
88
 
# File 'lib/farce/local/scheduler.rb', line 86

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.



145
 
# File 'lib/farce/local/scheduler.rb', line 145

def wraps_external? = !@constructor.nil?