example on stop accept

[matti]

How do I stop accepting connections?

In my server I want to have shutdown behavior that gracefully handles all existing connections, but does not accept new ones.

[ioquatix]

Good question

[ioquatix]

Graceful shutdown isn’t too complicated it could be done with a timeout around children tasks... would need to manage accept loop a bit more explicitly.... then on shutdown wait for all children tasks to complete.... after timeout invoke stop on them... then exit yourself.

[ioquatix]

Here is a rough idea of how to do it:

require 'async'
require 'async/io'

Async.logger.info!

GRACE_TIME = 5

def graceful_server(endpoint, condition = nil, &block)
	Async do |parent_task|
		server_tasks = []
		
		Async.logger.info "Binding to #{endpoint}..."
		
		endpoint.each do |endpoint|
			server = endpoint.bind
			server.listen(Socket::SOMAXCONN)
			
			Async.logger.info "Accepting connections from #{server}..."
			server_tasks << parent_task.async do
				server.accept_each(task: parent_task, &block)
			ensure
				server.close
			end
		end
		
		if condition
			Async.logger.info "Waiting on #{condition}..."
			condition.wait
			
			Async.logger.info("Task tree"){|buffer| parent_task.print_hierarchy(buffer)}
			
			Async.logger.info "Stopping #{server_tasks.count} accept loops..."
			server_tasks.each(&:stop)
			
			children = parent_task.children
			
			Async.logger.info "Stopping #{children.count} connections..."
			
			if children.any?
				Async.logger.warn("Waiting for #{children.count} connections...")
				
				parent_task.sleep GRACE_TIME
			end
			
			children = parent_task.children
			
			if children.any?
				Async.logger.warn("Stopping #{children.count} connections...")
				
				children.each(&:stop)
			end
		end
	end
end

def echo_server(endpoint)
	condition = Async do |task|
		task.sleep 60 while true
	end
	
	graceful_server(endpoint, condition) do |client, task:|
		# This is an asynchronous block within the current reactor:
		while data = client.read(512)
			# This produces out-of-order responses.
			task.sleep(rand * 0.01)
		
			client.write(data.reverse)
		end
	ensure
		client.close
	end
	
	return condition
end

def echo_client(endpoint, data)
	Async do |task|
		endpoint.connect do |peer|
			10.times do
				Async.logger.info "Client #{data}: sleeping"
				task.sleep 2
				
				result = peer.write(data)
				message = peer.read(512)
				
				Async.logger.info "Sent #{data}, got response: #{message}"
			end
		end
	end
end

Async do |task|
	endpoint = Async::IO::Endpoint.tcp('0.0.0.0', 9000)
	
	Async.logger.info "Starting server..."
	server = echo_server(endpoint)
	
	Async.logger.info "Clients connecting..."
	1.times.collect do |i|
		echo_client(endpoint, "Hello World #{i}")
	end
	
	task.sleep 5
	
	Async.logger.info "Stopping server..."
	server.stop
end

Async.logger.info "Finished..."

Honestly, it seems pretty complicated, but I guess graceful shutdown isn't that trivial.

[ioquatix]

It should work to put a timeout around the tasks, but I think there is a bug... needs to be fixed.

[ioquatix]

				parent_task.with_timeout(GRACE_TIME) do
					children.each(&:wait)
				rescue Async::TimeoutError
					Async.logger.warn("Could not terminate child connections...")
				end

Task#wait is not timeout safe. :( doh.

[ioquatix]

Okay, this is fixed in socketry/async@0b63195

I'll need to test a bit more.

[ioquatix]

Okay, this implementation works with async 1.15.1:

require 'async'
require 'async/io'

Async.logger.info!

GRACE_TIME = 5

def graceful_server(endpoint, condition = nil, &block)
	Async do |parent_task|
		server_tasks = []
		
		Async.logger.info "Binding to #{endpoint}..."
		
		endpoint.each do |endpoint|
			server = endpoint.bind
			server.listen(Socket::SOMAXCONN)
			
			Async.logger.info "Accepting connections from #{server}..."
			server_tasks << parent_task.async do
				server.accept_each(task: parent_task, &block)
			ensure
				server.close
			end
		end
		
		if condition
			Async.logger.info "Waiting on #{condition}..."
			condition.wait
			
			Async.logger.info("Task tree"){|buffer| parent_task.print_hierarchy(buffer)}
			
			Async.logger.info "Stopping #{server_tasks.count} accept loops..."
			server_tasks.each(&:stop)
			
			children = parent_task.children
			
			Async.logger.info "Stopping #{children.count} connections..."
			
			if children.any?
				Async.logger.warn("Waiting for #{children.count} connections...")
				
				parent_task.with_timeout(GRACE_TIME) do
					children.each(&:wait)
				rescue Async::TimeoutError
					Async.logger.warn("Could not terminate child connections...")
				end
			end
			
			children = parent_task.children
			
			if children.any?
				Async.logger.warn("Stopping #{children.count} connections...")
				
				children.each(&:stop)
			end
		end
	end
end

def echo_server(endpoint)
	condition = Async do |task|
		task.sleep 60 while true
	end
	
	graceful_server(endpoint, condition) do |client, task:|
		# This is an asynchronous block within the current reactor:
		while data = client.read(512)
			# This produces out-of-order responses.
			task.sleep(rand * 0.01)
		
			client.write(data.reverse)
		end
	ensure
		client.close
	end
	
	return condition
end

def echo_client(endpoint, data)
	Async do |task|
		endpoint.connect do |peer|
			10.times do
				Async.logger.info "Client #{data}: sleeping"
				task.sleep 2
				
				result = peer.write(data)
				message = peer.read(512)
				
				Async.logger.info "Sent #{data}, got response: #{message}"
			end
		end
	end
end

Async do |task|
	endpoint = Async::IO::Endpoint.tcp('0.0.0.0', 9000)
	
	Async.logger.info "Starting server..."
	server = echo_server(endpoint)
	
	Async.logger.info "Clients connecting..."
	1.times.collect do |i|
		echo_client(endpoint, "Hello World #{i}")
	end
	
	task.sleep 5
	
	Async.logger.info "Stopping server..."
	server.stop
end

Async.logger.info "Finished..."

[ioquatix]

I think you probably also want to tell the server tasks to signal the clients that shutdown is in progress...

To do this, you'll need some broadcast message. It will ultimately depend on how you implement your connection state... there is nothing wrong with making an array of connections and then just connections.each{|connection| connection.write("Goodbye!")} before waiting on the children tasks.

[ioquatix]

One more thing to keep in mind is if you have a graceful shutdown, you need to handle the situations where the graceful shutdown fails in some way and invoke #stop on everything.

[matti]

wow, thanks!

when are there multiple server_tasks ? to me it looks like there is just one server_task ?

			server_tasks << parent_task.async do
				server.accept_each(task: parent_task, &block)
			ensure
				server.close
			end

[ioquatix]

bind to localhost instead of 0.0.0.0 and you'll get IPv4 and IPv6

[matti]

what about threads (Async::Container) - I tried to follow falcon source with SharedEndpoint, but I can't come up with how to do it without server.accept_each that returns the task..

[ioquatix]

How would threads solve the problem? I think it only makes more problems.