CODEWITHSUNDEEP
pythonasynchronoustornado
cehmja
cehmja

Reputation: 140

Tornado asynchronous job in tornado gen coroutine

I've written an application which takes a job from a queue and executes it asynchronously.

def job(self):
    print 'In job'
    time.sleep(0.01)

@gen.coroutine
def start_jobs(self):
    jobs = filter(lambda x: x['status'] == 0, self.queue)
    for job in jobs:
        yield self.job()
    print 'exit from start job'

But, this code does not work.

Output:

In job

In job

In job etc

How do I do it correctly?

How do I make it work with Futures, and and is there a simpler way to do it with Tornado?

Upvotes: 0

Views: 303

Answers (1)

A. Jesse Jiryu Davis
A. Jesse Jiryu Davis

Reputation: 24027

Never call time.sleep in Tornado! Use yield gen.sleep instead.

Install Toro with pip install toro and use a JoinableQueue:

import random
from tornado import ioloop, gen
import toro


class C(object):
    def __init__(self):
        self.queue = toro.JoinableQueue()

    @gen.coroutine
    def start_jobs(self):
        while True:
            job_id = yield self.queue.get()
            self.job(job_id)

    @gen.coroutine
    def job(self, job_id):
        print 'job_id', job_id
        yield gen.sleep(random.random())
        print 'job_id', job_id, 'done'
        self.queue.task_done()


c = C()
for i in range(5):
    c.queue.put_nowait(i)

c.start_jobs()

io_loop = ioloop.IOLoop.instance()

# block until all tasks are done
c.queue.join().add_done_callback(lambda future: io_loop.stop())
io_loop.start()

Starting with Tornado 4.2, Toro is part of Tornado, so you can just do queue = tornado.queues.Queue() instead of using a Toro JoinableQueue:

http://tornado.readthedocs.org/en/latest/releases/v4.2.0.html#new-modules-tornado-locks-and-tornado-queues

Upvotes: 3

Related Questions