[Twisted-Python] Synchronization techniques

[Daniel Miller]

On Apr 3, 2007, at 8:42 PM, Itamar Shtull-Trauring wrote:

> On Tue, 2007-04-03 at 17:07 -0400, Daniel Miller wrote:
>
>>> twisted.internet.defer.DeferredLock and some of the related classes
>>> are
>>> what you ought to be using.
>>
>> Unfortunately that only gets me half way there. DeferredLock.acquire
>> () returns a deferred. How do I return the result of a deferred from
>> a PB remote_xxx() function?
>
> Just return the Deferred from the remote_xxx() function.
>

Thanks, I didn't know I could return a deferred from a PB remote_xxx 
() method. That detail doesn't seem to be documented in the  
Perspective Broker documentation, which I have read quite a few  
times. Maybe this could be highlighted in the "Complete Example" [0]  
section of the PB usage documentation? The examples use the  
TwistedQuotes application, and the IQuoter.getQuote() method always  
returns a string (at least I couldn't find any implementations that  
return a deferred). However, that would require rewriting most if not  
all implementations of IQuoter to return deferred's and/or the code  
that calls IQuoter.getQuote(), which demonstrates the viral nature of  
twisted when used in conjunction with other libraries.

So anyway, I rewrote my server-side library to do it the twisted way  
and return deferred's instead trying rig up some way of waiting for  
them. I still think it would be super useful to be able to pseudo- 
block on a deferred (i.e. allow the reactor to process other events  
while waiting for the deferred). It is very annoying to have to  
rewrite many layers of code when twisted is introduced into a  
program. I did find gthreadless.py, and maybe that would do it.  
Unfortunately discussion on that seems to have been dropped some time  
ago...

For the record, I've included updated versions of the previously  
posted code below. I'd be happy if someone pointed out if I'm doing  
anything wrong (with respect to twisted) in this code.

Thanks,
Daniel

[0] <http://twistedmatrix.com/projects/core/documentation/howto/pb- 
usage.html#auto1>

~~~~~~~~~

from twisted.internet.defer import Deferred, DeferredLock
from twisted.internet.error import ProcessTerminated
from twisted.internet.protocol import ProcessProtocol
from twisted.python.failure import Failure

class LockProxy(object):
     """A proxy that synchronizes method calls on a given subject"""

     def __init__(self, subject):
         self.lock = DeferredLock()
         self.subject = subject

     def __getattr__(self, name):
         value = getattr(self.subject, name)
         if hasattr(value, "im_self") and value.im_self is self.subject:
             def func(*args, **kw):
                 def callback(lock, *args, **kw):
                     try:
                         result = value(*args, **kw)
                     except:
                         lock.release()
                         raise
                     if isinstance(result, Deferred):
                         def release(arg, lock):
                             lock.release()
                             return arg
                         result.addBoth(release, lock)
                     else:
                         lock.release()
                     return result
                 dfr = self.lock.acquire()
                 dfr.addCallback(callback, *args, **kw)
                 return dfr
             return func
         return value

class ProcessError(Exception): pass

class ProcessErrorHandler(ProcessProtocol):

     def __init__(self, deferred):
         self.deferred = deferred
         self.errbuf = []
         self.errors = []

     def errReceived(self, text):
         self.errbuf.append(text)

     def processEnded(self, status):
         if self.errbuf:
             self.errors.append("".join(self.errbuf))
         if isinstance(status.value, ProcessTerminated):
             self.errors.append(status.getErrorMessage())
         if self.errors:
             err = Failure(ProcessError("\n".join(self.errors)))
             self.deferred.errback(err)
         else:
             self.deferred.callback(self)
         self.deferred = None

class TwistedProcess(object):

     def __init__(self, reactor=None, protocol=ProcessErrorHandler):
         if reactor is None:
             from twisted.internet import reactor
         self.reactor = reactor
         self.protocol = protocol

     def check_call(self, cmd):
         dfr = Deferred()
         protocol = self.protocol(dfr)
         self.reactor.spawnProcess(protocol, cmd[0], cmd,  
env=os.environ)
         return dfr