竜巻サーバーがスレッドモジュールと互換性がありません

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

トルネードイベントループ全体を、t.joinで100秒間ブロックします。 yield文がないか、コールバックをスケジュールして関数を終了しない限り、関数は非同期ではありません。あなたの関数 "on_message"が2つのスレッドを開始してからそれぞれt.joinを呼び出すことに注目してください - 関数がt.joinを待っている間、Tornadoのイベントループはどのように他の作業を行うことができますか？

代わりに、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)