Starting the feed
Start the source before the graph is finished
.start() immediately starts the thread that pushes the source into its queue, and returns the pipeline it was called on. The call is idempotent: a second .start() does not start the thread again. It can be called from any stage of the graph: feeding always starts at the root, and at every root merged in by .gather().
Signature
Pipeline.start() -> Pipeline[InPacket]from olympipe import Pipeline
source = Pipeline(range(4))
out = source.task(lambda x: x * 2)
# Nothing has been fed yet. start() returns this pipeline; the second call is a no-op.
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() # starts every root merged in by gather
print(sorted(merged.wait_for_result()))[0, 2, 4, 6]
[1, 2, 3, 4]How it works
.start() returns immediately. wait_for_result, wait_for_results, wait_for_completion and Pipeline.server start feeding on their own. .start() is for when the source has to run earlier — a graph built as you go, source first, consumers attached afterwards. The feeder thread blocks once the output queue is full (4 items) until a consumer reads it.
Attach consumers after the source
On a source that already exists, .start() keeps feeding while the rest of the graph does not exist yet. With more items than the queue holds (4), the thread blocks on the next put instead of closing the queue, and the step added afterwards consumes the whole stream.
from olympipe import Pipeline
def sensor():
for i in range(8):
yield i
source = Pipeline(sensor())
source.start() # the thread runs; it blocks once the queue is full
decoded = source.task(lambda x: x * 2)
print(sorted(decoded.wait_for_result()))[0, 2, 4, 6, 8, 10, 12, 14]