How to make worker threads quit after work is finished in a multithreaded producer-consumer pattern?

Question

I am trying to implement a multithreaded producer-consumer pattern using Queue.Queue in Python 2.7. I am trying to figure out how to make the consumers, i.e. the worker threads, stop once all required work is done.

See the second comment by Martin James to this answer: https://stackoverflow.com/a/19369877/1175080

Send an 'I am finished' task, instructing the pool threads to terminate. Any thread that gets such a task requeues it and then commits suicide.

But this does not work for me. See the following code for example.

import Queue
import threading
import time

def worker(n, q):
    # n - Worker ID
    # q - Queue from which to receive data
    while True:
        data = q.get()
        print 'worker', n, 'got', data
        time.sleep(1)  # Simulate noticeable data processing time
        q.task_done()
        if data == -1: # -1 is used to indicate that the worker should stop
            # Requeue the exit indicator.
            q.put(-1)
            # Commit suicide.
            print 'worker', n, 'is exiting'
            break

def master():
    # master() sends data to worker() via q.
    q = Queue.Queue()

    # Create 3 workers.
    for i in range(3):
        t = threading.Thread(target=worker, args=(i, q))
        t.start()

    # Send 10 items to work on.
    for i in range(10):
        q.put(i)
        time.sleep(0.5)

    # Send an exit indicator for all threads to consume.
    q.put(-1)

    print 'waiting for workers to finish ...'
    q.join()
    print 'done'

master()

This program hangs after all three workers have read the exit indicator, i.e. -1 from the queue, because each worker requeues -1 before exiting, so the queue never becomes empty and q.join() never returns.

I came up with the following but ugly solution where I send a -1 exit indicator for each worker via the queue, so that each worker can see it and commit suicide. But the fact that I have to send an exit indicator for each worker feels a little ugly.

import Queue
import threading
import time

def worker(n, q):
    # n - Worker ID
    # q - Queue from which to receive data
    while True:
        data = q.get()
        print 'worker', n, 'got', data
        time.sleep(1)  # Simulate noticeable data processing time
        q.task_done()
        if data == -1: # -1 is used to indicate that the worker should stop
            print 'worker', n, 'is exiting'
            break

def master():
    # master() sends data to worker() via q.
    q = Queue.Queue()

    # Create 3 workers.
    for i in range(3):
        t = threading.Thread(target=worker, args=(i, q))
        t.start()

    # Send 10 items to work on.
    for i in range(10):
        q.put(i)
        time.sleep(0.5)

    # Send one stop indicator for each worker.
    for i in range(3):
        q.put(-1)

    print 'waiting for workers to finish ...'
    q.join()
    print 'done'

master()

I have two questions.

1. Can the method of sending a single exit indicator for all threads (as explained in the second comment of https://stackoverflow.com/a/19369877/1175080 by Martin James) even work?
2. If the answer to the previous question is "No", is there a way to solve the problem in a way that I don't have to send a separate exit indicator for each worker thread?

Answer 1

Don't call it a special case for a task.

Use an Event instead, with non-blocking implementation for your workers.

stopping = threading.Event()

def worker(n, q, timeout=1):
    # run until the master thread indicates we're done
    while not stopping.is_set():
        try:
            # don't block indefinitely so we can return to the top
            # of the loop and check the stopping event
            data = q.get(True, timeout)
        # raised by q.get if we reach the timeout on an empty queue
        except queue.Empty:
            continue
        q.task_done()

def master():
    ...

    print 'waiting for workers to finish'
    q.join()
    stopping.set()
    print 'done'

Answer 2

Can the method of sending a single exit indicator for all threads (as explained in the second comment of https://stackoverflow.com/a/19369877/1175080 by Martin James) even work?

As you have notice it can't work, spreading the message will make the last thread to update the queue with one more item and since you are waiting for a queue that will never be empty, not with the code you have.

If the answer to the previous question is "No", is there a way to solve the problem in a way that I don't have to send a separate exit indicator for each worker thread?

You can join the threads instead of the queue:

def worker(n, q):
    # n - Worker ID
    # q - Queue from which to receive data
    while True:
        data = q.get()
        print 'worker', n, 'got', data
        time.sleep(1)  # Simulate noticeable data processing time
        q.task_done()
        if data == -1: # -1 is used to indicate that the worker should stop
            # Requeue the exit indicator.
            q.put(-1)
            # Commit suicide.
            print 'worker', n, 'is exiting'
            break

def master():
    # master() sends data to worker() via q.
    q = Queue.Queue()

    # Create 3 workers.
    threads = [threading.Thread(target=worker, args=(i, q)) for i in range(3)]
    for t in threads:
        threads.start()
    # Send 10 items to work on.
    for i in range(10):
        q.put(i)
        time.sleep(0.5)

    # Send an exit indicator for all threads to consume.
    q.put(-1)

    print 'waiting for workers to finish ...'
    for t in threads:
        t.join()
    print 'done'

master()

As the Queue documentation explain get method will rise an execption once its empty so if you know already the data to process you can fill the queue and then spam the threads:

import Queue
import threading
import time

def worker(n, q):
    # n - Worker ID
    # q - Queue from which to receive data
    while True:
        try:
            data = q.get(block=False, timeout=1)
            print 'worker', n, 'got', data
            time.sleep(1)  # Simulate noticeable data processing time
            q.task_done()
        except Queue.Empty:
            break


def master():
    # master() sends data to worker() via q.
    q = Queue.Queue()

    # Send 10 items to work on.
    for i in range(10):
        q.put(i)

    # Create 3 workers.
    for i in range(3):
        t = threading.Thread(target=worker, args=(i, q))
        t.start()

    print 'waiting for workers to finish ...'
    q.join()
    print 'done'

master()

Here you have a live example

Answer 3

Just for completeness sake: You could also enqueue a stop signal which is -(thread count). Each thread then can increment it by one and re queue it only if the stop signal is != 0.

    if data < 0: # negative numbers are used to indicate that the worker should stop
        if data < -1:
            q.put(data + 1)
        # Commit suicide.
        print 'worker', n, 'is exiting'
        break

But i'd personally go with Travis Mehlinger or Daniel Sanchez answer.

Answer 4

In addition to @DanielSanchez excellent answer, I propose to actually rely on a similar mechanism as a Java CountDownLatch.

The gist being,

• you create a latch that will open only after a certain counter went down,

• when the latch is opened, the thread(s) waiting on it will be allowed to proceed with their execution.

• I made an overly simple example, check here for a class like example of such a latch:

  import threading
  import Queue
  import time
  
  WORKER_COUNT = 3
  latch = threading.Condition()
  count = 3
  
  def wait():
      latch.acquire()
      while count > 0:
          latch.wait()
      latch.release()
  
  def count_down():
      global count
      latch.acquire()
      count -= 1
      if count <= 0:
          latch.notify_all()
      latch.release()
  
  def worker(n, q):
      # n - Worker ID
      # q - Queue from which to receive data
      while True:
          data = q.get()
          print 'worker', n, 'got', data
          time.sleep(1)  # Simulate noticeable data processing time
          q.task_done()
          if data == -1: # -1 is used to indicate that the worker should stop
              # Requeue the exit indicator.
              q.put(-1)
              # Commit suicide.
              count_down()
              print 'worker', n, 'is exiting'
              break
  
  # master() sends data to worker() via q.  
  
  def master():
      q = Queue.Queue()
  
      # Create 3 workers.
      for i in range(WORKER_COUNT):
          t = threading.Thread(target=worker, args=(i, q))
          t.start()
  
      # Send 10 items to work on.
      for i in range(10):
          q.put(i)
          time.sleep(0.5)
  
      # Send an exit indicator for all threads to consume.
      q.put(-1)
      wait()
      print 'done'
  
  master()