title: Self-joining thread pool: where's my race condition?
body: <p>Since I use a similar pattern in my work a lot, I decided to write a class that abstracts very simple worker concurrency via job queue / threading. I know there are already things out there that solve this, but I also wanted to use this as an opportunity to hone my multithreading skills.</p>
  
  <p>The main challenge I've given myself is that I want this to be able to let processes finish, even if they are not explicitly blocked by Queue.join(). "A process finishing" is defined by the input function returning a value (or None). The way I have attempted to accomplish this is by having each job create it's own results queue <code>rq</code>, which is then checked by <code>_wait_for_results</code> in a non-daemon thread, which blocks the automatic exit of all other daemonized threads until <code>rq</code> is filled by the worker in <code>add_to_queue</code>.</p>
  
  <p>Here is the full class:</p>
  
  <pre><code>class EasyPool(object):
      def __init__(self, concurrency, always_finish=True):
          def add_to_queue(q):
              while True:
                  func_data, rq = q.get()
                  func, args, kwargs = func_data
                  if not args:
                      args = []
                  if not kwargs:
                      kwargs = {}
                  result = func(*args, **kwargs)
                  rq.put(result)
                  q.task_done()
  
          self.rqs = []
          self.always_finish = always_finish
          self.q = Queue(maxsize=0)
          self.workers = []
          for i in range(concurrency):
              worker = Thread(target=add_to_queue, args=(self.q,))
              self.workers.append(worker)
              worker.setDaemon(True)
              worker.start()
  
      def _wait_for_results(self, rq):
          rq.not_empty.acquire()
          rq.not_empty.wait()
          rq.not_empty.notify()
          rq.not_empty.release()
  
      def add_job(self, func, *args, **kwargs):
          rq = Queue()
          if self.always_finish:
              blocker = Thread(target=self._wait_for_results, args=(rq,))
              blocker.setDaemon(False)
              blocker.start()
          to_add = []
          [ to_add.append(i) if i else to_add.append(None) for i in [func, args, kwargs] ]
          self.q.put((to_add, rq))
          return rq.get
  </code></pre>
  
  <p>When a job is created via the <code>.add_job</code> instance method, it immediately returns a promise-like object, which is a reference to the <code>.get</code> method of the results queue. The problem I'm facing is that there seems to be a race condition between this <code>.get</code> and the <code>_wait_for_results</code> method. I think the answer probably involves a Lock or a Condition, but I'm not really sure. Any help is much appreciated :)</p>