sample implementation of concurrent async-communication request to other server

concurrent_cache.py

#!/usr/bin/python
# -*- coding: utf-8 -*-

import tornado.web
import tornado.httpserver
import tornado.gen
import tornado.ioloop
import functools
import collections
import pprint
import tornadoasyncmemcache as memcache


class MyHandler(tornado.web.RequestHandler):

    @tornado.gen.engine
    @tornado.web.asynchronous
    def get(self):
        # hoge0 = hoge9
        print "got request"

        keys = []
        for i in range(10):
            key = "hoge%d" % i
            keys.append(key)

        print "keys=%s" % pprint.pformat(keys)

        result_dict = yield tornado.gen.Task(self._get_multi_concurrent, keys, concurrency=2)

        self.write(pprint.pformat(result_dict))
        self.finish()

    def _get_multi_concurrent(self, keys, concurrency, callback=None):
        servers = ['localhost:11211']
        client = memcache.Client(servers)
        uniq_keys = set(keys)
        key_queue = collections.deque(uniq_keys)

        expect_key_len = len(key_queue)
        _result_map = {}

        def __get_end_cb(value, thread_id, got_key=None):
            _result_map[got_key] = value

            if len(_result_map.keys()) == expect_key_len:
                # finish
                print "got all cache response!"
                callback(_result_map)
            else:
                if 0 < len(key_queue):
                    # more request
                    next_req_k = key_queue.popleft()
                    cb = functools.partial(__get_end_cb, thread_id=thread_id, got_key=next_req_k)
                    client.get(next_req_k, callback=cb)
                    print "[%d] cache get request started for key '%s'" % (thread_id, next_req_k)
                else:
                    # no need to request, request is already done
                    pass

        for i in range(concurrency):
            k = key_queue.popleft()
            cb = functools.partial(__get_end_cb, thread_id=i, got_key=k)
            client.get(k, callback=cb)
            print "[%d] cache get request started for key '%s'" % (i, k)


if __name__ == '__main__':
    app = tornado.web.Application(
        [(r'/', MyHandler)]
    )

    server = tornado.httpserver.HTTPServer(app)
    server.listen(8880)
    print "port=8880"
    tornado.ioloop.IOLoop.instance().start()

61-65: 最初に並列度concurrencyでスタート。その後順番にcallback(__get_end_cb関数(をpartialしたもの))でchainしていく。

thread_idは動作を可視化するのに使ってるだけで実際はいらない。