q.py

import Queue
import threading
import time
import os
import random
from logger import Logger
from config import Config

# input queue to be processed by many threads
q_in = Queue.Queue(maxsize=0)

# output queue to be processed by one thread
q_out = Queue.Queue(maxsize=0)

# q_sub = Queue.Queue(maxsize=0)

# number of worker threads to complete the processing
num_worker_threads = 10

epoch_start = int( time.time() )
random.seed()

# process that each worker thread will execute until the Queue is empty
def worker():
    l = Logger(Config)
    while True:
        # get item from queue, do work on it, let queue know processing is done for one item
        item = q_in.get()
        if random.randint(0, 1) == 1:
            l.debug( "QUEUE TEST: WRK %d is done (%d)" % (item, q_in.qsize()) )
            q_out.put(do_work(item))
            q_in.task_done()
        else:
            l.debug( "QUEUE TEST: WRK %d is slacking (%d)" % (item, q_in.qsize()) )
            time.sleep( 1 )
            # q_sub.put( item )
            q_in.put( item )
            q_in.task_done()

# squares a number and returns the number and its square
def do_work(item):
    return (item,item*item)

# another queued thread we will use to print output
def printer():
    l = Logger(Config)
    while True:
        # get an item processed by worker threads and print the result. Let queue know item has been processed
        item = q_out.get()
        l.debug( "QUEUE TEST: PRT " + ("%d squared is : %d" % item) + (" (%d)" % q_out.qsize() ) )
        q_out.task_done()

def manager ():
    l = Logger(Config)
    while True:
        item = q_sub.get()
        q_in.put( item )
        l.debug( "QUEUE TEST: SUB %d is handled (%d)" % (item, q_sub.qsize()) )
        q_sub.task_done()

# launch all of our queued processes
def main():
    # Launches a number of worker threads to perform operations using the queue of inputs
    for i in range(num_worker_threads):
         t = threading.Thread(target=worker)
         t.daemon = True
         t.start()

    # launches a single "printer" thread to output the result (makes things neater)
    t = threading.Thread(target=printer)
    t.daemon = True
    t.start()

    # _t = threading.Thread( target=manager )
    # _t.daemon = True
    # _t.start()

    # put items on the input queue (numbers to be squared)
    for item in range(10):
        q_in.put(item)

    # wait for two queues to be emptied (and workers to close)   
    q_in.join()       # block until all tasks are done
    q_out.join()
    # q_sub.join()

l = Logger(Config)
l.debug( "QUEUE TEST: Started at %i with PID: %i " % (epoch_start, os.getpid()) )
main()
epoch_end = int( time.time() )
l.debug( "QUEUE TEST: Process complete via %d with PID: %d" % ((epoch_end - epoch_start), os.getpid()) )