[Python-ideas] channel (synchronous queue)

[Arnaud Delobelle]

On 18 February 2012 15:38, Matt Joiner <anacrolix at gmail.com> wrote:
> Recently (for some) the CSP style of channel has become quite popular
> in concurrency implementations. This kind of channel allows sends that
> do not complete until a receiver has actually taken the item. The
> existing  queue.Queue would act like this if it didn't treat a queue
> size of 0 as infinite capacity.

I don't know if that's exactly what you have in mind, but you can
implement a channel very simply with a threading.Barrier object (new
in Python 3.2).  I'm no specialist of concurrency at all, but it seems
that this is what you are describing (what in the go language is
called a "synchronous channel" I think):

from threading import Barrier

class Channel:
    def __init__(self):
        self._sync = Barrier(2)
        self._values = [None, None]
    def send(self, value=None):
        i = self._sync.wait()
        self._values[i] = value
        self._sync.wait()
        return self._values[1 - i]
    def get(self):
        return self.send()

Then with the following convenience function to start a function in a
new thread:

from threading import Thread

def go(f, *args, **kwargs):
    thread = Thread(target=f, args=args, kwargs=kwargs)
    thread.start()
    return thread

You can have e.g. the scenario:

ch = Channel()

def produce(ch):
    for i in count():
        print("sending", i)
        ch.send(i)

def consume(ch, n):
    for i in range(n):
        print("getting", ch.get())

Giving you this:

>>> go(produce, ch)
sending 0
<Thread(Thread-103, started 4759019520)>
>>> go(consume, ch, 3)
<Thread(Thread-104, started 4763226112)>
getting 0
sending 1
getting 1
sending 2
getting 2
sending 3
>>> go(consume, ch, 5)
<Thread(Thread-105, started 4763226112)>
getting 3
sending 4
getting 4
sending 5
getting 5
sending 6
getting 6
sending 7
getting 7
sending 8
>>>

-- 
Arnaud