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.



142
143
144
 
# File 'lib/async/task.rb', line 142

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.



174
175
176
 
# File 'lib/async/task.rb', line 174

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) 


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

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 



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

def async(*arguments, **options, &block)
	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) 


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

def complete?
	@status == :complete
end

#current? ⇒ Boolean

Returns:

  • (Boolean) 


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

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

#failed? ⇒ Boolean

Returns:

  • (Boolean) 


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

def failed?
	@status == :failed
end

#finished? ⇒ Boolean

Whether we can remove this node from the reactor graph.

Returns:

  • (Boolean) 


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

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) 


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

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.



145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
 
# File 'lib/async/task.rb', line 145

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) 


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

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. @raises If the task’s fiber is the current fiber.



125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
 
# File 'lib/async/task.rb', line 125

def wait
	raise "Cannot wait on own fiber" if Fiber.current.equal?(@fiber)
	
	if running?
		@finished ||= Condition.new
		@finished.wait
	end
	
	case @result
	when 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