Почему этот вызов queue.join блокируется бесконечно?

Question

Я играю с личным проектом в python3.6, и я столкнулся со следующей проблемой, которая приводит к бесконечной блокировке вызова my_queue.join().Обратите внимание, что это не мой настоящий код, а минимальный пример, демонстрирующий проблему.

import threading
import queue

def foo(stop_event, my_queue):
  while not stop_event.is_set():
    try:
      item = my_queue.get(timeout=0.1)
      print(item) #Actual logic goes here
    except queue.Empty:
      pass
  print('DONE')

stop_event = threading.Event()
my_queue = queue.Queue()
thread = threading.Thread(target=foo, args=(stop_event, my_queue))
thread.start()

my_queue.put(1)
my_queue.put(2)
my_queue.put(3)

print('ALL PUT')

my_queue.join()

print('ALL PROCESSED')

stop_event.set()

print('ALL COMPLETE')

Я получаю следующий вывод (на самом деле он был непротиворечивым, но я понимаю, что порядок вывода может отличаться из-за многопоточности):

ALL PUT
1
2
3

Сколько бы я ни ждал, я никогда не увижу вывод ALL PROCESSED на консоль, так почему же my_queue.join() блокируется на неопределенный срок, когда все элементы были обработаны?

Answer 1

Из документов :

Количество незавершенных задач увеличивается при каждом добавлении элемента в очередь. Количество уменьшается, когда потребительский поток вызывает task_done (), чтобы указать, что элемент был получен и вся работа над ним завершена .Когда количество незавершенных задач падает до нуля, join () разблокирует.

Вы никогда не вызываете q.task_done() внутри своей функции foo.Функция foo должна выглядеть примерно так:

def worker():
    while True:
        item = q.get()
        if item is None:
            break
        do_work(item)
        q.task_done()