Threading.Condition problem

[Piet van Oostrum]

>>>>> Gabriel Rossetti <gabriel.rossetti at arimaz.com> (GR) wrote:

>GR> Sorry if this appears twice, I sent it once with an attachment and it never
>GR> arrived so maybe the attachment is posing problems. I inlined the code this
>GR> time (at the bottom), thank you,

>GR> Gabriel

>GR> ########################## Original message ############################

>GR> Hello everyone,

>GR> I wrote a small example that listens for xmpp msgs in a thread. The main
>GR> program calls a function that blocks (using Condition.wait) until a msg
>GR> has been received and then returns the msg. When a msg arrives, it is
>GR> put in a variable in the thread's object, it then calls the notify()
>GR> attr on the Condition object. For some reason, this doesn't work, the
>GR> thread gets the msg, tries to notify the Condition object, fails because
>GR> the lock has not been acquired yet and blocks. I tried ignoring the
>GR> failure, thinking that since it has not been acquired yet then when it
>GR> is, it will get the msg right away and never call Condition.wait, thus
>GR> not causing any problems, but this does not work either. Does someone
>GR> know what I am doing wrong? I attached the code to this msg.

The code that puts the message in the variable should also acquire the
lock:


     def onMessage(self, conn, msg):
         with self._cv:
             self.message = msg
             self._cv.notify()

A couple of remarks:

1. I think the code is neater if all manipulation with the condition is
   done in the same class (actually in the same instance -- making this
   instance into a monitor).

class Listener(Thread):
    def __init__(self, ws):
        Thread.__init__(self)
        self.interrupt = Event()
        self.message = None
        self._cv = Condition()
        self.client = ws._client
        self.client.RegisterHandler('message', self.onMessage)
    
    def onMessage(self, conn, msg):
        with self._cv:
            self.message = msg
            try:
                self._cv.notify()
            except RuntimeError:
                print "self._cv has not acquired the lock yet"
    
    def getMsg(self):
        with self._cv:
            while !self.message
                self._cv.wait()
            return self.message

class WS(object):
    def __init__(self, username, password, res):
        self._jid = xmpp.protocol.JID(username)
        self._client = xmpp.Client(self._jid.getDomain())
#        self._cv = Condition()
    
    def getMsg(self, mid=None):
        """
        """
        return self._listener.getMsg()

Of course I haven't tested this code as I don't have the context
modules.

2. I don't know if more than one message can be delivered in the same
   instance. If yes, than your code will not work, and neither will the
   code above as, the message instance variable is never cleared. So the
   next getMsg will be happy to deliver the previous one.
   You would have to clear it when returning this one.
    
    def getMsg(self):
        with self._cv:
            while !self.message
                self._cv.wait()
            msg = self.message
            self.message = None
            return msg

3. If the messages come in faster than they can be processed some will
   be lost as they will overwrite the previous one in the self.message
   variable. The solution is to use a threading.Queue to transfer the
   messages from one thread to the other. This also saves you the hassle
   of doing your own synchronisation like above. If you are not familiar
   with synchronising multithreaded applications it is very easy to make
   errors and even if you are it is quite easy to do them wrong. I have
   been involved in distributed programming courses at university level
   and I have seen many errors in this area.

-- 
Piet van Oostrum <piet at cs.uu.nl>
URL: http://pietvanoostrum.com [PGP 8DAE142BE17999C4]
Private email: piet at vanoostrum.org