Queue remote calls to a Python Twisted perspective broker?

asynchronous, python, twisted

Solution

One ready-made option that might help with this is `twisted.internet.defer.DeferredSemaphore`. This is the asynchronous version of the normal (counting) semaphore you might already know if you've done much threaded programming.

A (counting) semaphore is a lot like a mutex (a lock). But where a mutex can only be acquired once until a corresponding release, a (counting) semaphore can be configured to allow an arbitrary (but specified) number of acquisitions to succeed before any corresponding releases are required.

Here's an example of using `DeferredSemaphore` to run ten asynchronous operations, but to run at most three of them at once:

from twisted.internet.defer import DeferredSemaphore, gatherResults
from twisted.internet.task import deferLater
from twisted.internet import reactor


def async(n):
    print 'Starting job', n
    d = deferLater(reactor, n, lambda: None)
    def cbFinished(ignored):
        print 'Finishing job', n
    d.addCallback(cbFinished)
    return d


def main():
    sem = DeferredSemaphore(3)

    jobs = []
    for i in range(10):
        jobs.append(sem.run(async, i))

    d = gatherResults(jobs)
    d.addCallback(lambda ignored: reactor.stop())
    reactor.run()


if __name__ == '__main__':
    main()

`DeferredSemaphore` also has explicit `acquire` and `release` methods, but the `run` method is so convenient it's almost always what you want. It calls the `acquire` method, which returns a `Deferred`. To that first `Deferred`, it adds a callback which calls the function you passed in (along with any positional or keyword arguments). If that function returns a `Deferred`, then to that second `Deferred` a callback is added which calls the `release` method.

The synchronous case is handled as well, by immediately calling `release`. Errors are also handled, by allowing them to propagate but making sure the necessary `release` is done to leave the `DeferredSemaphore` in a consistent state. The result of the function passed to `run` (or the result of the `Deferred` it returns) becomes the result of the `Deferred` returned by `run`.

Another possible approach might be based on `DeferredQueue` and `cooperate`. `DeferredQueue` is mostly like a normal queue, but its `get` method returns a `Deferred`. If there happen to be no items in the queue at the time of the call, the `Deferred` will not fire until an item is added.

Here's an example:

from random import randrange

from twisted.internet.defer import DeferredQueue
from twisted.internet.task import deferLater, cooperate
from twisted.internet import reactor


def async(n):
    print 'Starting job', n
    d = deferLater(reactor, n, lambda: None)
    def cbFinished(ignored):
        print 'Finishing job', n
    d.addCallback(cbFinished)
    return d


def assign(jobs):
    # Create new jobs to be processed
    jobs.put(randrange(10))
    reactor.callLater(randrange(10), assign, jobs)


def worker(jobs):
    while True:
        yield jobs.get().addCallback(async)


def main():
    jobs = DeferredQueue()

    for i in range(10):
        jobs.put(i)

    assign(jobs)

    for i in range(3):
        cooperate(worker(jobs))

    reactor.run()


if __name__ == '__main__':
    main()

Note that the `async` worker function is the same as the one from the first example. However, this time, there's also a `worker` function which is explicitly pulling jobs out of the `DeferredQueue` and processing them with `async` (by adding `async` as a callback to the `Deferred` returned by `get`). The `worker` generator is driven by `cooperate`, which iterates it once after each `Deferred` it yields fires. The main loop, then, starts three of these worker generators so that three jobs will be in progress at any given time.

This approach involves a bit more code than the `DeferredSemaphore` approach, but has some benefits which may be interesting. First, `cooperate` returns a `CooperativeTask` instance which has useful methods like `pause`, `resume`, and a couple others. Also, all jobs assigned to the same cooperator will cooperate with each other in scheduling, so as not to overload the event loop (and this is what gives the API its name). On the `DeferredQueue` side, it's also possible to set limits on how many items are pending processing, so you can avoid completely overloading your server (for example, if your image processors get stuck and stop completing tasks). If the code calling `put` handles the queue overflow exception, you can use this as pressure to try to stop accepting new jobs (perhaps shunting them to another server, or alerting an administrator). Doing similar things with `DeferredSemaphore` is a bit trickier, since there's no way to limit how many jobs are waiting to be able to acquire the semaphore.

Problem

Original source