Class: Async::Task

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

Overview

Encapsulates the state of a running task and it’s result.

Instance Attribute Summary collapse

Attributes inherited from Node

#annotation, #children, #head, #parent, #tail

Class Method Summary collapse

Instance Method Summary collapse

Methods inherited from Node

#The parent node.=, #annotate, #children?, #consume, #description, #print_hierarchy, #root, #terminate, #transient?, #traverse

Constructor Details

#initialize(parent = Task.current?, finished: nil, **options, &block) ⇒ Task

Create a new task.



51
52
53
54
55
56
57
58
59
60
 
# File 'lib/async/task.rb', line 51

def initialize(parent = Task.current?, finished: nil, **options, &block)
	super(parent, **options)
	
	@status = :initialized
	@result = nil
	@finished = finished
	
	@block = block
	@fiber = nil
end

Instance Attribute Details

#fiber ⇒ Object (readonly) 



92
93
94
 
# File 'lib/async/task.rb', line 92

def fiber
  @fiber
end

#result ⇒ Object (readonly)

Access the result of the task without waiting. May be nil if the task is not completed. Does not raise exceptions.



147
148
149
 
# File 'lib/async/task.rb', line 147

def result
  @result
end

#status ⇒ Object (readonly) 



99
100
101
 
# File 'lib/async/task.rb', line 99

def status
  @status
end

Class Method Details

.current ⇒ Object

Lookup the Async::Task for the current fiber. Raise ‘RuntimeError` if none is available. @raises If task was not #set! for the current fiber.



179
180
181
 
# File 'lib/async/task.rb', line 179

def self.current
	Thread.current[:async_task] or raise RuntimeError, "No async task available!"
end

.current? ⇒ Boolean

Check if there is a task defined for the current fiber.

Returns:

  • (Boolean) 


185
186
187
 
# File 'lib/async/task.rb', line 185

def self.current?
	Thread.current[:async_task]
end

.yield ⇒ Object

Deprecated.

With no replacement.



44
45
46
 
# File 'lib/async/task.rb', line 44

def self.yield
	Fiber.scheduler.transfer
end

Instance Method Details

#alive? ⇒ Boolean

Returns:

  • (Boolean) 


94
95
96
 
# File 'lib/async/task.rb', line 94

def alive?
	@fiber&.alive?
end

#async(*arguments, **options, &block) ⇒ Object

Run an asynchronous task as a child of the current task.



115
116
117
118
119
120
121
122
123
 
# File 'lib/async/task.rb', line 115

def async(*arguments, **options, &block)
	raise "Cannot create child task within a task that has finished execution!" if self.finished?
	
	task = Task.new(self, **options, &block)
	
	task.run(*arguments)
	
	return task
end

#backtrace(*arguments) ⇒ Object 



67
68
69
 
# File 'lib/async/task.rb', line 67

def backtrace(*arguments)
	@fiber&.backtrace(*arguments)
end

#complete? ⇒ Boolean

Returns:

  • (Boolean) 


213
214
215
 
# File 'lib/async/task.rb', line 213

def complete?
	@status == :complete
end

#current? ⇒ Boolean

Returns:

  • (Boolean) 


189
190
191
 
# File 'lib/async/task.rb', line 189

def current?
	self.equal?(Thread.current[:async_task])
end

#failed? ⇒ Boolean

Returns:

  • (Boolean) 


205
206
207
 
# File 'lib/async/task.rb', line 205

def failed?
	@status == :failed
end

#finished? ⇒ Boolean

Whether we can remove this node from the reactor graph.

Returns:

  • (Boolean) 


201
202
203
 
# File 'lib/async/task.rb', line 201

def finished?
	super && @fiber.nil?
end

#reactor ⇒ Object 



62
63
64
 
# File 'lib/async/task.rb', line 62

def reactor
	self.root
end

#run(*arguments) ⇒ Object

Begin the execution of the task.



102
103
104
105
106
107
108
109
110
111
112
 
# File 'lib/async/task.rb', line 102

def run(*arguments)
	if @status == :initialized
		@status = :running
		
		schedule do
			@block.call(self, *arguments)
		end
	else
		raise RuntimeError, "Task already running!"
	end
end

#running? ⇒ Boolean

Check if the task is running.

Returns:

  • (Boolean) 


195
196
197
 
# File 'lib/async/task.rb', line 195

def running?
	@status == :running
end

#sleep(duration = nil) ⇒ Object

Deprecated.

Prefer Kernel#sleep except when compatibility with ‘stable-v1` is required.



77
78
79
 
# File 'lib/async/task.rb', line 77

def sleep(duration = nil)
	super
end

#stop(later = false) ⇒ Object

Stop the task and all of its children.



150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
 
# File 'lib/async/task.rb', line 150

def stop(later = false)
	if self.stopped?
		# If we already stopped this task... don't try to stop it again:
		return
	end
	
	if self.running?
		if self.current?
			if later
				Fiber.scheduler.push(Stop::Later.new(self))
			else
				raise Stop, "Stopping current task!"
			end
		elsif @fiber&.alive?
			begin
				Fiber.scheduler.raise(@fiber, Stop)
			rescue FiberError
				Fiber.scheduler.push(Stop::Later.new(self))
			end
		end
	else
		# We are not running, but children might be, so transition directly into stopped state:
		stop!
	end
end

#stopped? ⇒ Boolean

Returns:

  • (Boolean) 


209
210
211
 
# File 'lib/async/task.rb', line 209

def stopped?
	@status == :stopped
end

#to_s ⇒ Object 



72
73
74
 
# File 'lib/async/task.rb', line 72

def to_s
	"\#<#{self.description} (#{@status})>"
end

#wait ⇒ Object

Retrieve the current result of the task. Will cause the caller to wait until result is available. If the result was an exception, raise that exception.

Conceptually speaking, waiting on a task should return a result, and if it throws an exception, this is certainly an exceptional case that should represent a failure in your program, not an expected outcome. In other words, you should not design your programs to expect exceptions from ‘#wait` as a normal flow control, and prefer to catch known exceptions within the task itself and return a result that captures the intention of the failure, e.g. a `TimeoutError` might simply return `nil` or `false` to indicate that the operation did not generate a valid result (as a timeout was an expected outcome of the internal operation in this case).



131
132
133
134
135
136
137
138
139
140
141
142
143
144
 
# File 'lib/async/task.rb', line 131

def wait
	raise "Cannot wait on own fiber!" if Fiber.current.equal?(@fiber)
	
	if running?
		@finished ||= Condition.new
		@finished.wait
	end
	
	if @result.is_a?(Exception)
		raise @result
	else
		return @result
	end
end

#with_timeout(duration, exception = TimeoutError, message = "execution expired", &block) ⇒ Object

Execute the given block of code, raising the specified exception if it exceeds the given duration during a non-blocking operation.



82
83
84
 
# File 'lib/async/task.rb', line 82

def with_timeout(duration, exception = TimeoutError, message = "execution expired", &block)
	Fiber.scheduler.with_timeout(duration, exception, message, &block)
end

#yield ⇒ Object

Yield back to the reactor and allow other fibers to execute.



87
88
89
 
# File 'lib/async/task.rb', line 87

def yield
	Fiber.scheduler.yield
end