如何使用 tornado gen.Task / gen.coroutine 装饰器实现并行性

Question

在这种情况下，必须将并行性引入后端服务器。

我愿意查询 N 个 ELB，每个查询 5 个不同的查询，并将结果发送回 Web 客户端。

后端是Tornado，根据我在docs看过很多遍，以前用@gen.Task或者gen.coroutine应该可以并行处理好几个任务。

但是，我必须在这里遗漏一些东西，因为我所有的请求（20 个，4 个 elbs * 5 个查询）都是一个接一个地处理的。

def query_elb(fn, region, elb_name, period, callback):
    callback(fn (region, elb_name, period))

class DashboardELBHandler(RequestHandler):

    @tornado.gen.coroutine
    def get_elb_info(self, region, elb_name, period):
        elbReq = yield gen.Task(query_elb, ELBSumRequest, region, elb_name, period)
        elb2XX = yield gen.Task(query_elb, ELBBackend2XX, region, elb_name, period)
        elb3XX = yield gen.Task(query_elb, ELBBackend3XX, region, elb_name, period)
        elb4XX = yield gen.Task(query_elb, ELBBackend4XX, region, elb_name, period)
        elb5XX = yield gen.Task(query_elb, ELBBackend5XX, region, elb_name, period)

        raise tornado.gen.Return( 
            [
                elbReq,
                elb2XX,
                elb3XX,
                elb4XX,
                elb5XX,
            ]
        )

    @tornado.web.authenticated
    @tornado.web.asynchronous
    @tornado.gen.coroutine
    def post(self):
        ret = []

        period = self.get_argument("period", "5m")

        cloud_deployment = db.foo.bar.baz()
        for region, deployment in cloud_deployment.iteritems():

            elb_name = deployment["elb"][0]
            res = yield self.get_elb_info(region, elb_name, period)
            ret.append(res)

        self.push_json(ret)



def ELBQuery(region, elb_name,  range_name, metric, statistic, unit):
    dimensions = { u"LoadBalancerName": [elb_name] }

    (start_stop , period) = calc_range(range_name)

    cw = boto.ec2.cloudwatch.connect_to_region(region)
    data_points = cw.get_metric_statistics( period, start, stop, 
        metric, "AWS/ELB", statistic, dimensions, unit)    

    return data_points

ELBSumRequest   = lambda region, elb_name, range_name : ELBQuery(region, elb_name, range_name,  "RequestCount", "Sum", "Count")
ELBLatency      = lambda region, elb_name, range_name : ELBQuery(region, elb_name, range_name,  "Latency", "Average", "Seconds")
ELBBackend2XX   = lambda region, elb_name, range_name : ELBQuery(region, elb_name, range_name,  "HTTPCode_Backend_2XX", "Sum", "Count")
ELBBackend3XX   = lambda region, elb_name, range_name : ELBQuery(region, elb_name, range_name,  "HTTPCode_Backend_3XX", "Sum", "Count")
ELBBackend4XX   = lambda region, elb_name, range_name : ELBQuery(region, elb_name, range_name,  "HTTPCode_Backend_4XX", "Sum", "Count")
ELBBackend5XX   = lambda region, elb_name, range_name : ELBQuery(region, elb_name, range_name,  "HTTPCode_Backend_5XX", "Sum", "Count")

Answer 1

问题在于ELBQuery 是一个阻塞函数。如果它在某处没有yield 另一个协程，则协程调度程序无法交错调用。 （这就是协程的全部意义——它们是合作的，而不是抢占式的。）

如果问题类似于calc_range 调用，那可能很容易处理——将其分解成更小的部分，每个部分都让给下一个部分，这让调度程序有机会介入每个部分.

但最有可能的是，阻塞的是 boto 调用，您的函数的大部分时间都花在等待 get_metric_statistics 返回，而没有其他任何东西可以运行。

那么，你如何解决这个问题？

1. 为每个 boto 任务分拆一个线程。 Tornado 使透明地围绕线程或线程池任务包装协程变得非常容易，这神奇地解除了所有阻塞。当然，使用线程也是有代价的。
2. 将 boto 任务安排在线程池上，而不是每个线程上。与 #1 类似的权衡，尤其是在您只有少数任务的情况下。 （但如果您可以为 500 个不同的用户分别执行 5 个任务，您可能需要一个共享池。）
3. 重写或monkeypatch boto 以使用协程。这将是理想的解决方案……但它的工作量最大（并且最有可能破坏您不理解的代码，并且必须将其维护为 boto 更新等）。但是，有些人至少已经开始这样做了，例如 asyncboto 项目。
4. 使用greenlets 和monkeypatch 足够的库依赖项来欺骗它进入异步状态。这听起来很老套，但实际上可能是最好的解决方案；请参阅Marrying Boto to Tornado。
5. 使用greenlets 和monkeypatch 整个stdlib ala gevent 来欺骗boto 和tornado 一起工作，甚至没有意识到这一点。这听起来是个糟糕的主意。你最好将整个应用移植到gevent。
6. 使用一个单独的进程（甚至是它们的一个池），它使用类似gevent 的东西。

在不了解更多细节的情况下，我建议先查看 #2 和 #4，但我不能保证它们会成为您的最佳答案。