Limite & budget de temps

Arrêter un flux infini proprement

.limit(n) laisse passer les n premiers items puis arrête le pipeline, ce qui permet de consommer une source infinie (générateur, itertools.count()). .timeout(s) arrête l'étape une fois s secondes écoulées depuis son démarrage : c'est un budget de temps global, pas un délai d'inactivité entre deux items.

Signature

Pipeline.limit(packet_limit: int) -> Pipeline[InPacket]
Pipeline.timeout(timeout: float) -> Pipeline[InPacket]

python

import itertools
import time
from olympipe import Pipeline

# Une source infinie, coupée après 5 items
print(Pipeline(itertools.count()).task(lambda x: x * x).limit(5).wait_for_result())

def ticker():
    i = 0
    while True:
        yield i
        i += 1
        time.sleep(0.3)

# Un budget de 2 secondes sur un flux sans fin
start = time.time()
res = Pipeline(ticker()).timeout(2.0).wait_for_result()
print(res, f"{time.time() - start:.1f}s")

Sortie

[0, 1, 4, 9, 16]
[0, 1, 2, 3, 4, 5, 6, 7] 2.4s

Comment ça fonctionne

Quand la limite ou le budget est atteint, l'étape signale l'arrêt et la source cesse d'être lue. Les étapes en amont tournent en parallèle : la source a pu produire quelques items de plus que n, qui sont simplement ignorés. Si la source est bloquée (par exemple dans un sleep), l'arrêt complet attend qu'elle rende la main.

Précédent Réduction & checkpoint Suivant Récupérer les résultats