Class: AsyncFutures::ProcessExecutor

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

Overview

Executor implementation based on Process forking that uses up to max_workers to execute calls in parallel.

ProcessExecutor specific considerations:

The ProcessExecutor class is not required by default when loading the overall AsyncFutures gem.

# ProcessExecutor *NOT* loaded
require 'async_futures'

# ProcessExecutor loaded
require 'async_futures/process_executor'

This is because it depends on the 'base64' gem. This gem was bundled in Ruby 3.3 and prior, but was unbundled in 3.4 and later (even though it is still the Ruby core team that maintains this gem). One goal of AsyncFutures is to have no hard dependencies on code outside the standard library. Because this Executor does have a hard gem dependency, it is not loaded by default.

If you want to use this Executor in Ruby 3.4 or later, you will need to install the 'base64' gem as well.

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

Process workers are not reused for work like Threads and Ractors are. Each task gets a freshly forked process. This is because marshalling anonymous blocks is not trivial in Ruby; it is simpler to just fork after the block closure has been defined. Use ThreadExecutor or RactorExecutor for Executor implementations that support worker reuse.

Consequently, this executor is only really useful for expensive calculations where the startup time for a process is dwarfed by the time needed for the actual work. Although modern machines can fork a process thousands of times per second, this is very, very slow when machines can do billions of operations per second.

If RactorExecutor is available on your Ruby engine/version it is probably a better choice for parallel work.

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

Instance Method Summary collapse

Methods inherited from Executor

#map, #submit_concurrent

Constructor Details

#initialize(max_workers: nil, worker_name_prefix: nil, daemonize_workers: false) ⇒ ProcessExecutor

Create a new ProcessExecutor.

Uses a pool of up to max_workers to execute tasks in parallel. 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 worker names.

The parameter daemonize_workers, if set to true, causes workers to reparent under the init process and allow it to reap them. If set to false, this will cause the Executor instance to use Process.detach on the PID of each spawned worker, which will create an extra Ruby thread to reap the PID of each worker. It defaults to false.



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
 
# File 'lib/async_futures/process_executor.rb', line 91

def initialize(
  max_workers: nil,
  worker_name_prefix: nil,
  daemonize_workers: false
)
  super(worker_name_prefix: worker_name_prefix)

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

  # All private variables after this point
  # require synchronization to safely interact with.
  @futures = {}

  @pool = Set.new
  @pids = Set.new

  @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 { terminate_workers }
  at_exit { shutdown(wait: false, cancel_futures: true) }
end

Instance Method Details

#pool_size ⇒ Object 



147
148
149
 
# File 'lib/async_futures/process_executor.rb', line 147

def pool_size
  synchronize { @pool.size }
end

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

Shutdown ProcessExecutor instance.

See AsyncFutures::Executor.shutdown for full documentation.



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

def shutdown(wait: true, cancel_futures: false)
  yield(self) if block_given?
ensure
  at_first_shutdown do
    if cancel_futures
      while (task = @tasks.pop)
        future = task[0]
        future.cancel
      end
    end

    synchronize { wait_until { all_work_complete? } } 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) 


123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
 
# File 'lib/async_futures/process_executor.rb', line 123

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

  Future.new.tap do |future|
    task_ary = [future, block, args, kwargs]

    synchronize { @futures[future.object_id] = future } # rubocop:disable Lint/HashCompareByIdentity
    @tasks.push(task_ary)
    maybe_spawn_task_feeder
    maybe_spawn_result_feeder
  rescue ClosedQueueError
    synchronize { @futures.delete(future.object_id) }
    refute_shutdown
  end
end

#support_concurrency? ⇒ Boolean

Always returns true for ProcessExecutor.

Returns:

  • (Boolean) 


143
144
145
 
# File 'lib/async_futures/process_executor.rb', line 143

def support_concurrency?
  true
end