pg_lock.py

#!/usr/bin/env python

import sys
import time
import threading
import urllib

import psycopg2


DB_HOST = '127.0.0.1'
DB_USER = 'postgres'
DB_NAME = 'postgres'

HTTPS_URL = 'https://google.com/'
NUM_HTTPS_THREADS = 10


def connect():
    conn = psycopg2.connect(
        host=DB_HOST, user=DB_USER,
        database=DB_NAME, sslmode='require')
    return conn


class Worker(threading.Thread):

    def __init__(self):
        super(Worker, self).__init__()
        self._stop = threading.Event()

    def stop(self):
        self._stop.set()

    def stopped(self):
        return self._stop.isSet()

    def run(self):
        i = 0
        while not self.stopped():
            i += 1
            self.do_work(i)


class Libpq(Worker):

    def do_work(self, i):
        conn = connect()
        cur = conn.cursor()
        cur.execute('SELECT 1')
        cur.close()
        conn.close()
        if not i % 50:
            sys.stdout.write('+')


class HTTPS(Worker):

    def do_work(self, i):
        urllib.urlopen(HTTPS_URL).read()
        if not i % 50:
            sys.stdout.write('=')


def main():
    # make sure libpq has set its locking callbacks
    conn = connect()
    cur = conn.cursor()
    cur.execute('SELECT 1')
    cur.close()
    conn.close()

    threads = [Libpq()]
    threads += [HTTPS() for _ in range(NUM_HTTPS_THREADS)]

    print 'starting worker threads'

    try:
        for t in threads:
            t.start()
        while True:
            time.sleep(1000)
    except KeyboardInterrupt:
        for t in threads:
            t.stop()

    for t in threads:
        t.join()


if __name__ == "__main__":
    sys.exit(main())