title: Multiprocessing in a pipeline done right
body: <p>I'd like to know how multiprocessing is done right. Assuming I have a list <code>[1,2,3,4,5]</code> generated by function <code>f1</code> which is written to a <code>Queue</code> (left green circle). Now I start two processes pulling from that queue (by executing <code>f2</code> in the processes). They process the data, say: doubling the value, and write it to the second queue. Now, function <code>f3</code> reads this data and prints it out.</p>
  
  <p><img src="https://i.stack.imgur.com/D4PAd.png" alt="layout of the data flow"></p>
  
  <p>Inside the functions there is a kind of a loop, trying to read from the queue forever. How do I stop this process?</p>
  
  <p><strong>Idea 1</strong></p>
  
  <p><code>f1</code> does not only send the list, but also a <code>None</code> object or a custon object, <code>class PipelineTerminator: pass</code> or some such which is just being propagated all the way down. <code>f3</code> now waits for <code>None</code> to come, when it's there, it breaks out of the loop. Problem: it's possible that one of the two <code>f2</code>s  reads and propagates the <code>None</code> while the other one it still processing a number. Then the last value is lost.</p>
  
  <p><strong>Idea 2</strong></p>
  
  <p><code>f3</code> is <code>f1</code>. So the function <code>f1</code> generates the data and the pipes, spawns the processes with <code>f2</code> and feeds all data. After spawning and feeding, it listens on the second pipe, simply counting and processing the received objects. Because it knows how much data fed, it can terminate the processes executing <code>f2</code>. But if the target is to set up a processing pipeline, the different steps should be separable. So <code>f1</code>, <code>f2</code> and <code>f3</code> are different elements of a pipeline, and the expensive steps are done in parallel.</p>
  
  <p><strong>Idea 3</strong></p>
  
  <p><img src="https://i.stack.imgur.com/WtH1a.png" alt="pipeline idea 3"></p>
  
  <p>Each piece of the pipeline is a function, this function spawns processes as it likes to and is responsible to manage them. It knows, how much data came in and how much data has been returned (with <code>yield</code> maybe). So it's safe to propagate a <code>None</code> object.</p>
  
  <pre><code>setup child processes 
  
  execute thread one and two and wait until both finished
  
  thread 1:
      while True:
          pull from input queue
          if None: break and set finished_flag
          else: push to queue1 and increment counter1
  
  thread 2:
      while True:
          pull from queue2
          increment counter2
          yield result
          if counter1 == counter2 and finished_flag: break
  
  when both threads finished: kill process pool and return.
  </code></pre>
  
  <p>(Instead of using threads, maybe one can think of a smarter solution.)</p>
  
  <p><strong>So ...</strong></p>
  
  <p>I have implemented a solution following idea 2, feeding and waiting for the results to arrive, but it was not really a pipeline with independent functions plugged together. It worked for the task I had to manage, but was hard to maintain. </p>
  
  <p>I'd like to hear from you now how you implement pipelines (easy in one process with generator functions and so on, but with multiple processes?) and manage them usually.</p>