python - 重试 python App Engine 管道

Question

我有一个客户端管道，需要将请求发送到外部计算服务器，然后检索结果。计算需要一些时间，因此外部服务器在任务队列中处理它。客户端管道不确定结果何时准备好；相反，它需要轮询服务器 (get_status(id))，然后如果 status==Completed 检索结果 (get_results(id))

问题是服务器计算偶尔会失败，在这种情况下我需要客户端管道重试。我有一个基于这篇文章的管道模式:

Google AppEngine Pipelines API

如下所示:

class ReducePipeline(pipeline.Pipeline):

    def run(self, *args):
        results=[result for result in list(args)
                 if result]
        if results==[]:
            return None
        return results[0]

class CallbackPipeline(pipeline.Pipeline):

    def run(self, id):
        status=get_status(id) # call external server status
        results, future_results = None, None
        if status==Error:
            pass    
        elif status==Completed:
            results=get_results(id) # get results from external server
        elif status!=Completed:
            with pipeline.InOrder():
                yield Delay(seconds=CallbackWait) # NB
                future_results=yield CallbackPipeline(id)
        yield ReducePipeline(results, future_results)

class StartPipeline(pipeline.Pipeline):

    def run(self, request):
        id=start(request) # post request to external server; get job id in return
        yield CallbackPipeline(id)

    """
    is this really the best way to retry the pipeline ? dev_appserver.py results don't look promising :-(
    """

    def finalized(self):
        if not self.outputs.default:
            raise pipeline.Retry()

但这似乎不适用于 dev_appserver.py，只会导致 StartPipeline 最终阶段重复出现错误。我希望让整个 StartPipeline 重试。

任何人都可以建议我在收到 None 结果时重试 StartPipeline 的合理模式吗？

谢谢。

Answer 1

您可能需要第四个管道来解释、验证并确认 CallbackPipeline 的结果是您所需要的，并且您需要从该新管道的 run() 中引发 Retry。

您还可以在第四个管道中使用递归，而不是引发重试。您只需继续调用 CallbackPipeline 直到获得所需的结果。这需要异步完成，就像上面发布的 Anentropic 一样。