python threading in a loop
Question: I have a project that requires a bunch of large matrices, which are stored in
~200 MB files, to be cross-correlated (i.e. FFT * conj(FFT)) with each other.
The number of files is such that I can't just load them all up and then do my
processing. On the other hand, reading in each file as I need it is slower
than I'd like.

what I have so far is something like:

    
    
    result=0
    for i in xrange(N_files):
        f1 = file_reader(file_list[i])
    
        ############################################################################
        # here I want to have file_reader go start reading the next file I'll need #
        ############################################################################
    
        in_place_processing(f1)
        for j in xrange(i+1,N_files):
            f2 = file_reader(file_list[j])
    
            ##################################################################
            # here I want to have file_reader go start reading the next file #
            ##################################################################
    
            in_place_processing(f2)
            result += processing_function(f1,f2)
    

So basically, I just want to have two threads that will each read a file, give
it to me when I ask for it (or as soon as it's done after I ask for it), and
then go start reading the next file for when I ask for it. The object the
file_reader returns is rather large and complicated, so I'm not sure if
multiprocessing is the way to go here...

I've read about threading and queues but can't seem to figure out the part
where I ask the thread to go read the file and can proceed with the program
while it does. I don't want the threads to simply go about their business in
the background -- am I missing a detail here, or is threading not the way to
go?


Answer: Below is an example of using the `multiprocessing` module that will spawn off
child processes to call your `file_reader` method and queue up their results.
The queue should block when full, so you can control the number of read
ahead's you'd like to perform with the `QUEUE_SIZE` constant.

This utilizes a standard Producer/Consumer model of multiprocess
communication, with the child processes act as Producers, with the main thread
being the Consumer. The `join` method call in the class destructor ensures the
child process resources are cleaned up properly. There are some print
statements interspersed for demonstration purposes.

Additionally, I added the ability for the QueuedFileReader class to offload
work to a worker thread or run in the main thread, rather than using a child
process, for comparison. This is done by specifying the mode parameter at
class initialization to `MODE_THREADS` or `MODE_SYNCHRONOUS`, respectively.

    
    
    import multiprocessing as mp
    import Queue
    import threading
    import time
    
    QUEUE_SIZE = 2 #buffer size of queue
    
    ## Placeholder for your functions and variables
    N_files = 10
    file_list = ['file %d' % i for i in range(N_files)]
    
    def file_reader(filename):
        time.sleep(.1)
        result = (filename,'processed')
        return result
    
    def in_place_processing(f):
        time.sleep(.2)
    
    def processing_function(f1,f2):
        print f1, f2
        return id(f1) & id(f2)
    
    MODE_SYNCHRONOUS = 0  #file_reader called in main thread synchronously
    MODE_THREADS = 1      #file_reader executed in worker thread
    MODE_PROCESS = 2      #file_reader executed in child_process
    ##################################################
    ## Class to encapsulate multiprocessing objects.
    class QueuedFileReader():
        def __init__(self, idlist, mode=MODE_PROCESS):
            self.mode = mode
            self.idlist = idlist
            if mode == MODE_PROCESS:
                self.queue = mp.Queue(QUEUE_SIZE)
                self.process = mp.Process(target=QueuedFileReader.worker,
                                          args=(self.queue,idlist))
                self.process.start()
            elif mode == MODE_THREADS:
                self.queue = Queue.Queue(QUEUE_SIZE)
                self.thread = threading.Thread(target=QueuedFileReader.worker,
                                               args=(self.queue,idlist))
                self.thread.start()
    
        @staticmethod
        def worker(queue, idlist):
            for i in idlist:
                queue.put((i, file_reader(file_list[i])))
                print id(queue), 'queued', file_list[i]
            queue.put('done')
    
        def __iter__(self):
            if self.mode == MODE_SYNCHRONOUS:
                self.index = 0
            return self
    
        def next(self):
            if self.mode == MODE_SYNCHRONOUS:
                if self.index == len(self.idlist): raise StopIteration
                q = (self.idlist[self.index],
                     file_reader(file_list[self.idlist[self.index]]))
                self.index += 1
            else:
                q = self.queue.get()
                if q == 'done': raise StopIteration
            return q
    
        def __del__(self):
            if self.mode == MODE_PROCESS:
                self.process.join()
            elif self.mode == MODE_THREADS:
                self.thread.join()
    
    #mode = MODE_PROCESS
    mode = MODE_THREADS
    #mode = MODE_SYNCHRONOUS
    result = 0
    for i, f1 in QueuedFileReader(range(N_files),mode):
    
        in_place_processing(f1)
    
        for j, f2 in QueuedFileReader(range(i+1,N_files),mode):
            in_place_processing(f2)
            result += processing_function(f1,f2)
    

If your intermediate values are too large to pass through the Queue, you can
execute each iteration of the outer loop in its own process. A handy way to do
that would be using the `Pool` class in `multiprocessing` as in the example
below.

    
    
    import multiprocessing as mp
    import time
    
    ## Placeholder for your functions and variables
    N_files = 10
    file_list = ['file %d' % i for i in range(N_files)]
    
    def file_reader(filename):
        time.sleep(.1)
        result = (filename,'processed')
        return result
    
    def in_place_processing(f):
        time.sleep(.2)
    
    def processing_function(f1,f2):
        print f1, f2
        return id(f1) & id(f2)
    
    def file_task(file_index):
        print file_index
        f1 = file_reader(file_list[file_index])
        in_place_processing(f1)
        task_result = 0
        for j in range(file_index+1, N_files):
            f2 = file_reader(file_list[j])
            in_place_processing(f2)
            task_result += processing_function(f1,f2)
        return task_result
    
    
    
    pool = mp.Pool(processes=None) #processes default to mp.cpu_count()
    result = 0
    for file_result in pool.map(file_task, range(N_files)):
        result += file_result
    print 'result', result
    
    #or simply
    #result = sum(pool.map(file_task, range(N_files)))
    

