Démarrer l'alimentation
Lancez la source avant la fin du graphe
.start() lance tout de suite le thread qui pousse la source dans sa file, et renvoie le pipeline sur lequel on l'a appelé. L'appel est idempotent : un second .start() ne relance pas le thread. On peut l'appeler depuis n'importe quelle étape du graphe : l'alimentation démarre toujours à la racine, et à chaque racine fusionnée par .gather().
Signature
Pipeline.start() -> Pipeline[InPacket]from olympipe import Pipeline
source = Pipeline(range(4))
out = source.task(lambda x: x * 2)
# Rien n'a encore été poussé. start() renvoie ce pipeline ; le second appel est sans effet.
started = out.start()
started.start()
print(sorted(started.wait_for_result()))
left = Pipeline([1, 2])
right = Pipeline([3, 4])
merged = left.gather(right)
merged.start() # démarre chaque racine fusionnée par gather
print(sorted(merged.wait_for_result()))[0, 2, 4, 6]
[1, 2, 3, 4]Comment ça fonctionne
.start() rend la main tout de suite. wait_for_result, wait_for_results, wait_for_completion et Pipeline.server démarrent l'alimentation eux-mêmes. .start() sert quand la source doit tourner plus tôt — graphe construit au fil de l'eau, source d'abord, consommateurs branchés ensuite. Le thread d'alimentation bloque quand la file de sortie est pleine (4 items) tant qu'un consommateur ne lit pas.
Brancher les consommateurs après la source
Sur une source déjà construite, .start() fait tourner l'alimentation pendant que la suite du graphe n'existe pas encore. Avec plus d'items que la file (4), le thread bloque sur le put suivant au lieu de fermer la file, et l'étape ajoutée ensuite consomme tout le flux.
from olympipe import Pipeline
def sensor():
for i in range(8):
yield i
source = Pipeline(sensor())
source.start() # le thread tourne ; il bloquera file pleine
decoded = source.task(lambda x: x * 2)
print(sorted(decoded.wait_for_result()))[0, 2, 4, 6, 8, 10, 12, 14]