Class: AsyncFutures::FiberExecutor

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

Overview

Executor implementation based on Fiber primitives. Requires that Fiber.scheduler be set in order to work.

Several benefits of using FiberExecutor over using Fiber.schedule directly:

  • By default Fiber instances run via Fiber.schedule have no straightforward way of returning their final result upon completion like Thread and Ractor do (both support calling value to get the final result). Future makes this trivial to do for a scheduled Fiber.
  • Fiber instances cannot currently be shared across Thread instances (though this may change someday). However, Future instances can safely be shared across both Threads and Fibers (Ractors can share neither Fiber nor Future and that is unlikely to ever change due to their design).

FiberExecutor specific details for submission:

For FiberExecutor the tasks are run immediately upon submission using the Fiber.schedule method. This method will return as soon as the Fiber hits a blocking operation or runs the Fiber to completion. Thus it is completely possible that the returned Future is already completed by the time it is returned to the caller.

The FiberExecutor implementation does not guarantee that any particular task will be run concurrently with any other particular task; that is dependent on whether the submitted procs/blocks have blocking operations that yield control back to the Fiber::Scheduler and whether the Fiber::Scheduler properly implements Fiber switching for those operations.

Instance Method Summary collapse

Methods inherited from Executor

#map, #submit_concurrent

Constructor Details

#initialize(treat_as_concurrent: false, worker_name_prefix: nil) ⇒ FiberExecutor

Create a new FiberExecutor.

Spawns fibers via Fiber.schedule.

Raises AsyncFutures::Error unless Fiber.scheduler is set.

Because auto-fibers do not yield control unless they encounter a blocking operation, it is completely possible that the Fiber runs to completion upon submission. Thus, submit_concurrent fails by default unless the parameter treat_as_concurrent is set to true.

All internal state is protected via mutex, so it is safe to use a single FiberExecutor instance across multiple threads. However each thread must have its own Fiber::Scheduler set in order to successfully call submit.

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

Raises:



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

def initialize(treat_as_concurrent: false, worker_name_prefix: nil)
  raise Error.new('No Fiber.scheduler set') unless Fiber.scheduler

  super(worker_name_prefix: worker_name_prefix)
  @treat_as_concurrent = treat_as_concurrent
  @futures = SynchronizedDelegator.new(Set.new)

  at_exit { shutdown(wait: false, cancel_futures: true) }
end

Instance Method Details

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

Shutdown FiberExecutor instance.

See AsyncFutures::Executor.shutdown for full documentation.



116
117
118
119
120
121
122
123
124
125
126
127
128
 
# File 'lib/async_futures/fiber_executor.rb', line 116

def shutdown(wait: true, cancel_futures: false) # rubocop:disable Metrics/CyclomaticComplexity
  yield(self) if block_given?
ensure
  at_first_shutdown do
    if wait || cancel_futures
      futures_dup = @futures.dup.to_set
      futures_dup.reject!(&:cancel) if cancel_futures

      # This will deadlock outside a FiberScheduler,
      futures_dup.reject!(&:join) if wait
    end
  end
end

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

Asynchronously submit a task for execution.

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

Raises:

  • (ArgumentError) 


83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
 
# File 'lib/async_futures/fiber_executor.rb', line 83

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

  Future.new.tap do |future|
    synchronize do
      refute_shutdown

      # Need to set this immediately to ensure DeadlockError is raised appropriately.
      future.thread = Thread.current
      future.add_done_callback { |f| @futures.delete(f) }
      @futures.add(future)
    end

    Fiber.schedule do
      AsyncFutures.worker_name = new_worker_name
      future.complete(*args, **kwargs, &block)
    end
  end
end

#support_concurrency? ⇒ Boolean

Return true if treat_as_concurrent was passed as true to the FiberExecutor constructor.

Otherwise, return false.

Returns:

  • (Boolean) 


109
110
111
 
# File 'lib/async_futures/fiber_executor.rb', line 109

def support_concurrency?
  !!@treat_as_concurrent
end