nx_resource_arbiter.nx source
↩ module page · 337 lines · 12342 B
1// nx_resource_arbiter.nx -- allocation under contention with priority + treaty.
2//
3// Per CARDINAL [[feedback-conductor-heterogeneous-compute-no-second-class-
4// resources]] + [[feedback-parallel-companion-multimodal-dnd-real-time]]:
5// when multiple cells want the same scarce resource (VRAM bytes,
6// CPU microseconds, NVMe iops), the arbiter decides who gets it based
7// on priority + active treaty + current allocation share. REFUSES the
8// "first-come-first-served leads to starvation of low-rate streams" and
9// "highest-priority-always-wins leads to background-cell death" failure
10// modes.
11//
12// V1 ships sealed resource enum + request enum + grant verdict +
13// per-cell allocation tracking + max-share fairness floor (no cell gets
14// >X% in a single epoch) + per-resource budget pool.
15
16import "nx_syscalls.nx"
17import "nx_tier.nx"
18
19// ===== Sealed enum: NxResourceKind ================================
20
21const NX_RA_KIND_VRAM_BYTES: nx_int = 0
22const NX_RA_KIND_RAM_BYTES: nx_int = 1
23const NX_RA_KIND_DISK_BYTES: nx_int = 2
24const NX_RA_KIND_CPU_MICROSECONDS: nx_int = 3
25const NX_RA_KIND_GPU_MICROSECONDS: nx_int = 4
26const NX_RA_KIND_NETWORK_BYTES: nx_int = 5
27const NX_RA_KIND_FILE_HANDLES: nx_int = 6
28const NX_RA_KIND_N: nx_int = 7
29
30// ===== Sealed enum: NxGrantVerdict ================================
31
32const NX_RA_V_GRANTED_FULL: nx_int = 0
33const NX_RA_V_GRANTED_PARTIAL: nx_int = 1 // less than requested
34const NX_RA_V_DENIED_NO_BUDGET: nx_int = 2
35const NX_RA_V_DENIED_MAX_SHARE: nx_int = 3 // would exceed fairness cap
36const NX_RA_V_DENIED_PREEMPTED: nx_int = 4 // higher-pri cell took it back
37const NX_RA_V_INVALID: nx_int = 5
38const NX_RA_V_NULL: nx_int = 6
39const NX_RA_V_N: nx_int = 7
40
41// ===== Struct: NxAllocation =======================================
42
43struct NxAllocation {
44 cell_id: nx_int,
45 resource_kind: nx_int,
46 units_held: nx_size,
47 priority: nx_int,
48 granted_at_us: nx_size,
49 last_renewed_us: nx_size,
50 treaty_id: nx_int,
51}
52
53const NX_RA_ABI_CELL_BYTES: nx_int = __size_of(nx_int)
54const NX_RA_A_BYTES: nx_int = __size_of(NxAllocation)
55
56struct NxResourceArbiter {
57 allocations: *u8,
58 n_allocations: nx_int,
59 capacity: nx_int,
60 // Per-resource budget (i64[NX_RA_KIND_N])
61 budgets: *u8,
62 // Per-resource max-share-q10 (e.g. 717 = 70%)
63 max_share_q10: *u8,
64 grants_full: nx_int,
65 grants_partial: nx_int,
66 denials: nx_int,
67 preemptions: nx_int,
68}
69
70const NX_RA_BYTES: nx_int = __size_of(NxResourceArbiter)
71const NX_RA_Q10_ONE: nx_int = 1024
72const NX_RA_SIZE_MAX: nx_size = 9223372036854775807
73
74// ===== Validators =================================================
75
76func nx_ra_kind_is_valid(k: nx_int) -> nx_int {
77 if k < 0 { return 0 }
78 if k >= NX_RA_KIND_N { return 0 }
79 return 1
80}
81
82func nx_ra_v_is_valid(v: nx_int) -> nx_int {
83 if v < 0 { return 0 }
84 if v >= NX_RA_V_N { return 0 }
85 return 1
86}
87
88// ===== Constructor ================================================
89
90func nx_ra_new(capacity: nx_int) -> *NxResourceArbiter {
91 if capacity <= 0 { return 0 as *NxResourceArbiter }
92 if capacity > NX_RA_SIZE_MAX / NX_RA_A_BYTES { return 0 as *NxResourceArbiter }
93 let raw: *u8 = sys_mmap(NX_RA_BYTES)
94 if (raw as i64)<0 { return 0 as *NxResourceArbiter }
95 let a: *NxResourceArbiter = raw as *NxResourceArbiter
96 let allocations: *u8=sys_mmap(capacity*NX_RA_A_BYTES)
97 if (allocations as i64)<0 { sys_munmap(raw,NX_RA_BYTES);return 0 as *NxResourceArbiter }
98 let budgets: *u8=sys_mmap(NX_RA_KIND_N*__size_of(nx_size))
99 if (budgets as i64)<0 {
100 sys_munmap(allocations,capacity*NX_RA_A_BYTES);sys_munmap(raw,NX_RA_BYTES)
101 return 0 as *NxResourceArbiter
102 }
103 let shares: *u8=sys_mmap(NX_RA_KIND_N*__size_of(nx_int))
104 if (shares as i64)<0 {
105 sys_munmap(budgets,NX_RA_KIND_N*__size_of(nx_size))
106 sys_munmap(allocations,capacity*NX_RA_A_BYTES);sys_munmap(raw,NX_RA_BYTES)
107 return 0 as *NxResourceArbiter
108 }
109 a.allocations=allocations;a.n_allocations=0;a.capacity=capacity
110 a.budgets=budgets;a.max_share_q10=shares
111 let bud: *nx_size=budgets as *nx_size;let shr: *nx_int=shares as *nx_int
112 var i: nx_int=0
113 while i<NX_RA_KIND_N {
114 bud[i]=0
115 shr[i]=717 // Existing compatibility default; explicit policy setter overrides it.
116 i=i+1
117 }
118 a.grants_full=0;a.grants_partial=0;a.denials=0;a.preemptions=0
119 return a
120}
121
122func _ra_alloc_at(a: *NxResourceArbiter, idx: nx_int) -> *NxAllocation {
123 if idx < 0 { return 0 as *NxAllocation }
124 if idx >= a.n_allocations { return 0 as *NxAllocation }
125 let off: nx_int = idx * NX_RA_A_BYTES
126 return (a.allocations + off) as *NxAllocation
127}
128
129// ===== Budget config ==============================================
130
131func nx_ra_set_budget(a: *NxResourceArbiter,
132 resource_kind: nx_int,
133 budget: nx_size) -> nx_int {
134 if (a as i64) == 0 { return NX_RA_V_NULL }
135 if nx_ra_kind_is_valid(resource_kind) == 0 { return NX_RA_V_INVALID }
136 if budget < 0 { return NX_RA_V_INVALID }
137 let bud: *nx_size = a.budgets as *nx_size
138 bud[resource_kind] = budget
139 return NX_RA_V_GRANTED_FULL
140}
141
142func nx_ra_set_max_share(a: *NxResourceArbiter,
143 resource_kind: nx_int,
144 max_share_q10: nx_int) -> nx_int {
145 if (a as i64) == 0 { return NX_RA_V_NULL }
146 if nx_ra_kind_is_valid(resource_kind) == 0 { return NX_RA_V_INVALID }
147 if max_share_q10 <= 0 { return NX_RA_V_INVALID }
148 if max_share_q10 > NX_RA_Q10_ONE { return NX_RA_V_INVALID }
149 let shr: *nx_int = a.max_share_q10 as *nx_int
150 shr[resource_kind] = max_share_q10
151 return NX_RA_V_GRANTED_FULL
152}
153
154func nx_ra_get_budget(a: *NxResourceArbiter, resource_kind: nx_int) -> nx_size {
155 if (a as i64) == 0 { return 0 }
156 if nx_ra_kind_is_valid(resource_kind) == 0 { return 0 }
157 let bud: *nx_size = a.budgets as *nx_size
158 return bud[resource_kind]
159}
160
161// ===== Aggregations ===============================================
162
163func nx_ra_total_held(a: *NxResourceArbiter, resource_kind: nx_int) -> nx_size {
164 if (a as i64) == 0 { return 0 }
165 var sum: nx_size = 0
166 var i: nx_int = 0
167 while i < a.n_allocations {
168 let al: *NxAllocation = _ra_alloc_at(a, i)
169 if al.resource_kind == resource_kind { sum = sum + al.units_held }
170 i = i + 1
171 }
172 return sum
173}
174
175func nx_ra_held_by_cell(a: *NxResourceArbiter,
176 cell_id: nx_int,
177 resource_kind: nx_int) -> nx_size {
178 if (a as i64) == 0 { return 0 }
179 var sum: nx_size = 0
180 var i: nx_int = 0
181 while i < a.n_allocations {
182 let al: *NxAllocation = _ra_alloc_at(a, i)
183 if al.cell_id == cell_id {
184 if al.resource_kind == resource_kind { sum = sum + al.units_held }
185 }
186 i = i + 1
187 }
188 return sum
189}
190
191// ===== Request resource ===========================================
192//
193// Returns verdict + writes actual units_granted via *out_granted (0 on
194// deny / null). Granted_partial when budget allows < requested.
195
196func nx_ra_request(a: *NxResourceArbiter,
197 cell_id: nx_int,
198 resource_kind: nx_int,
199 units: nx_size,
200 priority: nx_int,
201 treaty_id: nx_int,
202 out_granted: *i64,
203 now_us: nx_size) -> nx_int {
204 if (out_granted as i64) == 0 { return NX_RA_V_NULL }
205 out_granted[0] = 0
206 if (a as i64) == 0 { return NX_RA_V_NULL }
207 if nx_ra_kind_is_valid(resource_kind) == 0 { return NX_RA_V_INVALID }
208 if cell_id == 0 { return NX_RA_V_INVALID }
209 if units <= 0 { return NX_RA_V_INVALID }
210 let budget: nx_size = nx_ra_get_budget(a,resource_kind)
211 let used: nx_size = nx_ra_total_held(a,resource_kind)
212 if budget <= 0 {
213 a.denials = a.denials + 1
214 return NX_RA_V_DENIED_NO_BUDGET
215 }
216 if used >= budget {
217 a.denials = a.denials + 1
218 return NX_RA_V_DENIED_NO_BUDGET
219 }
220 let free_pool: nx_size = budget - used
221 let shr: *nx_int = a.max_share_q10 as *nx_int
222 let share: nx_size = shr[resource_kind]
223 // Quotient/remainder decomposition avoids budget*share overflow and
224 // compares integer units directly instead of rounding usage down to Q10.
225 let cell_limit: nx_size = (budget / NX_RA_Q10_ONE) * share + ((budget % NX_RA_Q10_ONE) * share) / NX_RA_Q10_ONE
226 let held: nx_size = nx_ra_held_by_cell(a,cell_id,resource_kind)
227 if held > cell_limit {
228 a.denials = a.denials + 1
229 return NX_RA_V_DENIED_MAX_SHARE
230 }
231 if units > cell_limit - held {
232 a.denials = a.denials + 1
233 return NX_RA_V_DENIED_MAX_SHARE
234 }
235 var to_grant: nx_size = units
236 if to_grant > free_pool { to_grant = free_pool }
237 // Released/preempted slots remain reusable; capacity bounds simultaneous
238 // reservations, not the lifetime number of jobs.
239 var slot: nx_int = 0
240 while slot < a.n_allocations {
241 let existing: *NxAllocation = _ra_alloc_at(a,slot)
242 if existing.units_held == 0 { break }
243 slot = slot + 1
244 }
245 if slot >= a.capacity {
246 a.denials = a.denials + 1
247 return NX_RA_V_DENIED_NO_BUDGET
248 }
249 let al: *NxAllocation = (a.allocations + slot * NX_RA_A_BYTES) as *NxAllocation
250 al.cell_id = cell_id
251 al.resource_kind = resource_kind
252 al.units_held = to_grant
253 al.priority = priority
254 al.granted_at_us = now_us
255 al.last_renewed_us = now_us
256 al.treaty_id = treaty_id
257 if slot == a.n_allocations { a.n_allocations = a.n_allocations + 1 }
258 out_granted[0] = to_grant as i64
259 if to_grant == units {
260 a.grants_full = a.grants_full + 1
261 return NX_RA_V_GRANTED_FULL
262 }
263 a.grants_partial = a.grants_partial + 1
264 return NX_RA_V_GRANTED_PARTIAL
265}
266
267// ===== Release resource ===========================================
268
269func nx_ra_release(a: *NxResourceArbiter,
270 cell_id: nx_int,
271 resource_kind: nx_int) -> nx_size {
272 if (a as i64) == 0 { return 0 }
273 var released: nx_size = 0
274 var i: nx_int = 0
275 while i < a.n_allocations {
276 let al: *NxAllocation = _ra_alloc_at(a, i)
277 if al.cell_id == cell_id {
278 if al.resource_kind == resource_kind {
279 released = released + al.units_held
280 // Mark zero -- compaction deferred
281 al.units_held = 0
282 }
283 }
284 i = i + 1
285 }
286 return released
287}
288
289// ===== Preempt: take from lower-priority cell =====================
290
291func nx_ra_preempt(a: *NxResourceArbiter,
292 resource_kind: nx_int,
293 requesting_priority: nx_int,
294 units_needed: nx_size) -> nx_size {
295 if (a as i64) == 0 { return 0 }
296 if nx_ra_kind_is_valid(resource_kind) == 0 { return 0 }
297 var reclaimed: nx_size = 0
298 var i: nx_int = 0
299 while i < a.n_allocations {
300 if reclaimed < units_needed {
301 let al: *NxAllocation = _ra_alloc_at(a, i)
302 if al.resource_kind == resource_kind {
303 if al.priority < requesting_priority {
304 if al.units_held > 0 {
305 reclaimed = reclaimed + al.units_held
306 al.units_held = 0
307 a.preemptions = a.preemptions + 1
308 }
309 }
310 }
311 }
312 i = i + 1
313 }
314 return reclaimed
315}
316
317// ===== Stats accessors ============================================
318
319func nx_ra_grants_full(a: *NxResourceArbiter) -> nx_int {
320 if (a as i64) == 0 { return 0 }
321 return a.grants_full
322}
323
324func nx_ra_grants_partial(a: *NxResourceArbiter) -> nx_int {
325 if (a as i64) == 0 { return 0 }
326 return a.grants_partial
327}
328
329func nx_ra_denials(a: *NxResourceArbiter) -> nx_int {
330 if (a as i64) == 0 { return 0 }
331 return a.denials
332}
333
334func nx_ra_preemptions(a: *NxResourceArbiter) -> nx_int {
335 if (a as i64) == 0 { return 0 }
336 return a.preemptions
337}