Using a queue for synchronous multiprocessing
Question: I have a python program which reads a line from a file, processes the line and
then writes it to a new file. It repeats this for all lines in the file.
Essentially:

    
    
    for i in range(nlines):
        line = read_line(line_number = i)
        processed_line = process_line(line)
        write_line(line)
    

I'd like to multiprocess it so that one process takes care of the reading and
writing and another process takes care of the processing:

    
    
    read line 1 -> read line 2 -> write line 1 -> read line 3 -> write line 2 --> etc
                  process line 1 --------------> process line 2 ----------------> etc
    

I guess that I would need to utilise two queues to pass the data back and
forth, although I don't really know how to implement this in practice. Do you
have any ideas of how I might split this problem across 2 processes using
multiprocessing?


Answer: 
    from multiprocessing import Process, Queue as mpq
    
    def worker(qIn, qOut):
        for fpath,i,line in iter(qIn.get, None):
            qOut.put((fpath,i, process_line(line)))
    
    def main(infilepath, outfilepath):
        qIn, qOut = mpq(), mpq()
        Process(target=worker, args=(qIn, qOut)).start()
        with open(infilepath) as infile, open(outfilepath, 'w') as outfile:
            numLines = 0
            for tup in enumerate(infile):
                qIn.put(tup)
                numLines += 1
            qIn.put(None)
            retlines = {}
            top = -1
            for _ in range(numLines)
                i,line = qOut.get))
                retlines[i] = line
                if i+1 in retlines:
                    outfile.write(retlines.pop(i+1))
                    i += 1
    

Of course, this waits to finish reading the input file before it starts
writing to the output file, which is an efficiency bottleneck. I would do this
instead:

    
    
    def procWorker(qIn, qOut, numWriteWorkers):
        for fpath,i,line in iter(qIn.get, None):
            qOut.put((fpath,i, process_line(line)))
        for _ in range(numWriteWorkers):
            qOut.put(None)
    
    def readWorker(qIn, qOut, numProcWorkers):
        for infilepath in iter(qIn.get, None):
            with open(infilepath) as infile:
                for line in infile:
                    qOut.put((infilepath, i, line))
        for _  in range(numProcWorkers):
            qOut.put(None)
    
    def writeWorker(qIn, qOut):
        outfilepaths = {"test1.in" : "test1.out"}  # dict that maps input filepaths to corresponding output filepaths
        lines = collections.defaultdict(dict)
        inds = collections.defaultdict(lamnda : -1)
        for fpath,i,line in iter(qIn.get, None):
            if i == inds[fpath] + 1:
                inds[fpath] += 1
                with open(outfilepaths[fpath], 'a') as outfile:
                    outfile.write(line)
                    qOut.put((fpath, i))
            else:
                lines[fpath][i] = line
        for fpath in lines:
            with open(outfilepaths[fpath], 'a') as outfile:
                for i in sorted(fpath):
                    outfile.write(lines[fpath][i])
                    qOut.put((fpath, i))
        qOut.put(None)
    
    def main(infilepaths):
        readqIn, readqOut, procqOut, writeqOut = [Queue for _ in range(4)]
        numReadWorkers = 1  # fiddle to taste
        numWriteWorkers = 1  # fiddle to taste
        numProcWorkers = 1  # fiddle to taste
    
        for _ in range(numReadWorkers):
            Process(target=readWorker, args=(readqIn, readqOut, numProcWorkers)).start()
        for infilepath in infilepaths:
            readqIn.put(infilepath)
        for _ in range(numReadWorkers):
            readqIn.put(None)
    
        for _ in range(numProcWorkers):
            Process(target=procWorker, args=(readqOut, procqOut, numWriteWorkers)).start()
    
        for _ in range(numWriteWorkers):
            Process(target=writeWorker, args=(procqOut, writeqOut)).start()
    
        writeStops = 0
        while True:
            if writeStops == numWriteWorkers:
                break
            msg = writeqOut.get()
            if msg == None:
                writeStops += 1
            else:
                fpath, i = msg
                print("line #%d was written to file %s" %(i, fpath))
    

Note that this allows for the possibility of multiple readers and writers.
Normally, this would be pointless, as there is only on head on a hard drive.
However, if you are on some distributed filesystem or your files reside on
multiple hard drives, then you could increase the number of read/write workers
to increase your efficiency. Assuming a trivial `process_line` function,
`numReadWorkers` \+ `numWriteWorkers` should equal the number of magnetic
heads you have on all your hard drives. You could balance files on drives (_a
la_ raid) to reach some optimum here, but a lot depends on file sizes,
read/write speeds, caching, etc.

Really, the first speedups you should get are from fiddling with
`numProcWorkers`, which should give you a nice linear increase in efficiency,
bounded of course, by the number of logical core processors on your machine

