Class: AsyncFutures::RactorExecutor

Inherits:
Object
  • Object
show all
Includes:
Executor
Defined in:
lib/async_futures/ractor_executor.rb

Overview

Executor implementation based on Thread primitives that uses a pool of up to max_workers to execute calls concurrently.

RactorExecutor specific submission considerations:

For RactorExecutor the tasks are never run immediately upon submission. They are placed into a work queue to be picked up later by worker threads.

This does not guarantee that any particular task will be run concurrently with any other particular task; that is dependent on how many worker threads and tasks there are at any given point in time.

Instance Method Summary collapse

Methods included from Executor

#map, map, shutdown, submit, #submit_concurrent, submit_concurrent, support_concurrency?

Constructor Details

#initialize(max_workers: nil, worker_name_prefix: nil, move_result: false, move_args: false, make_args_shareable: false, copy_args: false) ⇒ RactorExecutor

Create a new RactorExecutor.

Uses a pool of up to max_workers to execute tasks concurrently. If no value is given for max_workers it will default to [32, Etc.nprocessors + 4].min. Workers are spawned lazily as needed when tasks are added to the work queue.

The parameter worker_name_prefix can be used to optionally add a prefix to generated Thread names.

If the move_result keyword argument is true, results from worker ractors will be moved instead of copied. Moving is faster than copying, but less safe if the worker ractor keeps the values around for some reason (in a cache, for example). If you aren't doing something like caching inside workers you are probably safe to set this to true.

If the move_args keyword argument is true, args and kwargs will be moved instead of copied from the submitting ractor to the worker ractors. Moving is faster than copying, but is even less safe than move_result because the submitting ractor is more likely to have kept references to the submitted values. You should only set this to true if you are absolutely certain that submitted values have no remaining references in the submitting ractor otherwise the submitting ractor will error when accessing them later.



66
67
68
69
70
71
72
73
74
75
76
77
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
 
# File 'lib/async_futures/ractor_executor.rb', line 66

def initialize( # rubocop:disable Metrics/AbcSize,Metrics/ParameterLists
  max_workers: nil,
  worker_name_prefix: nil,
  move_result: false,
  move_args: false,
  make_args_shareable: false,
  copy_args: false
)
  if copy_args && !make_args_shareable
    raise ArgumentError.new('`copy_args` cannot be true unless `make_args_shareable` is also true')
  end

  @max_workers = (max_workers || [32, Etc.nprocessors + 4].min).to_i
  @worker_name_prefix = worker_name_prefix

  # This value is passed into worker Ractors.
  # If the caller passed something not shareable,
  # it would error when we spawn workers later.
  # Boolean values are always safely shareable
  # and since we only care about the truthiness of this value
  # double negation makes sense here.
  @move_result = !!move_result # rubocop:disable Style/DoubleNegation

  @move_args = move_args
  @make_args_shareable = make_args_shareable
  @copy_args = copy_args
  @mutex = Thread::Mutex.new
  @condition = Thread::ConditionVariable.new
  @tasks = Thread::Queue.new
  @worker_tasks_ports = Thread::Queue.new

  # All private variables after this point
  # require synchronization to safely interact with.
  @results_ports = {}
  @worker_to_task_port_map = ObjectSpace::WeakKeyMap.new
  @futures = {}

  @pool = Set.new
  @worker_count = 0

  @task_feeder = nil
  @result_feeder = nil

  # The inter-thread communication between these is necessary for shutdown,
  # so even if nothing is submitted, we still need these to exist for now.
  maybe_spawn_task_feeder
  maybe_spawn_result_feeder

  at_exit { shutdown(wait: false) }
end

Instance Method Details

#shutdown(wait: true, cancel_futures: false, &block) ⇒ Object

Shutdown RactorExecutor instance.

See AsyncFutures::Executor.shutdown for full documentation.



157
158
159
160
161
162
163
164
165
166
167
168
169
170
 
# File 'lib/async_futures/ractor_executor.rb', line 157

def shutdown(wait: true, cancel_futures: false, &block)
  block&.call(self)
ensure
  unless check_and_set_shutdown!
    if cancel_futures
      while (task = @tasks.pop)
        future = task[0]
        future.cancel
      end
    end

    synchronize { wait_until { @worker_tasks_ports.closed? && @worker_tasks_ports.empty? } } if wait
  end
end

#submit(*args, **kwargs, &block) ⇒ Object

Asynchronously submit a task for execution.

See AsyncFutures::Executor.submit method for full documentation.

Raises:

  • (ArgumentError) 


120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
 
# File 'lib/async_futures/ractor_executor.rb', line 120

def submit(*args, **kwargs, &block)
  raise ArgumentError.new('No block given') unless block

  Future.new.tap do |future|
    # Attempt to make everything shareable upon submit
    # so that if making shareable would raise an exception
    # the caller can know immediately
    # that a given value won't work.
    # Otherwise the errors could easily get swallowed in a background thread
    # and the caller would never know.
    sh_block = Ractor.shareable_proc(&block)
    sh_args = @make_args_shareable ? Ractor.make_shareable(args, copy: @copy_args) : args
    sh_kwargs = @make_args_shareable ? Ractor.make_shareable(kwargs, copy: @copy_args) : kwargs
    ractor_task = [future, sh_block, sh_args, sh_kwargs]

    @tasks.push(ractor_task)
    maybe_spawn_worker
    maybe_spawn_task_feeder
    maybe_spawn_result_feeder
  rescue ClosedQueueError
    raise 'RactorExecutor instance is shutdown'
  end
end

#support_concurrency? ⇒ Boolean

Always returns true for RactorExecutor.

Returns:

  • (Boolean) 


148
149
150
 
# File 'lib/async_futures/ractor_executor.rb', line 148

def support_concurrency?
  true
end