python - 惰性延迟列表达到最大递归深度

Question

我有一个很大的文档列表要更新插入到 MongoDB 中(可能 n > 100000)。我不想一次创建 100000 个延迟，但我不想顺序执行和等待每个查询，因为我有一个到 MongoDB 的连接池，并且我想充分利用它。所以我有一个生成器函数，它将产生由 DeferredLazyList 消耗的延迟。 .

def generate_update_deferreds(collection, many_docs):
    for doc in many_docs:
        d = collection.update({'_id': doc['_id']}, doc, upsert=True)
        yield d

这是链接生成延迟更新插入的代码，以及 DeferredLazyList .

@defer.inlineCallbacks
def update_docs(collection, many_docs):
    gen_deferreds = generate_update_deferreds(collection, many_docs)
    results = yield DeferredLazyList(gen_deferreds, count=pool_size, consume_errors=True)

DeferredLazyList类似于 DeferredList ，但不是接受延迟列表来等待，而是接受迭代器。延迟是从迭代器中检索的，同时只有 count延迟同时激活。这用于有效地批量延迟，因为它们是在产生时创建的。

class DeferredLazyList(defer.Deferred):
    """
    The ``DeferredLazyList`` class is used for collecting the results of
    many deferreds. This is similar to ``DeferredList``
    (``twisted.internet.defer.DeferredList``) but works with an iterator
    yielding deferreds. This will only maintain a certain number of
    deferreds simultaneously. Once one of the deferreds finishes, another
    will be obtained from the iterator.
    """

    def __init__(self, deferreds, count=None, consume_errors=None):
        defer.Deferred.__init__(self)

        if count is None:
            count = 1

        self.__consume_errors = bool(consume_errors)

        self.__iter = enumerate(deferreds)
        self.__results = []
        for _i in xrange(count):
            # Start specified number of simultaneous deferreds.
            if not self.called:
                self.__next_save_result(None, None, None)
            else:
                break

    def __next_save_result(self, result, success, index):
        """
        Called when a deferred completes.
        """
        # Make sure we can save result at index.
        if index is not None:
            results_len = len(self.__results)
            if results_len <= index:
                self.__results += [NO_RESULT] * (index - results_len + 1)
            # Save result.
            self.__results[index] = (success, result)

        # Get next deferred.
        try:
            i, d = self.__iter.next()
            d.addCallbacks(self.__next_save_result, self.__next_save_result, callbackArgs=(True, i), errbackArgs=(False, i))

        except StopIteration:
            # Iterator is exhausted, callback self with results.
            self.callback(self.__results)

        # Pass through result.
        return result if success or not self.__consume_errors else None

问题是当延迟从 generate_update_deferreds() 产生时他们的.called已设置为True这导致DeferredLazyList递归调用自身。

发生的事情是:

1. 在 DeferredLazyList.__init__() , self.__next_save_result()称为 count次(比如 5 次)。

2. 每次调用self.__next_save_result()消耗 1 延迟于 self.__iter ，并且其本身作为回调添加。

3. 因为产生的 deferred 有 .called设置为True , d.addCallbacks(self.__next_save_result, ...)立即调用self.__next_save_result()这个循环一直持续到 RuntimeError由于已达到递归深度而引发。

我在达到递归限制之前打印了堆栈跟踪，以确认这是问题的原因:

File "/home/caleb/it/Development/projects/python/amazon/bin/feeds-daemon/lib/server.py", line 937, in update_many_docs
    results = yield DeferredLazyList(gen_deferreds, count=self.mongo_connections, consume_errors=True, return_results=True)
File "/home/caleb/it/Development/projects/python/amazon/bin/feeds-daemon/lib/twisted.py", line 157, in __init__
    self.__next_save_result(None, None, None)
File "/home/caleb/it/Development/projects/python/amazon/bin/feeds-daemon/lib/twisted.py", line 222, in __next_save_result
    d.addCallbacks(self.__next_save_result, self.__next_save_result, callbackArgs=(True, i), errbackArgs=(False, i))
File "/usr/lib/python2.7/dist-packages/twisted/internet/defer.py", line 290, in addCallbacks
    self._runCallbacks()
File "/usr/lib/python2.7/dist-packages/twisted/internet/defer.py", line 551, in _runCallbacks
    current.result = callback(current.result, *args, **kw)
File "/home/caleb/it/Development/projects/python/amazon/bin/feeds-daemon/lib/twisted.py", line 222, in __next_save_result
    d.addCallbacks(self.__next_save_result, self.__next_save_result, callbackArgs=(True, i), errbackArgs=(False, i))
File "/usr/lib/python2.7/dist-packages/twisted/internet/defer.py", line 290, in addCallbacks
    self._runCallbacks()
File "/usr/lib/python2.7/dist-packages/twisted/internet/defer.py", line 551, in _runCallbacks
    current.result = callback(current.result, *args, **kw)
File "/home/caleb/it/Development/projects/python/amazon/bin/feeds-daemon/lib/twisted.py", line 222, in __next_save_result
    d.addCallbacks(self.__next_save_result, self.__next_save_result, callbackArgs=(True, i), errbackArgs=(False, i))
# Repeated until the RuntimeError
exceptions.RuntimeError: maximum recursion depth exceeded

任何帮助将不胜感激。顺便说一句，我正在运行 Python 2.7.3 和 Twisted 12.1.0，MongoDB 的东西实际上只与理解上下文相关。

---

我想要每个延迟的结果，但是 cooperate()不返回这些，所以我在将它们传递给 CooperativeTask 之前为每个 deferred 添加了一个回调s:

from twisted.internet.defer import DeferredList, inlineCallbacks
from twisted.internet.task import cooperate

NO_RESULT = object()

def generate_update_deferreds(collection, many_docs, save_results):
    for i, doc in enumerate(update_docs):
        d = collection.update({'_id': doc['_id']}, doc, upsert=True)
        d.addBoth(save_result, i, save_results) # Save result
        yield d

def save_result(result, i, save_results):
    save_results[i] = result

@inlineCallbacks
def update_docs(collection, many_docs):
    save_results = [NO_RESULT] * len(many_docs)
    gen_deferreds = generate_update_deferreds(collection, many_docs, save_results))
    workers = [cooperate(gen_deferreds).whenDone() for _i in xrange(count)]
    yield defer.DeferredList(workers)
    # Handle save_results...

Answer 1

Twisted 中有一些工具可以帮助您更轻松地完成此操作。例如合作:

from twisted.internet.task import cooperate

def generate_update_deferreds(collection, many_docs):
    for doc in update_docs:
        d = collection.update({'_id': doc['_id']}, doc, upsert=True)
        yield d

work = generate_update_deferreds(...)
worker_tasks = []
for i in range(count):
    task = cooperate(work)
    worker_tasks.append(task)

all_done_deferred = DeferredList([task.whenDone() for task in worker_tasks])