em-pg-client

The Ruby EventMachine driver interface to the PostgreSQL RDBMS. It is based on ruby-pg.

Author: Rafał Michalski (rafal at yeondir dot com)

Description

em-pg-client provides PG::EM::Client} class which inherits {PG::Connection. You can work with {PG::EM::Client} almost the same way you would work with PG::Connection.

The real difference begins when you turn the EventMachine reactor on.

require 'pg/em'
  
  pg = PG::EM::Client.new dbname: 'test'

  # no async
  pg.query('select * from foo') do |result|
    puts Array(result).inspect
  end

  # asynchronous
  EM.run do
    Fiber.new do
      pg.query('select * from foo') do |result|
        puts Array(result).inspect
      end
      EM.stop
    end.resume
  end

  # asynchronous + deferrable
  EM.run do
    df = pg.query_defer('select * from foo')
    df.callback { |result|
      puts Array(result).inspect
      EM.stop
    }
    df.errback {|ex|
      raise ex
    }
    puts "sent"
  end

Features

Requirements

Install

$ [sudo] gem install em-pg-client

Gemfile

gem "em-pg-client", "~> 0.3.4"

Github

git clone git://github.com/royaltm/ruby-em-pg-client.git

Usage

PG::Connection commands adapted to the EventMachine

Asynchronous, the EventMachine style:

For arguments of these methods consult their original (without the _defer suffix) counterparts in the PG::Connection manual.

Use callback with a block on the returned deferrable object to receive the result. In case of connect_defer and reset_defer the result is an instance of the PG::EM::Client}. The received client is in connected state and ready for the queries. Otherwise an instance of the {PG::Result is received. You may clear the obtained result object or leave it to gc.

To detect an error in the executed command call errback on the deferrable with a block. You should expect an instance of the raised Exception (usually PG::Error) as the block argument.

Reactor sensing methods, EM-Synchrony style:

The above methods call *_defer counterparts of themselves and yield from the current fiber awaiting for the result. The PG::Result instance (or PG::EM::Client for new) is then returned to the caller. If a code block is given, it will be passed the result as an argument. In that case the value of the block is returned instead and the result is being cleared (or in case of new - client is being closed) after block terminates.

These methods check if EventMachine’s reactor is running and the current fiber is not a root fiber. Otherwise the parent (thread-blocking) PG::Connection methods are being called.

You can call asynchronous, fiber aware and blocking methods without finishing the connection. You only need to start/stop EventMachine in between the asynchronous calls.

Although the em-synchrony provides very nice set of tools for the untangled EventMachine, you don’t really require it to fully benefit from the PG::EM::Client. Just wrap your asynchronous code in a fiber:

Fiber.new { ... }.resume

Special options

There are four special connection options and one of them is a standard pg option used by the async methods. You may pass them as one of the hash options to {PG::EM::Client.new} or {PG::EM::Client.connect_defer} or simply use the accessor methods to change them on the fly.

The options are:

Only connect_timeout is a standard libpq option, although changing it with the accessor method affects asynchronous functions only. See {PG::EM::Client} for more details.

Handling errors

Exactly like in pg:

EM.synchrony do
    begin
      pg.query('smellect 1')
    rescue => e
      puts "error: #{e.inspect}"
    end
    EM.stop
  end

with *_defer methods:

EM.run do
    pg.query_defer('smellect 1') do |ret|
      if ret.is_a?(Exception)
        puts "PSQL error: #{ret.inspect}"
      end
    end
  end

or

EM.run do
    pg.query_defer('smellect 1').callback do |ret|
      puts "do something with #{ret}"
    end.errback do |err|
      puts "PSQL error: #{err.inspect}"
    end
  end

Auto re-connecting in asynchronous mode

Connection reset is done in a non-blocking manner using reset_defer internally.

EM.run do
    Fiber.new do
      pg = PG::EM::Client.new async_autoreconnect: true

      try_query = lambda do
        pg.query('select * from foo') do |result|
          puts Array(result).inspect
        end
      end

      try_query.call
      system 'pg_ctl stop -m fast'
      system 'pg_ctl start -w'
      try_query.call

      EM.stop
    end.resume
  end

to enable this feature call:

pg.async_autoreconnect = true

Additionally the on_autoreconnect callback may be set on the connection. It’s being invoked after successfull connection restart, just before the pending command is sent again to the server.

Server-sent notifications - async style

Not surprisingly, there are two possible ways to wait for notifications, one with a deferrable:

pg = PG::EM::Client.new
  EM.run do
    pg.wait_for_notify_defer(7).callback do |notify|
      if notify
        puts "Someone spoke to us on channel: #{notify[:relname]} from #{notify[:be_pid]}"
      else
        puts "Too late, 7 seconds passed"
      end
    end.errback do |ex|
      puts "Connection to deep space lost..."
    end
    pg.query_defer("LISTEN deep_space") do
      pg.query_defer("NOTIFY deep_space") do
        puts "Reaching out... to the other worlds"
      end
    end
  end

and the other, using fibers:

EM.synchrony do
    pg = PG::EM::Client.new
    EM::Synchrony.next_tick do
      pg.query('LISTEN "some channel"')
      pg.query('SELECT pg_notify($1::text,$2::text)', ['some channel', 'with some message'])
    end
    pg.wait_for_notify(10) do |channel, pid, payload|
      puts "I've got notification on #{channel} #{payload}."
    end.tap do |name|
      puts "Whatever, I've been waiting too long already" if name.nil?
    end
  end

As you might have noticed, one does not simply wait for notifications, but one can also run some queries on the same connection at the same time, if one wishes so.

Connection Pool

Forever alone? Not anymore! There is a dedicated {PG::EM::ConnectionPool} class with dynamic pool for both types of asynchronous commands (deferral and fiber-synchronized).

It also provides a transaction method which locks the in-transaction connection to the calling fiber and allows to execute commands on the same connection within a transaction block. The transactions may be nested. See also docs for the {PG::EM::Client#transaction} method.

Parallel async queries

require 'pg/em/connection_pool'
  require 'em-synchrony'

  EM.synchrony do
    pg = PG::EM::ConnectionPool.new(size: 2, dbname: 'test')

    multi = EM::Synchrony::Multi.new
    multi.add :foo, pg.query_defer('select pg_sleep(1)')
    multi.add :bar, pg.query_defer('select pg_sleep(1)')

    start = Time.now
    res = multi.perform
    # around 1 sec.
    puts Time.now - start

    EM.stop
  end

Fiber Concurrency

require 'pg/em/connection_pool'
  require 'em-synchrony'
  require "em-synchrony/fiber_iterator"

  EM.synchrony do
    concurrency = 5
    queries = (1..10).map {|i| "select pg_sleep(1); select #{i}" }

    pg = PG::EM::ConnectionPool.new(size: concurrency, dbname: 'test')

    start = Time.now
    EM::Synchrony::FiberIterator.new(queries, concurrency).each do |query|
      pg.query(query) do |result|
        puts "recv: #{result.getvalue(0,0)}"
      end
    end
    # around 2 secs.
    puts Time.now - start

    EM.stop
  end

API Changes

0.2.x -> 0.3.x

There is a substantial difference in the API between this and the previous releases. The idea behind it was to make this implementation as much compatible as possible with the threaded pg interface. E.g. the #async_exec is now an alias to #exec.

The other reason was to get rid of the ugly em / em-synchrony duality.

as their names were confusing due to the unfortunate #async_exec.

0.1.x -> 0.2.x

Bugs/Limitations

TODO:

More Info

This implementation makes use of non-blocking: PGConn#is_busy and PGConn#consume_input methods. Depending on the size of queried results and the concurrency level, the gain in overall speed and responsiveness of your application might be actually quite huge. See {file:BENCHMARKS.md BENCHMARKING}.

Thanks

The greetz go to: