| 1 | module async |
| 2 | |
| 3 | import context |
| 4 | import sync |
| 5 | import time |
| 6 | |
| 7 | enum PeriodicResultKind { |
| 8 | completed |
| 9 | job_error |
| 10 | context_error |
| 11 | } |
| 12 | |
| 13 | struct PeriodicResult { |
| 14 | kind PeriodicResultKind |
| 15 | err IError = none |
| 16 | } |
| 17 | |
| 18 | // PeriodicHandle owns one detached periodic loop started by start_every(). |
| 19 | // |
| 20 | // The handle exposes only explicit lifecycle operations: stop() requests |
| 21 | // cooperative shutdown, and wait() consumes the loop result once. |
| 22 | @[heap] |
| 23 | pub struct PeriodicHandle { |
| 24 | mut: |
| 25 | parent context.Context |
| 26 | ctx context.Context |
| 27 | cancel context.CancelFn = unsafe { nil } |
| 28 | result_ch chan PeriodicResult |
| 29 | mutex &sync.Mutex = sync.new_mutex() |
| 30 | stopped bool |
| 31 | stop_owns_cancel bool |
| 32 | waited bool |
| 33 | } |
| 34 | |
| 35 | // every runs f repeatedly with interval spacing until ctx is canceled or f fails. |
| 36 | // |
| 37 | // This helper is intentionally blocking: it does not start a hidden scheduler or |
| 38 | // background loop. The first iteration runs after one interval. Iterations never |
| 39 | // overlap; a slow job delays the next interval. Cancellation is cooperative, so |
| 40 | // a running job must observe `ctx.done()` if it needs to stop before returning. |
| 41 | pub fn every(parent context.Context, interval time.Duration, f JobFn) ! { |
| 42 | if interval.nanoseconds() <= 0 { |
| 43 | return error(err_interval_invalid) |
| 44 | } |
| 45 | if f == unsafe { nil } { |
| 46 | return error(err_nil_job) |
| 47 | } |
| 48 | |
| 49 | mut ctx := parent |
| 50 | initial_err := ctx.err() |
| 51 | if initial_err !is none { |
| 52 | return initial_err |
| 53 | } |
| 54 | |
| 55 | done_ch := ctx.done() |
| 56 | mut watch_done := true |
| 57 | // context.background().done() is closed in V but err() remains none. Treat |
| 58 | // that as a non-cancelable context instead of returning immediately. |
| 59 | select { |
| 60 | _ := <-done_ch { |
| 61 | err := ctx.err() |
| 62 | if err !is none { |
| 63 | return err |
| 64 | } |
| 65 | watch_done = false |
| 66 | } |
| 67 | else {} |
| 68 | } |
| 69 | |
| 70 | for { |
| 71 | if watch_done { |
| 72 | select { |
| 73 | _ := <-done_ch { |
| 74 | err := ctx.err() |
| 75 | if err !is none { |
| 76 | return err |
| 77 | } |
| 78 | watch_done = false |
| 79 | } |
| 80 | interval { |
| 81 | run_periodic_iteration(mut ctx, f)! |
| 82 | } |
| 83 | } |
| 84 | } else { |
| 85 | time.sleep(interval) |
| 86 | run_periodic_iteration(mut ctx, f)! |
| 87 | } |
| 88 | } |
| 89 | } |
| 90 | |
| 91 | // start_every starts f in one detached periodic loop and returns its lifecycle handle. |
| 92 | // |
| 93 | // The first iteration runs after one interval. Iterations never overlap; a slow |
| 94 | // job delays the next interval. The returned handle must be stopped and waited |
| 95 | // on by its owner, otherwise the detached loop can keep running. |
| 96 | pub fn start_every(parent context.Context, interval time.Duration, f JobFn) !&PeriodicHandle { |
| 97 | if interval.nanoseconds() <= 0 { |
| 98 | return error(err_interval_invalid) |
| 99 | } |
| 100 | if f == unsafe { nil } { |
| 101 | return error(err_nil_job) |
| 102 | } |
| 103 | |
| 104 | mut parent_ctx := parent |
| 105 | initial_err := parent_ctx.err() |
| 106 | if initial_err !is none { |
| 107 | return initial_err |
| 108 | } |
| 109 | |
| 110 | ctx, cancel := new_cancel_context(parent) |
| 111 | mut handle := &PeriodicHandle{ |
| 112 | parent: parent |
| 113 | ctx: context.Context(ctx) |
| 114 | cancel: cancel |
| 115 | result_ch: chan PeriodicResult{cap: 1} |
| 116 | mutex: sync.new_mutex() |
| 117 | } |
| 118 | spawn run_periodic_handle(mut handle, interval, f) |
| 119 | return handle |
| 120 | } |
| 121 | |
| 122 | // stop requests cooperative shutdown of the detached periodic loop. |
| 123 | // |
| 124 | // stop is idempotent and non-blocking. A running job still has to return |
| 125 | // naturally; wait() is responsible for observing the final result. |
| 126 | pub fn (mut h PeriodicHandle) stop() { |
| 127 | mut parent := h.parent |
| 128 | parent_err := parent.err() |
| 129 | h.mutex.lock() |
| 130 | if h.stopped { |
| 131 | h.mutex.unlock() |
| 132 | return |
| 133 | } |
| 134 | h.stopped = true |
| 135 | h.stop_owns_cancel = parent_err is none |
| 136 | should_cancel := h.stop_owns_cancel |
| 137 | cancel := h.cancel |
| 138 | h.mutex.unlock() |
| 139 | if should_cancel { |
| 140 | cancel() |
| 141 | } |
| 142 | } |
| 143 | |
| 144 | // wait blocks until the detached periodic loop exits. |
| 145 | // |
| 146 | // wait is one-shot. It returns job errors and parent cancellation errors. A |
| 147 | // normal stop() request is reported as successful completion. |
| 148 | pub fn (mut h PeriodicHandle) wait() ! { |
| 149 | h.mutex.lock() |
| 150 | if h.waited { |
| 151 | h.mutex.unlock() |
| 152 | return error(err_periodic_wait_called) |
| 153 | } |
| 154 | h.waited = true |
| 155 | h.mutex.unlock() |
| 156 | |
| 157 | result := <-h.result_ch |
| 158 | if result.err !is none { |
| 159 | if result.kind == .context_error && h.is_stop_result() { |
| 160 | return |
| 161 | } |
| 162 | return result.err |
| 163 | } |
| 164 | } |
| 165 | |
| 166 | fn run_periodic_iteration(mut ctx context.Context, f JobFn) ! { |
| 167 | f(mut ctx)! |
| 168 | err := ctx.err() |
| 169 | if err !is none { |
| 170 | return err |
| 171 | } |
| 172 | } |
| 173 | |
| 174 | fn run_periodic_handle(mut h PeriodicHandle, interval time.Duration, f JobFn) { |
| 175 | mut ctx := h.ctx |
| 176 | result := run_detached_periodic_loop(mut ctx, interval, f) |
| 177 | h.cancel() |
| 178 | h.result_ch <- result |
| 179 | } |
| 180 | |
| 181 | fn run_detached_periodic_loop(mut ctx context.Context, interval time.Duration, f JobFn) PeriodicResult { |
| 182 | done_ch := ctx.done() |
| 183 | mut watch_done := true |
| 184 | select { |
| 185 | _ := <-done_ch { |
| 186 | err := ctx.err() |
| 187 | if err !is none { |
| 188 | return PeriodicResult{ |
| 189 | kind: .context_error |
| 190 | err: err |
| 191 | } |
| 192 | } |
| 193 | watch_done = false |
| 194 | } |
| 195 | else {} |
| 196 | } |
| 197 | |
| 198 | for { |
| 199 | if watch_done { |
| 200 | select { |
| 201 | _ := <-done_ch { |
| 202 | err := ctx.err() |
| 203 | if err !is none { |
| 204 | return PeriodicResult{ |
| 205 | kind: .context_error |
| 206 | err: err |
| 207 | } |
| 208 | } |
| 209 | watch_done = false |
| 210 | } |
| 211 | interval { |
| 212 | result := run_detached_periodic_iteration(mut ctx, f) |
| 213 | if result.err !is none { |
| 214 | return result |
| 215 | } |
| 216 | } |
| 217 | } |
| 218 | } else { |
| 219 | time.sleep(interval) |
| 220 | result := run_detached_periodic_iteration(mut ctx, f) |
| 221 | if result.err !is none { |
| 222 | return result |
| 223 | } |
| 224 | } |
| 225 | } |
| 226 | return PeriodicResult{} |
| 227 | } |
| 228 | |
| 229 | fn run_detached_periodic_iteration(mut ctx context.Context, f JobFn) PeriodicResult { |
| 230 | f(mut ctx) or { return PeriodicResult{ |
| 231 | kind: .job_error |
| 232 | err: err |
| 233 | } } |
| 234 | err := ctx.err() |
| 235 | if err !is none { |
| 236 | return PeriodicResult{ |
| 237 | kind: .context_error |
| 238 | err: err |
| 239 | } |
| 240 | } |
| 241 | return PeriodicResult{} |
| 242 | } |
| 243 | |
| 244 | fn (mut h PeriodicHandle) is_stop_result() bool { |
| 245 | h.mutex.lock() |
| 246 | stop_owns_cancel := h.stop_owns_cancel |
| 247 | h.mutex.unlock() |
| 248 | return stop_owns_cancel |
| 249 | } |
| 250 | |