Multithreading in Python with the threading and queue modules
Question: I have a file with hundreds of thousands of lines, each line of which needs to
be undergo the same process (calculating a co-variance). I was going to
multithread because it takes pretty long as is. All the examples/tutorials I
have seen have been fairly complicated for what I want to do, however. If
anyone could point me to a good tutorial that explains how to use the two
modules together that would be great.


Answer: Whenever I have to process something in parallel, I use something similar to
this (I just ripped this out of an existing script):

    
    
    #!/usr/bin/env python2
    # This Python file uses the following encoding: utf-8
    
    import os, sys, time
    from multiprocessing import Queue, Manager, Process, Value, Event, cpu_count
    
    class ThreadedProcessor(object):
      def __init__(self, parser, input_file, output_file, threads=cpu_count()):
        self.parser = parser
    
        self.num_processes = threads
        self.input_file = input_file
        self.output_file = output_file
    
        self.shared_proxy = Manager()
    
        self.input_queue = Queue()
        self.output_queue = Queue()
    
        self.input_process = Process(target=self.parse_input)
        self.output_process = Process(target=self.write_output)
    
        self.processes = [Process(target=self.process_row) for i in range(self.num_processes)]
    
        self.input_process.start()
        self.output_process.start()
    
        for process in self.processes:
          process.start()
    
        self.input_process.join()
    
        for process in self.processes:
          process.join()
    
        self.output_process.join()
    
      def parse_input(self):
        for index, row in enumerate(self.input_file):
          self.input_queue.put([index, row])
    
        for i in range(self.num_processes):
          self.input_queue.put('STOP')
    
      def process_row(self):
        for index, row in iter(self.input_queue.get, 'STOP'):
          self.output_queue.put([index, row[0], self.parser.parse(row[1])])
    
        self.output_queue.put('STOP')
    
      def write_output(self):
        current = 0
        buffer = {}
    
        for works in range(self.num_processes):
          for index, id, row in iter(self.output_queue.get, 'STOP'):
            if index != current:
              buffer[index] = [id] + row
            else:
              self.output_file.writerow([id] + row)
              current += 1
    
              while current in buffer:
                self.output_file.writerow(buffer[current])
                del buffer[current]
                current += 1
    

Basically, you have two processes managing the reading/writing of the file.
One reads and parses the input, the other reads from the "done" queue and
writes to your output file. The other processes are spawned (in this case the
number is equal to the number of total processor cores your CPU has) and they
all process elements from the input queue.

