python - 如何在同步环境中执行 Tornado 协程？

Question

我有一些Tornado与协程相关的问题。

有一些Python模型A，它能够执行某些功能。该函数可以从模型外部设置。我无法更改模型本身，但我可以传递任何我想要的函数。我试图通过我的函数教它与 Tornado 的 ioloop 一起工作，但我做不到。

这是片段:

import functools
import pprint
from tornado import gen
from tornado import ioloop

class A:
    f = None
    def execute(self):
      return self.f()
    pass

@gen.coroutine
def genlist():
    raise gen.Return(range(1, 10))

@gen.coroutine
def some_work():
    a = A()
    a.f = functools.partial(
        ioloop.IOLoop.instance().run_sync,
        lambda: genlist())
    print "a.f set"
    raise gen.Return(a)

@gen.coroutine
def main():
    a = yield some_work()
    retval = a.execute()
    raise gen.Return(retval)


if __name__ == "__main__":
    pprint.pprint(ioloop.IOLoop.current().run_sync(main))

所以问题是我在代码的一部分中设置了该函数，但在另一部分中使用模型的方法执行它。

现在，Tornado 4.2.1 给了我“IOLoop 已在运行”，但在 Tornado 3.1.1 中它可以工作(但我不知 Prop 体如何)。

我知道接下来的事情:

1. 我可以创建新的 ioloop，但我想使用现有的 ioloop。
2. 我可以用一些函数包装 genlist，该函数知道 genlist 的结果是 Future，但我不知道如何阻止执行，直到 future 的结果被设置在同步函数内部。

此外，我不能使用 a.execute() 的结果作为 future 对象，因为 a.execute() 可以从代码的其他部分调用，即它应该返回列表实例.

所以，我的问题是:是否有机会使用当前 IOLoop 从同步模型的方法执行异步 genlist？

Answer 1

您无法在此处重新启动外部 IOLoop。您有三个选择:

1. 随处使用异步接口(interface):将 a.execute() 以及堆栈顶部的所有内容更改为协程。这是基于 Tornado 的应用程序的常见模式；试图跨越同步和异步世界是很困难的，最好留在一边或另一边。
2. 在临时 IOLoop 上使用 run_sync()。这就是 Tornado 的同步 tornado.httpclient.HTTPClient 所做的事情，这使得从另一个 IOLoop 中进行调用变​​得安全。但是，如果您这样做，外部 IOLoop 仍处于阻塞状态，因此您通过使 genlist 异步不会获得任何好处。

3. 在单独的线程上运行a.execute并回调主IOLoop的线程以获取内部函数。如果 a.execute 无法异步，这是避免 IOLoop 运行时阻塞的唯一方法。

   executor = concurrent.futures.ThreadPoolExecutor(8)
   
   @gen.coroutine
   def some_work():
       a = A()
       def adapter():
           # Convert the thread-unsafe tornado.concurrent.Future
           # to a thread-safe concurrent.futures.Future.
           # Note that everything including chain_future must happen
           # on the IOLoop thread.
           future = concurrent.futures.Future()
           ioloop.IOLoop.instance().add_callback(
               lambda: tornado.concurrent.chain_future(
                   genlist(), future)
           return future.result()
       a.f = adapter
       print "a.f set"
       raise gen.Return(a)
   
   @gen.coroutine
   def main():
       a = yield some_work()
       retval = yield executor.submit(a.execute)
       raise gen.Return(retval)