Class: Phronomy::Task

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

Overview

A thread-free asynchronous completion handle.

Task no longer executes work. Execution belongs to EventLoop/FSMSession or OffloadPool. Task only represents completion, failure, cancellation, callbacks and a blocking wait for external callers.

Constant Summary collapse

STATES = 
%i[pending completed failed cancelled].freeze

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(name: nil, parent: nil) ⇒ Task

Returns a new instance of Task.



20
21
22
23
24
25
26
27
28
29
30
31
 
# File 'lib/phronomy/engine/task.rb', line 20

def initialize(name: nil, parent: nil)
  @name = name
  @parent = parent
  @status = :pending
  @value = nil
  @error = nil
  @mutex = Mutex.new
  @cond = ConditionVariable.new
  @children = []
  @on_complete_callbacks = []
  parent&.register_child(self)
end

Instance Attribute Details

#name ⇒ Object (readonly)

Returns the value of attribute name.



18
19
20
 
# File 'lib/phronomy/engine/task.rb', line 18

def name
  @name
end

#parent ⇒ Object (readonly)

Returns the value of attribute parent.



18
19
20
 
# File 'lib/phronomy/engine/task.rb', line 18

def parent
  @parent
end

Class Method Details

.deferred(name: nil, parent: nil) ⇒ Object 



14
15
16
 
# File 'lib/phronomy/engine/task.rb', line 14

def self.deferred(name: nil, parent: nil)
  new(name: name, parent: parent)
end

Instance Method Details

#alive? ⇒ Boolean

Returns:

  • (Boolean) 


41
42
43
 
# File 'lib/phronomy/engine/task.rb', line 41

def alive?
  !done?
end

#cancel!(error = Phronomy::CancellationError.new("Task cancelled")) ⇒ Object 



125
126
127
128
129
130
131
132
 
# File 'lib/phronomy/engine/task.rb', line 125

def cancel!(error = Phronomy::CancellationError.new("Task cancelled"))
  changed = settle!(:cancelled, error: error)
  if changed
    children = @mutex.synchronize { @children.dup }
    children.each(&:cancel!)
  end
  self
end

#complete(value = nil) ⇒ Object 



116
117
118
 
# File 'lib/phronomy/engine/task.rb', line 116

def complete(value = nil)
  settle!(:completed, value: value)
end

#done? ⇒ Boolean

Returns:

  • (Boolean) 


37
38
39
 
# File 'lib/phronomy/engine/task.rb', line 37

def done?
  @mutex.synchronize { TERMINAL_STATES.include?(@status) }
end

#fail(error) ⇒ Object

Raises:

  • (ArgumentError) 


120
121
122
123
 
# File 'lib/phronomy/engine/task.rb', line 120

def fail(error)
  raise ArgumentError, "error is required" unless error
  settle!(:failed, error: error)
end

#join(limit = nil) ⇒ Object

Compatibility wait that does not re-raise the task error. Returns self when settled, nil on timeout.



76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
 
# File 'lib/phronomy/engine/task.rb', line 76

def join(limit = nil)
  if Phronomy::Runtime.in_event_loop_context? && !done?
    raise Phronomy::EventLoopReentrancyError,
      "Task#join cannot block the EventLoop thread; continue via an event"
  end

  deadline = limit && monotonic_now + limit.to_f
  @mutex.synchronize do
    until TERMINAL_STATES.include?(@status)
      if deadline
        remaining = deadline - monotonic_now
        return nil if remaining <= 0
        @cond.wait(@mutex, remaining)
      else
        @cond.wait(@mutex)
      end
    end
  end
  self
end

#map(&block) ⇒ Object

Raises:

  • (ArgumentError) 


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

def map(&block)
  raise ArgumentError, "map requires a block" unless block

  mapped = self.class.deferred(name: "#{@name}-mapped", parent: @parent)
  on_complete do |value, error|
    if error
      mapped.fail(error)
      next
    end

    begin
      mapped.complete(block.call(value))
    rescue => mapped_error
      mapped.fail(mapped_error)
    end
  end
  mapped
end

#on_complete(&callback) ⇒ Object

Registers an independent completion notification.

A callback failure is logged and does not suppress delivery to other completion callbacks or change the Task's already-settled result.

Raises:

  • (ArgumentError) 


101
102
103
104
105
106
107
108
109
110
111
112
113
114
 
# File 'lib/phronomy/engine/task.rb', line 101

def on_complete(&callback)
  raise ArgumentError, "on_complete requires a block" unless callback

  fire_args = nil
  @mutex.synchronize do
    if TERMINAL_STATES.include?(@status)
      fire_args = [@value, @error]
    else
      @on_complete_callbacks << callback
    end
  end
  deliver_completion_callback(callback, *fire_args) if fire_args
  self
end

#status ⇒ Object 



33
34
35
 
# File 'lib/phronomy/engine/task.rb', line 33

def status
  @mutex.synchronize { @status }
end

#wait_result(timeout: nil) ⇒ Object

Blocks the calling thread until settlement. EventLoop is never allowed to wait for a Task; it must continue through explicit events instead.



47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
 
# File 'lib/phronomy/engine/task.rb', line 47

def wait_result(timeout: nil)
  if Phronomy::Runtime.in_event_loop_context? && !done?
    raise Phronomy::EventLoopReentrancyError,
      "Task#wait_result cannot block the EventLoop thread; continue via an event"
  end

  deadline = timeout && monotonic_now + timeout.to_f
  value, error = @mutex.synchronize do
    until TERMINAL_STATES.include?(@status)
      if deadline
        remaining = deadline - monotonic_now
        if remaining <= 0
          raise Phronomy::TimeoutError,
            "timed out waiting for Task #{@name || "(unnamed)"}"
        end
        @cond.wait(@mutex, remaining)
      else
        @cond.wait(@mutex)
      end
    end
    [@value, @error]
  end

  raise error if error
  value
end