async_rt.nx source

1// async_rt.nx -- cooperative async task runtime. 2// 3// Phase A: runtime data structures + scheduler. Language-level 4// `async fn` / `await` (EFFICIENCY_ROADMAP §4.3) is a parser 5// change pending; this module gives callers enough machinery 6// to hand-code stackful coroutines today + get lifted into real 7// async/await syntax later without breaking the API. 8// 9// Design: single-threaded cooperative scheduler on one kernel 10// thread. Each task = (function_pointer, state_blob). 11// Scheduler runs ready queue to exhaustion, then epolls for IO 12// readiness + wakes blocked tasks. 13// 14// Why single-threaded: thread safety without atomics. Multi- 15// thread scheduling adds complexity that belongs in a later 16// phase once the sequential model is proven. 17// 18// Used by: HTTP server accept loops, DNS resolver, file watcher, 19// anything that would otherwise thread-per-connection. 20// 21// Invariants: 22// AR1 Scheduler `run_until_idle` returns when all tasks are 23// blocked waiting for IO + nothing's ready. Caller 24// does the epoll_wait before re-entering. 25// AR2 `task_spawn` always succeeds if slot cap not hit; 26// returns task id. 27// AR3 `task_yield` marks the calling task ready + saves its 28// state via the caller-supplied resume hook. 29 30import "syscalls.nx" 31 32const AR_MAX_TASKS: i64 = 256 33const AR_ERR_FULL: i64 = -1 34const AR_ERR_NO_SUCH: i64 = -2 35 36const TASK_READY: i64 = 0 37const TASK_BLOCKED: i64 = 1 38const TASK_DONE: i64 = 2 39 40struct AsyncTask { 41 id: i64, 42 state: i64, // TASK_* 43 resume_fn: i64, // function-pointer (i64-encoded) 44 state_blob: i64, // opaque pointer for user state 45 wait_fd: i64, // fd being waited on (-1 if none) 46 wait_ev: i64, // events mask (1=read, 2=write) 47} 48 49struct Runtime { 50 tasks: *AsyncTask, 51 n_tasks: i64, 52 next_id: i64, 53} 54 55func rt_new() -> *Runtime { 56 let raw: *u8 = sys_mmap(32) 57 let rt: *Runtime = raw as *Runtime 58 let t_raw: *u8 = sys_mmap(AR_MAX_TASKS * 64 + 64) 59 rt.tasks = t_raw as *AsyncTask 60 rt.n_tasks = 0 61 rt.next_id = 1 62 return rt 63} 64 65// Spawn a new task. Returns task id >= 1 on success. 66func task_spawn(rt: *Runtime, resume_fn: i64, state_blob: i64) -> i64 { 67 if rt.n_tasks >= AR_MAX_TASKS { return AR_ERR_FULL } 68 let base: i64 = rt.tasks as i64 69 let t: *AsyncTask = (base + rt.n_tasks * 64) as *AsyncTask 70 t.id = rt.next_id 71 t.state = TASK_READY 72 t.resume_fn = resume_fn 73 t.state_blob = state_blob 74 t.wait_fd = -1 75 t.wait_ev = 0 76 rt.next_id = rt.next_id + 1 77 rt.n_tasks = rt.n_tasks + 1 78 return t.id 79} 80 81// Find a task by id. Returns index or AR_ERR_NO_SUCH. 82func task_find(rt: *Runtime, id: i64) -> i64 { 83 var i: i64 = 0 84 while i < rt.n_tasks { 85 let base: i64 = rt.tasks as i64 86 let t: *AsyncTask = (base + i * 64) as *AsyncTask 87 if t.id == id { return i } 88 i = i + 1 89 } 90 return AR_ERR_NO_SUCH 91} 92 93// Mark a task ready (unblock). 94func task_wake(rt: *Runtime, id: i64) -> i64 { 95 let idx: i64 = task_find(rt, id) 96 if idx < 0 { return idx } 97 let base: i64 = rt.tasks as i64 98 let t: *AsyncTask = (base + idx * 64) as *AsyncTask 99 t.state = TASK_READY 100 t.wait_fd = -1 101 t.wait_ev = 0 102 return 0 103} 104 105// Block the task on a given fd for a given event mask. 106func task_block(rt: *Runtime, id: i64, fd: i64, events: i64) -> i64 { 107 let idx: i64 = task_find(rt, id) 108 if idx < 0 { return idx } 109 let base: i64 = rt.tasks as i64 110 let t: *AsyncTask = (base + idx * 64) as *AsyncTask 111 t.state = TASK_BLOCKED 112 t.wait_fd = fd 113 t.wait_ev = events 114 return 0 115} 116 117// Mark a task done (scheduler will reap). 118func task_complete(rt: *Runtime, id: i64) -> i64 { 119 let idx: i64 = task_find(rt, id) 120 if idx < 0 { return idx } 121 let base: i64 = rt.tasks as i64 122 let t: *AsyncTask = (base + idx * 64) as *AsyncTask 123 t.state = TASK_DONE 124 return 0 125} 126 127// Count how many tasks are in a given state. 128func rt_count_state(rt: *Runtime, state: i64) -> i64 { 129 var count: i64 = 0 130 var i: i64 = 0 131 while i < rt.n_tasks { 132 let base: i64 = rt.tasks as i64 133 let t: *AsyncTask = (base + i * 64) as *AsyncTask 134 if t.state == state { count = count + 1 } 135 i = i + 1 136 } 137 return count 138} 139 140// Compile-only smoke. 141func main() -> i64 { 142 let rt: *Runtime = rt_new() 143 let id1: i64 = task_spawn(rt, 0x1000, 0) 144 if id1 != 1 { return 1 } 145 let id2: i64 = task_spawn(rt, 0x2000, 0) 146 if id2 != 2 { return 2 } 147 148 if rt_count_state(rt, TASK_READY) != 2 { return 3 } 149 150 task_block(rt, id1, 3, 1) 151 if rt_count_state(rt, TASK_READY) != 1 { return 4 } 152 if rt_count_state(rt, TASK_BLOCKED) != 1 { return 5 } 153 154 task_wake(rt, id1) 155 if rt_count_state(rt, TASK_READY) != 2 { return 6 } 156 157 task_complete(rt, id2) 158 if rt_count_state(rt, TASK_DONE) != 1 { return 7 } 159 return 0 160}