Werk begrenzen met een queue en tegendruk

Een producent kan sneller nieuwe opdrachten aanleveren dan een verwerker ze afhandelt. Zonder afspraak blijft de voorraad groeien. Een begrensde async queue geeft de producent een reden om even te wachten wanneer de tussenbuffer vol is.

Een wachtrij tussen twee rollen

Een asyncio.Queue bewaart items in FIFO-volgorde: het eerst ingevoegde item wordt het eerst opgehaald. De producent voegt werk toe; een consumer haalt het op en behandelt het. Die rollen kunnen ieder hun eigen coroutine zijn.

Met await queue.put(item) bied je een item aan. Met await queue.get() wacht een consumer op een beschikbaar item. Bij maxsize=1 past er één item in de buffer; een volgende put moet wachten totdat een consumer ruimte maakt.

Dat wachten heet tegendruk of backpressure. De producent kan dan niet onbeperkt vooruitlopen. Het beperkt de hoeveelheid werk in deze tussenbuffer, niet automatisch al het geheugen dat een compleet programma gebruikt.

import asyncio

async def main():
    queue = asyncio.Queue(maxsize=1)
    await queue.put("eerste")
    begun = asyncio.Event()
    finished = asyncio.Event()
    async def producer():
        begun.set()
        await queue.put("tweede")
        finished.set()
    task = asyncio.create_task(producer())
    await begun.wait()
    print("producent klaar:", finished.is_set())
    print(await queue.get())
    queue.task_done()
    await task
    print(await queue.get())
    queue.task_done()
    await queue.join()

asyncio.run(main())

producent klaar: False
eerste
tweede

Het voorbeeld begint met een volle buffer. Een tweede producentaanroep is werkelijk begonnen, maar wordt pas afgerond nadat het eerste item is opgehaald. Daarvoor is geen vaste slaaptijd of snelheidsschatting nodig.

Ophalen is nog niet behandelen

Na get is het item uit de buffer verdwenen, maar de behandeling kan nog lopen. De queue houdt daarom ook bij hoeveel ingevoegde items nog niet zijn afgemeld. Met queue.task_done() meldt de consumer één opgehaald item af.

Doe dat na de behandeling, ook als die behandeling faalt. Zet de get vóór de try/finally rond de verwerking: als wachten op een lege queue wordt geannuleerd, is er geen item om af te melden. Dat finally-blok voorkomt dat een fout tijdens de behandeling het afmelden overslaat. Roep task_done niet direct vóór een nog lopende verwerking aan: anders lijkt het werk al afgehandeld.

Met await queue.join() wacht je tot alle ingevoegde items zijn afgemeld. De queue kan dus leeg zijn terwijl join nog wacht. Join controleert de afmeldingen; zij bewijst niet dat iedere bedrijfsbewerking succesvol was.

Een consumer laten stoppen

Een queue weet niet vanzelf dat er geen nieuwe opdrachten meer komen. In deze les gebruiken we None als stopbericht. Dat is een afspraak tussen onze producent en consumer, geen ingebouwde speciale queuewaarde.

import asyncio

async def consumer(queue):
    while True:
        value = await queue.get()
        try:
            if value is None:
                return
            print("werk:", value)
        finally:
            queue.task_done()

async def main():
    queue = asyncio.Queue()
    worker = asyncio.create_task(consumer(queue))
    await queue.put(0)
    await queue.put(None)
    await queue.join()
    await worker
    print("afgerond")

asyncio.run(main())

werk: 0
afgerond

Het stopbericht is ook met put ingevoegd. Een consumer die het ophaalt moet het dus eveneens eenmaal afmelden. Daarna kan zij terugkeren in plaats van opnieuw op get te blijven wachten.

Nul, False en een lege tekst zijn hier gewone gegevens. Controleer daarom item is None om te stoppen. Een algemene waarheidstest zoals if not item zou ook geldig werk kunnen weggooien.

Veelgemaakte fout: join wacht voor altijd

Je ziet dat de buffer leeg is en verwacht dat join terugkomt. Als een opgehaald item geen task_done heeft gekregen, blijft er echter nog werk geregistreerd staan. Controleer dan de afmeldingen, vooral het foutpad en het stopbericht.

Te vaak afmelden is evenmin goed: task_done kan dan een ValueError geven. Het vaste patroon is één get, daarna de behandeling of het stopbesluit, en precies één afmelding voor dat opgehaalde item. De producent meldt haar puts niet zelf af.

Samenwerken zonder werk achter te laten

De oefening heeft drie functies. De producent zet alle meegegeven waarden in de queue zonder de invoerlijst te wijzigen. Een consumerfunctie behandelt één item; een andere blijft behandelen tot zij None ontvangt.

De meegegeven process-functie is async. Wacht op haar echte resultaat of fout voordat je het item afmeldt. Een gewone fout moet als hetzelfde exceptionobject doorgaan; een queue-afmelding maakt zo'n fout niet ineens opgelost.

Een gestopte consumer en een afgeronde join zijn verschillende afspraken. In een programma met meerdere consumers heb je ook voor iedere consumer een stopbesluit nodig. De volgende les brengt producent, meerdere workers en hun levensduur samen in één begrensde pipeline.

Samenvatting

• Put en get werken samen via een FIFO-buffer; een positieve maxsize geeft tegendruk.
• Meld ieder opgehaald item na behandeling precies eenmaal af, ook bij fouten en stoppen.
• Join wacht op afmeldingen; een stopbericht beëindigt volgens onze afspraak een consumer.