Pipelines and concurrency

Connect a pipeline

Join commands with |. The result is a PipelineResult with one CommandResult per stage. Only the final stage's stdout is captured; interior stdout feeds the next stage and appears as None on that stage's result.

import sys

from cuprum import Program, ProgramCatalogue, sh

catalogue = ProgramCatalogue.from_programs(sys.executable, name="pipeline")
python = sh.make(Program(sys.executable), catalogue=catalogue)
producer = python("-c", "print('hello')")
consumer = python("-c", "import sys; print(sys.stdin.read().upper(), end='')")
result = (producer | consumer).run_sync()
assert result.ok
assert result.stdout == "HELLO\n"
assert [stage.exit_code for stage in result.stages] == [0, 0]

Find the stage that failed

The final stage can succeed after an earlier one has failed, so check ok or failure_index, never only the last exit code. When a stage fails while others are still running, Cuprum terminates those peers. The first observed failure is the one recorded; when two stages finish at almost the same moment, stage order breaks the tie. The pipeline job sheet shows a three-stage run failing in the middle.

Run commands side by side

run_concurrent is async; run_concurrent_sync serves synchronous callers. Bound the number of simultaneous children with ConcurrentConfig.concurrency. Results come back in submission order.

import sys

from cuprum import Program, ConcurrentConfig, ProgramCatalogue, run_concurrent_sync, sh

catalogue = ProgramCatalogue.from_programs(sys.executable, name="concurrent")
python = sh.make(Program(sys.executable), catalogue=catalogue)
commands = [python("-c", f"print({number})") for number in (1, 2)]
result = run_concurrent_sync(*commands, config=ConcurrentConfig(concurrency=2))
assert result.ok
assert [item.stdout for item in result.results] == ["1\n", "2\n"]
assert result.submission_indices == (0, 1)

With fail_fast=True, the remaining commands are cancelled after the first non-zero exit, so some may be missing from the results and a command that had already finished can still appear. failures indexes the returned tuple; submission_indices and failure_submission_indices map each result back to the command that produced it. Use those rather than assuming a result's position is its submission index.