jobqueue.py

import gevent
import gevent.queue
import gevent.lock
from functools import partial

class JobQueue:

    class Job:
        def __init__(self, run=None, *args, **kwargs):
            self.callback = kwargs.pop('dispatch_callback', None)
            self._run = partial(run, *args, **kwargs)

        def start(self):
            return self._run()


    def __init__(self):
        self.g = gevent.spawn(self._run)
        self.queue = gevent.queue.Queue()

    def dispatch(self, run, *args, **kwargs):
        dispatch_wait = kwargs.pop('dispatch_wait', False)
        job = self.Job(run, *args, **kwargs)
        self.queue.put(job)
        if dispatch_wait:
            self.dispatch_barrier()

    def dispatch_barrier(self):
        def dispatch_barrier_lock(lock):
            lock.release()

        lock = gevent.lock.Semaphore(0)
        job = self.Job(dispatch_barrier_lock, lock)
        self.queue.put(job, lock)
        lock.acquire()



    def _run(self):
        for job in self.queue:
            res = job.start()
            if job.callback:
                job.callback(res)

test.py

import gevent
from random import random
from jobqueue import JobQueue

def print_num(num):
    gevent.sleep(random() * 5)
    print "job %d done" % num
    return num

def callback_num(num):
    print 'callback %d' % num

print '*** Starting JobQueue test'
print

print '5 completely async jobs'
for i in range(5):
    print "dispatching %d" % i
    a = gevent.spawn(print_num, i)
gevent.wait()
print

q = JobQueue()
print '5 queued async jobs w/o wait'
for i in range(5):
    print "dispatching %d" % i
    q.dispatch(print_num, i)
q.dispatch_barrier()
print

q = JobQueue()
print '5 queued async jobs w/ wait'
for i in range(5):
    print "dispatching %d" % i
    q.dispatch(print_num, i, dispatch_wait=True)
q.dispatch_barrier()
print

q = JobQueue()
print '5 queued async jobs w/o wait - callback'
for i in range(5):
    print "dispatching %d" % i
    q.dispatch(print_num, i, dispatch_callback=callback_num)
q.dispatch_barrier()
print

q = JobQueue()
print '5 queued async jobs w/ wait - callback'
for i in range(5):
    print "dispatching %d" % i
    q.dispatch(print_num, i, dispatch_callback=callback_num, dispatch_wait=True)
q.dispatch_barrier()
print