EM, em-http-request and amqp

[AsyncSubscriber]

Hi,

I wrote a program that upon receiving a message from AMQP (rabbitmq) would then go use EM and em-http-request to fetch files. The code looks like

require 'minion'
include Minion

Minion.amqp_url = ...

puts "Listening on queue: #{queue}"

job queue do |args|
  pid = fork do # the fork was there originally
      ActiveRecord::Base.connection_pool.disconnect!
      process(args['id'])
  end
  Process.waitpid(pid)
end

# the process() looks like
def process(id)
   ... load data for id ...
   EM.run { log;

          req = EventMachine::HttpRequest.new(url, :connect_timeout => timeout, :inactivity_timeout => timeout)
          http = req.get
          http.errback { log; ... }
          http.callback { log; ... }
   }
end

The process method runs very well by itself in test. Now, run the whole script under amqp, I can see that it gets into the EM.run block and that the requests have been created. However, no http request seems to be executed at all (errback / callback) and the loop just exit.

I check posts relating to EM/AMQP and tried comment out the EM.kqueue call, but nothing works.

What can go wrong here?

Thanks a lot for any help!

[AsyncSubscriber]

OK, I wrote a number of test cases to reproduce the problem. It appears to do with nested EM.run and forking. (AMQP/Minon runs in an EM loop.)

First, I would need to use EM.fork_reactor rather than just fork (despite EM.run's implementation seems trying to take care of the same situation,) else the inner EM.run's stuff won't be executed at all. After I use EM.fork_reactor, the async requests will be made, however, the top-level code Client.dl would exit right the way before the inner EM.run returns. Apparently, this part is due to the nested EM.run(), even though I use a new Thread to run the inner loop.

So the questions are

1) How do I run nested inner loop where the invoker of the inner EM.run expect synchronous completion of the call. Why wouldn't using a new Thread to run the inner loop work?

2) The reason there is a fork call in the first place is that the process codes (upto now) do not need to run in EM loop; what's the proper way to terminate the EM loop in a forked process and continue? Would that help (1)?

Thanks

require 'eventmachine'
require 'em-http-request'

module Client
  def self.dl
    result = ''
    EM.run do
      req = EventMachine::HttpRequest.new('http://www.google.com')
      http = req.get :redirects => 5
      http.errback {
        puts "\n        dl got error"
        EM.stop_event_loop
      }
      http.stream {|chunk|
        if result > ''
          print " #{chunk.size}" if chunk
        else
          print "        dl got data size #{chunk.size}" if chunk
        end
        result += chunk
      }
      http.callback {
        puts "\n        dl got response #{http.response_header.status}"
        EM.stop_event_loop
      }
    end
    puts "        dl returning data size #{result.nil? ? 'NIL' : result.size}"
    return result # RETURN
  end
end

puts "\nasyncdl test fork"
EM.run do
  puts "outer EM.run"
  EM.next_tick {
    puts "from to #{Process.pid}"
    pid = fork do
      puts " forked to #{Process.pid}. EM.run #{EM.reactor_running?}. reactor thread #{EM.reactor_thread?}"
      thr = Thread.new do
        puts "  in thread EM.run #{EM.reactor_running?}. reactor thread #{EM.reactor_thread?}"
        res = Client.dl
        puts "  result size #{res.nil? ? 'NIL' : res.size}"
      end
      thr.join
      puts " thread returns"
    end
    Process.waitpid(pid)
    EM.add_timer(2) {
      puts "in timer"
      EM.stop
    }
  }
end

The output:

puts "\nasyncdl test EM.fork_reactor"
EM.run do
  EM.next_tick {
    puts "from to #{Process.pid}"
    pid = EM.fork_reactor do
      puts " forked to #{Process.pid}. EM.run #{EM.reactor_running?}. reactor thread #{EM.reactor_thread?}"
      thr = Thread.new do
        puts "  in thread EM.run #{EM.reactor_running?}. reactor thread #{EM.reactor_thread?}"
        res = Client.dl
        puts "  result size #{res.nil? ? 'NIL' : res.size}"
      end
      thr.join
      puts " thread returns"
    end
    Process.waitpid(pid)
    EM.add_timer(2) {
      puts "in timer"
      EM.stop
    }
  }
end

asyncdl test fork
outer EM.run
from to 20271
 forked to 20410. EM.run true. reactor thread true
  in thread EM.run true. reactor thread false
        dl returning data size 0 # nothing in the inner EM.run runs
  result size 0 # and we drop out
 thread returns
in timer

asyncdl test EM.fork_reactor
from to 20271
 forked to 20411. EM.run true. reactor thread true
  in thread EM.run true. reactor thread false
        dl returning data size 0 # stuff in the inner EM.run runs, but after the caller exists
  result size 0
 thread returns
        dl got data size 306 11255
        dl got response 200
in timer

[Iñaki Baz Castillo]

2012/4/20 AsyncSubscriber <winson...@gmail.com>:

> First, I would need to use EM.fork_reactor rather than just fork (despite
> EM.run's implementation seems trying to take care of the same situation,)
> else the inner EM.run's stuff won't be executed at all.

> After I use
> EM.fork_reactor, the async requests will be made, however, the top-level
> code Client.dl would exit right the way before the inner EM.run returns.

But this makes sense, am I wrong? If process A forks to B, then
process A continues and will terminate (if responsability of the
developer not tu finish process A until B terminates).

> 1) How do I run nested inner loop where the invoker of the inner EM.run
> expect synchronous completion of the call. Why wouldn't using a new Thread
> to run the inner loop work?
>
> 2) The reason there is a fork call in the first place is that the process
> codes (upto now) do not need to run in EM loop;

> what's the proper way to terminate the EM loop in a forked process and continue?

I don't see the problem. Just call "exit" of the forked process which
runs EM loop and you are done.

BTW I see a "fork" call within the EM loop in your code. I consider
that really annoying. Why do you want that?

Being honest I don't understand properly your question or your issue.
However I can tell you that I use EM and fork in a personal project as
follows:

- Process A runs and performs some stuff (reading conf file, and so).

- Process A then forks and process B appears.

- Process A runs a EM loop with some listeners and event handlers.

- Process B also starts its own EM loop with different listeners.

- When Process A receives a TERM/QUIT/INT signal it kill process B and
exists (so it avoids process B to remain alive as an orphan process).

Not sure if this helps, I just mean that it works perfectly.

Regards.

--
Iñaki Baz Castillo
<i...@aliax.net>

[AsyncSubscriber]

The fork was there for legacy reason (some worker codes have some subtle dependency on the fork, like not closing open file descriptor causing an FD leak.)

It turns out EM and EM.fork_reactor are working fine but it is the peculiar intricacy with AMQP that causes the problems. I post the problem to to the AMQP group and the answer is that I will be out of luck unless I remove the forking. See

  https://groups.google.com/forum/?fromgroups#!topic/ruby-amqp/MQ7rW_cQ4WQ

Thanks!

[Iñaki Baz Castillo]

2012/4/23 AsyncSubscriber <winson...@gmail.com>:

> It turns out EM and EM.fork_reactor are working fine but it is the peculiar
> intricacy with AMQP that causes the problems. I post the problem to to the
> AMQP group and the answer is that I will be out of luck unless I remove the
> forking. See
>
>   https://groups.google.com/forum/?fromgroups#!topic/ruby-amqp/MQ7rW_cQ4WQ

So in short:

"To avoid both problems, start the EventMachine reactor and AMQP
connection after the master process forks workers."

And this is the reason I have no issues, because I run two instances
of EM in different process but both EM instances start running after
creating the fork :)