[Twisted-Python] Mimicking a blocking API using Twisted

Paul Reznicek maillists at ergo-pc.net
Sun Mar 25 21:27:42 EDT 2012

I have similar needs and did not found usable answers, so I wrote some
q&d hacks for doctests and also for UI interactions, where the user should
be blocked until something in non-blocking deferred finish or "time-outed".

Below hacks are working with twisted 10.0.0, no idea about more actual versions.

1. Multiple start/stop of reactor during doctests works when using "reactor.crash"
   Example - 2 tested class methods in one class:
# ---------------------
def method_a(self, parm):
    >>> g = GateWay('TEST')
    >>> s1 = reactor.callWhenRunning(lambda: print(g.method_a('X') ) )
    >>> s2 = reactor.callWhenRunning(lambda: print(g.method_a('Y') ) )
    >>> s3 = reactor.callWhenRunning(lambda: print(g.method_a('Z') ) )
    >>> s0 = reactor.callLater(11.1, reactor.crash); reactor.run()  # doctest:+ELLIPSIS
    {u'result of X...}
    {u'result of Y...}
    {u'result of Z...}

def method_b(self):
    >>> g = GateWay('TEST')
    >>> s1 = reactor.callWhenRunning(lambda: print(g.method_b() ) )
    >>> s0 = reactor.callLater(5.1, reactor.crash); reactor.run()   # doctest:+ELLIPSIS
    {u'bytes': ..., u'packets': ..., u'duration': ...}
# ---------------------
This way, it is possible to use doctests with reactor start/stop in several
methods/functions in a class/module and also in external doctests files.

2. Sleep w/o blocking reactor:
# ---------------------
def nonBlockingSleep(delay):
    '''Hack to allow sleep for a period of time w/o blocking the reactor
    Do *block* the caller for `delay` seconds.
    if delay > 90:
        print('WARNING nonBlockingSleep delay={0} sec. dangerous, reset to 90s'.format(delay) )
        delay = 90.0
    toTime = time.time() + delay
    while toTime >= time.time():
        except Exception as err:
# ---------------------
The values of .iterate() and .sleep() are the best, I found.
time.sleep(0.001) is important, it is not reliable w/o + CPU load is high.

3. Wait until deferred return a result or timeout expire
# ---------------------
def sleepUntilDFRDresult(dfrd, timeout=60):
    '''Hack to make non-blocking deferred waiting for its result or error for
    in `timeout` specified number of seconds, but not to block reactor loop.
    This wrapper *block* the caller until deferred.result or timeout appears.

    >>> dfrd = defer.Deferred() #.addBoth(cbReactorStop)
    >>> sleepUntilDFRDresult(dfrd)       # doctest:+ELLIPSIS
    Exception(u'ERROR sleepUntilDFRDresult server not running for: <Deferred at 0x...> from <doctest
__main__/<module> #L1',)

    >>> startTime = time.time()
    >>> f = lambda: print(sleepUntilDFRDresult(dfrd, timeout=999), 'in:', time.time()-startTime,
'sec.' )
    >>> s1 = reactor.callWhenRunning( f );
    >>> dc2 = reactor.callLater(2, dfrd.callback, 'CALLED')
    >>> dc0 = reactor.callLater(3, reactor.crash); reactor.run()   # doctest:+ELLIPSIS
    WARNING sleepUntilDFRDresult timeout=999 sec. dangerous
    CALLED in: 2.0... sec.
    if timeout > 600:
        print('WARNING sleepUntilDFRDresult timeout={0} sec. dangerous'.format(timeout) )

    err = None
    if not isinstance(dfrd, defer.Deferred):
        err = 'ERROR sleepUntilDFRDresult dfrd "{0}" not Deferred: '.format(type(dfrd))
    elif not reactor.running:
        err = 'ERROR sleepUntilDFRDresult server not running for: '
        toTime = time.time() + timeout
        while not dfrd.called and toTime >= time.time():
            except Exception as err:
        if dfrd.called and hasattr(dfrd, 'result'):
            return dfrd.result
    func_name, module_name, line_no = tools.getCallerNameLocation()
    repr_dfrd = '{0} from {2}/{1} #L{3}'.format(dfrd, func_name, module_name, line_no)
    err = Exception( (err or 'ERROR TIMEOUT sleepUntilDFRDresult for: ') + repr_dfrd )
##     print(err)
    return err
# ---------------------
This hack is not 100% reliable, but I did not found a better solution.
There is sometime an "Already called..." error in the logs, when server is under
high load with lot of connections, but since around 2 years I did not observed
a data loss or other severer problem.

I hope it helps.
If somebody have a better (less "hacky") solution for above 3 tasks,
please send it to this list.


Laurens Van Houtven wrote:
> The thing I'm documenting is a server, I'm documenting it by interacting with it as a client. I realize that doesn't entirely detract from your point -- you might still be introducing problems that would not affect a "real" client.
> My intention is to write BDD-ish stuff (except not with the usual cucumber-style scenario text).
> cheers
> lvh
> On 25 Mar 2012, at 23:44, Itamar Turner-Trauring wrote:
>> On 03/25/2012 05:02 PM, Laurens Van Houtven wrote:
>>> Hi,
>>> I'm trying to find out if there's a reasonable way to mimic a blocking 
>>> API with an existing non-blocking API. I want to do this so I can 
>>> write doctests.
>>> For example, I want to make a remote AMP call. It returns a deferred. 
>>> Instead of returning a deferred, I want it to block until the deferred 
>>> fires.
>> Doctests are supposed to document and test, or perhaps have tested 
>> documentation. If the API you're presenting doesn't match actual usage, 
>> that's bad documentation...

More information about the Twisted-Python mailing list