2

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)

これはDeferredListDeferredLazyListに似ていますが、遅延のリストを受け入れる代わりに、それを待つためにイテレータを受け入れます。延期されたものはイテレータから取得されますが、同時にアクティブなのは延期されたものだけです。これは、遅延が生成されるときに作成されるため、遅延を効果的にバッチ処理するために使用されます。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()ものから生成され、それが再帰的に自分自身を呼び出す原因になっている場合です。.calledTrueDeferredLazyList

何が起こっているのですか:

  1. DeferredLazyList.__init__()self.__next_save_result()は回と呼ばれcountます(たとえば5)。

  2. への各呼び出しは、self.__next_save_result()から延期された1を消費しself.__iter、それ自体がコールバックとして追加されます。

  3. 生成されたdeferredがに.called設定されているためTrued.addCallbacks(self.__next_save_result, ...)すぐに呼び出し、再帰の深さに達したためself.__next_save_result()にaが発生するまでこのループが続きます。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

どんな助けでも大歓迎です。ちなみに、私はTwisted12.1.0でPython2.7.3を実行していますが、MongoDBのものは、実際にはコンテキストを理解するためにのみ関連しています。


それぞれの延期からの結果が欲しかったのですが、cooperate()それらを返さないので、CooperativeTasksに渡す前に各延期にコールバックを追加しました。

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...
4

1 に答える 1

3

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])
于 2013-03-26T13:18:25.740 に答える