Class: AsyncFutures::ProcessExecutor

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

Overview

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

ProcessExecutor specific submission considerations:

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. Each task gets a freshly forked process. This is because marshalling anonymous blocks is not trivial; 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. If RactorExecutor is available on your Ruby version it is almost certainly a better choice than this.

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, #support_concurrency?, support_concurrency?

Constructor Details

#initialize(max_workers: nil, worker_name_prefix: '', reap_after: nil) ⇒ ProcessExecutor

Create a new ProcessExecutor.

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 reap_after keyword argument is given, worker threads will be shut down if they haven't received any work after this amount of seconds. If it is nil or not given, they will not be reaped until the ProcessExecutor instance is shutdown.



56
57
58
59
60
61
62
63
64
65
 
# File 'lib/async_futures/process_executor.rb', line 56

def initialize(max_workers: nil, worker_name_prefix: '', reap_after: nil)
  @max_workers = (max_workers || [32, Etc.nprocessors + 4].min).to_i
  @worker_name_prefix = worker_name_prefix.to_s
  @reap_after = reap_after
  @mutex = Thread::Mutex.new
  @tasks = Thread::Queue.new
  @pool = Set.new

  at_exit { shutdown(wait: false) }
end

Instance Method Details

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

Shutdown ProcessExecutor instance.

See AsyncFutures::Executor.shutdown for full documentation.



87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
 
# File 'lib/async_futures/process_executor.rb', line 87

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

    if wait
      synchronize { @pool.dup }.each do |thread|
        thread.join
        synchronize { @pool.delete(thread) }
      end
    end
  end
end

#submit(*args, **kwargs, &block) ⇒ Object Also known as: submit_concurrent

Asynchronously submit a task for execution.

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

Raises:

  • (ArgumentError) 


70
71
72
73
74
75
76
77
78
 
# File 'lib/async_futures/process_executor.rb', line 70

def submit(*args, **kwargs, &block)
  raise ArgumentError.new('No block given') unless block
  raise 'ProcessExecutor instance is shutdown' if @tasks.closed?

  Future.new.tap do |future|
    @tasks.push([future, block, args, kwargs])
    maybe_spawn_worker
  end
end