This post originated from an RSS feed registered with Ruby Buzz by Guy Naor. Original Post: Improving Postgres LISTEN/NOTIFY Support in Ruby/Rails Feed Title: Famundo - The Dev Blog Feed URL: http://devblog.famundo.com/xml/rss/feed.xml Feed Description: A blog describing the development and related technologies involved in creating famundo.com - a family management sytem written using Ruby On Rails and postgres Latest Ruby Buzz Posts Latest Ruby Buzz Posts by Guy Naor Latest Posts From Famundo - The Dev Blog

The Postgres DB server has a very cool mechanism for passing messages between the backend and clients and between the clients, using a simple LISTEN and NOTIFY system.

The way it works is that any client can register to LISTEN to a named event, and whenever any other client (or the backend using triggers/functions) issues a NOTIFY command for the named event, all listening clients will be notified of it. You can find the full documentation here and here.

A usage scenario is a background process that need to act at specific times based on database changes. The process can sleep not consuming any CPU, until a notification arrives, it wakes up does it's thing, and go back to sleep. It can also be used to keep cache consistency or to manage locks between clients. And many other scenarios.

The ruby postgres driver (C based version) has some support for this using the get_notify method. But it has some limitations which I wanted to remove:

1. Notifications arrive at the client whenever they are sent, but to know they arrived you need to poll the library, in effect making it again a poll solution. I wanted to change it to a solution that uses no CPU while waiting, and also isn't dependant on polling intervals.

2. Currently you have to issue a query to know that the notification arrived. This makes the polling solution even less attractive as you have to generate some query to know if something arrived.

The Postgres library provides a function called PQconsumeInput that solves the problem of having to send a query to see if a notification arrives. Once executed, calling get_notify will return any waiting notifications (call it multiple times until a nil is returned, as each call returns one notification).

To completely remove the need to poll (even with PQconsumeInput you still need to poll) I added another function based on an example from the Postgres documentation. This functions will get the current connection socket, and using a select() (system one, not sql) wait for any new activity on the socket. Once a notification arrives, the select() will stop blocking, and we can call consume_input and then get_notify() to get any waiting notifications. Put it in a loop and go back to waiting after handling the incoming events, and you have a process that waits for notifications without using any CPU time while waiting.

Please note that in order to use those new functions in rails applications you have to use the raw_connection:

ActiveRecord::Base.connection.raw_connection.consume_input

Here is diff to the Ruby driver (you will need to recompile it). If you would like to also have the patched C file, let me know.

--- ruby-postgres-0.7.1/postgres.c   2003-01-05 17:38:20.000000000 -0800
+++ postgres.c  2006-12-04 13:04:55.000000000 -0800
@@ -439,6 +439,40 @@
     return ary;
 }

+
+static VALUE
+pgconn_consume_input(obj)
+    VALUE obj;
+{
+    PGconn *conn = get_pgconn(obj);
+    if (PQconsumeInput(conn) == 0)
+      rb_raise(rb_ePGError, PQerrorMessage(conn));
+    return Qnil;
+}
+
+
+static VALUE
+pgconn_wait_for_activity(obj)
+    VALUE obj;
+{
+    fd_set input_mask;
+    PGconn *conn = get_pgconn(obj);
+    int    sock  = PQsocket(conn);
+
+    if (sock < 0)
+         rb_raise(rb_ePGError, "Bad connection socket");
+
+    FD_ZERO(&input_mask);
+    FD_SET(sock, &input_mask);
+
+    /* Wait for something to happen on the socket */
+    if (select(sock + 1, &input_mask, NULL, NULL, NULL) < 0)
+        rb_raise(rb_ePGError, "Socket select() failed");
+
+    return Qnil;
+}
+
+
 static VALUE pg_escape_regex;
 static VALUE pg_escape_str;
 static ID    pg_gsub_bang_id;
@@ -1443,6 +1477,8 @@
     rb_define_method(rb_cPGconn, "getline", pgconn_getline, 0);
     rb_define_method(rb_cPGconn, "endcopy", pgconn_endcopy, 0);
     rb_define_method(rb_cPGconn, "notifies", pgconn_notifies, 0);
+    rb_define_method(rb_cPGconn, "consume_input", pgconn_consume_input, 0);
+    rb_define_method(rb_cPGconn, "wait_for_activity", pgconn_wait_for_activity, 0);

 #ifdef HAVE_PQSETCLIENTENCODING
     rb_define_method(rb_cPGconn, "client_encoding",pgconn_client_encoding, 0);