Implementing a Distributed Queue

Table of contents:

Credit: James Edward Gray II

Problem

You want to use a central server as a workhorse, queueing up requests from remote clients and handling them one at a time.

Solution

Heres a method that shares a Queue object with clients. Clients put job objects into the queue, and the server handles them by yielding them to a code block. #!/usr/bin/ruby

	#!/usr/bin/ruby
	# queue_server.rb

	require 	hread # For Rubys thread-safe Queue
	require drb

	$SAFE = 1 # Minimum acceptable paranoia level when sharing code!

	def run_queue(url=druby://127.0.0.1:61676)
	 queue = Queue.new # Containing the jobs to be processed

	 # Start up DRb with URI and object to share
	 DRb.start_service(url, queue)
	 puts Listening for connection…
	 while job = queue.deq
	 yield job
	 end
	end

Have your server call run_queue, passing in a code block that handles a single job. Every time one of your clients puts a job into the server queue, the server passes the job into the code block. Heres a sample code block that can handle a fast-running job ("Report") or a slow-running job ("Process"):

	run_queue do |job|
	 case job[
equest]
	 when Report
	 puts "Reporting for #{job[from]}… Done."
	 when Process
	 puts "Processing for #{job[from]}…"
	 sleep 3 # Simulate real work
	 puts Processing complete.
	 end
	end

If we get a couple of clients sending in requests, output might look like this:

	$ ruby queue_server.rb
	Listening for connection…

	Processing for Client 1…
	Processing complete.
	Processing for Client 2…
	Processing complete.
	Reporting for Client 1… Done.
	Reporting for Client 2… Done.
	Processing for Client 1…
	Processing complete.
	Reporting for Client 2… Done.
	…

Discussion

A client for the queue server defined in the Solution simply needs to connect to the DRB server and add a mix of "Report" and "Process" jobs to the queue. Heres a client that connects to the DRb server and adds 20 jobs to the queue at random:

	#!/usr/bin/ruby
	# queue_client.rb

	require 	hread
	require drb

	# Get a unique name for this client
	NAME = ARGV.shift or raise "Usage: #{File.basename($0)} CLIENT_NAME"

	DRb.start_service
	queue = DRbObject.new_with_uri("druby://127.0.0.1:61676")

	20.times do
	 queue.enq(
equest => [Report, Process][rand(2)], from => NAME)
	 sleep 1 # simulating network delays
	end

Everything from Recipe 16.10 applies here. The major difference is that Ruby ships with a thread-safe Queue. That saves us the trouble of building our own.

Problem

Solution

Discussion

See Also