nx_pipeline.nx source
↩ module page · 245 lines · 9251 B
1// nx_pipeline.nx -- N-stage MIMD pipeline with bounded backpressure.
2//
3// Each stage runs the same number of workers; each worker pulls
4// from the stage's input MPMC channel, applies the stage's
5// transform fn, pushes to the stage's output channel. Bounded
6// channels between stages naturally provide backpressure: when
7// the downstream queue fills, upstream's nx_chan_send blocks
8// (spin-on-full today; futex-blocking when nx_chan grows that).
9//
10// Shutdown: caller calls nx_pipeline_finish which pushes
11// n_workers_per_stage sentinels (NX_PIPE_SENTINEL = i64::MIN) into
12// stage-0's input channel; each worker that pulls a sentinel
13// forwards one sentinel to the next stage and exits. Stage K
14// receives exactly n_workers sentinels, so each of its workers
15// gets one and exits. Requires uniform worker count across
16// stages -- elastic per-stage sizing is a follow-up evolution.
17//
18// Composes against: [[nx_thread_pool_shared_queue]] (worker
19// execution), [[vyukov_mpmc_channel]] (per-stage queue),
20// [[nx_hw_dynamic_probes]] (worker sizing default),
21// [[fn_ptr_indirect_call]] (typed transform dispatch).
22
23// nx_safety_envelope:
24// intended_use: AUTO_APPLIED -- primitive-specific tuning queued
25// sil_target: SIL1
26// evidence: [bulk_applied_2026-05-16, see-file-comment-for-detail]
27// verdict: NOT_YET_EVALUATED
28
29import "nx_syscalls.nx"
30import "nx_atom.nx"
31import "nx_thread.nx"
32import "nx_chan.nx"
33import "nx_thread_pool.nx"
34import "nx_hw.nx"
35
36// Sentinel value: i64::MIN. -9223372036854775808 in source is awkward;
37// build it as -(2^62) - (2^62) ... actually -1 shifted left 63 = i64::MIN.
38// NishiLang doesn't have <<-on-literal evaluator yet -- compute via two
39// constants: SHL of 1 by 63. Pre-compute as a static.
40const NX_PIPE_SENTINEL: i64 = -9223372036854775807 // close enough; user data
41 // must not equal this.
42
43// Each stage carries its in/out chan pointers + transform fn.
44// The pipeline owns a flat array of these. Workers receive the
45// stage struct's address as their ctx.
46struct NxPipelineStage {
47 in_chan_addr: i64, // *NxChan (input to this stage)
48 out_chan_addr: i64, // *NxChan (output of this stage)
49 transform_fn: func(i64) -> i64,
50 n_workers: i64,
51}
52
53const NX_PIPE_STAGE_BYTES: i64 = 32
54
55struct NxPipeline {
56 pool_addr: i64, // *NxThreadPool
57 stages_base: i64, // *NxPipelineStage[max_stages]
58 n_stages: i64,
59 max_stages: i64,
60 n_workers_per_stage: i64,
61 queue_cap: i64,
62 input_chan_addr: i64, // *NxChan (= stage[0].in_chan)
63 output_chan_addr: i64, // *NxChan (= stage[N-1].out_chan)
64 started: i64,
65}
66
67// Compute address of stage[idx].
68func _nx_pipe_stage_at(pl: *NxPipeline, idx: i64) -> *NxPipelineStage {
69 return (pl.stages_base + idx * NX_PIPE_STAGE_BYTES) as *NxPipelineStage
70}
71
72// Worker body: ctx is the stage address. Loop recv -> transform -> send;
73// exit on sentinel (which it also forwards to the next stage).
74func _nx_pipe_worker(ctx: i64) -> i64 {
75 let stage: *NxPipelineStage = ctx as *NxPipelineStage
76 let in_c: *NxChan = stage.in_chan_addr as *NxChan
77 let out_c: *NxChan = stage.out_chan_addr as *NxChan
78 let fp: func(i64) -> i64 = stage.transform_fn
79
80 var running: i64 = 1
81 while running == 1 {
82 let v: i64 = nx_chan_recv(in_c)
83 if v == NX_PIPE_SENTINEL {
84 nx_chan_send(out_c, NX_PIPE_SENTINEL)
85 running = 0
86 } else {
87 let r: i64 = fp(v)
88 nx_chan_send(out_c, r)
89 }
90 }
91 return 0
92}
93
94// ---- public API -------------------------------------------------
95
96// Create a new pipeline. Allocates space for `max_stages`; the
97// actual count grows via nx_pipeline_add_stage. queue_cap is the
98// per-stage MPMC channel depth (controls backpressure tightness).
99// n_workers_per_stage defaults to nx_hw_worker_count() if <= 0.
100func nx_pipeline_new(pool: *NxThreadPool, max_stages: i64,
101 n_workers_per_stage: i64, queue_cap: i64) -> *NxPipeline {
102 if n_workers_per_stage < 1 { n_workers_per_stage = nx_hw_worker_count() }
103 if queue_cap < 1 { queue_cap = 16 }
104
105 let raw: *u8 = sys_mmap(128)
106 let pl: *NxPipeline = raw as *NxPipeline
107 let stages_raw: *u8 = sys_mmap(max_stages * NX_PIPE_STAGE_BYTES)
108 pl.pool_addr = pool as i64
109 pl.stages_base = stages_raw as i64
110 pl.n_stages = 0
111 pl.max_stages = max_stages
112 pl.n_workers_per_stage = n_workers_per_stage
113 pl.queue_cap = queue_cap
114 pl.started = 0
115 return pl
116}
117
118// Append a stage with transform fn. Stage 0 creates the input
119// channel; every stage also creates its output channel (which the
120// next stage uses as its input). Returns stage index or -1 on
121// capacity exhaustion.
122func nx_pipeline_add_stage(pl: *NxPipeline, fn: func(i64) -> i64) -> i64 {
123 if pl.n_stages >= pl.max_stages { return -1 }
124 let idx: i64 = pl.n_stages
125 let stage: *NxPipelineStage = _nx_pipe_stage_at(pl, idx)
126
127 // Input chan: stage 0 creates a fresh one (the pipeline input);
128 // stage K>=1 reuses stage K-1's output.
129 if idx == 0 {
130 let in_c: *NxChan = nx_chan_new(pl.queue_cap)
131 stage.in_chan_addr = in_c as i64
132 pl.input_chan_addr = in_c as i64
133 } else {
134 let prev: *NxPipelineStage = _nx_pipe_stage_at(pl, idx - 1)
135 stage.in_chan_addr = prev.out_chan_addr
136 }
137
138 // Output chan: always fresh. Last stage's becomes the pipeline output.
139 let out_c: *NxChan = nx_chan_new(pl.queue_cap)
140 stage.out_chan_addr = out_c as i64
141 stage.transform_fn = fn
142 stage.n_workers = pl.n_workers_per_stage
143 pl.output_chan_addr = out_c as i64
144 pl.n_stages = idx + 1
145 return idx
146}
147
148// Spawn workers for every stage. After this call the pipeline is
149// live: feed it with nx_pipeline_push and drain with nx_pipeline_recv.
150func nx_pipeline_start(pl: *NxPipeline) -> i64 {
151 if pl.started == 1 { return 0 }
152 let pool: *NxThreadPool = pl.pool_addr as *NxThreadPool
153 var s: i64 = 0
154 while s < pl.n_stages {
155 let stage: *NxPipelineStage = _nx_pipe_stage_at(pl, s)
156 var w: i64 = 0
157 while w < pl.n_workers_per_stage {
158 nx_pool_submit(pool, _nx_pipe_worker, stage as i64)
159 w = w + 1
160 }
161 s = s + 1
162 }
163 pl.started = 1
164 return 0
165}
166
167// Push a value into the pipeline's input. Blocks if the input
168// channel is full (backpressure).
169func nx_pipeline_push(pl: *NxPipeline, v: i64) -> i64 {
170 let in_c: *NxChan = pl.input_chan_addr as *NxChan
171 return nx_chan_send(in_c, v)
172}
173
174// Receive a value from the pipeline's output. Blocks if empty.
175// Returns NX_PIPE_SENTINEL when the pipeline has been fully drained
176// after finish().
177func nx_pipeline_recv(pl: *NxPipeline) -> i64 {
178 let out_c: *NxChan = pl.output_chan_addr as *NxChan
179 return nx_chan_recv(out_c)
180}
181
182// Non-blocking recv variant -- returns 0 + writes *out on success,
183// returns -1 if empty. Useful for drain loops that also need to
184// watch a deadline.
185func nx_pipeline_try_recv(pl: *NxPipeline, out: *i64) -> i64 {
186 let out_c: *NxChan = pl.output_chan_addr as *NxChan
187 if nx_chan_try_recv(out_c, out) == 1 { return 0 }
188 return -1
189}
190
191// Signal the pipeline to drain and shut down. Pushes n_workers
192// sentinels into the input channel; each propagates through every
193// stage causing exactly one worker per stage to exit on receipt.
194func nx_pipeline_finish(pl: *NxPipeline) -> i64 {
195 var k: i64 = 0
196 while k < pl.n_workers_per_stage {
197 nx_pipeline_push(pl, NX_PIPE_SENTINEL)
198 k = k + 1
199 }
200 return 0
201}
202
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 }
205
206// ---- self-test --------------------------------------------------
207
208func _pipe_self_test_inc(x: i64) -> i64 { return x + 1 }
209func _pipe_self_test_double(x: i64) -> i64 { return x * 2 }
210
211func main() -> i64 {
212 let pool: *NxThreadPool = nx_pool_new(4, 32)
213 let pl: *NxPipeline = nx_pipeline_new(pool, 4, 1, 8)
214 nx_pipeline_add_stage(pl, _pipe_self_test_inc)
215 nx_pipeline_add_stage(pl, _pipe_self_test_double)
216 if pl.n_stages != 2 { return __syscall(93, 1, 0, 0, 0, 0, 0) }
217 nx_pipeline_start(pl)
218
219 // Push 5 values; each goes through inc + double.
220 var i: i64 = 0
221 while i < 5 { nx_pipeline_push(pl, i); i = i + 1 }
222
223 // Recv 5 values; verify (i+1)*2 = (0,2,4,6,8) -> (2,4,6,8,10).
224 var seen: i64 = 0
225 var j: i64 = 0
226 while j < 5 {
227 let v: i64 = nx_pipeline_recv(pl)
228 let expected: i64 = (j + 1) * 2
229 if v != expected { return __syscall(93, 10 + j, 0, 0, 0, 0, 0) }
230 seen = seen + 1
231 j = j + 1
232 }
233 if seen != 5 { return __syscall(93, 50, 0, 0, 0, 0, 0) }
234
235 nx_pipeline_finish(pl)
236 // Drain residual sentinels.
237 var k: i64 = 0
238 while k < pl.n_workers_per_stage {
239 nx_pipeline_recv(pl) // each sentinel pushed will eventually emerge
240 k = k + 1
241 }
242
243 nx_pool_shutdown(pool)
244 return 0
245}