在后台线程上执行长时间运行的计算密集型操作答案

【问题标题】：Performing long running compute intensive operation on background thread在后台线程上执行长时间运行的计算密集型操作
【发布时间】：2019-03-01 12:15:21
【问题描述】：

我在 Python Tornado 框架中编写了一个 REST API，它可以预测给定段落中问题的答案。

这是 Tornado 处理程序的 Python 代码：

def post(self):
    """
    This function predicts the response from the pre-trained Allen model
    """    
    try:
        request_payload = tornado.escape.json_decode(self.request.body)

        if (request_payload is None):
            return self._return_response(self, { "message": "Invalid request!" }, 400)

        context = request_payload["context"]
        question = request_payload["question"]

        if(context is None or not context):
            return self._return_response(self, { "message": "Context is not provided!" }, 400)

        if(question is None or not question):
            return self._return_response(self, { "message": "Question is not provided!" }, 400)

        # Compute intensive operation which blocks the main thread
        answer_prediction = predictor.predict(passage=str(context), question=str(question))
        best_answer = answer_prediction["best_span_str"] or "Sorry, no answer found for your question!"

        return self._return_response(self, { "answer": best_answer }, 200)

    except KeyError:
        #Return bad request if any of the keys are missing
        return self._return_response(self, { "message": 'Some keys are missing from the request!' }, 400)

    except json.decoder.JSONDecodeError:
        return self._return_response(self, { "message": 'Cannot decode request body!' }, 400)

    except Exception as ex:
        return self._return_response(self, { "message": 'Could not complete the request because of some error at the server!', "cause": ex.args[0], "stack_trace": traceback.format_exc(sys.exc_info()) }, 500)

问题在于：

answer_prediction = predictor.predict(passage=str(context), 问题=str（问题））

为传入请求阻塞主线程并等待该长时间运行的操作完成，同时阻塞其他请求并有时使当前请求超时。

我已阅读this 的答案，详细说明了将长时间运行的操作置于队列中的解决方案，但我没有得到它。

另外，由于 Python 的 GIL 只能同时运行一个线程，这迫使我产生一个单独的进程来处理它，因为进程成本高昂，是否有任何可行的解决方案来解决我的问题以及如何处理这个问题种情况。

这是我的问题：

如何安全地将计算密集型操作卸载到后台线程
如何优雅地处理超时和异常
如何维护队列结构以检查长时间运行的操作是否已完成。

【问题讨论】：

您是否将任务计划程序检查为APScheduler
不，我没有，但它是如何关联的？
如果当前任务繁重，可以添加到队列并进行并发操作。
@adnbsr 你可以举个例子
查看文档，他们更有成果。

标签： python tornado

【解决方案1】：

在单独的线程中运行阻塞代码。使用IOLoop.run_in_executor。

例子：

from functools import partial

async def post(self):
    ...

    # create a partial object with the keyword arguments
    predict_partial = partial(predictor.predict, passage=str(context), question=str(question))

    answer_prediction = await IOLoop.current().run_in_executor(None, predict_partial)

    ...

【讨论】：

所以run_in_executor 的来源说max_workers=(cpu_count() * 5) 这是什么意思？
@KunalMukherjee 这是最大线程数。如果你有一个双核 CPU，这个值将等于 10。这意味着你的服务器将最多运行 10 个线程来同时为 10 个单独的请求计算 predictor.predict。如果有更多请求进来，他们将不得不等到一个线程空闲。即使所有线程都被占用，您的服务器也不会阻塞。

【解决方案2】：

我认为您应该将此 API 调用转换为异步调用并立即使用令牌返回给调用者。

令牌稍后将使用令牌来检查（使用另一个 API 调用）操作是否完成。

【讨论】：