Как я могу отменить зависающую асинхронную задачу в торнадо с тайм-аутом?

Question

Моя установка - сервер торнадо python, который асинхронно обрабатывает задачи с помощью ThreadPoolExecutor. В некоторых условиях задача может превратиться в бесконечный цикл. С помощью декоратора with_timeout мне удалось поймать исключение тайм-аута и вернуть клиенту результат ошибки. Проблема в том, что задача все еще работает в фоновом режиме. Как можно остановить выполнение задачи в ThreadPoolExecutor? Или можно отменить Future? Вот код, который воспроизводит проблему. Выполните код с торнадо 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:

отменить()

Попытка отменить вызов. Если вызов в настоящее время выполняется и не может быть отменен, тогда метод вернет False, , иначе вызов будет отменен, и метод вернет 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), 
])