queue.task_done records that a worker has finished processing one item previously retrieved from an asyncio.Queue. Each queue retrieval increases the unfinished-task count; task_done decreases it. This bookkeeping lets queue.join determine when all queued work is complete, not merely when the queue contains no items. section 12.1
Supermarket checkout workers
Several cashier tasks retrieve customers from a queue and process their products. After a customer is checked out, the worker calls queue.task_done. The main coroutine can then await customer_queue.join(); without task_done, join may never terminate because the queue retains an unfinished-work count. section 12.1
Listing 12.1 A supermarket checkout queue
import asyncio
from asyncio import Queue
from random import randrange
from typing import List
class Product:
def __init__(self, name: str, checkout_time: float):
self.name = name
self.checkout_time = checkout_time
class Customer:
def __init__(self, customer_id: int, products: List[Product]):
self.customer_id = customer_id
self.products = products
async def checkout_customer(queue: Queue, cashier_number: int):
while not queue.empty(): #1
customer: Customer = queue.get_nowait()
print(f'Cashier {cashier_number} '
f'checking out customer '
f'{customer.customer_id}')
for product in customer.products: #2
print(f"Cashier {cashier_number} "
f"checking out customer "
f"{customer.customer_id}'s {product.name}")
await asyncio.sleep(product.checkout_time)
print(f'Cashier {cashier_number} '
f'finished checking out customer '
f'{customer.customer_id}')
queue.task_done()
async def main():
customer_queue = Queue()
all_products = [Product('beer', 2),
Product('bananas', .5),
Product('sausage', .2),
Product('diapers', .2)]
for i in range(10): #3
products = [all_products[randrange(len(all_products))]
for _ in range(randrange(10))]
customer_queue.put_nowait(Customer(i, products))
cashiers = [asyncio.create_task(checkout_customer(customer_queue, i)) #4
for i in range(3)] #4
await asyncio.gather(customer_queue.join(), *cashiers)
asyncio.run(main())
Inspect the complete supermarket checkout listing to see task_done placed after a worker finishes its current customer. section 12.1
