[Twisted-Python] I could swear I've seen this pattern *somewhere* in Twisted...

[Jp Calderone]

On Mon, 09 May 2005 21:02:42 -0400, "Mike C. Fletcher" <mcfletch at rogers.com> wrote:
>Itamar Shtull-Trauring wrote:
>
>>On Mon, 2005-05-09 at 13:42 -0400, Mike C. Fletcher wrote:
>>
>>
>>>I have a little utility module "parallel", that dispatches a callable
>>>for "count" items in parallel, collecting them as would a DeferredList
>>>(attached for reference).  The idea here is that you want only count
>>>items to be actively processing at any given time, and you want to be
>>>notified when the entire suite of items completes processing (just like
>>>a DeferredList).
>>>
>>>
>>
>>Hm. Would t.i.defer.DeferredSemaphore do what you want? There's also
>>DeferredQueue (courtesy of JP, 2.0 and later).
>>
>>
>DeferredSemaphore certainly makes it much simpler!  Attached so that
>others can see how much it simplifies the pattern. Not quite short
>enough to be "too short to bother wrapping in a function", but far less
>involved all-in-all, and really too short even to bother with a class.
>
>Thanks,
>Mike

  I'd be interested to hear about the relative performance of your two solutions.  I would expect DeferredSemaphore to create many more Deferreds, and thus perhaps perform less well.

  Also, Glyph and I did some work for Quotient today on some code which may be applicable to this kind of problem.  Here's an example:

    from twisted.internet import defer
    from atop.tpython import Cooperator

    def parallel(iterable, count, callable, *args, **named):
        source = iter(iterable)
        def work():
            for elem in source:
                yield callable(elem, *args, **named)
        coop = Cooperator()
        tasks = []
        for i in range(count):
            tasks.append(coop.coiterate(work()))
        return defer.DeferredList(tasks)

  Note that this returns a DeferredList of each "task", rather than of each result.  If results are desired as well, adding a callback (before yielding) to the result of callable() inside work() which saved the result, then adding a callback to the DeferredList which discarded the list of task results and returned the saved results would accomplish this.

  Note also that Cooperator is in a branch at the moment.  It'll most likely be in Quotient trunk sometime tomorrow.

  Jp