Class: AsyncFutures::ThreadExecutor

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

Overview

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

ThreadExecutor specific submission considerations:

For ThreadExecutor, with no arguments passed, the tasks are not 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 and whether the strict_concurrency argument is passed.

Instance Method Summary collapse

Methods included from Executor

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

Constructor Details

#initialize(max_workers: nil, strict_concurrency: false, reap_after: nil, worker_name_prefix: nil) ⇒ ThreadExecutor

Create a new ThreadExecutor.

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 strict_concurrency argument changes the behavior of both submit and submit_concurrent. When the argument is false then both methods will just put tasks on a queue and assume they will get picked up later. When the argument is true then the following happens:

  • submit: if adding another task to the queue would make more tasks than available (or potential) workers, then the task is run immediately and a completed future is returned to the caller.
  • submit_concurrent: if adding another task to the queue would make more tasks than available (or potential) workers, then a NoConcurrencyError is raised.

With strict_concurrency: false you can do interesting/dangerous things. For example, you can add tasks to the executor from within a executor worker thread, even if the max worker count is only 1. If you think through this scenario you will realize that joining on the returned future will deadlock the worker thread.

This defaults to false precisely because scenarios like this are uncommon. The most common scenario is firing of many tasks from the main thread of execution that do not interact other than to return a value to the main thread.

A scenario where you might want strict_concurrency to be true: you have client and server tasks and they must run concurrent to each other in order to work correctly.

Consider this pseudocode, for example:

ThreadExecutor.new(max_workers: 1, strict_concurrency: true).shutdown do |executor|
  # ... other work, potentially using executor ...

  # This can fail if all workers are busy. Good! We want it to.
  # It doesn't make sense to run the client code afterward
  # if the server isn't first running concurrently.
  executor.submit_concurrent { Server.new.listen() }

  # The client doesn't *need* to run concurrently;
  # It is logically correct to run it either concurrently OR immediately,
  # so we use `submit` instead of `submit_concurrent` for client code.
  executor.submit do
    client = Client.new
    client.ping()
    # ... interact with `client` while server runs in background concurrently ...
  ensure
    client.signal_server_shutdown()
  end

  # ... maybe more code after ...
end

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 ThreadExecutor instance is shutdown.

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



106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
 
# File 'lib/async_futures/thread_executor.rb', line 106

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

  # Set Hash value to `true` when a worker is running
  # and `false` otherwise.
  @pool = {}
  @worker_count = 0

  at_exit { shutdown(wait: false) }
end

Instance Method Details

#pool_size ⇒ Object

Return the current size of the worker pool



163
164
165
 
# File 'lib/async_futures/thread_executor.rb', line 163

def pool_size
  synchronize { @pool.size }
end

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

Shutdown ThreadExecutor instance.

See AsyncFutures::Executor.shutdown for full documentation.



180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
 
# File 'lib/async_futures/thread_executor.rb', line 180

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

Asynchronously submit a task for execution.

May run task immediately and return a completed Future under certain circumstances.

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

Raises:

  • (ArgumentError) 


134
135
136
137
138
139
140
141
142
 
# File 'lib/async_futures/thread_executor.rb', line 134

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

  Future.new.tap do |f|
    f.complete(*args, **kwargs, &block) unless queue_task(f, *args, **kwargs, &block)
  rescue ClosedQueueError
    raise 'ThreadExecutor instance is shutdown'
  end
end

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

Submit a task for concurrent execution.

Will raise NoConcurrencyError if it is not possible to run the task concurrently with other already scheduled tasks.

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

Raises:

  • (ArgumentError) 


152
153
154
155
156
157
158
159
160
 
# File 'lib/async_futures/thread_executor.rb', line 152

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

  Future.new.tap do |f|
    raise NoConcurrencyError.new('Tasks exceed potential workers') unless queue_task(f, *args, **kwargs, &block)
  rescue ClosedQueueError
    raise 'ThreadExecutor instance is shutdown'
  end
end

#support_concurrency? ⇒ Boolean

Always returns true for ThreadExecutor.

Returns:

  • (Boolean) 


171
172
173
 
# File 'lib/async_futures/thread_executor.rb', line 171

def support_concurrency?
  true
end