Bounded flow control

Opening your lesson

A fast producer can fill memory while a slower consumer works correctly on every individual item. Bounding the queue makes the producer wait, and completion accounting tells the owner when accepted work has finished. These are service-level correctness rules.

You run a real asyncio queue, compare it with an explicit nonblocking admission model and finish a downstream batching contract. A waiting put and a rejected put are different policies, so each prompt names which one applies. Flush the final short batch. Acknowledge only successful effects, and retain failures for a later retry. The supplied values include zero and empty identifiers because sentinels must be chosen deliberately rather than borrowed from ordinary data.

import json

async def process_payload(payload: str) -> str:
    data = json.loads(payload)
    import asyncio
    queue = asyncio.Queue(maxsize=data['capacity'])
    result = []
    async def produce():
        for value in data['values']:
            await queue.put(value)
        await queue.put(None)
    async def consume():
        while True:
            value = await queue.get()
            try:
                if value is None:
                    return
                result.append(value * 2)
            finally:
                queue.task_done()
    producer = asyncio.create_task(produce())
    consumer = asyncio.create_task(consume())
    await producer
    await queue.join()
    await consumer
    return json.dumps(result,separators=(',',':'))

A bounded producer must yield when full; the consumer preserves FIFO values including zero. Start producer and consumer together so a full bounded queue can make progress. Choose a sentinel outside the integer data domain. The queue holds at most capacity pending values; output is FIFO and includes valid zero values.

Choose your character

Your Work Gym account