Problems with async callbacks from within an async callback

[Adam Watson]

So this question isn't specific to Goliath, but it is a pattern I'm
trying to use within my Goliath-based app via EM-Http-Request.

I have an AMQP server exchange to which I have a consumer queue bound.
In the subscribe method, I want to be able to invoke http requests to
a Goliath server using em-http-request, and provide a callback to
process the response asynchronously.

The pattern works fine as long as the number of incoming messages is
less than 1000 or so. Otherwise, it seems as if the em-http-request
callbacks are never given a chance to be called, since the incoming
amqp messages have priority and keep pouring in.

Here is a gist as an example of what I am doing: https://gist.github.com/1384414

If you run the gist, after around 2300 or so messages are consumed,
the system will segfault with this error:

--------- (snip) -----------

/Users/awatson/.rvm/gems/ruby-1.9.2-p180@app/gems/
eventmachine-1.0.0.beta.3/lib/eventmachine.rb:199: [BUG] Segmentation
fault
ruby 1.9.2p180 (2011-02-18 revision 30909) [x86_64-darwin10.7.3]

-- control frame ----------
c:0005 p:---- s:0016 b:0016 l:000015 d:000015 CFUNC :run_machine
c:0004 p:0248 s:0013 b:0013 l:000012 d:000012 METHOD /Users/
awatson/.rvm/gems/ruby-1.9.2-p180@app/gems/eventmachine-1.0.0.beta.3/
lib/eventmachine.rb:199
c:0003 p:0039 s:0006 b:0006 l:0014e8 d:001668 EVAL amqptest.rb:4
c:0002 p:---- s:0004 b:0004 l:000003 d:000003 FINISH
c:0001 p:0000 s:0002 b:0002 l:0014e8 d:0014e8 TOP
---------------------------
-- Ruby level backtrace information
----------------------------------------
amqptest.rb:4:in `<main>'
/Users/awatson/.rvm/gems/ruby-1.9.2-p180@app/gems/
eventmachine-1.0.0.beta.3/lib/eventmachine.rb:199:in `run'
/Users/awatson/.rvm/gems/ruby-1.9.2-p180@app/gems/
eventmachine-1.0.0.beta.3/lib/eventmachine.rb:199:in `run_machine'

-- C level backtrace information
-------------------------------------------

[NOTE]
You may have encountered a bug in the Ruby interpreter or extension
libraries.
Bug reports are welcome.
For details: http://www.ruby-lang.org/bugreport.html

Abort trap

--------- (snip) -----------

I've tried several workarounds, including wrapping my handle_message
method in its own fiber, to no avail. I've also tried using an
EM::Synchrony.sync call on an em-synchrony-http request object which
also does not work. The only way I can get this to function as
expected is to either limit the total number of incoming messages in
the queue to less than 2000 or so (which I won't be able to control in
a production environment) or to use a blocking http request via
Net::Http inside the handle_message method, which I'd rather not have
to do.

Does anyone have any idea how else I could approach this in order to
asynchronously handle the http request within the async consumer
callback?

Thanks,
Adam

[dan sinclair]

For the queue processing we were doing at PostRank we found we had to switch from the subscribe to the pop method of AMQP. With subscribe, we'd never know how many messages were piled up on our consumer. With pop, we had a lot more control of when new messages arrived.

dan

[Nathan Stults]

Will creating your channel with the prefetch option, subscribing with :ack => true, and acking your messages after the http request is complete solve your problem? This is what we use to control the rate of flow of messages into our message consumers.

http://rubydoc.info/github/ruby-amqp/amqp/master/file/docs/Queues.textile#QoS___Prefetching_messages

 

Nathan

[Adam Watson]

Hi Dan and Nathan,

Awesome, thanks for the great responses guys! After an initial test,
both of these methods do indeed work. I've gone ahead and implemented
the queue.pop approach for now, but am diving into further load tests
with the prefetch solution that Nathan suggests, since this will be
even more flexible when traffic is ramped up and we spin up multiple
consumers.

Cheers... I owe you guys a beer :)

Adam