Class: Phronomy::TaskGroup

Inherits:
Object
  • Object
show all
Defined in:
lib/phronomy/engine/task_group.rb

Overview

Manages a bounded set of concurrent Tasks with structured concurrency.

Constant Summary collapse

FAILURE_POLICIES = 
%i[fail_fast collect_all skip_failed].freeze

Instance Method Summary collapse

Constructor Details

#initialize(runtime:, limit: Float::INFINITY, failure_policy: :fail_fast) ⇒ TaskGroup

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns a new instance of TaskGroup.

Parameters:

  • limit (Integer, Float::INFINITY) (defaults to: Float::INFINITY)
  • failure_policy (Symbol) (defaults to: :fail_fast)
  • runtime (Runtime) —

    runtime authority used to spawn every child Task



12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
 
# File 'lib/phronomy/engine/task_group.rb', line 12

def initialize(runtime:, limit: Float::INFINITY, failure_policy: :fail_fast)
  unless FAILURE_POLICIES.include?(failure_policy)
    raise ArgumentError, "unknown failure_policy: #{failure_policy}"
  end
  unless runtime
    raise ArgumentError, "runtime is required"
  end

  @limit = limit
  @failure_policy = failure_policy
  @runtime = runtime
  @tasks = []
  @mutex = Mutex.new
  @cond = ConditionVariable.new
  @active = 0
end

Instance Method Details

#active_task_count ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



155
156
157
 
# File 'lib/phronomy/engine/task_group.rb', line 155

def active_task_count
  @mutex.synchronize { @active }
end

#await_all ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



44
45
46
47
48
49
50
51
52
53
 
# File 'lib/phronomy/engine/task_group.rb', line 44

def await_all
  tasks = @mutex.synchronize { @tasks.dup }
  return [] if tasks.empty?

  if Phronomy::Runtime::Scheduler.current
    _await_all_cooperative(tasks)
  else
    _await_all_threaded(tasks)
  end
end

#cancel_all! ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
 
# File 'lib/phronomy/engine/task_group.rb', line 132

def cancel_all!
  tasks = @mutex.synchronize { @tasks.dup }
  tasks.each(&:cancel!)
  tasks.each do |task|
    task.join
  rescue
    nil
  end

  scheduler = Phronomy::Runtime::Scheduler.current
  if scheduler && @coop_signal
    @active = 0
    scheduler.raise_signal_all(@coop_signal)
  else
    @mutex.synchronize do
      @active = 0
      @cond.broadcast
    end
  end
  self
end

#spawn(&block) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



30
31
32
33
34
35
36
37
38
39
40
41
 
# File 'lib/phronomy/engine/task_group.rb', line 30

def spawn(&block)
  wait_for_slot!

  task = @runtime.spawn(name: "task-group-worker") do
    block.call
  ensure
    release_slot!
  end

  @mutex.synchronize { @tasks << task }
  task
end