Class: Smith::Workflow::Parallel::ExecutionContext

Inherits:
Object
  • Object
show all
Extended by:
Dry::Initializer
Defined in:
lib/smith/workflow/parallel/execution_context.rb

Constant Summary collapse

THREAD_KEY = 
:smith_parallel_execution_context
DEPTH_KEY = 
:smith_parallel_execution_depth

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ ExecutionContext

Returns a new instance of ExecutionContext.



20
21
22
23
24
25
26
 
# File 'lib/smith/workflow/parallel/execution_context.rb', line 20

def initialize(...)
  super
  @pending_top_level = top_level_branch_count
  @active_top_level = 0
  @reserved_nested_workers = 0
  @mutex = Mutex.new
end

Class Method Details

.current ⇒ Object 



28
29
30
 
# File 'lib/smith/workflow/parallel/execution_context.rb', line 28

def self.current
  Fiber[THREAD_KEY]
end

.current_depth ⇒ Object 



32
33
34
 
# File 'lib/smith/workflow/parallel/execution_context.rb', line 32

def self.current_depth
  Fiber[DEPTH_KEY] || 0
end

Instance Method Details

#next_nesting_depth! ⇒ Object

Raises:



51
52
53
54
55
56
 
# File 'lib/smith/workflow/parallel/execution_context.rb', line 51

def next_nesting_depth!
  depth = self.class.current_depth + 1
  return depth if depth <= nesting_limit

  raise WorkflowError, "parallel nesting exceeds configured limit #{nesting_limit}"
end

#raise_if_cancelled! ⇒ Object 



47
48
49
 
# File 'lib/smith/workflow/parallel/execution_context.rb', line 47

def raise_if_cancelled!
  raise(signal.reason || Cancellation.new("cancelled")) if signal.cancelled?
end

#release_workers(count) ⇒ Object 



82
83
84
85
86
87
 
# File 'lib/smith/workflow/parallel/execution_context.rb', line 82

def release_workers(count)
  @mutex.synchronize do
    @reserved_nested_workers -= count
    validate_worker_count!
  end
end

#reserve_workers(requested) ⇒ Object 



73
74
75
76
77
78
79
80
 
# File 'lib/smith/workflow/parallel/execution_context.rb', line 73

def reserve_workers(requested)
  @mutex.synchronize do
    reserved = [requested, available_workers].min
    @reserved_nested_workers += reserved
    validate_worker_count!
    reserved
  end
end

#top_level_finished! ⇒ Object 



66
67
68
69
70
71
 
# File 'lib/smith/workflow/parallel/execution_context.rb', line 66

def top_level_finished!
  @mutex.synchronize do
    @active_top_level -= 1
    validate_worker_count!
  end
end

#top_level_started! ⇒ Object 



58
59
60
61
62
63
64
 
# File 'lib/smith/workflow/parallel/execution_context.rb', line 58

def top_level_started!
  @mutex.synchronize do
    @pending_top_level -= 1
    @active_top_level += 1
    validate_worker_count!
  end
end

#within(depth: self.class.current_depth) ⇒ Object 



36
37
38
39
40
41
42
43
44
45
 
# File 'lib/smith/workflow/parallel/execution_context.rb', line 36

def within(depth: self.class.current_depth)
  previous = self.class.current
  previous_depth = Fiber[DEPTH_KEY]
  Fiber[THREAD_KEY] = self
  Fiber[DEPTH_KEY] = depth
  yield
ensure
  Fiber[THREAD_KEY] = previous
  Fiber[DEPTH_KEY] = previous_depth
end