| 1 | import context |
| 2 | import time |
| 3 | import x.async as xasync |
| 4 | |
| 5 | fn test_with_cancel_sets_error_and_closes_done() { |
| 6 | mut ctx, cancel := xasync.with_cancel() |
| 7 | cancel() |
| 8 | done := ctx.done() |
| 9 | select { |
| 10 | _ := <-done { |
| 11 | assert ctx.err().msg() == 'context canceled' |
| 12 | } |
| 13 | 1 * time.second { |
| 14 | assert false, 'cancel did not close the context done channel' |
| 15 | } |
| 16 | } |
| 17 | } |
| 18 | |
| 19 | fn test_group_success() { |
| 20 | parent := context.background() |
| 21 | mut group := xasync.new_group(parent) |
| 22 | done := chan int{cap: 2} |
| 23 | group.go(fn [done] (mut ctx context.Context) ! { |
| 24 | _ = ctx |
| 25 | done <- 1 |
| 26 | })! |
| 27 | group.go(fn [done] (mut ctx context.Context) ! { |
| 28 | _ = ctx |
| 29 | done <- 2 |
| 30 | })! |
| 31 | group.wait()! |
| 32 | assert (<-done) + (<-done) == 3 |
| 33 | } |
| 34 | |
| 35 | fn test_group_error_returns_first_error() { |
| 36 | parent := context.background() |
| 37 | mut group := xasync.new_group(parent) |
| 38 | group.go(fn (mut ctx context.Context) ! { |
| 39 | _ = ctx |
| 40 | return error('group failed') |
| 41 | })! |
| 42 | group.wait() or { |
| 43 | assert err.msg() == 'group failed' |
| 44 | return |
| 45 | } |
| 46 | assert false |
| 47 | } |
| 48 | |
| 49 | fn test_group_default_does_not_collect_errors() { |
| 50 | mut group := xasync.new_group(context.background()) |
| 51 | group.go(fn (mut ctx context.Context) ! { |
| 52 | _ = ctx |
| 53 | return error('group failed') |
| 54 | })! |
| 55 | group.wait() or { |
| 56 | assert err.msg() == 'group failed' |
| 57 | assert group.errors().len == 0 |
| 58 | return |
| 59 | } |
| 60 | assert false |
| 61 | } |
| 62 | |
| 63 | fn test_group_rejects_invalid_error_collection_config() { |
| 64 | xasync.new_group_with_config(context.background(), collect_errors: true, max_errors: 0) or { |
| 65 | assert err.msg() == 'async: group max_errors must be positive' |
| 66 | return |
| 67 | } |
| 68 | assert false |
| 69 | } |
| 70 | |
| 71 | fn test_group_wait_without_tasks() { |
| 72 | parent := context.background() |
| 73 | mut group := xasync.new_group(parent) |
| 74 | group.wait()! |
| 75 | } |
| 76 | |
| 77 | fn test_group_refuses_go_after_wait() { |
| 78 | parent := context.background() |
| 79 | mut group := xasync.new_group(parent) |
| 80 | group.wait()! |
| 81 | group.go(fn (mut ctx context.Context) ! { |
| 82 | _ = ctx |
| 83 | }) or { |
| 84 | assert err.msg() == 'async: group does not accept new tasks after wait starts' |
| 85 | return |
| 86 | } |
| 87 | assert false |
| 88 | } |
| 89 | |
| 90 | fn test_group_refuses_second_wait() { |
| 91 | parent := context.background() |
| 92 | mut group := xasync.new_group(parent) |
| 93 | group.wait()! |
| 94 | group.wait() or { |
| 95 | assert err.msg() == 'async: group wait was already called' |
| 96 | return |
| 97 | } |
| 98 | assert false |
| 99 | } |
| 100 | |
| 101 | fn test_group_refuses_nil_job() { |
| 102 | parent := context.background() |
| 103 | mut group := xasync.new_group(parent) |
| 104 | nil_job := unsafe { xasync.JobFn(nil) } |
| 105 | group.go(nil_job) or { |
| 106 | assert err.msg() == 'async: job function is nil' |
| 107 | return |
| 108 | } |
| 109 | assert false |
| 110 | } |
| 111 | |
| 112 | fn test_group_cancels_siblings_cooperatively() { |
| 113 | parent := context.background() |
| 114 | mut group := xasync.new_group(parent) |
| 115 | cancelled := chan bool{cap: 1} |
| 116 | group.go(fn (mut ctx context.Context) ! { |
| 117 | _ = ctx |
| 118 | return error('stop siblings') |
| 119 | })! |
| 120 | group.go(fn [cancelled] (mut ctx context.Context) ! { |
| 121 | done := ctx.done() |
| 122 | select { |
| 123 | _ := <-done { |
| 124 | cancelled <- true |
| 125 | } |
| 126 | 1 * time.second { |
| 127 | cancelled <- false |
| 128 | } |
| 129 | } |
| 130 | })! |
| 131 | group.wait() or { assert err.msg() == 'stop siblings' } |
| 132 | assert <-cancelled |
| 133 | } |
| 134 | |
| 135 | fn test_group_first_error_remains_stable_with_concurrent_secondary_errors() { |
| 136 | parent := context.background() |
| 137 | mut group := xasync.new_group(parent) |
| 138 | secondary_ready := chan bool{cap: 8} |
| 139 | for _ in 0 .. 8 { |
| 140 | group.go(fn [secondary_ready] (mut ctx context.Context) ! { |
| 141 | secondary_ready <- true |
| 142 | done := ctx.done() |
| 143 | select { |
| 144 | _ := <-done { |
| 145 | return error('secondary error after cancellation') |
| 146 | } |
| 147 | 1 * time.second { |
| 148 | return error('secondary error timeout') |
| 149 | } |
| 150 | } |
| 151 | })! |
| 152 | } |
| 153 | for _ in 0 .. 8 { |
| 154 | assert <-secondary_ready |
| 155 | } |
| 156 | group.go(fn (mut ctx context.Context) ! { |
| 157 | _ = ctx |
| 158 | return error('primary error') |
| 159 | })! |
| 160 | group.wait() or { |
| 161 | assert err.msg() == 'primary error' |
| 162 | return |
| 163 | } |
| 164 | assert false |
| 165 | } |
| 166 | |
| 167 | fn test_group_collects_errors_without_changing_wait_error() { |
| 168 | mut group := xasync.new_group_with_config(context.background(), |
| 169 | collect_errors: true |
| 170 | max_errors: 4 |
| 171 | )! |
| 172 | secondary_ready := chan bool{cap: 3} |
| 173 | for i in 0 .. 3 { |
| 174 | group.go(fn [secondary_ready, i] (mut ctx context.Context) ! { |
| 175 | secondary_ready <- true |
| 176 | done := ctx.done() |
| 177 | select { |
| 178 | _ := <-done { |
| 179 | return error('secondary collected error ${i}') |
| 180 | } |
| 181 | 1 * time.second { |
| 182 | return error('secondary timeout ${i}') |
| 183 | } |
| 184 | } |
| 185 | })! |
| 186 | } |
| 187 | for _ in 0 .. 3 { |
| 188 | assert <-secondary_ready |
| 189 | } |
| 190 | group.go(fn (mut ctx context.Context) ! { |
| 191 | _ = ctx |
| 192 | return error('primary collected error') |
| 193 | })! |
| 194 | group.wait() or { |
| 195 | assert err.msg() == 'primary collected error' |
| 196 | errs := group.errors() |
| 197 | assert errs.len == 4 |
| 198 | assert error_messages_contain(errs, 'primary collected error') |
| 199 | return |
| 200 | } |
| 201 | assert false |
| 202 | } |
| 203 | |
| 204 | fn test_group_error_collection_is_bounded_and_includes_first_error() { |
| 205 | mut group := xasync.new_group_with_config(context.background(), |
| 206 | collect_errors: true |
| 207 | max_errors: 2 |
| 208 | )! |
| 209 | secondary_ready := chan bool{cap: 4} |
| 210 | for i in 0 .. 4 { |
| 211 | group.go(fn [secondary_ready, i] (mut ctx context.Context) ! { |
| 212 | secondary_ready <- true |
| 213 | done := ctx.done() |
| 214 | select { |
| 215 | _ := <-done { |
| 216 | return error('bounded secondary error ${i}') |
| 217 | } |
| 218 | 1 * time.second { |
| 219 | return error('bounded secondary timeout ${i}') |
| 220 | } |
| 221 | } |
| 222 | })! |
| 223 | } |
| 224 | for _ in 0 .. 4 { |
| 225 | assert <-secondary_ready |
| 226 | } |
| 227 | group.go(fn (mut ctx context.Context) ! { |
| 228 | _ = ctx |
| 229 | return error('bounded primary error') |
| 230 | })! |
| 231 | group.wait() or { |
| 232 | assert err.msg() == 'bounded primary error' |
| 233 | errs := group.errors() |
| 234 | assert errs.len == 2 |
| 235 | assert error_messages_contain(errs, 'bounded primary error') |
| 236 | return |
| 237 | } |
| 238 | assert false |
| 239 | } |
| 240 | |
| 241 | fn test_group_errors_returns_snapshot_copy() { |
| 242 | mut group := xasync.new_group_with_config(context.background(), |
| 243 | collect_errors: true |
| 244 | max_errors: 1 |
| 245 | )! |
| 246 | group.go(fn (mut ctx context.Context) ! { |
| 247 | _ = ctx |
| 248 | return error('snapshot source error') |
| 249 | })! |
| 250 | group.wait() or { |
| 251 | assert err.msg() == 'snapshot source error' |
| 252 | mut snapshot := group.errors() |
| 253 | assert snapshot.len == 1 |
| 254 | snapshot[0] = error('mutated snapshot error') |
| 255 | second_snapshot := group.errors() |
| 256 | assert second_snapshot.len == 1 |
| 257 | assert second_snapshot[0].msg() == 'snapshot source error' |
| 258 | return |
| 259 | } |
| 260 | assert false |
| 261 | } |
| 262 | |
| 263 | fn test_group_many_short_jobs_return_first_error() { |
| 264 | jobs := 64 |
| 265 | mut group := xasync.new_group(context.background()) |
| 266 | done := chan int{cap: jobs} |
| 267 | for i in 0 .. jobs { |
| 268 | group.go(fn [done, i] (mut ctx context.Context) ! { |
| 269 | _ = ctx |
| 270 | done <- i |
| 271 | })! |
| 272 | } |
| 273 | group.go(fn (mut ctx context.Context) ! { |
| 274 | _ = ctx |
| 275 | return error('group stress failure') |
| 276 | })! |
| 277 | group.wait() or { |
| 278 | assert err.msg() == 'group stress failure' |
| 279 | mut seen := []bool{len: jobs} |
| 280 | for _ in 0 .. jobs { |
| 281 | select { |
| 282 | i := <-done { |
| 283 | assert i >= 0 |
| 284 | assert i < jobs |
| 285 | assert !seen[i] |
| 286 | seen[i] = true |
| 287 | } |
| 288 | 1 * time.second { |
| 289 | assert false, 'group short job did not finish' |
| 290 | } |
| 291 | } |
| 292 | } |
| 293 | for was_seen in seen { |
| 294 | assert was_seen |
| 295 | } |
| 296 | return |
| 297 | } |
| 298 | assert false |
| 299 | } |
| 300 | |
| 301 | fn test_group_parent_cancellation_is_observed_by_cooperative_job() { |
| 302 | parent_ctx, cancel := xasync.with_cancel() |
| 303 | mut group := xasync.new_group(parent_ctx) |
| 304 | observed := chan string{cap: 1} |
| 305 | group.go(fn [observed] (mut ctx context.Context) ! { |
| 306 | done := ctx.done() |
| 307 | select { |
| 308 | _ := <-done { |
| 309 | err := ctx.err() |
| 310 | observed <- err.msg() |
| 311 | return err |
| 312 | } |
| 313 | 1 * time.second { |
| 314 | observed <- 'not canceled' |
| 315 | return error('parent cancellation was not observed') |
| 316 | } |
| 317 | } |
| 318 | })! |
| 319 | cancel() |
| 320 | group.wait() or { assert err.msg() == 'context canceled' } |
| 321 | select { |
| 322 | msg := <-observed { |
| 323 | assert msg == 'context canceled' |
| 324 | } |
| 325 | 2 * time.second { |
| 326 | assert false, 'cooperative group job did not observe parent cancellation' |
| 327 | } |
| 328 | } |
| 329 | } |
| 330 | |
| 331 | fn error_messages_contain(errs []IError, msg string) bool { |
| 332 | for err in errs { |
| 333 | if err.msg() == msg { |
| 334 | return true |
| 335 | } |
| 336 | } |
| 337 | return false |
| 338 | } |
| 339 | |