Class: Farce::Abstract::Executor Abstract

Inherits:
Object
  • Object
show all
Includes:
Internal::Noncopyable
Defined in:
lib/farce/abstract/executor.rb

Overview

This class is abstract.

Use a scheduler or pool class.

A task executor that can submit work or wait for it to finish.

#schedule submits a task. #execute waits for the task to finish and may run it directly when execution is local. Concrete executors decide where tasks run and how their arguments are transferred.

Direct Known Subclasses

Pool, Scheduler, ThreadExecutor

Instance Method Summary collapse

Methods included from Internal::Noncopyable

#duplicable?

Instance Method Details

#close ⇒ self

This method is abstract.

This method must be implemented by concrete executor subclasses.

Closes the executor, releasing any resources it holds. Once closed, the executor cannot be used to schedule new tasks. Will drain any pending tasks before fully closing. If tasks are running indefinitely, it may block.

Returns:

  • (self) 


78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
 
# File 'lib/farce/abstract/executor.rb', line 78

class Executor
  include Internal::Noncopyable

  # An error that occurred during the executor's operation, if any.
  # This is primarily for scheduling errors, not execution errors.
  # You can check this if the {#state} is `:error`.
  def error = nil

  # @param wait [Boolean] Whether to wait for the owner to be non-nil before checking if the executor is local.
  # @return [Boolean] Whether the executor is local for the current Ractor.
  def local?(wait: true) = false # rubocop:disable Lint/UnusedMethodArgument

  # Attempts to submit a task. Queue-based executors return `nil` when full
  # instead of waiting for space. Other executors behave like {#schedule}.
  #
  # @param args (see #schedule)
  # @param mode (see #schedule)
  # @param auto_local (see #schedule)
  # @return [self, nil] self when submitted, or nil when the queue is full.
  def try_schedule(...) = schedule(...)

  # Closes the executor in every scope.
  # For executors without scopes, this behaves like {#close}.
  #
  # @return [self]
  def close_all = close

  # Like {#schedule}, with two differences:
  # 1. If {#local?} returns true, and it would run in local mode (either due to `auto_local` or explicitly set
  #    `mode: :local`), it executes the block directly without enqueuing it.
  # 2. It blocks until the block has been executed, either immediately in local mode or after being scheduled.
  # A `timeout` only limits scheduling waits. Once submitted, this waits for completion without a timeout.
  #
  # @param (see #schedule)
  # @return [self, nil] self after execution, or nil when the scheduling timeout expires.
  def execute(*, mode: :smart, auto_local: true, timeout: nil, &callback)
    raise LocalJumpError, "Cannot yield without a block" unless block_given?
    raise SchedulerClosedError, "cannot execute task on a closed scheduler" if closed?

    deadline = scheduling_deadline(timeout)
    if (auto_local || mode == :local) && local?(wait: timeout.nil?)
      raise SchedulerClosedError, "cannot execute task on a closed scheduler" if closed?
      yield(*)
      return self
    end

    ran      = Farce::Flag.new(false)
    signal   = Signal.new
    callback = Ractor.shareable_proc(&callback) unless Ractor.shareable?(callback)

    submitted = schedule(callback, ran, signal, *, mode:, auto_local:,
      timeout: Clock.remaining(deadline)) do |callback, ran, signal, *args|
      callback.call(*args)
    ensure
      ran.set
      signal.broadcast
    end
    return unless submitted

    signal.wait_until { ran.value }
    self
  end

  private

  def scheduling_deadline(timeout)
    return if timeout.nil?
    timeout = Float(timeout)
    unless timeout.finite? && !timeout.negative?
      raise ArgumentError, "timeout must be a finite, non-negative number or nil"
    end
    Clock.timeout(timeout)
  end
end

#close_all ⇒ self

Closes the executor in every scope. For executors without scopes, this behaves like #close.

Returns:

  • (self) 


103
 
# File 'lib/farce/abstract/executor.rb', line 103

def close_all = close

#closed? ⇒ Boolean

This method is abstract.

This method must be implemented by concrete executor subclasses.

Returns Whether the executor is closed.

Returns:

  • (Boolean) —

    Whether the executor is closed.



78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
 
# File 'lib/farce/abstract/executor.rb', line 78

class Executor
  include Internal::Noncopyable

  # An error that occurred during the executor's operation, if any.
  # This is primarily for scheduling errors, not execution errors.
  # You can check this if the {#state} is `:error`.
  def error = nil

  # @param wait [Boolean] Whether to wait for the owner to be non-nil before checking if the executor is local.
  # @return [Boolean] Whether the executor is local for the current Ractor.
  def local?(wait: true) = false # rubocop:disable Lint/UnusedMethodArgument

  # Attempts to submit a task. Queue-based executors return `nil` when full
  # instead of waiting for space. Other executors behave like {#schedule}.
  #
  # @param args (see #schedule)
  # @param mode (see #schedule)
  # @param auto_local (see #schedule)
  # @return [self, nil] self when submitted, or nil when the queue is full.
  def try_schedule(...) = schedule(...)

  # Closes the executor in every scope.
  # For executors without scopes, this behaves like {#close}.
  #
  # @return [self]
  def close_all = close

  # Like {#schedule}, with two differences:
  # 1. If {#local?} returns true, and it would run in local mode (either due to `auto_local` or explicitly set
  #    `mode: :local`), it executes the block directly without enqueuing it.
  # 2. It blocks until the block has been executed, either immediately in local mode or after being scheduled.
  # A `timeout` only limits scheduling waits. Once submitted, this waits for completion without a timeout.
  #
  # @param (see #schedule)
  # @return [self, nil] self after execution, or nil when the scheduling timeout expires.
  def execute(*, mode: :smart, auto_local: true, timeout: nil, &callback)
    raise LocalJumpError, "Cannot yield without a block" unless block_given?
    raise SchedulerClosedError, "cannot execute task on a closed scheduler" if closed?

    deadline = scheduling_deadline(timeout)
    if (auto_local || mode == :local) && local?(wait: timeout.nil?)
      raise SchedulerClosedError, "cannot execute task on a closed scheduler" if closed?
      yield(*)
      return self
    end

    ran      = Farce::Flag.new(false)
    signal   = Signal.new
    callback = Ractor.shareable_proc(&callback) unless Ractor.shareable?(callback)

    submitted = schedule(callback, ran, signal, *, mode:, auto_local:,
      timeout: Clock.remaining(deadline)) do |callback, ran, signal, *args|
      callback.call(*args)
    ensure
      ran.set
      signal.broadcast
    end
    return unless submitted

    signal.wait_until { ran.value }
    self
  end

  private

  def scheduling_deadline(timeout)
    return if timeout.nil?
    timeout = Float(timeout)
    unless timeout.finite? && !timeout.negative?
      raise ArgumentError, "timeout must be a finite, non-negative number or nil"
    end
    Clock.timeout(timeout)
  end
end

#error ⇒ BasicObject

An error that occurred during the executor's operation, if any. This is primarily for scheduling errors, not execution errors. You can check this if the #state is :error.



84
 
# File 'lib/farce/abstract/executor.rb', line 84

def error = nil

#execute(mode: :smart, auto_local: true, timeout: nil, &callback) ⇒ self?

Like #schedule, with two differences:

  1. If #local? returns true, and it would run in local mode (either due to auto_local or explicitly set mode: :local), it executes the block directly without enqueuing it.
  2. It blocks until the block has been executed, either immediately in local mode or after being scheduled. A timeout only limits scheduling waits. Once submitted, this waits for completion without a timeout.

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 after execution, or nil when the scheduling timeout expires.

Raises:

  • (LocalJumpError) 


113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
 
# File 'lib/farce/abstract/executor.rb', line 113

def execute(*, mode: :smart, auto_local: true, timeout: nil, &callback)
  raise LocalJumpError, "Cannot yield without a block" unless block_given?
  raise SchedulerClosedError, "cannot execute task on a closed scheduler" if closed?

  deadline = scheduling_deadline(timeout)
  if (auto_local || mode == :local) && local?(wait: timeout.nil?)
    raise SchedulerClosedError, "cannot execute task on a closed scheduler" if closed?
    yield(*)
    return self
  end

  ran      = Farce::Flag.new(false)
  signal   = Signal.new
  callback = Ractor.shareable_proc(&callback) unless Ractor.shareable?(callback)

  submitted = schedule(callback, ran, signal, *, mode:, auto_local:,
    timeout: Clock.remaining(deadline)) do |callback, ran, signal, *args|
    callback.call(*args)
  ensure
    ran.set
    signal.broadcast
  end
  return unless submitted

  signal.wait_until { ran.value }
  self
end

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



88
 
# File 'lib/farce/abstract/executor.rb', line 88

def local?(wait: true) = false # rubocop:disable Lint/UnusedMethodArgument

#ractor_shareable? ⇒ Boolean

This method is abstract.

Executors should include Shareable or Unshareable, which implement this method.

Returns Whether the executor is safe to use across Ractors.

Returns:

  • (Boolean) —

    Whether the executor is safe to use across Ractors.



78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
 
# File 'lib/farce/abstract/executor.rb', line 78

class Executor
  include Internal::Noncopyable

  # An error that occurred during the executor's operation, if any.
  # This is primarily for scheduling errors, not execution errors.
  # You can check this if the {#state} is `:error`.
  def error = nil

  # @param wait [Boolean] Whether to wait for the owner to be non-nil before checking if the executor is local.
  # @return [Boolean] Whether the executor is local for the current Ractor.
  def local?(wait: true) = false # rubocop:disable Lint/UnusedMethodArgument

  # Attempts to submit a task. Queue-based executors return `nil` when full
  # instead of waiting for space. Other executors behave like {#schedule}.
  #
  # @param args (see #schedule)
  # @param mode (see #schedule)
  # @param auto_local (see #schedule)
  # @return [self, nil] self when submitted, or nil when the queue is full.
  def try_schedule(...) = schedule(...)

  # Closes the executor in every scope.
  # For executors without scopes, this behaves like {#close}.
  #
  # @return [self]
  def close_all = close

  # Like {#schedule}, with two differences:
  # 1. If {#local?} returns true, and it would run in local mode (either due to `auto_local` or explicitly set
  #    `mode: :local`), it executes the block directly without enqueuing it.
  # 2. It blocks until the block has been executed, either immediately in local mode or after being scheduled.
  # A `timeout` only limits scheduling waits. Once submitted, this waits for completion without a timeout.
  #
  # @param (see #schedule)
  # @return [self, nil] self after execution, or nil when the scheduling timeout expires.
  def execute(*, mode: :smart, auto_local: true, timeout: nil, &callback)
    raise LocalJumpError, "Cannot yield without a block" unless block_given?
    raise SchedulerClosedError, "cannot execute task on a closed scheduler" if closed?

    deadline = scheduling_deadline(timeout)
    if (auto_local || mode == :local) && local?(wait: timeout.nil?)
      raise SchedulerClosedError, "cannot execute task on a closed scheduler" if closed?
      yield(*)
      return self
    end

    ran      = Farce::Flag.new(false)
    signal   = Signal.new
    callback = Ractor.shareable_proc(&callback) unless Ractor.shareable?(callback)

    submitted = schedule(callback, ran, signal, *, mode:, auto_local:,
      timeout: Clock.remaining(deadline)) do |callback, ran, signal, *args|
      callback.call(*args)
    ensure
      ran.set
      signal.broadcast
    end
    return unless submitted

    signal.wait_until { ran.value }
    self
  end

  private

  def scheduling_deadline(timeout)
    return if timeout.nil?
    timeout = Float(timeout)
    unless timeout.finite? && !timeout.negative?
      raise ArgumentError, "timeout must be a finite, non-negative number or nil"
    end
    Clock.timeout(timeout)
  end
end

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

This method is abstract.

This method must be implemented by concrete executor subclasses.

Schedules a block for execution. What actually happens depends on the concrete executor implementation. Some executors may ignore the auto_local option. Some might treat every mode as :local.

The executor may choose to block until the task has been executed, but most will add it to a queue of pending tasks, only blocking when that queue's capacity has been reached.

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) —

    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.

Yields:

  • (*args) —

    The block to be executed by the scheduler.

Yield Parameters:

  • args (Array) —

    The arguments passed to the block.

Returns:

  • (self, nil) —

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

Yield Receiver:

  • (BasicObject, nil) —

    Local mode will not touch the block's binding (self stays the same), but other modes may change it to nil.



78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
 
# File 'lib/farce/abstract/executor.rb', line 78

class Executor
  include Internal::Noncopyable

  # An error that occurred during the executor's operation, if any.
  # This is primarily for scheduling errors, not execution errors.
  # You can check this if the {#state} is `:error`.
  def error = nil

  # @param wait [Boolean] Whether to wait for the owner to be non-nil before checking if the executor is local.
  # @return [Boolean] Whether the executor is local for the current Ractor.
  def local?(wait: true) = false # rubocop:disable Lint/UnusedMethodArgument

  # Attempts to submit a task. Queue-based executors return `nil` when full
  # instead of waiting for space. Other executors behave like {#schedule}.
  #
  # @param args (see #schedule)
  # @param mode (see #schedule)
  # @param auto_local (see #schedule)
  # @return [self, nil] self when submitted, or nil when the queue is full.
  def try_schedule(...) = schedule(...)

  # Closes the executor in every scope.
  # For executors without scopes, this behaves like {#close}.
  #
  # @return [self]
  def close_all = close

  # Like {#schedule}, with two differences:
  # 1. If {#local?} returns true, and it would run in local mode (either due to `auto_local` or explicitly set
  #    `mode: :local`), it executes the block directly without enqueuing it.
  # 2. It blocks until the block has been executed, either immediately in local mode or after being scheduled.
  # A `timeout` only limits scheduling waits. Once submitted, this waits for completion without a timeout.
  #
  # @param (see #schedule)
  # @return [self, nil] self after execution, or nil when the scheduling timeout expires.
  def execute(*, mode: :smart, auto_local: true, timeout: nil, &callback)
    raise LocalJumpError, "Cannot yield without a block" unless block_given?
    raise SchedulerClosedError, "cannot execute task on a closed scheduler" if closed?

    deadline = scheduling_deadline(timeout)
    if (auto_local || mode == :local) && local?(wait: timeout.nil?)
      raise SchedulerClosedError, "cannot execute task on a closed scheduler" if closed?
      yield(*)
      return self
    end

    ran      = Farce::Flag.new(false)
    signal   = Signal.new
    callback = Ractor.shareable_proc(&callback) unless Ractor.shareable?(callback)

    submitted = schedule(callback, ran, signal, *, mode:, auto_local:,
      timeout: Clock.remaining(deadline)) do |callback, ran, signal, *args|
      callback.call(*args)
    ensure
      ran.set
      signal.broadcast
    end
    return unless submitted

    signal.wait_until { ran.value }
    self
  end

  private

  def scheduling_deadline(timeout)
    return if timeout.nil?
    timeout = Float(timeout)
    unless timeout.finite? && !timeout.negative?
      raise ArgumentError, "timeout must be a finite, non-negative number or nil"
    end
    Clock.timeout(timeout)
  end
end

#state ⇒ Symbol

This method is abstract.

This method must be implemented by concrete executor subclasses.

Returns The current state of the executor.

Common states include:

  • :initialized, :launching and :setup for the setup phase.
  • :running for the execution phase.
  • :closing, :closed, and :error for the termination phase.

Not all executors will implement every state.

Returns:

  • (Symbol) —

    The current state of the executor.

    Common states include:

    • :initialized, :launching and :setup for the setup phase.
    • :running for the execution phase.
    • :closing, :closed, and :error for the termination phase.

    Not all executors will implement every state.



78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
 
# File 'lib/farce/abstract/executor.rb', line 78

class Executor
  include Internal::Noncopyable

  # An error that occurred during the executor's operation, if any.
  # This is primarily for scheduling errors, not execution errors.
  # You can check this if the {#state} is `:error`.
  def error = nil

  # @param wait [Boolean] Whether to wait for the owner to be non-nil before checking if the executor is local.
  # @return [Boolean] Whether the executor is local for the current Ractor.
  def local?(wait: true) = false # rubocop:disable Lint/UnusedMethodArgument

  # Attempts to submit a task. Queue-based executors return `nil` when full
  # instead of waiting for space. Other executors behave like {#schedule}.
  #
  # @param args (see #schedule)
  # @param mode (see #schedule)
  # @param auto_local (see #schedule)
  # @return [self, nil] self when submitted, or nil when the queue is full.
  def try_schedule(...) = schedule(...)

  # Closes the executor in every scope.
  # For executors without scopes, this behaves like {#close}.
  #
  # @return [self]
  def close_all = close

  # Like {#schedule}, with two differences:
  # 1. If {#local?} returns true, and it would run in local mode (either due to `auto_local` or explicitly set
  #    `mode: :local`), it executes the block directly without enqueuing it.
  # 2. It blocks until the block has been executed, either immediately in local mode or after being scheduled.
  # A `timeout` only limits scheduling waits. Once submitted, this waits for completion without a timeout.
  #
  # @param (see #schedule)
  # @return [self, nil] self after execution, or nil when the scheduling timeout expires.
  def execute(*, mode: :smart, auto_local: true, timeout: nil, &callback)
    raise LocalJumpError, "Cannot yield without a block" unless block_given?
    raise SchedulerClosedError, "cannot execute task on a closed scheduler" if closed?

    deadline = scheduling_deadline(timeout)
    if (auto_local || mode == :local) && local?(wait: timeout.nil?)
      raise SchedulerClosedError, "cannot execute task on a closed scheduler" if closed?
      yield(*)
      return self
    end

    ran      = Farce::Flag.new(false)
    signal   = Signal.new
    callback = Ractor.shareable_proc(&callback) unless Ractor.shareable?(callback)

    submitted = schedule(callback, ran, signal, *, mode:, auto_local:,
      timeout: Clock.remaining(deadline)) do |callback, ran, signal, *args|
      callback.call(*args)
    ensure
      ran.set
      signal.broadcast
    end
    return unless submitted

    signal.wait_until { ran.value }
    self
  end

  private

  def scheduling_deadline(timeout)
    return if timeout.nil?
    timeout = Float(timeout)
    unless timeout.finite? && !timeout.negative?
      raise ArgumentError, "timeout must be a finite, non-negative number or nil"
    end
    Clock.timeout(timeout)
  end
end

#try_schedule ⇒ self?

Attempts to submit a task. Queue-based executors return nil when full instead of waiting for space. Other executors behave like #schedule.

Parameters:

  • args (Array) —

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

  • mode (Symbol) —

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

  • auto_local (Boolean) —

    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.



97
 
# File 'lib/farce/abstract/executor.rb', line 97

def try_schedule(...) = schedule(...)