Using the Postgresql Gem Async

using the postgresql gem async

The pg library provides full support for PostgreSQL's asynchronous API. I've added an example of how to use it to the samples/ directory:

#!/usr/bin/env ruby

require 'pg'

# This is a example of how to use the asynchronous API to query the
# server without blocking other threads. It's intentionally low-level;
# if you hooked up the PGconn#socket to some kind of reactor, you
# could make this much nicer.

TIMEOUT = 5.0 # seconds to wait for an async operation to complete
CONN_OPTS = {
    :host     => 'localhost',
    :dbname   => 'test',
    :user     => 'jrandom',
    :password => 'banks!stealUR$',
}

# Print 'x' continuously to demonstrate that other threads aren't
# blocked while waiting for the connection, for the query to be sent,
# for results, etc. You might want to sleep inside the loop or 
# comment this out entirely for cleaner output.
progress_thread = Thread.new { loop { print 'x' } }

# Output progress messages
def output_progress( msg )
    puts "\n>>> #{msg}\n"
end

# Start the connection
output_progress "Starting connection..."
conn = PGconn.connect_start( CONN_OPTS ) or 
    abort "Unable to create a new connection!"
abort "Connection failed: %s" % [ conn.error_message ] if
    conn.status == PGconn::CONNECTION_BAD

# Now grab a reference to the underlying socket so we know when the
# connection is established
socket = IO.for_fd( conn.socket )

# Track the progress of the connection, waiting for the socket to 
# become readable/writable before polling it
poll_status = PGconn::PGRES_POLLING_WRITING
until poll_status == PGconn::PGRES_POLLING_OK ||
      poll_status == PGconn::PGRES_POLLING_FAILED

    # If the socket needs to read, wait 'til it becomes readable to
    # poll again
    case poll_status
    when PGconn::PGRES_POLLING_READING
        output_progress "  waiting for socket to become readable"
        select( [socket], nil, nil, TIMEOUT ) or
            raise "Asynchronous connection timed out!"

    # ...and the same for when the socket needs to write
    when PGconn::PGRES_POLLING_WRITING
        output_progress "  waiting for socket to become writable"
        select( nil, [socket], nil, TIMEOUT ) or
            raise "Asynchronous connection timed out!"
    end

    # Output a status message about the progress
    case conn.status
    when PGconn::CONNECTION_STARTED
        output_progress "  waiting for connection to be made."
    when PGconn::CONNECTION_MADE
        output_progress "  connection OK; waiting to send."
    when PGconn::CONNECTION_AWAITING_RESPONSE
        output_progress "  waiting for a response from the server."
    when PGconn::CONNECTION_AUTH_OK
        output_progress "  received authentication; waiting for " +
                        "backend start-up to finish."
    when PGconn::CONNECTION_SSL_STARTUP
        output_progress "  negotiating SSL encryption."
    when PGconn::CONNECTION_SETENV
        output_progress "  negotiating environment-driven " +
                        "parameter settings."
    end

    # Check to see if it's finished or failed yet
    poll_status = conn.connect_poll
end

abort "Connect failed: %s" % [ conn.error_message ] unless 
    conn.status == PGconn::CONNECTION_OK

output_progress "Sending query"
conn.send_query( "SELECT * FROM pg_stat_activity" )

# Fetch results until there aren't any more
loop do
    output_progress "  waiting for a response"

    # Buffer any incoming data on the socket until a full result 
    # is ready. 
    conn.consume_input
    while conn.is_busy
        select( [socket], nil, nil, TIMEOUT ) or
            raise "Timeout waiting for query response."
        conn.consume_input
    end

    # Fetch the next result. If there isn't one, the query is 
    # finished
    result = conn.get_result or break

    puts "\n\nQuery result:\n%p\n" % [ result.values ]
end

output_progress "Done."
conn.finish

if defined?( progress_thread )
    progress_thread.kill
    progress_thread.join
end

I'd recommend that you read the documentation on the PQconnectStart function and the Asynchronous Command Processing section of the PostgreSQL manual, and then compare that with the sample above.

I haven't used EventMachine before, but if it lets you register a socket and callbacks for when it becomes readable/writable, I'd think it'd be fairly easy to integrate database calls into it.

I've been meaning to use the ideas in Ilya Grigorik's article on using Fibers to clean up evented code to make the async API easier to use, but that's a ways off. I do have a ticket open to track it if you're interested/motivated to do it yourself.

Ruby EventMachine with PostgreSQL

This looks like it works with ActiveRecord:

https://github.com/mperham/em_postgresql

Asynchronous database access with EventMachine

Personally I prefer to use DB drivers directly in EM based servers.

The reason for this is that the type of projects that require EM servers also require high performance but fairly minimal amount of processing. As such, it is often simpler to put together few SQL statements rather than configure ORM properly and have it work with EM.

With ORM, especially advanced ones, it is far too easy to introduce a bottleneck of some kind. Additionaly, EM is still not very widespread so the chances of hitting some weird bug are larger if you try to use ORM over a simple driver.

Mysql2 gem has builtin support for async operations. Postgresql I haven't used lately so not sure.

How to execute an async Query with libpq

You misunderstand something here: PQsendQuery() sends an asynchronous query, meaning, that the command does not block until the results are available. However, you may not call PQsendQuery() again before all results were read!

From the manpage:

After successfully calling PQsendQuery, call PQgetResult one or more times to obtain the results. PQsendQuery may not be called again (on the same connection) until PQgetResult has returned a null pointer, indicating that the command is done.

As the manpage indicates, you would have to open several connections to run several concurrent queries. The different results are then associated with the different database handles.

The PostgreSQLc async io api does not provide callback mechanisms; however, this is not too hard to build yourself. libevent can be used, since you can retrieve the socket file descriptor with PQsocket(). However, quite some glue is needed to make it work.

Rails serialization error for async geocoding with ahoy gem using mongoid

Try adding the GlobalId mixin to your model

class Visit
  include GlobalID::Identification
end

postgresql_adapter async_exec error on rails rspec test using Capybara + FactoryGirl

Ok I figured it out following the answer here: FactoryGirl screws up rake db:migrate process

So the problem was coming from factorygirl. I updated my Gemfile to look like this:

gem "factory_girl_rails", :require => false

And then also add this:

require 'factory_girl_rails'

to my spec_helper.rb file and that fixed all the issues both localy and on circleci :)

---

Related Topics