background_task.py

from __future__ import absolute_import, unicode_literals

import logging

from itertools import count
from time import sleep
from tornado.ioloop import IOLoop
from tornado.web import asynchronous, RequestHandler
from multiprocessing.pool import ThreadPool


logger = logging.getLogger(__name__)


workers = ThreadPool(10)
idx_generator = count(0)


def run_background(func, callback, args=None, kwargs=None):
    def _callback(result):
        IOLoop.instance().add_callback(lambda: callback(result))
    if args is None:
        args = ()
    if kwargs is None:
        kwargs = {}
    workers.apply_async(func, args, kwargs, _callback)


def blocking_task(idx, s):
    logger.info("Starting blocking_task %s", idx)
    for i in range(s):
        logger.info("idx=%s i=%s", idx, i)
        sleep(1)
    return idx


class AsyncTaskHandler(RequestHandler):
    @asynchronous
    def post(self):
        idx = idx_generator.next()
        run_background(blocking_task, self.on_complete, (idx, 5,))

    def on_complete(self, result):
        self.write("Complete {0}".format(result))
        self.finish()


class BackgroundTaskHandler(RequestHandler):
    @asynchronous
    def post(self):
        idx = idx_generator.next()
        run_background(blocking_task, self.on_complete, (idx, 5,))
        self.finish()

    def on_complete(self, result):
        logger.info("Complete %s", result)

background_task_threadpoolexecutor.py

from __future__ import absolute_import, unicode_literals

import logging

from itertools import count
from time import sleep

from concurrent.futures import ThreadPoolExecutor
from tornado.concurrent import run_on_executor
from tornado.ioloop import IOLoop
from tornado import gen, web


logger = logging.getLogger(__name__)


pool = ThreadPoolExecutor(max_workers=10)
idx_generator = count(0)


def blocking_task(idx, s):
    logger.info("Starting blocking_task %s", idx)
    for i in range(s):
        logger.info("idx=%s i=%s", idx, i)
        sleep(1)
    return idx


class BackgroundTaskHandler(web.RequestHandler):
    executor = pool

    @gen.coroutine
    def post(self):
        idx = idx_generator.next()
        IOLoop.instance().add_callback(
            lambda: self.background_task(blocking_task, self.on_complete, idx, 5))
        self.finish()

    @run_on_executor
    def background_task(self, func, callback, *args, **kwargs):
        result = func(*args, **kwargs)
        callback(result)

    def on_complete(self, result):
        logger.info("Complete %s", result)

References:

• https://gist.github.com/methane/2185380