Een begrensde pipeline samenstellen

Je wilt een reeks bronnen verwerken zonder voor iedere bron meteen een nieuwe taak te maken. Een grote invoer mag niet automatisch een even grote verzameling actieve taken of wachtende werkitems veroorzaken. De technieken uit deze module vormen samen een kleine pipeline met duidelijke grenzen.

Drie rollen en één eigenaar

De producent haalt items geleidelijk uit de invoer en zet ze in een queue. Een vast aantal workers haalt daar werk uit en wacht op de meegegeven process-functie. De TaskGroup beheert de levensduur van producent en workers samen.

Het patroon is: producent → begrensde queue → vaste workers. Het aantal workers begrenst hoeveel process-aanroepen tegelijk actief zijn. De capaciteit van de queue begrenst daarnaast de tussenvoorraad; dat zijn twee verschillende instellingen.

import asyncio

async def producer(queue):
    for name in ("A", "B"):
        await queue.put(name)
    await queue.put(None)

async def worker(queue):
    while True:
        name = await queue.get()
        try:
            if name is None:
                return
            await asyncio.sleep(0)
            print("klaar:", name)
        finally:
            queue.task_done()

async def main():
    queue = asyncio.Queue(maxsize=1)
    async with asyncio.TaskGroup() as group:
        group.create_task(producer(queue))
        group.create_task(worker(queue))
    print("werkplaats gesloten")

asyncio.run(main())

klaar: A
klaar: B
werkplaats gesloten

Het voorbeeld gebruikt één worker en twee vaste opdrachten. Let op de positieve maxsize en het stopbericht. Een grotere echte pipeline hoeft niet voor iedere nieuwe opdracht een extra worker aan te maken.

De bron geleidelijk gebruiken

De oefenfunctie ontvangt een volgordelijke invoer met len en iteratie, zoals een lijst of een eigen object met __len__ en __iter__. Indexeren met items[i] hoeft de bron niet te ondersteunen. Je hoeft die bron niet eerst naar een nieuwe lijst te kopiëren. Zo'n volledige kopie zou immers alle items al ophalen voordat de begrensde verwerking begint.

De producent haalt normaal één item op en probeert het daarna in de queue te zetten. Als de buffer vol is, wacht zij met dat ene item totdat er ruimte komt. Zij loopt daardoor maar beperkt vooruit op de workers.

Deze afspraak maakt niet het hele programma constant in geheugen. De uiteindelijke resultatenlijst bevat nog steeds een plek voor ieder resultaat. Je begrenst hier het aantal uitvoeringen en de tussenvoorraad, niet de omvang van het gevraagde eindantwoord.

Een index bewaart de betekenis

Geef ieder werkitem zijn oorspronkelijke index mee: (index, value). Een worker schrijft het resultaat op die index in de resultatenlijst. Zo blijft de uitvoer in invoervolgorde staan wanneer een latere bron eerder klaar is.

results = [None] * 3
finished = [(2, "C"), (0, "A"), (1, "B")]
for index, value in finished:
    results[index] = value
print(results)

['A', 'B', 'C']

Het voorbeeld is bewust een gewone lijstbewerking: het laat alleen de indexafspraak zien. De concurrency verandert niet waar een resultaat thuishoort. Een nul, None of lege tekst uit de process-functie blijft daarbij een geldige uitkomst.

Ook stoppen is onderdeel van het werk

Na alle gewone items zet de producent voor iedere worker een los None-stopbericht in de queue. Een gewone invoerwaarde None is geen stopbericht: zij zit samen met haar index in een tuple. De worker herkent alleen het losse stopbericht met job is None.

Meld ieder opgehaald queue-item met task_done af, ook bij een stopbericht of procesfout. Een finally-blok houdt dat patroon op één plaats. Wacht op de producent voordat je join als bevestiging van alle afmeldingen gebruikt; een aanvankelijk lege queue kan anders te vroeg klaar lijken.

Veelgemaakte fout: begrenzen na het plannen

Voor alle items alvast een Task maken en alleen de process-aanroep achter een semaphore zetten, kan de actieve verwerking beperken. De verzameling geplande taken groeit dan echter nog steeds met de hele invoer. Hier gebruik je daarom echte vaste workers met een begrensde tussenbuffer.

Bij een gewone procesfout laat je de TaskGroup haar andere werk annuleren en afhandelen. Geef de foutgroep door in plaats van een gedeeltelijke lijst als geslaagd antwoord terug te geven. Een extern stopverzoek blijft na de eigen taakafhandeling een annulering.

Configuratie vóór uitvoering

Workers en capacity moeten positieve gehele getallen zijn; booleans zijn hier geen limieten. Een queuecapaciteit nul betekent in asyncio juist onbegrensd. Meld ongeldige configuratie daarom met ValueError voordat je invoer ophaalt of taken start.

Samenvatting

• Vaste workers begrenzen de verwerking; een positieve queuecapaciteit begrenst de tussenvoorraad.
• Geleidelijk ophalen voorkomt een onbeperkte voorsprong; indices bewaren de resultaatvolgorde.
• Stopberichten, afmeldingen en de TaskGroup sluiten de levenscyclus van het eigen werk.

Hierna onderzoek je hoe idiomatische Pythonpatronen en gerichte metingen een werkende oplossing verder kunnen verbeteren.