Tornado IOLoop Exception in callback None in Celer

问题:

I am using tornado.ioloop inside celery worker because I need to use mongodb.

class WorkerBase():
    @gen.engine
    def foo(self,args,callback)
        bar = ['Python','Celery','Javascript','HTML']

        # ... process something ....

        callback(bar)

    @gen.engine
    def RunMyTask(self,args):

        result = yield gen.Task(self.foo,args=args)
        # Stop IOLoop instance
        IOLoop.instance().stop()


@task(name="MyWorker",base=WorkerBase)
def CeleryWorker(args):
    # This works because i'm adding base as WorkerBase
    CeleryWorker.RunMyTask(args)
    IOLoop.instance().start()
    return True

When I am invoking a task it gives an error saying:

[2014-10-02 12:12:11,561: ERROR/Worker-4] Exception in callback None
Traceback (most recent call last):
    File "/var/www/myapp/env/local/lib/python2.7/site-packages/tornado/ioloop.py", line 832, in start
fd_obj, handler_func = self._handlers[fd]
KeyError: 16

or

[2014-10-02 12:12:11,561: ERROR/Worker-4] Exception in callback None
Traceback (most recent call last):
    File "/var/www/myapp/env/local/lib/python2.7/site-packages/tornado/ioloop.py", line 832, in start
fd_obj, handler_func = self._handlers[fd]
KeyError: 14

These errors are not consistent. Is there any raise condition?

回答1:

This looks like a threading problem. I'm not familiar with celery's threading model but it looks like it's starting multiple copies of CeleryWorker, each of which is trying to run the same singleton IOLoop.instance(). Each worker thread needs its own IOLoop if you're going to run it like this - look at what the synchronous tornado.httpclient.HTTPClient does to create and run a temporary IOLoop

回答2:

Looks like your worker task just return and being treated as finished before ioloop stops, so gen.engine's callback cannot find the original stack_context I guess.

@task(name="MyWorker",base=WorkerBase)
def CeleryWorker(args):
    # This works because i'm adding base as WorkerBase
    CeleryWorker.RunMyTask(args)
    IOLoop.instance().start()
    return True

I have some suggestion for you

1) remove return

@task(name="MyWorker",base=WorkerBase)
def CeleryWorker(args):
    # This works because i'm adding base as WorkerBase
    CeleryWorker.RunMyTask(args)
    IOLoop.instance().start()

2) use run_sync

import functools

@task(name="MyWorker",base=WorkerBase)
def CeleryWorker(args):
    # This works because i'm adding base as WorkerBase
    func = functools.partial(CeleryWorker.RunMyTask, args)
    IOLoop.instance().run_sync(func)