title: Python multiprocessing and handling exceptions in workers
body: <p>I use python multiprocessing library for an algorithm in which I have many workers processing certain data and returning result to the parent process. I use multiprocessing.Queue for passing jobs to workers, and second to collect results.</p>
  
  <p>It all works pretty well, until worker fails to process some chunk of data. In the simplified example below each worker has two phases:</p>
  
  <ul>
  <li>initialization - can fail, in this case worker should be destroyed</li>
  <li>data processing - processing a chunk of data can fail, in this case worker should skip this chunk and continue with next data.</li>
  </ul>
  
  <p>When either of this phases fails I get a deadlock after script completion. This code simulates my problem:</p>
  
  <pre><code>import multiprocessing as mp
  import random
  
  workers_count = 5
  # Probability of failure, change to simulate failures
  fail_init_p = 0.2
  fail_job_p = 0.3
  
  
  #========= Worker =========
  def do_work(job_state, arg):
      if random.random() &lt; fail_job_p:
          raise Exception("Job failed")
      return "job %d processed %d" % (job_state, arg)
  
  def init(args):
      if random.random() &lt; fail_init_p:
          raise Exception("Worker init failed")
      return args
  
  def worker_function(args, jobs_queue, result_queue):
      # INIT
      # What to do when init() fails?
      try:
          state = init(args)
      except:
          print "!Worker %d init fail" % args
          return
      # DO WORK
      # Process data in the jobs queue
      for job in iter(jobs_queue.get, None):
          try:
              # Can throw an exception!
              result = do_work(state, job)
              result_queue.put(result)
          except:
              print "!Job %d failed, skip..." % job
          finally:
              jobs_queue.task_done()
      # Telling that we are done with processing stop token
      jobs_queue.task_done()
  
  
  
  #========= Parent =========
  jobs = mp.JoinableQueue()
  results = mp.Queue()
  for i in range(workers_count):
      mp.Process(target=worker_function, args=(i, jobs, results)).start()
  
  # Populate jobs queue
  results_to_expect = 0
  for j in range(30):
      jobs.put(j)
      results_to_expect += 1
  
  # Collecting the results
  # What if some workers failed to process the job and we have
  # less results than expected
  for r in range(results_to_expect):
      result = results.get()
      print result
  
  #Signal all workers to finish
  for i in range(workers_count):
      jobs.put(None)
  
  #Wait for them to finish
  jobs.join()
  </code></pre>
  
  <p>I have two question about this code:</p>
  
  <ol>
  <li>When <code>init()</code> fails, how to detect that worker is invalid and not to wait for it to finish?</li>
  <li>When <code>do_work()</code> fails, how to notify parent process that less results should be expected in the results queue? </li>
  </ol>
  
  <p>Thank you for help!</p>