重试一个python App Engine管道 [英] Retrying a python App Engine pipeline

问题描述

我有一个客户端流水线，需要向外部计算服务器发送请求，然后检索结果。计算需要一些时间，所以外部服务器会在任务队列中处理它。客户渠道不确定结果何时准备好;相反，它需要轮询服务器（get_status（id）），然后如果status == Completed检索结果（get_results（id））

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

Google AppEngine Pipelines API

如下所示：

  ReducePipeline（pipeline.Pipeline）：
 
 def run（self，* args）：
 results = [结果列表（args）
 if result] 
 if results []：
 return None 
 return results [0] 
 $ b $ class CallbackPipeline def run（self，id）：
 status = get_status（id）＃调用外部服务器状态
 results，future_results = None，None 
 if status ==错误：
 pass 
 elif status ==完成：
 results = get_results（id）＃从外部服务器获取结果
 elif status！=完成：
 with pipeline.InOrder（）：
产量延迟（秒= CallbackWait）＃NB 
 future_results = yield CallbackPipeline（id）
 yield ReducePipeline（results，future_results）
 
 class StartPipeline（pipeline.Pipeline）：
 
 def run（self，request）：
 id = start（request）＃向外部服务器发送请求;获得作业ID作为回报
产量CallbackPipeline（id）
 

这真的是重试管道的最佳方式吗？dev_appserver.py结果看起来并不怎么样： - （

 
 def finalized（self）：
如果不是self.outputs.default：
 raise pipeline.Retry（）
  

但是这似乎不适用于dev_appserver.py，只是在StartPipeline完成阶段造成重复的错误。以获得整个StartPipeline重试。

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

谢谢。

解决方案

您可能需要第四条管道来解释，验证和确认CallbackPipeline的结果是你所需要的，你需要在新的管道的run（）中提高Retry。

你也可以使用rec在第四条管道中提升Retry。你只需要继续调用CallbackPipeline，直到你获得了你正在寻找的结果。这需要异步完成，就像上面贴出的Anentropic一样。

I have a client pipeline which needs to post a request to an external calculation server, then retrieve the results. The calculation takes some time, so the external server processes it in a task queue. The client pipeline isn't exactly sure when the results will be ready; instead it needs to poll the server (get_status(id)), then if status==Completed retrieve the results (get_results(id))

Problem is that occasionally the server calculation fails, in which case I need the client pipeline to retry. I have a pipeline pattern based on this post:

Google AppEngine Pipelines API

which looks as follows:

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()

but this doesn't seem to work on dev_appserver.py, simply causing repeated errors at the StartPipeline finalisation stage. I was hoping to get the entire StartPipeline to retry instead.

Can anyone advise me on a sensible pattern for retrying the StartPipeline on receipt of None results ?

Thanks.

解决方案

You'll likely need a 4th pipeline that interprets, validates and confirms that the results of CallbackPipeline are what you need and you'll need to raise the Retry from within the run() of that new pipeline.

You could also use recursion in that 4th pipeline versus raising a Retry. You'd simply keep calling the CallbackPipeline until you did achieve the results you were looking for. This would need to be done asynchronously, like Anentropic posted above.

这篇关于重试一个python App Engine管道的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！