python - Twisted:WAITING子任务完成

Question

在我的代码中，我有两个假设任务:一个从生成器获取 url 并使用 Twisted 的 Cooperator 批量下载它们，另一个获取下载的源并异步解析它。我试图将所有获取和解析任务封装到一个 Deferred 对象中，该对象在所有页面都已下载且所有源都已解析时回调。

我想出了以下解决方案:

from twisted.internet import defer, task, reactor, threads
from twisted.web.client import getPage


BATCH_SIZE = 5

def main_task():
    result = defer.Deferred()
    state = {'count': 0, 'done': False}

    def on_parse_finish(r):
        state['count'] -= 1
        if state['done'] and state['count'] == 0:
            result.callback(True)

    def process(source):
        deferred = parse(source)
        state['count'] += 1
        deferred.addCallback(on_parse_finish)

    def fetch_urls():
        for url in get_urls():
            deferred = getPage(url)
            deferred.addCallback(process)
            yield deferred

    def on_finish(r):
        state['done'] = True

    deferreds = []

    coop = task.Cooperator()
    urls = fetch_urls()
    for _ in xrange(BATCH_SIZE):
        deferreds.append(coop.coiterate(urls))

    main_tasks = defer.DeferredList(deferreds)
    main_tasks.addCallback(on_finish)

    return defer.DeferredList([main_tasks, result])

# `main_task` is meant to be used with `blockingCallFromThread`
# The following should block until all fetch/parse tasks are completed:
# threads.blockingCallFromThread(reactor, main_task)

代码有效，但我觉得我要么遗漏了一些明显的东西，要么不知道一个简单的 Twisted 模式，这会使它变得更简单。有没有更好的方法来返回一个在所有提取和解析完成后回调的 Deferred？

Answer 1

如目前所写，在我看来，这段代码的并行下载数量有限，但并行解析作业的数量不受限制。那是故意的吗？我将假设“否”，因为如果您的网络恰好很快而您的解析器恰好很慢，因为 URL 的数量接近无穷大，那么您的内存使用量也是如此:)。

所以这里有一个并行性有限但通过下载顺序执行解析的东西:

from twisted.internet import defer, task
from twisted.web.client import getPage

BATCH_SIZE = 5

def main_task(reactor):
    def fetch_urls():
        for url in get_urls():
            yield getPage(url).addCallback(parse)

    coop = task.Cooperator()
    urls = fetch_urls()

    return (defer.DeferredList([coop.coiterate(urls)
                               for _ in xrange(BATCH_SIZE)])
            .addCallback(task_finished))

task.react(main_task)

之所以可行，是因为 parse(显然)返回了一个 Deferred，将其作为回调添加到 getPage 返回的结果中Deferred 在 parse 完成其业务之前不会调用由 coiterate 添加的回调。

既然你问的是惯用的 Twisted 代码，我也冒昧地对其进行了一些现代化改造(使用 task.react 而不是手动运行 react 器，内联表达式使事情更简洁等等)。

如果您确实希望并行解析多于并行提取，那么这样的方法可能会更好:

from twisted.internet import defer, task
from twisted.web.client import getPage

PARALLEL_FETCHES = 5
PARALLEL_PARSES = 10

def main_task(reactor):
    parseSemaphore = defer.DeferredSemaphore(PARALLEL_PARSES)

    def parseWhenReady(r):
        def parallelParse(_):
            parse(r).addBoth(
                lambda result: parseSemaphore.release().addCallback(
                    lambda _: result
                )
            )
        return parseSemaphore.acquire().addCallback(parallelParse)

    def fetch_urls():
        for url in get_urls():
            yield getPage(url).addCallback(parseWhenReady)

    coop = task.Cooperator()
    urls = fetch_urls()

    return (defer.DeferredList([coop.coiterate(urls)
                               for _ in xrange(PARALLEL_FETCHES)])
            .addCallback(lambda done:
                         defer.DeferredList(
                            [parseSemaphore.acquire()
                             for _ in xrange(PARALLEL_PARSES)]
                         ))
            .addCallback(task_finished))

task.react(main_task)

您可以看到 parseWhenReady 返回从 acquire 返回的 Deferred，因此只要并行解析可以，并行提取就会继续 begin，因此即使解析器过载，您也不会继续不加选择地获取。但是，parallelParse 谨慎地避免返回 parse 或 release 返回的 Deferred，因为提取应该能够继续那些正在进行中。

(请注意，由于您的初始示例不可运行，因此我根本没有测试过其中任何一个。希望即使存在错误，意图也很清楚。)