Using Skytools PGQ for targeted copying of data in cluster

In a clustered database where some data on one node is shared with some
other nodes there is a need to provide targeted replication on row level.
It is not possible with Slon replication for replication copies all data.
Using PGQ in a star schema where all nodes are interconnected would require
<number of nodes>^2 processes to run (high number for even a 256-node
cluster).

Solution: Push-Pull queue schema using pgq.RemoteConsumer to send copy
messages of object data.

There are 2 ticker processes (push_queue and pull_queue) and two consumer
processes per node making total number of processes <number of nodes>X2+2.

On push queue side sending node inserts one event per target node:
pgq.insert_event('push_queue','copy', '', <node
number>,'action=ins&id=1&name=test', '' ,'' );

Push queue consumer calls function to relay same event into pull queue
pgq.insert_event('pull_queue','copy', '', <node
number>,'action=ins&id=1&name=test', '' ,'' )

Pull queue consumer before issuing "ev.done" checks that the node it is
running on is the event's intended recipient. It ensures against data loss
should a node become unresponsive as failed event is kept in queue until it
reconnects.

## pull.py
import sys, os, pgq, skytools
import psycopg2

class Copier(pgq.RemoteConsumer):
def __init__(self, args):
pgq.RemoteConsumer.__init__(self, "pull_ticker", "src_db",
"dst_db", args)
### GET "node" SETTING FROM .ini
### MUST ADD "node = <node number>" TO .ini
self.node = self.cf.get("node")

def process_remote_batch(self, db, batch_id, event_list, dst_db):
for ev in event_list:
if ev.type <> 'copy':
ev.tag_done()
### CHECK TARGET IS CURRENT NODE
if self.node == ev.ev_extra1:
### COMMIT EVENT IF IT IS PROCESSING ON TARGET SERVER
ev.tag_done()
cur = dst_db.cursor()
cur.execute("select __data_copier('%s', '%s')" %
(ev.ev_extra1, ev.ev_extra2))
records = cur.fetchall()
dst_db.commit()

if __name__ == '__main__':
script = Copier(sys.argv[1:])
script.start()