code wiki / _hdl_build / nx_clockjobs.nx
nx_clockjobs.nx source
↩ module page · 214 lines · 13052 B
1// nx_clock.nx -- THE ONE NISHI CLOCK (operator: "we should be s class exceed like how pulses work on clocks ...
2// so we avoid issues ... dont have a million pulses and daemons"). Researched (knowledge/fetched/sched_*.raw:
3// clock-distribution = ONE oscillator -> a tree of DIVIDERS; PLL derives every frequency from ONE reference;
4// cron = ONE daemon reads ONE table of timed jobs; tickless = don't burn a constant tick when idle). The S-class
5// pattern is identical: ONE tick source + ONE durable JOB REGISTRY + ONE dispatcher. Every periodic capability
6// REGISTERS a job (name, interval-in-ticks) = a divider off the single clock -- it does NOT spin up its own
7// daemon/pulse loop. So N capabilities cost ONE loop, not N. This library is the registry + dispatcher; the single
8// tick source drives it (one external spark, like a crystal). license_tier: ORIGINAL
9import "nx_syscalls.nx"
10import "nx_store_seed_lib.nx"
11
12const CLK_MAXJOBS: i64 = 128
13const CLK_NAMEW: i64 = 48 // bytes per job name slot
14const CLK_REG: *u8 = "knowledge/sched/jobs.tsv" // the ONE durable registry (the "crontab")
15
16func clk_slot(names: *u8, i: i64) -> *u8 { return ((names as i64) + i*CLK_NAMEW) as *u8 }
17func clk_streq(a: *u8, b: *u8) -> i64 { var i: i64=0; while a[i]!=(0 as u8){ if a[i]!=b[i]{return 0} i=i+1 } if b[i]!=(0 as u8){return 0} return 1 }
18
19func clk_find(names: *u8, n: i64, name: *u8) -> i64 { var i: i64=0; while i<n { if clk_streq(clk_slot(names,i), name)==1 { return i } i=i+1 } return 0-1 }
20
21// REGISTER a periodic job = add a divider off the one clock. Idempotent: re-registering the same name is a no-op
22// (so capabilities can declare their job every boot without ever creating a duplicate pulse). returns 1 if added.
23func clk_register(names: *u8, organs: *u8, intervals: *i64, next_due: *i64, np: *i64, name: *u8, organ: *u8, interval: i64) -> i64 {
24 let n: i64 = np[0]
25 if clk_find(names, n, name) >= 0 { return 0 }
26 if n >= CLK_MAXJOBS { return 0 }
27 var d: *u8 = clk_slot(names, n); var i: i64=0
28 while i<CLK_NAMEW-1 { if name[i]==(0 as u8){ d[i]=0 as u8; i=CLK_NAMEW } else { d[i]=name[i]; i=i+1 } }
29 if i==CLK_NAMEW-1 { d[i]=0 as u8 }
30 var e: *u8 = clk_slot(organs, n); var j: i64=0
31 while j<CLK_NAMEW-1 { if organ[j]==(0 as u8){ e[j]=0 as u8; j=CLK_NAMEW } else { e[j]=organ[j]; j=j+1 } }
32 if j==CLK_NAMEW-1 { e[j]=0 as u8 }
33 var iv: i64 = interval; if iv < 1 { iv = 1 }
34 intervals[n] = iv; next_due[n] = iv; np[0] = n + 1
35 return 1
36}
37
38// THE DISPATCHER: advance to logical tick `now`; mark + return how many jobs are DUE (next_due<=now), advancing
39// each due job's next_due by its interval (catch-up safe: a job never fires more than once for a missed window).
40func clk_tick(intervals: *i64, next_due: *i64, n: i64, now: i64, fired: *i64) -> i64 {
41 var count: i64 = 0; var i: i64 = 0
42 while i < n {
43 if next_due[i] <= now {
44 fired[i] = 1; count = count + 1
45 while next_due[i] <= now { next_due[i] = next_due[i] + intervals[i] } // re-arm past now (no runaway on a skipped tick)
46 } else { fired[i] = 0 }
47 i = i + 1
48 }
49 return count
50}
51
52// THE FUNCTIONAL DISPATCH: advance to `now` and actually RUN each due job by fork+exec of its organ elf path
53// (the same fork+exec idiom nx_god_pulse / nx_aw_hostctl use). One dispatcher runs N jobs at their divided rates;
54// there is NO per-job daemon. Parent waits each child so a slow job can't be lost (a real scheduler can make this
55// bounded-concurrent / fire-and-forget). `organs` is a parallel slot array (organ[i] = an executable path). Returns
56// the number of jobs dispatched this tick.
57func clk_dispatch_run(organs: *u8, intervals: *i64, next_due: *i64, n: i64, now: i64, fired: *i64) -> i64 {
58 var count: i64 = 0; var i: i64 = 0
59 while i < n {
60 if next_due[i] <= now {
61 fired[i] = 1
62 while next_due[i] <= now { next_due[i] = next_due[i] + intervals[i] } // re-arm past now (catch-up safe)
63 let path: *u8 = clk_slot(organs, i)
64 __syscall(90, path as i64, 0x1ed, 0, 0, 0, 0) // chmod 0755 first: recv-shipped organs aren't reliably +x (HC_MGMT_CMD chmods for the same reason) -> without this execve fails 127 = silent no-dispatch
65 let pid: i64 = sys_fork()
66 if pid == 0 {
67 let argv: *i64 = sys_mmap(32) as *i64; argv[0] = path as i64; argv[1] = 0
68 let envp: *i64 = sys_mmap(16) as *i64; envp[0] = "PATH=/usr/bin:/bin" as *u8 as i64; envp[1] = 0
69 sys_execve(path, argv, envp)
70 sys_exit(127)
71 }
72 let st: *i64 = sys_mmap(16) as *i64; st[0] = 0
73 sys_wait4(pid, st, 0)
74 let raw: i64 = st[0]; let sig: i64 = raw & 0x7f; let code: i64 = (raw >> 8) & 0xff
75 // HONEST count: a job whose organ failed to exec (child exit 127 = missing/broken organ) did NOT run,
76 // so it is not counted as dispatched (it gets fired[]=1 so a caller can flag it, mirroring the
77 // publisher's dead-letter). A crash (signalled) or any other exit means the organ DID run.
78 if sig != 0 { count = count + 1 } else { if code != 127 { count = count + 1 } }
79 } else { fired[i] = 0 }
80 i = i + 1
81 }
82 return count
83}
84
85// TICKLESS run (sched_tickless lesson): from start_tick, run up to maxbeats beats, but before each beat SLEEP
86// exactly until the MINIMUM next_due across all jobs -- skipping every idle tick -- then dispatch the due organs.
87// Advances next_due[] in place (the caller persists for resume). out[0..3] = beats, dispatches, skipped_idle,
88// final_tick. ONE implementation, shared by the driver (nx_clock_tickless) and its gate (no parallel copy).
89func clk_run_tickless(organs: *u8, intervals: *i64, next_due: *i64, n: i64, start_tick: i64, maxbeats: i64, tick_ms: i64, out: *i64) -> i64 {
90 let fired: *i64 = sys_mmap(CLK_MAXJOBS*8) as *i64
91 var T: i64 = start_tick; var beats: i64 = 0; var disp: i64 = 0; var skipped: i64 = 0
92 while beats < maxbeats {
93 var minnd: i64 = next_due[0]; var i: i64 = 1
94 while i < n { if next_due[i] < minnd { minnd = next_due[i] } i = i + 1 }
95 if minnd <= T { minnd = T + 1 }
96 let skip: i64 = minnd - T
97 if skip > 1 { skipped = skipped + (skip - 1) }
98 let totms: i64 = skip * tick_ms
99 let ts: *i64 = sys_mmap(16) as *i64; ts[0] = totms / 1000; ts[1] = (totms - (totms/1000)*1000) * 1000000
100 __syscall(35, ts as i64, 0, 0, 0, 0, 0) // nanosleep until the next due event (tickless)
101 T = minnd
102 disp = disp + clk_dispatch_run(organs, intervals, next_due, n, T, fired)
103 beats = beats + 1
104 }
105 out[0] = beats; out[1] = disp; out[2] = skipped; out[3] = T
106 return disp
107}
108
109// STATUS helpers (the consolidation payoff: ONE place shows every periodic job). clk_due_in = ticks until a job
110// fires (<=0 means due now). clk_twin_ok = 1 iff the job's organ elf actually exists (a 0 means it is registered
111// but un-blessed -> the dispatcher would honestly report it didn't run -> surface it in status, do not hide it).
112func clk_due_in(next_due_i: i64, now: i64) -> i64 { return next_due_i - now }
113func clk_twin_ok(organ: *u8) -> i64 { let fd: i64 = sys_openat_rd(organ); if fd < 0 { return 0 } sys_close(fd); return 1 }
114
115func clk_itoa(buf: *u8, o: i64, v: i64) -> i64 { var w: i64=o; var m: i64=v; if m==0{buf[w]=48 as u8;return w+1} if m<0{buf[w]=45 as u8;w=w+1;m=0-m} let t:*u8=sys_mmap(24); var k:i64=0; while m>0{t[k]=(48+(m%10)) as u8;m=m/10;k=k+1} var j:i64=0; while j<k{buf[w]=t[k-1-j];w=w+1;j=j+1} return w }
116
117// persist the registry (the ONE crontab): name<TAB>interval<TAB>next_due per line. atomic via tmp+rename.
118func clk_save(path: *u8, names: *u8, organs: *u8, intervals: *i64, next_due: *i64, n: i64) -> i64 {
119 let buf: *u8 = sys_mmap(CLK_MAXJOBS*160); var o: i64 = 0
120 var i: i64 = 0
121 while i < n {
122 let nm: *u8 = clk_slot(names, i); var k: i64=0; while nm[k]!=(0 as u8){ buf[o]=nm[k]; o=o+1; k=k+1 }
123 buf[o]=9 as u8; o=o+1; o = clk_itoa(buf, o, intervals[i]); buf[o]=9 as u8; o=o+1; o = clk_itoa(buf, o, next_due[i]); buf[o]=9 as u8; o=o+1
124 let og: *u8 = clk_slot(organs, i); var g: i64=0; while og[g]!=(0 as u8){ buf[o]=og[g]; o=o+1; g=g+1 }
125 buf[o]=10 as u8; o=o+1
126 i = i + 1
127 }
128 sys_mkdir("knowledge" as *u8, 0x1ed); sys_mkdir("knowledge/sched" as *u8, 0x1ed)
129 let tmp: *u8 = sys_mmap(256); var t: i64=0; let pp: *u8 = path; while pp[t]!=(0 as u8){ tmp[t]=pp[t]; t=t+1 } tmp[t]=46 as u8; tmp[t+1]=116 as u8; tmp[t+2]=109 as u8; tmp[t+3]=112 as u8; tmp[t+4]=0 as u8 // path + ".tmp"
130 let fd: i64 = sys_openat_wr(tmp, 0x1a4); if fd<0 { return 0 } sys_write(fd, buf, o); sys_close(fd)
131 __syscall(82, tmp as i64, path as i64, 0, 0, 0, 0) // atomic rename .tmp -> registry
132 return n
133}
134
135// MERGE new job declarations from an nx_store PLANE (off-tsv, additive) into the in-memory registry.
136// THE CLOBBER FIX: the clock is READ-ONLY on this plane -- external adds go via
137// `nx_store_put <plane> put <actor> <name> <interval> <organ>` (upsert-by-name, additive) and can NEVER be
138// clobbered by the clock's own (now clock-private) tsv save. Plane row cols (TAB): 0=name 1=interval(sec) 2=organ.
139// A job already present (by name) is skipped (idempotent -- re-merge is a no-op). New jobs arm at base_tick+interval.
140// Returns the number of NEW jobs added this call.
141func clk_merge_store(prefix: *u8, names: *u8, organs: *u8, intervals: *i64, next_due: *i64, np: *i64, base_tick: i64) -> i64 {
142 let buf: *u8 = sys_mmap(CLK_MAXJOBS*160)
143 let dn: i64 = sts_load(prefix, buf, CLK_MAXJOBS*160)
144 if dn <= 0 { return 0 }
145 var added: i64 = 0
146 var i: i64=0; var ls: i64=0
147 while i < dn {
148 if buf[i] == (10 as u8) {
149 if i > ls {
150 let line: *u8 = ((buf as i64)+ls) as *u8; let ll: i64 = i-ls
151 let nm: *u8 = sys_mmap(CLK_NAMEW); let og: *u8 = sys_mmap(CLK_NAMEW)
152 var f: i64=0; var p: i64=0; var iv: i64=0; var w: i64=0; var g: i64=0
153 while p < ll {
154 if line[p]==(9 as u8) { f=f+1 }
155 else {
156 if f==0 { if w<CLK_NAMEW-1 { nm[w]=line[p]; w=w+1 } }
157 else { if f==1 { if line[p]>=(48 as u8) { if line[p]<=(57 as u8) { iv=iv*10+((line[p]-(48 as u8)) as i64) } } }
158 else { if f==2 { if g<CLK_NAMEW-1 { og[g]=line[p]; g=g+1 } } } }
159 }
160 p=p+1
161 }
162 nm[w]=0 as u8; og[g]=0 as u8
163 if w > 0 { if g > 0 {
164 let n: i64 = np[0]
165 if clk_find(names, n, nm) < 0 { if n < CLK_MAXJOBS {
166 var d: *u8=clk_slot(names,n); var k: i64=0; while k<CLK_NAMEW-1 { if nm[k]==(0 as u8){d[k]=0 as u8;k=CLK_NAMEW} else {d[k]=nm[k];k=k+1} } if k==CLK_NAMEW-1 {d[k]=0 as u8}
167 var e: *u8=clk_slot(organs,n); var j: i64=0; while j<CLK_NAMEW-1 { if og[j]==(0 as u8){e[j]=0 as u8;j=CLK_NAMEW} else {e[j]=og[j];j=j+1} } if j==CLK_NAMEW-1 {e[j]=0 as u8}
168 if iv<1 { iv=1 }
169 intervals[n]=iv; next_due[n]=base_tick+iv; np[0]=n+1; added=added+1
170 } }
171 } }
172 }
173 ls=i+1
174 }
175 i=i+1
176 }
177 return added
178}
179
180// load the registry into the arrays. returns #jobs (also written to np[0]).
181func clk_load(path: *u8, names: *u8, organs: *u8, intervals: *i64, next_due: *i64, np: *i64) -> i64 {
182 np[0] = 0
183 let lenp: *i64 = sys_mmap(8) as *i64
184 let data: *u8 = sys_read_file(path, lenp)
185 if (data as i64) == 0 { return 0 }
186 let dn: i64 = lenp[0]; var i: i64=0; var ls: i64=0; var n: i64=0
187 while i < dn {
188 if data[i] == (10 as u8) {
189 if i > ls { if n < CLK_MAXJOBS {
190 let line: *u8 = ((data as i64)+ls) as *u8; let ll: i64 = i-ls
191 let d: *u8 = clk_slot(names, n); let e: *u8 = clk_slot(organs, n)
192 var f: i64=0; var p: i64=0; var iv: i64=0; var nd: i64=0; var w: i64=0; var g: i64=0 // f: 0=name 1=interval 2=next_due 3=organ
193 while p < ll {
194 if line[p] == (9 as u8) { f = f + 1 }
195 else {
196 if f == 0 { if w < CLK_NAMEW-1 { d[w]=line[p]; w=w+1 } }
197 else { if f == 3 { if g < CLK_NAMEW-1 { e[g]=line[p]; g=g+1 } }
198 else { if line[p] >= (48 as u8) { if line[p] <= (57 as u8) {
199 let dig: i64 = (line[p]-(48 as u8)) as i64
200 if f == 1 { iv = iv*10 + dig } else { nd = nd*10 + dig }
201 } } } }
202 }
203 p = p + 1
204 }
205 d[w] = 0 as u8; e[g] = 0 as u8
206 intervals[n]=iv; next_due[n]=nd; n=n+1
207 } }
208 ls = i + 1
209 }
210 i = i + 1
211 }
212 np[0] = n
213 return n
214}