Reputation: 1810
I have a consumer which pulls messages off of a queue via an evented subscription. It takes those messages and then connects with a rather slow http interface. I have a worker pool of 8 and once those are all filled up I need to stop pulling requests from the queue and have the fibers that are working on the http jobs keep working. Here is an example I've thrown together.
def send_request(callback)
EM.synchrony do
while $available <= 0
sleep 2
puts "sleeping"
end
url = 'http://example.com/api/Restaurant/11111/images/?image%5Bremote_url%5D=https%3A%2F%2Firs2.4sqi.net%2Fimg%2Fgeneral%2Foriginal%2F8NMM4yhwsLfxF-wgW0GA8IJRJO8pY4qbmCXuOPEsUTU.jpg&image%5Bsource_type_enum%5D=3'
result = EM::Synchrony.sync EventMachine::HttpRequest.new(url, :inactivity_timeout => 0).send("apost", :head => {:Accept => 'services.v1'})
callback.call(result.response)
end
end
def display(value)
$available += 1
puts value.inspect
end
$available = 8
EM.run do
EM.add_periodic_timer(0.001) do
$available -= 1
puts "Available: #{$available}"
puts "Tick ..."
puts send_request(method(:display))
end
end
I have found that if I call sleep within a while loop in the synchrony block, the reactor loop gets stuck. If I call sleep within an if statement(sleeping just once) then most times it is enough time for the requests to finish but it is unreliable at best. If I use EM::Synchrony.sleep, then the main reactor loop will keep creating new requests.
Is there a way to pause the main loop but have the fibers finish their execution?
Upvotes: 1
Views: 1319
Reputation: 4900
sleep 2
...
add_periodic_timer(0.001)
Are you serious?
Have you ever though how many send_request
's are sleeping in the loop? And it's adding 1000 every second.
What about this:
require 'eventmachine'
require 'em-http'
require 'fiber'
class Worker
URL = 'http://example.com/api/whatever'
def initialize callback
@callback = callback
end
def work
f = Fiber.current
loop do
http = EventMachine::HttpRequest.new(URL).get :timeout => 20
http.callback do
@callback.call http.response
f.resume
end
http.errback do
f.resume
end
Fiber.yield
end
end
end
def display(value)
puts "Done: #{value.size}"
end
EventMachine.run do
8.times do
Fiber.new do
Worker.new(method(:display)).work
end.resume
end
end
Upvotes: 1