[Python-ideas] channel (synchronous queue)

[Sturla Molden]

Den 19.02.2012 17:01, skrev Antoine Pitrou:

> Even for multiprocessing? (I realize we didn't implement a Barrier in 
> multiprocessing; patches welcome :-)) Regards Antoine. 
> _______________________________________________ Python-ideas mailing 
> list Python-ideas at python.org 
> http://mail.python.org/mailman/listinfo/python-ideas 


Here is a sceleton (replace self._data with some IPC mechanism for 
multiprocessing).

I'll post a barrier for multiprocessing asap, I happen to have one ;-)

Sturla


from threading import Lock, Event

class Channel(object):

     def __init__(self):
         self._writelock = Lock()
         self._readlock = Lock()
         self._new_data = Event()
         self._recv_data = Event()
         self._data = None

     def put(self, msg):
         with self._writelock:
             self._data = msg
             self._new_data.set()
             self._recv_data.wait()
             self._recv_data.clear()

     def get(self):
         with self._readlock:
             self._new_data.wait()
             msg = self._data
             self._data = None
             self._new_data.clear()
             self._recv_data.set()
         return msg


if __name__ == "__main__":

     from threading import Thread
     from sys import stdout

     def thread2(channel):
         for i in range(1000):
             msg = channel.get()
             stdout.flush()
             print "Thread 2 received '%s'\n" % msg,
             stdout.flush()

     def thread1(channel):
         for i in range(1000):
             stdout.flush()
             print "Thread 1 preparing to send 'message %d'\n" % i,
             stdout.flush()
             msg = channel.put(("message %d" % i,))
             stdout.flush()
             print "Thread 1 finished sending 'message %d'\n" % i,
             stdout.flush()

     channel = Channel()
     t2 = Thread(target=thread2, args=(channel,))
     t2.start()
     thread1(channel)
     t2.join()