techrec/server/processqueue.py

import multiprocessing


class JobQueue(object):
    def __init__(self):
        self.pool = multiprocessing.Pool(processes=1)
        self.last_job_id = 0
        self.jobs = {}  # job_id: AsyncResult

    def submit(self, function, *args, **kwargs):
        self.last_job_id += 1
        job_id = self.last_job_id

        def clean_jobs(res):
            '''this callback will remove the job from the queue'''
            del self.jobs[job_id]
        self.jobs[job_id] = self.pool.apply_async(function, args, kwargs,
                                                  clean_jobs)
        return job_id

    def check_job(self, job_id):
        '''
        If the job is running, return the asyncResult.
        If it has already completed, returns True.
        If no such job_id exists at all, returns False
        '''
        if job_id <= 0:
            raise ValueError("non-valid job_id")
        if self.last_job_id < job_id:
            return False
        if job_id in self.jobs:
            return self.jobs[job_id]
        return True

    def join(self):
        self.pool.close()
        self.pool.join()
        self.pool = None


def simulate_long_job(recid=None, starttime=None, endtime=None, name='', filename=None):
    from time import sleep
    print "evviva " + name
    sleep(2)
    print "lavoro su " + name
    sleep(2)
    print "done su " + name
_queue = None


def get_process_queue():
    global _queue
    if _queue is None:
        _queue = JobQueue()
    return _queue

if __name__ == '__main__':
    from datetime import datetime
    n = datetime.now()

    def sleep(n):
        import time
        print "Inizio %d" % n
        time.sleep(n)
        print "Finisco %d" % n
        return n

    get_process_queue().submit(sleep, 3)
    get_process_queue().submit(sleep, 3)
    get_process_queue().join()
    print get_process_queue().jobs
    delta = (datetime.now() - n).total_seconds()
    print delta
    assert 5 < delta < 7