vxx / vlib / x / async / group_test.v
338 lines · 321 sloc · 7.46 KB · a66aa5b0fe618b32988f4fd9b235017c336940ba
Raw
1import context
2import time
3import x.async as xasync
4
5fn 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
19fn 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
35fn 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
49fn 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
63fn 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
71fn test_group_wait_without_tasks() {
72 parent := context.background()
73 mut group := xasync.new_group(parent)
74 group.wait()!
75}
76
77fn 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
90fn 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
101fn 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
112fn 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
135fn 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
167fn 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
204fn 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
241fn 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
263fn 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
301fn 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
331fn 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