mongrel unicorn - Nov 2010 - AMQP and Unicorn (mq gem)

Hi all,

I''m having issues with Unicorn and connecting to 
RabbitMQ using the tmm1-amqp gem.

 I''ve tried lots of  approaches.

The classic initializer with Thread.new { EM.run } 
for the Rails app and even tried using 
the Qusion library. 

 (https://github.com/danielsdeleo/qusion)

I''ve made a simple mod to Qusion for it to monkey patch 
unicorn too. 

The code looks like:

/vendor/plugins/qusion/lib/amqp.rb

module AMQP
  def self.start_web_dispatcher(amqp_settings={})
    @settings = settings.merge(amqp_settings)
    case Qusion::ServerSpy.server_type
    when :passenger
      PhusionPassenger.on_event(:starting_worker_process) do |forked| 
        if forked
          EM.kill_reactor
          Thread.current[:mq], @conn = nil, nil
        end 
        Thread.new { start }
        die_gracefully_on_signal
      end
    when :standard
      Thread.new { start }
      die_gracefully_on_signal
    when :evented
      die_gracefully_on_signal
    when :none # << HERE
      Thread.new { start }
      die_gracefully_on_signal
    else
      raise ArgumentError, "AMQP#start_web_dispatcher
 requires an argument of 
[:standard|:evented|:passenger|:none]"
    end
  end
  
  def self.die_gracefully_on_signal
    Signal.trap("INT")  { AMQP.stop { EM.stop } }
    Signal.trap("TERM") { AMQP.stop { EM.stop } }
  end
end

See the :none case? It used to be empty and return nil. 

Now it has the same behavior as the non-
evented :standard server. 

The problem is that the "server_spy" wasn''t 
recognizing Unicorn. So far so 
good.

Originally, the AMQP/Qusion conf goes into config/environment.rb

   config.after_initialize do
      
      Qusion.start(:user => ''guest'',:pass =>
''mypass'')
 end


However, when I publish messages to the queue,
 it just isn''t posted. It silently fails. 

I''ve then tried to put this on config/unicorn.rb:

before_fork do |server,worker|
 Qusion.start(:user => ''guest'',:pass =>
''mypass'')
 MQ.new.queue(''jobs'').publish(''hey!'')
end

And the hey! message is posted. 

However, it only works from here, the MQ.new.queue on the app code 
fails silently.

Any ideas on what might be happenig?

Marcelo de Moraes Serpa <fullofcaffeine at gmail.com>
wrote:
> Hi all,
> 
> I''m having issues with Unicorn and connecting to 
> RabbitMQ using the tmm1-amqp gem.
> 
>  I''ve tried lots of  approaches.
> 
> The classic initializer with Thread.new { EM.run } 
> for the Rails app and even tried using 
> the Qusion library. 
> 
>  (https://github.com/danielsdeleo/qusion)
> 
> I''ve made a simple mod to Qusion for it to monkey patch 
> unicorn too. 
> 
> The code looks like:
> 
> /vendor/plugins/qusion/lib/amqp.rb
> 
> module AMQP
>   def self.start_web_dispatcher(amqp_settings={})
>     @settings = settings.merge(amqp_settings)
>     case Qusion::ServerSpy.server_type
>     when :passenger
>       PhusionPassenger.on_event(:starting_worker_process) do |forked| 
<side note>maybe adding a similar hook to Unicorn for library/app authors
to use would be a good idea.</side note>
>         if forked
>           EM.kill_reactor
>           Thread.current[:mq], @conn = nil, nil
>         end 
>         Thread.new { start }
>         die_gracefully_on_signal
>       end
>     when :standard
>       Thread.new { start }
>       die_gracefully_on_signal
>     when :evented
>       die_gracefully_on_signal
>     when :none # << HERE
>       Thread.new { start }
>       die_gracefully_on_signal
>     else
>       raise ArgumentError, "AMQP#start_web_dispatcher
>  requires an argument of 
> [:standard|:evented|:passenger|:none]"
>     end
>   end
>   
>   def self.die_gracefully_on_signal
>     Signal.trap("INT")  { AMQP.stop { EM.stop } }
>     Signal.trap("TERM") { AMQP.stop { EM.stop } }
>   end
Registering a standard at_exit {} block in the Unicorn
after_fork hook would be a good idea than trapping INT
and TERM, which Unicorn uses.
> end
> 
> See the :none case? It used to be empty and return nil. 
> 
> Now it has the same behavior as the non-
> evented :standard server. 
> 
> The problem is that the "server_spy" wasn''t 
> recognizing Unicorn. So far so 
> good.
> 
> Originally, the AMQP/Qusion conf goes into config/environment.rb
> 
>    config.after_initialize do
>       
>       Qusion.start(:user => ''guest'',:pass =>
''mypass'')
>  end
> 
> 
> However, when I publish messages to the queue,
>  it just isn''t posted. It silently fails. 
> 
> I''ve then tried to put this on config/unicorn.rb:
> 
> before_fork do |server,worker|
>  Qusion.start(:user => ''guest'',:pass =>
''mypass'')
>  MQ.new.queue(''jobs'').publish(''hey!'')
> end
> 
> And the hey! message is posted. 
> 
> However, it only works from here, the MQ.new.queue on the app code 
> fails silently.
> 
> Any ideas on what might be happenig?
Is preload_app true?  Then you''re probably trying to share sockets
across processes and confusing EM or AMQP.

Instead of using before_fork above, you should probably
be using after_fork:

  before_fork do |server,worker|
    Qusion.start(:user => ''guest'',:pass =>
''mypass'')
    MQ.new.queue(''jobs'').publish(''hey!'')
  end

I''m not very familiar with AMQP, but the Unicorn process is strictly an
AMQP sender/client and doesn''t listen as a server on a port, right?
If so, the above before_fork config *should* work for you.

Let us know how it goes!

-- 
Eric Wong

We use unicorn + sinatra + tmm1/amqp, and I can say for a fact that you must
initialize the AMQP connection in after_fork.  Eric is probably right about the
fd sharing from before_fork.

In the rails-specific context, if preload_app is true, it''s possible
that using config.after_initialize {} will still result in the AMQP
initialization occurring before the fork.

cheers,
--jordan


On Nov 9, 2010, at 4:55 PM, Eric Wong wrote:
> Marcelo de Moraes Serpa <fullofcaffeine at gmail.com> wrote:
>> Hi all,
>> 
>> I''m having issues with Unicorn and connecting to 
>> RabbitMQ using the tmm1-amqp gem.
>> 
>> I''ve tried lots of  approaches.
>> 
>> The classic initializer with Thread.new { EM.run } 
>> for the Rails app and even tried using 
>> the Qusion library. 
>> 
>> (https://github.com/danielsdeleo/qusion)
>> 
>> I''ve made a simple mod to Qusion for it to monkey patch 
>> unicorn too. 
>> 
>> The code looks like:
>> 
>> /vendor/plugins/qusion/lib/amqp.rb
>> 
>> module AMQP
>>  def self.start_web_dispatcher(amqp_settings={})
>>    @settings = settings.merge(amqp_settings)
>>    case Qusion::ServerSpy.server_type
>>    when :passenger
>>      PhusionPassenger.on_event(:starting_worker_process) do |forked| 
> 
> <side note>maybe adding a similar hook to Unicorn for library/app
authors
> to use would be a good idea.</side note>
> 
>>        if forked
>>          EM.kill_reactor
>>          Thread.current[:mq], @conn = nil, nil
>>        end 
>>        Thread.new { start }
>>        die_gracefully_on_signal
>>      end
>>    when :standard
>>      Thread.new { start }
>>      die_gracefully_on_signal
>>    when :evented
>>      die_gracefully_on_signal
>>    when :none # << HERE
>>      Thread.new { start }
>>      die_gracefully_on_signal
>>    else
>>      raise ArgumentError, "AMQP#start_web_dispatcher
>> requires an argument of 
>> [:standard|:evented|:passenger|:none]"
>>    end
>>  end
>> 
>>  def self.die_gracefully_on_signal
>>    Signal.trap("INT")  { AMQP.stop { EM.stop } }
>>    Signal.trap("TERM") { AMQP.stop { EM.stop } }
>>  end
> 
> Registering a standard at_exit {} block in the Unicorn
> after_fork hook would be a good idea than trapping INT
> and TERM, which Unicorn uses.
> 
>> end
>> 
>> See the :none case? It used to be empty and return nil. 
>> 
>> Now it has the same behavior as the non-
>> evented :standard server. 
>> 
>> The problem is that the "server_spy" wasn''t 
>> recognizing Unicorn. So far so 
>> good.
>> 
>> Originally, the AMQP/Qusion conf goes into config/environment.rb
>> 
>>   config.after_initialize do
>> 
>>      Qusion.start(:user => ''guest'',:pass =>
''mypass'')
>> end
>> 
>> 
>> However, when I publish messages to the queue,
>> it just isn''t posted. It silently fails. 
>> 
>> I''ve then tried to put this on config/unicorn.rb:
>> 
>> before_fork do |server,worker|
>> Qusion.start(:user => ''guest'',:pass =>
''mypass'')
>>
MQ.new.queue(''jobs'').publish(''hey!'')
>> end
>> 
>> And the hey! message is posted. 
>> 
>> However, it only works from here, the MQ.new.queue on the app code 
>> fails silently.
>> 
>> Any ideas on what might be happenig?
> 
> Is preload_app true?  Then you''re probably trying to share sockets
> across processes and confusing EM or AMQP.
> 
> Instead of using before_fork above, you should probably
> be using after_fork:
> 
>  before_fork do |server,worker|
>    Qusion.start(:user => ''guest'',:pass =>
''mypass'')
>    MQ.new.queue(''jobs'').publish(''hey!'')
>  end
> 
> I''m not very familiar with AMQP, but the Unicorn process is
strictly an
> AMQP sender/client and doesn''t listen as a server on a port,
right?
> If so, the above before_fork config *should* work for you.
> 
> Let us know how it goes!