Source Code for Module commons.async

 1  # -*- mode: python; tab-width: 4; indent-tabs-mode: nil; py-indent-offset: 4; -*- 
 2  # vim:ft=python:et:sw=4:ts=4 
 3   
 4  """ 
 5  Asynchronous utilities for use with the 
 6  U{Twisted<http://twistedmatrix.com/>} library. 
 7  """ 
 8   
 9  from twisted.internet import defer, reactor 
10   
11 -def async_sleep( seconds ): 
12      """ 
13      Acts as the asynchronous version of C{time.sleep}. 
14   
15      @param seconds: The amount of time in seconds to sleep. 
16      @type seconds: float 
17   
18      @return: The C{defer.Deferred} that will be fired on wake-up. 
19      @rtype: defer.Deferred 
20      """ 
21      d = defer.Deferred() 
22      reactor.callLater( seconds, lambda: d.callback( None ) ) 
23      return d 
24   
25 -def asynchronized( func ): 
26      """ 
27      Acts as the asynchronous version of a C{synchronized} decorator 
28      (inspired by Java's keyword). Ensures that no event-handling path 
29      (the asynchronous equivalent to threads) will enter the function 
30      if another one is currently in it (which can happen if the current 
31      "thread" is itself waiting on a deferred, thus giving the reactor 
32      an opportunity to execute the competing "thread"). 
33   
34      Provides additional protection against U{Twisted bug 
35      411<http://twistedmatrix.com/trac/ticket/411>}. In this case, if 
36      too many acquisition requests are chained onto the deferred which 
37      is fired on release, then the stack will overflow. This is 
38      circumvented by maintaining a queue of closures and deferreds 
39      which is checked on each release, to keep the callback chain 
40      "flat." 
41   
42      @param func: The function to decorate. 
43      @type func: function 
44   
45      @return: The "thread-safe" version of the function. 
46      @rtype: function 
47      """ 
48      def wrapper( self, *args, **kwargs ): 
49          try: 
50              q = self._sync_queue 
51          except AttributeError: 
52              q = self._sync_queue = [] 
53          def release( result ): 
54              invoke, d = q[0] 
55              # actually perform the callback on the outer 
56              # deferred; the deferred running this cb() should 
57              # just terminate and be discarded 
58              d.callback( result ) 
59               
60              del q[0] 
61              if len( q ) > 0: 
62                  invoke, d = q[0] 
63                  # we're tempted to do just run invoke() directly, but 
64                  # that would lead to the same stack explosion 
65                  reactor.callLater( 0, invoke ) 
66              return result 
67          def invoke(): 
68              return defer.maybeDeferred( func, self, *args, **kwargs ).addCallback( release ) 
69          d = defer.Deferred() 
70          q.append( ( invoke, d ) ) 
71          if len( q ) == 1: 
72              reactor.callLater( 0, invoke ) 
73          return d 
74      wrapper.__name__ = func.__name__ 
75      wrapper.__dict__ = func.__dict__ 
76      wrapper.__doc__ = func.__doc__ 
77      return wrapper 
78