[Python-ideas] channel (synchronous queue)

[Massimo Di Pierro]

On Feb 19, 2012, at 3:29 PM, Sturla Molden wrote:
> On the other hand, if we put in N**2 pipes (or channels), we could achieve the same atomicity of transaction by having an index for sender and receiver of a message. This is what MPI does in the functions MPI_Send and MPI_Recv. But then I will be scolded for using to many semaphores on FreeBSD again :-(

I like this a lot.

Below is some toy code I use in my parallel algorithms class (I removed the global communications broadcast, scatter, gather, reduce and I removed logging, network topology constraints, and checks).

class PSim(object):
    def __init__(self,p):
        """                                                                                       
        forks p-1 processes and creates p*p pipes                                                       
        """
        self.nprocs = p
        self.pipes = {}
        for i in range(p):
            for j in range(p):
                self.pipes[i,j] = os.pipe()
        self.rank = 0
        for i in range(1,p):
            if not os.fork():
                self.rank = i
   def send(self,j,data):
        s = cPickle.dumps(data)
        os.write(self.pipes[self.rank,j][1], string.zfill(str(len(s)),10))
        os.write(self.pipes[self.rank,j][1], s)
   def recv(self,j):
        size=int(os.read(self.pipes[j,self.rank][0],10))
        s=os.read(self.pipes[j,self.rank][0],size)
        data=cPickle.loads(s)
        return data

if __name__ == '__main__':
    comm = PSim(2)
    if comm.rank == 0: comm.send(1,'hello world')
    else: print comm.recv(0)

It would be very useful to have something like these channels built-in. Notice that using OS pipes have the problem of a OS dependent size. send is non-blocking for small data-size but becomes blocking for large data sizes. Using OS mkfifo or multiprocessing Queue is better but the OS limits the number of files open by one program.