From: Robert Klemme <shortcutter@...>
Date: 2010-01-07T22:45:05+09:00
Subject: Re: Non-blocking communication between Ruby processes

On 01/07/2010 02:18 PM, Iņaki Baz Castillo wrote:
> Hi, I run Unicorn which is a Rack http server using N forked worker processes. 
> I need the following:
> 
> - When a worker processes a HTTP request it must notify some data to other 
> independent Ruby process XXX (different than Unicorn).
> 
> - This communication must be non-blocking, this is, the Unicorn worker process 
> sends the notification and doesn't wait for response from the process XXX, so 
> the Unicorn worker can, at the moment, generate the HTTP response and send 
> back to the client, getting free to handle new HTTP requests.
> 
> - The ruby process XXX should use some kind of queue system to store 
> notifications and handle them. In fact, it should take them periodically and 
> send via TCP (but not HTTP) to other server.
> 
> 
> Which is the best approach to design such communication? perhaps using 
> something as EventMachine for the XXX process and Unix/TCP socket 
> communication between Unicorn processes and XXX process? any other alternative 
> or suggestion?
> 
> Thanks a lot.

I would probably first try a simple setup: make process XXX publish a 
Queue via DRb on a well known port and have one or more threads fetching 
from the queue and processing data.  If you fear resource exhaustion, 
you can make the queue size limited.  E.g.:

x.rb server
c.rb client


robert@fussel:~$ cat x.rb
#!/usr/local/bin/ruby19

require 'thread'
require 'drb'

QUEUE_SIZE = 1024
THREAD_COUNT = 5
URI="druby://localhost:8787"

QUEUE = SizedQueue.new QUEUE_SIZE

threads = (1..THREAD_COUNT).map do
   Thread.new do
     while msg = QUEUE.deq
       p msg
     end
   end
end

DRb.start_service(URI, QUEUE)
DRb.thread.join

robert@fussel:~$ cat c.rb
#!/usr/local/bin/ruby19

require 'drb/drb'
require 'benchmark'

SERVER_URI="druby://localhost:8787"

QUEUE = DRbObject.new_with_uri(SERVER_URI)

10.times do |i|
   puts Benchmark.times do
     QUEUE.enq(sprintf("msg %4d at %-20s", i, Time.now))
   end
end
robert@fussel:~$

Of course you can as well use a named pipe for the communication.  But 
then demarcation of message boundaries might be more difficult etc.

Kind regards

	robert


-- 
remember.guy do |as, often| as.you_can - without end
http://blog.rubybestpractices.com/