Class: Async::Container::Threaded::Child

Inherits:
Channel
  • Object
show all
Defined in:
lib/async/container/threaded.rb

Overview

Represents a running child thread from the point of view of the parent container.

Defined Under Namespace

Classes: Exit, Instance, Status

Instance Attribute Summary

Attributes inherited from Channel

#in, #out

Class Method Summary collapse

Instance Method Summary collapse

Methods inherited from Channel

#close_read, #close_write, #receive

Constructor Details

#initialize(name: nil, **options) ⇒ Child

Initialize the thread.



128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
 
# File 'lib/async/container/threaded.rb', line 128

def initialize(name: nil, **options)
	super(**options)
	
	@status = nil
	
	@thread = yield(self)
	@thread.report_on_exception = false
	@thread.name = name
	
	@waiter = ::Thread.new do
		begin
			@thread.join
		rescue Exit => exit
			finished(exit.error)
		rescue Interrupt
			# Graceful shutdown.
			finished
		rescue Exception => error
			finished(error)
		else
			finished
		end
	end
end

Class Method Details

.fork(**options) ⇒ Object

Start a new child thread and execute the provided block in it.



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

def self.fork(**options)
	self.new(**options) do |thread|
		::Thread.new do
			# This could be a configuration option (see forked implementation too):
			::Thread.handle_interrupt(SignalException => :immediate) do
				yield Instance.for(thread)
			end
		end
	end
end

Instance Method Details

#as_json ⇒ Object

Convert the child process to a hash, suitable for serialization.



156
157
158
159
160
161
 
# File 'lib/async/container/threaded.rb', line 156

def as_json(...)
	{
		name: @thread.name,
		status: @status&.as_json,
	}
end

#close ⇒ Object

Invoke #terminate! and then #wait for the child thread to exit.



189
190
191
192
193
194
 
# File 'lib/async/container/threaded.rb', line 189

def close
	self.terminate!
	self.wait
ensure
	super
end

#interrupt! ⇒ Object

Raise Interrupt in the child thread.



197
198
199
 
# File 'lib/async/container/threaded.rb', line 197

def interrupt!
	@thread.raise(Interrupt)
end

#kill! ⇒ Object

Invoke Thread#kill on the child thread.



207
208
209
210
211
212
 
# File 'lib/async/container/threaded.rb', line 207

def kill!
	# Killing a thread does not raise an exception in the thread, so we need to handle the status here:
	@status = Status.new(:killed)
	
	@thread.kill
end

#name ⇒ Object

Get the name of the thread.



178
179
180
 
# File 'lib/async/container/threaded.rb', line 178

def name
	@thread.name
end

#name=(value) ⇒ Object

Set the name of the thread.



172
173
174
 
# File 'lib/async/container/threaded.rb', line 172

def name= value
	@thread.name = value
end

#restart! ⇒ Object

Raise Restart in the child thread.



215
216
217
 
# File 'lib/async/container/threaded.rb', line 215

def restart!
	@thread.raise(Restart)
end

#terminate! ⇒ Object

Raise Async::Container::Terminate in the child thread.



202
203
204
 
# File 'lib/async/container/threaded.rb', line 202

def terminate!
	@thread.raise(Terminate)
end

#to_json ⇒ Object

Convert the request to JSON.



166
167
168
 
# File 'lib/async/container/threaded.rb', line 166

def to_json(...)
	as_json.to_json(...)
end

#to_s ⇒ Object

A human readable representation of the thread.



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

def to_s
	"\#<#{self.class} #{@thread.name}>"
end

#wait(timeout = 0.1) ⇒ Object

Wait for the thread to exit and return he exit status.



224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
 
# File 'lib/async/container/threaded.rb', line 224

def wait(timeout = 0.1)
	if @waiter
		Console.debug(self, "Waiting for thread to exit...", child: {thread_id: @thread.object_id}, timeout: timeout)
		
		unless @waiter.join(timeout)
			Console.warn(self, "Thread is blocking, sending kill signal...", child: {thread_id: @thread.object_id}, timeout: timeout)
			self.kill!
			@waiter.join
		end
		
		@waiter = nil
	end
	
	Console.debug(self, "Thread exited.", child: {thread_id: @thread.object_id, status: @status})
	
	return @status
end