igrigorik / em-synchrony

Fiber aware EventMachine clients and convenience classes
http://www.igvita.com/2010/03/22/untangling-evented-code-with-ruby-fibers
MIT License
1.04k stars 151 forks source link

EM-Synchrony

Gem Version Analytics Build Status

Collection of convenience classes and primitives to help untangle evented code, plus a number of patched EM clients to make them Fiber aware. To learn more, please see: Untangling Evented Code with Ruby Fibers.

Supported clients:

Other clients with native Fiber support:

Fiber-aware Iterator: mixing sync / async code

Allows you to perform each, map, inject on a collection of any asynchronous tasks. To advance the iterator, simply call iter.next, or iter.return(result). The iterator will not exit until you advance through the entire collection. Additionally, you can specify the desired concurrency level! Ex: crawling a web-site, but you want to have at most 5 connections open at any one time.

require "em-synchrony"
require "em-synchrony/em-http"

EM.synchrony do
    concurrency = 2
    urls = ['http://url.1.com', 'http://url2.com']

    # iterator will execute async blocks until completion, .each, .inject also work!
    results = EM::Synchrony::Iterator.new(urls, concurrency).map do |url, iter|

        # fire async requests, on completion advance the iterator
        http = EventMachine::HttpRequest.new(url).aget
        http.callback { iter.return(http) }
        http.errback { iter.return(http) }
    end

    p results # all completed requests
    EventMachine.stop
end

Or, you can use FiberIterator to hide the async nature of em-http:

require "em-synchrony"
require "em-synchrony/em-http"
require "em-synchrony/fiber_iterator"

EM.synchrony do
    concurrency = 2
    urls = ['http://url.1.com', 'http://url2.com']
    results = []

    EM::Synchrony::FiberIterator.new(urls, concurrency).each do |url|
        resp = EventMachine::HttpRequest.new(url).get
    results.push resp.response
    end

    p results # all completed requests
    EventMachine.stop
end

Fiber-aware ConnectionPool shared by a fiber:

Allows you to create a pool of resources which are then shared by one or more fibers. A good example is a collection of long-lived database connections. The pool will automatically block and wake up the fibers as the connections become available.

require "em-synchrony"
require "em-synchrony/mysql2"

EventMachine.synchrony do
    db = EventMachine::Synchrony::ConnectionPool.new(size: 2) do
        Mysql2::EM::Client.new
    end

    multi = EventMachine::Synchrony::Multi.new
    multi.add :a, db.aquery("select sleep(1)")
    multi.add :b, db.aquery("select sleep(1)")
    res = multi.perform

    p "Look ma, no callbacks, and parallel MySQL requests!"
    p res

    EventMachine.stop
end

Fiber-aware Multi interface: parallel HTTP requests

Allows you to fire simultaneous requests and wait for all of them to complete (success or error) before advancing. Concurrently fetching many HTTP pages at once is a good example; parallel SQL queries is another. Technically, this functionality can be also achieved by using the Synchrony Iterator shown above.

require "em-synchrony"
require "em-synchrony/em-http"
EventMachine.synchrony do
    multi = EventMachine::Synchrony::Multi.new
    multi.add :a, EventMachine::HttpRequest.new("http://www.postrank.com").aget
    multi.add :b, EventMachine::HttpRequest.new("http://www.postrank.com").apost
    res = multi.perform

    p "Look ma, no callbacks, and parallel HTTP requests!"
    p res

    EventMachine.stop
end

Fiber-aware & EventMachine backed TCPSocket:

This is dangerous territory - you've been warned. You can patch your base TCPSocket class to make any/all libraries depending on TCPSocket be actually powered by EventMachine and Fibers under the hood.

require "em-synchrony"
require "lib/em-synchrony"
require "net/http"

EM.synchrony do
  # replace default Socket code to use EventMachine Sockets instead
  TCPSocket = EventMachine::Synchrony::TCPSocket

  Net::HTTP.get_print 'www.google.com', '/index.html'
  EM.stop
end

Inline synchronization & Fiber sleep:

Allows you to inline/synchronize any callback interface to behave as if it was a blocking call. Simply pass any callback object to Synchrony.sync and it will do the right thing: the fiber will be resumed once the callback/errback fires. Likewise, use Synchrony.sleep to avoid blocking the main thread if you need to put one of your workers to sleep.

require "em-synchrony"
require "em-synchrony/em-http"
EM.synchrony do
  # pass a callback enabled client to sync to automatically resume it when callback fires
  result = EM::Synchrony.sync EventMachine::HttpRequest.new('http://www.gooogle.com/').aget
  p result

  # pause execution for 2 seconds
  EM::Synchrony.sleep(2)

  EM.stop
end

Async ActiveRecord:

Allows you to use async ActiveRecord within Rails and outside of Rails (see async-rails). If you need to control the connection pool size, use rack/fiber_pool.

require "em-synchrony"
require "em-synchrony/mysql2"
require "em-synchrony/activerecord"

ActiveRecord::Base.establish_connection(
  :adapter => 'em_mysql2',
  :database => 'widgets'
)

result = Widget.all.to_a

Hooks

em-synchrony already provides fiber-aware calls for sleep, system and defer. When mixing fiber-aware code with other gems, these might use non-fiber-aware versions which result in unexpected behavior: calling sleep would pause the whole reactor instead of a single fiber. For that reason, hooks into the Kernel are provided to override the default behavior to e.g. add logging or redirect these calls their fiber-aware versions.

# Adding logging but still executes the actual sleep
require "em-synchrony"
log = Logger.new(STDOUT)
EM::Synchrony.on_sleep do |*args|
  log.warn "Kernel.sleep called by:"
  caller.each { |line| log.warn line }
  sleep(*args) # Calls the actual sleep
end
# Redirects to EM::Synchrony.sleep
require "em-synchrony"
log = Logger.new(STDOUT)
EM::Synchrony.on_sleep do |*args|
  EM::Synchrony.sleep(*args)
end

License

The MIT License - Copyright (c) 2011 Ilya Grigorik