Class: Async::Barrier

Inherits:
Object
  • Object
show all
Defined in:
lib/async/barrier.rb

Overview

A barrier is used to synchronize multiple tasks, waiting for them all to complete before continuing.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(parent: nil) ⇒ Barrier

Returns a new instance of Barrier.



28
29
30
31
32
 
# File 'lib/async/barrier.rb', line 28

def initialize(parent: nil)
  @tasks = []
  
  @parent = parent
end

Instance Attribute Details

#tasks ⇒ Object (readonly)

All tasks which have been invoked into the barrier.



35
36
37
 
# File 'lib/async/barrier.rb', line 35

def tasks
  @tasks
end

Instance Method Details

#async(*arguments, parent: (@parent or Task.current), **options, &block) ⇒ Object 



41
42
43
44
45
46
47
 
# File 'lib/async/barrier.rb', line 41

def async(*arguments, parent: (@parent or Task.current), **options, &block)
  task = parent.async(*arguments, **options, &block)
  
  @tasks << task
  
  return task
end

#empty? ⇒ Boolean

Returns:

  • (Boolean) 


49
50
51
 
# File 'lib/async/barrier.rb', line 49

def empty?
  @tasks.empty?
end

#size ⇒ Object 



37
38
39
 
# File 'lib/async/barrier.rb', line 37

def size
  @tasks.size
end

#stop ⇒ Object 



72
73
74
75
76
 
# File 'lib/async/barrier.rb', line 72

def stop
  # We have to be careful to avoid enumerating tasks while adding/removing to it:
  tasks = @tasks.dup
  tasks.each(&:stop)
end

#wait ⇒ Object

Wait for all tasks.



55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
 
# File 'lib/async/barrier.rb', line 55

def wait
  # TODO: This would be better with linked list.
  while @tasks.any?
    task = @tasks.first
    
    begin
      task.wait
    ensure
      # We don't know for sure that the exception was due to the task completion.
      unless task.running?
        # Remove the task from the waiting list if it's finished:
        @tasks.shift if @tasks.first == task
      end
    end
  end
end