A decorator for adding timeouts to Deferred calls that don't offer an alternative

Twisted Timeout Decorator.py

#
# This gist is released under Creative Commons Public Domain Dedication License CC0 1.0
# http://creativecommons.org/publicdomain/zero/1.0/
#

from twisted.internet import defer, reactor

class TimeoutError(Exception):
    """Raised when time expires in timeout decorator"""

def timeout(secs):
    """
    Decorator to add timeout to Deferred calls
    """
    def wrap(func):
        @defer.inlineCallbacks
        def _timeout(*args, **kwargs):
            rawD = func(*args, **kwargs)
            if not isinstance(rawD, defer.Deferred):
                defer.returnValue(rawD)

            timeoutD = defer.Deferred()
            timesUp = reactor.callLater(secs, timeoutD.callback, None)

            try:
                rawResult, timeoutResult = yield defer.DeferredList([rawD, timeoutD], fireOnOneCallback=True, fireOnOneErrback=True, consumeErrors=True)
            except defer.FirstError, e:
                #Only rawD should raise an exception
                assert e.index == 0
                timesUp.cancel()
                e.subFailure.raiseException()
            else:
                #Timeout
                if timeoutD.called:
                    rawD.cancel()
                    raise TimeoutError("%s secs have expired" % secs)

            #No timeout
            timesUp.cancel()
            defer.returnValue(rawResult)
        return _timeout
    return wrap

Hi Roger,

I simply made my function test to replace your get_nearby as follows:

@defer.inlineCallbacks
def test(secs,counter):
sleep(secs)
tmpstr = ""
for i in range(counter):
tmpstr += str(i)
print tmpstr
return "no timeout"

@defer.inlineCallbacks
def main():
timeout_test = timeout(5)(test)
try:
start = time()
result = yield timeout_test(6,10)
end = time()
print "use time:",end-start
finally:
reactor.stop()
print result

main()
reactor.run()

The program just hangs up.
When we comment reactor.stop(), we can run the program with an error of "
Unhandled error in Deferred:
Unhandled Error
Traceback (most recent call last):
File "/usr/local/lib/python2.7/dist-packages/Twisted-11.0.0-py2.7-linux-x86_64.egg/twisted/internet/defer.py", line 1141, in unwindGenerator
return _inlineCallbacks(None, f(_args, *_kwargs), Deferred())
File "/usr/local/lib/python2.7/dist-packages/Twisted-11.0.0-py2.7-linux-x86_64.egg/twisted/internet/defer.py", line 1020, in _inlineCallbacks
result = g.send(result)
File "twisted_timeout_decorator.py", line 14, in _timeout
rawD = func(_args, *_kwargs)
File "/usr/local/lib/python2.7/dist-packages/Twisted-11.0.0-py2.7-linux-x86_64.egg/twisted/internet/defer.py", line 1141, in unwindGenerator
return _inlineCallbacks(None, f(_args, *_kwargs), Deferred())
--- ---
File "/usr/local/lib/python2.7/dist-packages/Twisted-11.0.0-py2.7-linux-x86_64.egg/twisted/internet/defer.py", line 1020, in _inlineCallbacks
result = g.send(result)
exceptions.AttributeError: 'str' object has no attribute 'send' "

Could you please provide a workable/debugable example to invoke timeout function above?

Thanks
Yong

Your test function needs to return a Deferred, instead of calling sleep().

    def test(secs): 
        print "TEST %s" % secs
        d = defer.Deferred()
        reactor.callLater(secs, lambda: d.callback(None))
    return d

Here's the whole thing:

    from twisted.internet import defer, reactor
    import time

    class TimeoutError(Exception):
        """Raised when time expires in timeout decorator"""

    def timeout(secs):
        """
        Decorator to add timeout to Deferred calls
        """
        def wrap(func):
            @defer.inlineCallbacks
            def _timeout(*args, **kwargs):
                rawD = func(*args, **kwargs)
                if not isinstance(rawD, defer.Deferred):
                    defer.returnValue(rawD)

                timeoutD = defer.Deferred()
                timesUp = reactor.callLater(secs, timeoutD.callback, None)

                try:
                    rawResult, timeoutResult = yield defer.DeferredList([rawD, timeoutD], fireOnOneCallback=True, fireOnOneErrback=True, consumeErrors=True)
                except defer.FirstError, e:
                    #Only rawD should raise an exception
                    assert e.index == 0
                    timesUp.cancel()
                    e.subFailure.raiseException()
                else:
                    #Timeout
                    if timeoutD.called:
                        rawD.cancel()
                        raise TimeoutError("%s secs have expired" % secs)

                #No timeout
                timesUp.cancel()
                defer.returnValue(rawResult)
            return _timeout
        return wrap


    def test(secs):
        print "TEST %s" % secs
        d = defer.Deferred()
        reactor.callLater(secs, lambda: d.callback(None))
        return d

    @defer.inlineCallbacks
    def main():
        timeout_test = timeout(5)(test)
        try:
            start = time.time()
            result = yield timeout_test(1)
            end = time.time()
            print "use time:",end-start
            print result
        except Exception, e:
            print e
        finally:
            try:
                reactor.stop()
            except Exception, e:
                print e

    main()
    reactor.run()

Hi Roger,

Thanks a lot for your explanation.

Regards,
Yong

Roger, can you clarify the copyright/license for this code? Thanks.

It's public domain. I do not wish to retain any copyrights. What's the best way to indicate that?

Roger,

CC0 is intended to allow one to place a work as nearly as possible into the public domain, worldwide. I believe that it would be sufficient to add a comment along the lines of:

"This gist is released under the Creative Commons 0 license. See http://creativecommons.org/publicdomain/zero/1.0/ for more information"

I appreciate it.

Thanks @htoothrot