When an asynchronous application shuts down, queue.join lets it wait for queued orders to finish before stopping workers and closing cleanly. section 12.1.1
queue.join waits until every queued item has been processed and marked complete. Workers call queue.task_done after each item; join returns only when all outstanding work is finished. section 12.1.1
A reliable queue lifecycle follows this pattern:
- Create the queue and start worker tasks that retrieve items with queue.get. section 12.1.1
- Process each item and call queue.task_done when its work finishes. section 12.1.1
- Stop accepting new work, then await queue.join so existing items and in-progress work can finish. section 12.1.1
- Wrap the wait in a timeout when shutdown must not be blocked indefinitely. section 12.1.1
Draining a web crawler
A crawler seeds a queue with a root URL, lets workers download pages and enqueue discovered links, then awaits queue.join. Once no queued work remains and all workers have completed their items, it cancels the workers. section 12.1.2
Listing 12.3 Queues with a web application
import asyncio
from asyncio import Queue, Task
from typing import List
from random import randrange
from aiohttp import web
from aiohttp.web_app import Application
from aiohttp.web_request import Request
from aiohttp.web_response import Response
routes = web.RouteTableDef()
QUEUE_KEY = 'order_queue'
TASKS_KEY = 'order_tasks'
async def process_order_worker(worker_id: int, queue: Queue): #1
while True:
print(f'Worker {worker_id}: Waiting for an order...')
order = await queue.get()
print(f'Worker {worker_id}: Processing order {order}')
await asyncio.sleep(order)
print(f'Worker {worker_id}: Processed order {order}')
queue.task_done()
@routes.post('/order')
async def place_order(request: Request) -> Response:
order_queue = app[QUEUE_KEY]
await order_queue.put(randrange(5)) #2
return Response(body='Order placed!')
async def create_order_queue(app: Application): #3
print('Creating order queue and tasks.')
queue: Queue = asyncio.Queue(10)
app[QUEUE_KEY] = queue
app[TASKS_KEY] = [asyncio.create_task(process_order_worker(i, queue))
for i in range(5)]
async def destroy_queue(app: Application): #4
order_tasks: List[Task] = app[TASKS_KEY]
queue: Queue = app[QUEUE_KEY]
print('Waiting for pending queue workers to finish....')
try:
await asyncio.wait_for(queue.join(), timeout=10)
finally:
print('Finished all pending items, canceling worker tasks...')
[task.cancel() for task in order_tasks]
app = web.Application()
app.on_startup.append(create_order_queue)
app.on_shutdown.append(destroy_queue)
app.add_routes(routes)
web.run_app(app)
Shows queue.get, item processing, task_done, and the worker structure used with queue.join during application shutdown. section 12.1.1
Listing 12.4 A queue-based crawler
import asyncio
import aiohttp
import logging
from asyncio import Queue
from aiohttp import ClientSession
from bs4 import BeautifulSoup
class WorkItem:
def __init__(self, item_depth: int, url: str):
self.item_depth = item_depth
self.url = url
async def worker(worker_id: int, queue: Queue, session: ClientSession, max_depth: int):
print(f'Worker {worker_id}')
while True: #1
work_item: WorkItem = await queue.get()
print(f'Worker {worker_id}: Processing {work_item.url}')
await process_page(work_item, queue, session, max_depth)
print(f'Worker {worker_id}: Finished {work_item.url}')
queue.task_done()
async def process_page(work_item: WorkItem, queue: Queue, session: ClientSession, max_depth: int): #2
try:
response = await asyncio.wait_for(session.get(work_item.url), timeout=3)
if work_item.item_depth == max_depth:
print(f'Max depth reached, '
f'for {work_item.url}')
else:
body = await response.text()
soup = BeautifulSoup(body, 'html.parser')
links = soup.find_all('a', href=True)
for link in links:
queue.put_nowait(WorkItem(work_item.item_depth + 1,
link['href']))
except Exception as e:
logging.exception(f'Error processing url {work_item.url}')
async def main(): #3
start_url = 'http:/ /example.com'
url_queue = Queue()
url_queue.put_nowait(WorkItem(0, start_url))
async with aiohttp.ClientSession() as session:
workers = [asyncio.create_task(worker(i, url_queue, session, 3))
for i in range(100)]
await url_queue.join()
[w.cancel() for w in workers]
asyncio.run(main())
Shows crawler workers marking URLs complete while new links enter the queue before queue.join drains the expanding workload. section 12.1.2
