code wiki / (root) / nx_pipeline.nx

nx_pipeline.nx

buildroot/runtime/nx_pipeline.nx

9251 B245 linesdepth 7pulls 9 transitivereach 1 importersview sourcekind tooltopic pipeline
docsdependenciesstructsconstsfunctions

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

nx_syscalls.nx nx_atom.nx nx_thread.nx nx_chan.nx nx_thread_pool.nx nx_hw.nx nx_pipeline.nx nx_pipeline_test.nx

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

main nx_pool_new nx_hw_worker_count nx_hw_cpu_count sys_mmap sys_munmap sys_mmap ↻ nx_chan_new sys_mmap ↻ _nx_chan_cell sys_thread_create nx_thread_spawn sys_mmap ↻ nx_thread_spawn_fn sys_mmap ↻ nx_pipeline_new nx_hw_worker_count ↻ sys_mmap ↻ nx_pipeline_add_stage _nx_pipe_stage_at nx_chan_new ↻ nx_pipeline_start _nx_pipe_stage_at ↻ nx_pool_submit nx_atom_faa_i64 nx_chan_send nx_chan_try_send _nx_chan_cell ↻ nx_thread_yield _pool_futex_wake_all sys_futex_wake nx_pipeline_push nx_chan_send ↻ nx_pipeline_recv nx_chan_recv sys_mmap ↻ nx_chan_try_recv _nx_chan_cell ↻ nx_thread_yield ↻ nx_pipeline_finish

structs

46struct NxPipelineStage
55struct NxPipeline

consts

40const NX_PIPE_SENTINEL: i64 = -9223372036854775807 // close enough; user data
53const NX_PIPE_STAGE_BYTES: i64 = 32

functions

68func _nx_pipe_stage_at(pl: *NxPipeline, idx: i64) -> *NxPipelineStage
74func _nx_pipe_worker(ctx: i64) -> i64
100func nx_pipeline_new(pool: *NxThreadPool, max_stages: i64,
called by 2: mainmain calls 2: nx_hw_worker_countsys_mmap
122func nx_pipeline_add_stage(pl: *NxPipeline, fn: func(i64) -> i64) -> i64
150func nx_pipeline_start(pl: *NxPipeline) -> i64
169func nx_pipeline_push(pl: *NxPipeline, v: i64) -> i64
177func nx_pipeline_recv(pl: *NxPipeline) -> i64
called by 2: mainmain calls 1: nx_chan_recv
185func nx_pipeline_try_recv(pl: *NxPipeline, out: *i64) -> i64
194func nx_pipeline_finish(pl: *NxPipeline) -> i64
called by 2: mainmain calls 1: nx_pipeline_push
203func nx_pipeline_n_stages(pl: *NxPipeline) -> i64 { return pl.n_stages }
204func nx_pipeline_n_workers_per_stage(pl: *NxPipeline) -> i64 { return pl.n_workers_per_stage }
208func _pipe_self_test_inc(x: i64) -> i64 { return x + 1 }
209func _pipe_self_test_double(x: i64) -> i64 { return x * 2 }
211func main() -> i64