python - 如何在超时的情况下取消 Tornado 中挂起的异步任务？

Question

我的设置是 python tornado 服务器，它使用 ThreadPoolExecutor 异步处理任务。在某些情况下，任务可能会变成无限循环。使用 with_timeout 装饰器，我成功地捕获了超时异常并将错误结果返回给客户端。问题是任务仍在后台运行。如何停止任务在 ThreadPoolExecutor 中运行？或者是否可以取消 Future？这是重现问题的代码。使用 tornado 4 和 concurrent.futures 库运行代码并转到 http://localhost:8888/test

from tornado.concurrent import run_on_executor
from tornado.gen import with_timeout
from tornado.ioloop import IOLoop
import tornado.web
from tornado import gen
from concurrent.futures import ThreadPoolExecutor
import datetime
MAX_WAIT_SECONDS = 10

class MainHandler(tornado.web.RequestHandler):
    executor = ThreadPoolExecutor(2)

    @run_on_executor
    def test_func(self):
        ...
        #infinite loop might be here
        ...

    @tornado.gen.coroutine
    def get(self):
        future = self.test_func()
        try:
            result_search_struct = yield with_timeout(datetime.timedelta(seconds=MAX_WAIT_SECONDS), future )
            self.write({'status' : 0})
            self.finish()
        except Exception, e:
            #how to cancel the task here if it was timeout
            future.cancel() # <-- Does not work
            self.write({'status' : 100})
            self.finish()

application = tornado.web.Application([
    (r"/test", MainHandler),
])
application.listen(8888)
IOLoop.instance().start()

Answer 1

Future 实例本身一旦实际执行就无法取消，只有在处于挂起状态时才能取消。这是注意到in the docs :

cancel()

Attempt to cancel the call. If the call is currently being executed and cannot be cancelled then the method will return False, otherwise the call will be cancelled and the method will return True.

因此，中止在后台运行的方法的唯一方法是实际将逻辑插入到潜在的无限循环中，以便在您告诉它时可以中止。对于您的示例，您可以使用 threading.Event :

class MainHandler(tornado.web.RequestHandler):
    executor = ThreadPoolExecutor(2)

    @run_on_executor
    def test_func(self, event):
        i = 0
        while not event.is_set():
            print i
            i = i + 1

    @tornado.gen.coroutine
    def get(self):
        event = threading.Event()
        future = self.test_func(event)
        try:
            result_search_struct = yield with_timeout(datetime.timedelta(seconds=MAX_WAIT_SECONDS), future )
            self.write({'status' : 0})
            self.finish()
        except Exception, e:
            future.cancel() # Might not work, depending on how busy the Executor is
            event.set()
            self.write({'status' : 100})
            self.finish()

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