GDP

#!/usr/bin/env python

import gdp

import sys

import threading

import time

import pprint

# Assumes that logs are already created, and are named as follows:

#   PREFIXlog1, PREFIXlog2, PREFIXlog3...

# Global variables

PREFIX = "edu.eecs.berkeley.mor.aug30.threading-test."

lock = None

start = None

def localPrint(s):

    global lock

    global start

    lock.acquire()

    print "%9.6f ## %s] %s" % ((time.time()-start),

                            threading.current_thread().getName(), s)

    sys.stdout.flush()

    lock.release()

def sleep(t):

    # localPrint("Sleeping for %i" % t)

    time.sleep(t)

def appendThread(logname, sleepBefore, numRecords):

    h = gdp.GDP_GCL(gdp.GDP_NAME(PREFIX + logname), gdp.GDP_MODE_AO)

    localPrint(str(h))

    sleep(sleepBefore)

    for i in xrange(numRecords):

        sleep(1)

        data = "\"%s-%s-%d\"" % (logname,

                        threading.current_thread().getName(), i)

        localPrint("Appending %s to %s" % (data, logname))

        h.append({"data": data})

def subscribeThread(logname):

    h = gdp.GDP_GCL(gdp.GDP_NAME(PREFIX + logname), gdp.GDP_MODE_RO)

    localPrint(str(h))

    h.subscribe(0, 0, None)

    while True:

        event = h.get_next_event(None)

        localPrint("Got data: %s on %s" % (event["datum"]["data"], logname))

        # pprint.pprint(event)

def main(numA, numS, delay, numR):

    gdp.gdp_init()

    # gdp.dbg_set("*=15")

    global start

    start = time.time()

    global lock

    lock = threading.Lock()

    for i in range(numA):

        localPrint("Starting AThread%d" % (i+1))

        logname = "log"+str(i+1)

        t = threading.Thread(target=appendThread,

                            name="AThread"+str(i+1),

                            args=(logname, delay*(i+1), numR))

        t.daemon = False

        t.start()

    for i in range(numS):

        localPrint("Starting SThread%d" % (i+1))

        logname = "log"+str(i+1) 

        t = threading.Thread(target=subscribeThread,

                            name="SThread"+str(i+1),

                            args=(logname,))

        t.daemon = True

        t.start()

USAGE = """

    Runs `numA` threads appending data to `numA` logs (one log per

    thread). Also starts `numS` subscriber threads (again, one log

    per thread). `delay` and numRec` control the execution pattern.

    `delay` is the delay in seconds before starting to append data

    in an append thread, `numRec` is the number of records that are

    appended in each thread (one second delay between consecutive

    appends within a thread)

    Assumes that the logs are pre-created

"""

if __name__=="__main__":

    if len(sys.argv)<5:

        print "Uusage: %s numA numS delay numRec" % sys.argv[0]

        print USAGE

        sys.exit(1)

    main(int(sys.argv[1]), int(sys.argv[2]), int(sys.argv[3]), int(sys.argv[4]))