Design

A picture worth a thousand words.

A few observations and notes:

1. Messages are off-loaded to IO loops immediately but task and result queues are protected by semaphores to not off-load more tasks on IO loops than the queue size.

2. In fact there are n task queue, a queue per worker. Task messages are distributed evenly between workers.

3. add_callback [2] is safe (and only safe) method to pass control from other thread to IO loop’s thread.

4. Because queue message handlers (fn1 and fn2) are called from IO loop they can be coroutines.

5. Pool stops its workers by sending PoisonPill task message per worker.

Worker

There is a base class for a Worker. It represents a process worker that runs its own Tornado IO loop. It processes task messages and puts them in result queue (or just puts without receiving anything). It requires override of _process_message(self, task) in a subclass and mandates that once the task is done, self._task_done() is called (directly in try-finally or with @task_done), like:

def _process_message(self, task):
    try:
        result = 'some processing'
        self._put_nowait(result)
    finally:
        self._task_done()

WorkerSubclass._process_message may be plain function or coroutine. More details are available in the package’s unit test module [5].

Installation

pip install Looppool

Usage

#!/usr/bin/env python3


import looppool
from tornado import gen, ioloop, httpclient


class FetchWorker(looppool.Worker):

    _http_client = None
    '''Tornado asynchronous HTTP client'''


    def _initialise(self):
        self._http_client = httpclient.AsyncHTTPClient()

    @looppool.task_done
    @gen.coroutine
    def _process_message(self, url):
        response = yield self._http_client.fetch(url)
        self._put_nowait((url, response.headers.get('server')))


@looppool.task_done
def process_message(result, result_done):
    print(result)


@gen.coroutine
def main():
    loop = ioloop.IOLoop.instance()
    pool = looppool.Pool(loop, pool_size = 4, worker_class = FetchWorker)
    pool.process_message = process_message
    pool.start()

    urls = [
        'https://python.org/',
        'http://tornadoweb.org/',
        'https://google.com/',
        'https://stackoverflow.com/',
    ]
    list(map(pool.put_nowait, urls))

    pool.stop()


if __name__ == '__main__':
    ioloop.IOLoop.instance().run_sync(main)

Maintenance

Maintaining a process group instead of one process is more tricky thing to do. Initially, you may want to see if your pool instance has actually spawned any processes. Here’s what you can do visualise your process tree, which has main process, one or two (depending on start method) multiprocessing helper processes, and processes of your looppool pools (you can use different pools for different purposes):

htop -p $(pgrep -d"," -g $(pgrep -f "main-process-name-or-its-start-args"))

top will also work but is limited to 20 PIDs. Enforcing the process tree stop is also different. If something goes wrong process tree should be killed like:

kill -9 -- -$(pgrep -f "main-process-name-or-its-start-args")

Killing only main process will leave helper and worker processes running.

You can improve the names of your worker processes by setting them in worker’s initialiser with setproctitle [6] (see example below).

#!/usr/bin/env python3


import logging

import looppool
from looppool.utility import ResourceReporter
from statsd import StatsClient
from raven import Client
from raven.exceptions import InvalidDsn
from raven.handlers.logging import SentryHandler
from setproctitle import setproctitle


class InstrumentatedWorker(looppool.Worker):

    _resource_reporter = None
    '''CPU, RSS and IO loop stats reporter'''


    def _initialise(self):
        setproctitle('python APP_NAME POOL_NAME pool worker')

        statsd = StatsClient('localhost', 8125, 'APP_PREFIX')
        self._resource_reporter = ResourceReporter(self._ioloop, statsd,
            'worker.instrumentated.process.{}'.format(self._number))

        try:
            handler = SentryHandler(Client('SENTRY_DSN'))
            handler.setLevel(logging.WARNING)
        except InvalidDsn:
            logging.exception('Cannot configure Sentry handler')
        else:
            logging.basicConfig(handlers=[handler], level=logging.WARNING)

    @looppool.task_done
    def _process_message(self, message):
        self._put_nowait((message, self._number))

    def start(self):
        super().start()

        self._resource_reporter.start()

    def join(self):
        self._resource_reporter.stop()

        super().join()