Сервер tornado несовместим с модулем с резьбой

Question

Я использую торнадо с потоками.

Короче говоря, каждый раз, когда обработчик websocket получает запросы, он начинает выполнять задачу, которая может занять несколько минут.

Однако, как только клиент подключен, ни один другой клиент не может быть подключен, пока первый не отключится.

Любые идеи?

Я приложил минимальный пример, который использует time.sleep для имитации длительных задач.

import tornado.web 
import tornado.websocket 
import tornado.httpserver 
import tornado.ioloop 
import time 
import json 
import threading 

class TaskHandler(tornado.websocket.WebSocketHandler): 
    def open(self): 
     pass 

    def check_origin(self, origin): 
     return True 

    def on_message(self, message): 
     try: 
      print 'received: ', message 
      self.write_message(json.dumps({'status': 'running'})) 

      def worker_A(kwargs): 
       time.sleep(100) 
       pass 

      def worker_B(kwargs): 
       time.sleep(100) 
       pass 

      threads = [] 
      for target in [worker_A, worker_B]: 
       t = threading.Thread(target = target, args = ({'xxx': 'yyy'},)) 
       t.daemon = True 
       t.start() 
       threads.append(t) 

      for t in threads: 
       t.join() 

     except Exception, e: 
      print 'TaskHandler: exception: ', e 
      pass 

     self.write_message(json.dumps({'status': 'done'})) 

    def on_close(self): 
     pass 

class Server(tornado.web.Application): 
    def __init__(self): 
     handlers = [ 
      ('/task', TaskHandler), 
     ] 

     tornado.web.Application.__init__(self, handlers) 

if __name__ == '__main__': 
    server = tornado.httpserver.HTTPServer(Server()) 
    server.listen(8765, address = '127.0.0.1') 
    tornado.ioloop.IOLoop.instance().start() 

Answer 1

Вы блокировать весь цикл событий Торнадо в течение 100 секунд в t.join. Если у вас нет инструкции yield или не планируйте обратный вызов и не выходите из функции, ваша функция не является асинхронной. Обратите внимание, как ваша функция «on_message» начинает два потока, а затем называет t.join - как может цикл событий Tornado выполнять любую другую работу, пока ваша функция ждет t.join?

Вместо этого используйте ThreadPoolExecutor что-то вроде этого:

thread_pool = ThreadPoolExecutor(4) 

class TaskHandler(tornado.websocket.WebSocketHandler): 
    # Make this an asynchronous coroutine 
    @gen.coroutine 
    def on_message_coroutine(self, message): 
     print 'received: ', message 
     self.write_message(json.dumps({'status': 'running'})) 

     def worker_A(kwargs): 
      time.sleep(100) 
      pass 

     def worker_B(kwargs): 
      time.sleep(100) 
      pass 

     futures = [] 
     for target in [worker_A, worker_B]: 
      f = thread_pool.submit(target, {'xxx': 'yyy'}) 
      futures.append(future) 

     # Now the event loop can do other things 
     yield futures 

    def on_message(self, message): 
     IOLoop.current().spawn_callback(self.on_message_coroutine, 
             message)