Утечка памяти Tornado при сброшенных соединениях

У меня есть настройка, где Tornado используется как проход для рабочих. Запрос получен от Tornado, который отправляет этот запрос N работникам, агрегирует результаты и отправляет их клиенту. Что отлично работает, за исключением случаев, когда по какой-то причине происходит тайм-аут. то у меня утечка памяти.

У меня есть настройка, которая похожа на этот псевдокод:

workers = ["http://worker1.example.com:1234/",
           "http://worker2.example.com:1234/", 
           "http://worker3.example.com:1234/" ...]

class MyHandler(tornado.web.RequestHandler):
    @tornado.web.asynchronous
    def post(self):
        responses = []

        def __callback(response):
            responses.append(response)
            if len(responses) == len(workers):
                self._finish_req(responses)

        for url in workers:
            async_client = tornado.httpclient.AsyncHTTPClient()
            request = tornado.httpclient.HTTPRequest(url, method=self.request.method, body=body)
            async_client.fetch(request, __callback) 

    def _finish_req(self, responses):
        good_responses = [r for r in responses if not r.error]
        if not good_responses:
            raise tornado.web.HTTPError(500, "\n".join(str(r.error) for r in responses))
        results = aggregate_results(good_responses)
        self.set_header("Content-Type", "application/json")
        self.write(json.dumps(results))
        self.finish()

application = tornado.web.Application([
    (r"/", MyHandler),
])

if __name__ == "__main__":
    ##.. some locking code 
    application.listen()
    tornado.ioloop.IOLoop.instance().start()

Что я делаю неправильно? Откуда возникает утечка памяти?

Ответ 1

Я не знаю источник проблемы, и кажется, что gc должен уметь позаботиться об этом, но есть две вещи, которые вы можете попробовать.

Первый способ заключается в упрощении некоторых ссылок (похоже, что все еще могут быть ссылки на responses, когда RequestHandler завершает):

class MyHandler(tornado.web.RequestHandler):
    @tornado.web.asynchronous
    def post(self):
        self.responses = []

        for url in workers:
            async_client = tornado.httpclient.AsyncHTTPClient()
            request = tornado.httpclient.HTTPRequest(url, method=self.request.method, body=body)
            async_client.fetch(request, self._handle_worker_response) 

    def _handle_worker_response(self, response):
        self.responses.append(response)
        if len(self.responses) == len(workers):
            self._finish_req()

    def _finish_req(self):
        ....

Если это не работает, вы всегда можете вручную вызвать сборку мусора:

import gc
class MyHandler(tornado.web.RequestHandler):
    @tornado.web.asynchronous
    def post(self):
        ....

    def _finish_req(self):
        ....

    def on_connection_close(self):
        gc.collect()

Ответ 2

Код выглядит хорошо. Вероятно, утечка находится внутри "Торнадо".

Я только наткнулся на эту строку:

async_client = tornado.httpclient.AsyncHTTPClient()

Знаете ли вы о магии создания экземпляра в этом конструкторе? Из документов:

"""
The constructor for this class is magic in several respects:  It actually
creates an instance of an implementation-specific subclass, and instances
are reused as a kind of pseudo-singleton (one per IOLoop).  The keyword
argument force_instance=True can be used to suppress this singleton
behavior.  Constructor arguments other than io_loop and force_instance
are deprecated.  The implementation subclass as well as arguments to
its constructor can be set with the static method configure()
"""

Таким образом, вам не нужно делать это внутри цикла. (С другой это не должно навредить.) Но какая реализация вы используя CurlAsyncHTTPClient или SimpleAsyncHTTPClient?

Если это SimpleAsyncHTTPClient, обратите внимание на этот комментарий в коде:

"""
This class has not been tested extensively in production and
should be considered somewhat experimental as of the release of
tornado 1.2. 
"""

Вы можете попробовать переключиться на CurlAsyncHTTPClient. Или следовать Предложение Николая Фоминых и проследить звонки __callback().