nx_pipeline.nx
buildroot/runtime/nx_pipeline.nx
about
nx_pipeline.nx -- N-stage MIMD pipeline with bounded backpressure.
Each stage runs the same number of workers; each worker pulls
from the stage's input MPMC channel, applies the stage's
transform fn, pushes to the stage's output channel. Bounded
channels between stages naturally provide backpressure: when
the downstream queue fills, upstream's nx_chan_send blocks
(spin-on-full today; futex-blocking when nx_chan grows that).
Shutdown: caller calls nx_pipeline_finish which pushes
n_workers_per_stage sentinels (NX_PIPE_SENTINEL = i64::MIN) into
stage-0's input channel; each worker that pulls a sentinel
forwards one sentinel to the next stage and exits. Stage K
receives exactly n_workers sentinels, so each of its workers
gets one and exits. Requires uniform worker count across
stages -- elastic per-stage sizing is a follow-up evolution.
Composes against: [[nx_thread_pool_shared_queue]] (worker
execution), [[vyukov_mpmc_channel]] (per-stage queue),
[[nx_hw_dynamic_probes]] (worker sizing default),
[[fn_ptr_indirect_call]] (typed transform dispatch).
dependencies 6 imports · 1 importers
imports: nx_syscalls.nxnx_atom.nxnx_thread.nxnx_chan.nxnx_thread_pool.nxnx_hw.nx
imported by: nx_pipeline_test.nx
call flow from main pre-order; caps 40 nodes / depth 6 declared; ↻ = already shown
structs
| 46 | struct NxPipelineStage |
| 55 | struct NxPipeline |
consts
| 40 | const NX_PIPE_SENTINEL: i64 = -9223372036854775807 // close enough; user data |
| 53 | const NX_PIPE_STAGE_BYTES: i64 = 32 |
functions
| 68 | func _nx_pipe_stage_at(pl: *NxPipeline, idx: i64) -> *NxPipelineStage |
| 74 | func _nx_pipe_worker(ctx: i64) -> i64 |
| 100 | func nx_pipeline_new(pool: *NxThreadPool, max_stages: i64, |
| 122 | func nx_pipeline_add_stage(pl: *NxPipeline, fn: func(i64) -> i64) -> i64 |
| 150 | func nx_pipeline_start(pl: *NxPipeline) -> i64 |
| 169 | func nx_pipeline_push(pl: *NxPipeline, v: i64) -> i64 |
| 177 | func nx_pipeline_recv(pl: *NxPipeline) -> i64 |
| 185 | func nx_pipeline_try_recv(pl: *NxPipeline, out: *i64) -> i64 calls 1: nx_chan_try_recv |
| 194 | func nx_pipeline_finish(pl: *NxPipeline) -> i64 |
| 203 | func nx_pipeline_n_stages(pl: *NxPipeline) -> i64 { return pl.n_stages } |
| 204 | func nx_pipeline_n_workers_per_stage(pl: *NxPipeline) -> i64 { return pl.n_workers_per_stage } |
| 208 | func _pipe_self_test_inc(x: i64) -> i64 { return x + 1 } |
| 209 | func _pipe_self_test_double(x: i64) -> i64 { return x * 2 } |
| 211 | func main() -> i64 |