Class: SolidQueue::FiberPool

Inherits:
Pool
  • Object
show all
Defined in:
lib/solid_queue/fiber_pool.rb

Instance Attribute Summary

Attributes inherited from Pool

#size

Instance Method Summary collapse

Methods inherited from Pool

build, #idle?, #type

Methods included from AppExecutor

#create_thread, #handle_thread_error, #wrap_in_app_executor

Constructor Details

#initialize(size, on_idle: nil) ⇒ FiberPool

Returns a new instance of FiberPool.



5
6
7
8
9
10
11
12
13
14
 
# File 'lib/solid_queue/fiber_pool.rb', line 5

def initialize(size, on_idle: nil)
  super

  @state_mutex = Mutex.new
  @shutdown = false
  @fatal_error = nil
  @boot_queue = Thread::Queue.new
  @pending_executions = Thread::Queue.new
  @reactor_thread = nil
end

Instance Method Details

#available_capacity ⇒ Object 



23
24
25
26
 
# File 'lib/solid_queue/fiber_pool.rb', line 23

def available_capacity
  raise_if_fatal_error!
  super
end

#post(execution) ⇒ Object

Raises:

  • (RuntimeError) 


16
17
18
19
20
21
 
# File 'lib/solid_queue/fiber_pool.rb', line 16

def post(execution)
  raise_if_fatal_error!
  raise RuntimeError, "Execution pool is shutting down" if shutdown?

  super
end

#shutdown ⇒ Object 



28
29
30
31
32
33
34
35
36
37
38
 
# File 'lib/solid_queue/fiber_pool.rb', line 28

def shutdown
  state_mutex.synchronize do
    next false if @shutdown

    @shutdown = true
  end.tap do |shut_down|
    # Wake the reactor: already-queued executions are drained before the
    # blocked pop in +wait_for_executions+ returns nil
    pending_executions.close if shut_down
  end
end

#shutdown? ⇒ Boolean

Returns:

  • (Boolean) 


40
41
42
 
# File 'lib/solid_queue/fiber_pool.rb', line 40

def shutdown?
  state_mutex.synchronize { @shutdown }
end

#wait_for_termination(timeout) ⇒ Object 



44
45
46
 
# File 'lib/solid_queue/fiber_pool.rb', line 44

def wait_for_termination(timeout)
  reactor_thread&.join(timeout)
end