From a66aa5b0fe618b32988f4fd9b235017c336940ba Mon Sep 17 00:00:00 2001 From: GGRei Date: Mon, 22 Jun 2026 08:28:41 +0200 Subject: [PATCH] async: add bounded pool, periodic, and group helpers (#27524) --- vlib/x/async/README.md | 149 +++++-- vlib/x/async/benchmarks/README.md | 8 +- vlib/x/async/benchmarks/async_benchmark.v | 97 +++++ .../x/async/benchmarks/run_async_benchmark.sh | 6 +- vlib/x/async/errors.v | 2 + vlib/x/async/examples/README.md | 3 + vlib/x/async/examples/bounded_pool_submit.v | 77 ++++ vlib/x/async/group.v | 49 ++- vlib/x/async/group_test.v | 127 ++++++ vlib/x/async/periodic.v | 181 +++++++++ vlib/x/async/periodic_test.v | 208 ++++++++++ vlib/x/async/pool.v | 131 ++++++- vlib/x/async/pool_test.v | 364 ++++++++++++++++++ 13 files changed, 1354 insertions(+), 48 deletions(-) create mode 100644 vlib/x/async/examples/bounded_pool_submit.v diff --git a/vlib/x/async/README.md b/vlib/x/async/README.md index 6ddb04222..05ba503e8 100644 --- a/vlib/x/async/README.md +++ b/vlib/x/async/README.md @@ -21,10 +21,13 @@ context and function-type helpers: sibling jobs cooperatively. - `Task[T]`: run one value-returning job and wait for its result once. - `Pool`: run accepted jobs with a fixed concurrency limit and bounded backlog. -- `every`: run a periodic job without overlapping iterations. +- periodic jobs: run a blocking `every()` loop or a detached + `PeriodicHandle` without overlapping iterations. - `with_timeout` / `with_timeout_context`: run one job with a bounded deadline. Ticker objects and server-specific helpers are not part of this API. +`PeriodicHandle` is only an explicit `stop()` / `wait()` lifecycle handle for +one detached periodic loop. ## Why @@ -137,9 +140,9 @@ fn worker(mut ctx context.Context) ! { ``` If a job ignores `ctx.done()`, `Group.wait()` will still wait for it to return. -`Pool.close()` and `every()` also wait for running non-cooperative jobs to return. -`with_timeout()` returns when the timeout expires, but the ignored job may keep -running until it finishes naturally. +`Pool.close()`, `every()`, and `PeriodicHandle.wait()` also wait for running +non-cooperative jobs to return. `with_timeout()` returns when the timeout +expires, but the ignored job may keep running until it finishes naturally. ## Group @@ -175,6 +178,20 @@ Guarantees: keeps the lifecycle simple: create a group, submit jobs, wait once, then discard the group. +Use `new_group_with_config()` only when callers need a bounded snapshot of job +errors in addition to the first error returned by `wait()`: + +```v ignore +mut group := async.new_group_with_config(parent, collect_errors: true, max_errors: 8)! +``` + +Error collection is opt-in. `new_group()` does not collect errors, and +`errors()` returns an empty array for that default mode. When collection is +enabled, `max_errors` must be positive; collection stores at most `max_errors` +observed job errors and `errors()` returns a copy of that bounded snapshot. +`wait()` still returns only the first observed error and never returns an +aggregate. Concurrent error order is not guaranteed. + ## Task `Task[T]` represents one concurrent computation that returns either a value or @@ -252,6 +269,35 @@ Backpressure is explicit. `try_submit()` never waits for backlog space: - if the pool is closed or already waiting, it returns `async: pool is closed`; - if the job function is nil, it returns `async: job function is nil`. +Use `submit_with_context()` when the caller should wait for backlog space, but +only while a parent context remains active: + +```v ignore +parent_ctx, cancel := async.with_cancel() +defer { + cancel() +} +pool.submit_with_context(parent_ctx, fn (mut ctx context.Context) ! { + process_message(mut ctx)! +})! +``` + +Use `submit_with_timeout()` when admission should wait only up to a fixed +duration: + +```v ignore +pool.submit_with_timeout(250 * time.millisecond, fn (mut ctx context.Context) ! { + process_message(mut ctx)! +})! +``` + +The context or timeout bounds admission only. Once the job is accepted, it +receives the pool's context, the same as `try_submit()`. If the parent context is +canceled before acceptance, `submit_with_context()` returns that parent context +error. If the timeout expires before acceptance, `submit_with_timeout()` returns +`async: timeout`. If `wait()` or `close()` starts while a caller is waiting for +admission, the submit call returns `async: pool is closed`. + Lifecycle: - `workers` and `queue_size` must be positive and are fixed at creation. @@ -273,7 +319,7 @@ held. ## Periodic Jobs `every()` runs a job repeatedly until its context is canceled or the job returns -an error. +an error. It is blocking and does not start background work: ```v ignore ctx, cancel := async.with_cancel() @@ -296,10 +342,57 @@ Guarantees: - a job error stops the loop and is returned unchanged. - context cancellation stops the loop and returns the context error. -`every()` is not a scheduler and does not expose a ticker handle. If a running -job ignores cancellation, `every()` cannot return until that job returns +Use `start_every()` when the periodic loop should run in the background with an +explicit lifecycle handle: + +```v ignore +ctx, cancel := async.with_cancel() +defer { + cancel() +} + +mut handle := async.start_every(ctx, 5 * time.second, fn (mut ctx context.Context) ! { + cleanup_stale_clients(mut ctx)! +})! +defer { + handle.stop() + handle.wait() or {} +} +``` + +`start_every()` has the same interval, nil-job, first-tick, no-overlap, and job +error rules as `every()`, but returns immediately after starting one detached +loop. If the parent context is already canceled, it returns the parent context +error and does not start a worker. + +`PeriodicHandle` lifecycle: + +- `stop()` is idempotent and requests cooperative shutdown of the detached loop; +- `stop()` is non-blocking and does not kill a running job; +- `wait()` blocks until the detached loop exits; +- `wait()` is one-shot; a second call returns + `async: periodic wait was already called`; +- a normal `stop()` makes `wait()` return successfully; +- a job error is returned unchanged by `wait()`, even if its message matches + `context canceled`; +- parent cancellation is returned by `wait()` as the parent context error. + +The detached loop publishes its final result through a bounded channel with +capacity 1, so it can exit even if the owner has not called `wait()` yet. + +A normal stop is recognized only when the loop itself observes the handle-owned +context cancellation outside a user job error. If parent cancellation is already +observable when `stop()` is called, parent cancellation wins and `wait()` returns +the parent context error. + +If the owner never calls `stop()` or the parent context is never canceled, the +detached loop can keep running. If the owner calls `stop()` but never calls +`wait()`, a running non-cooperative job may still continue until it returns naturally. +Periodic helpers are not schedulers and do not expose ticker objects. They run +one serial loop; a slow iteration delays the next one. + ## Timeout `with_timeout()` runs one job with a background context and a timeout: @@ -329,13 +422,20 @@ async.with_timeout_context(parent, 250 * time.millisecond, fn (mut ctx context.C Error behavior: -- if the job returns first, the job error is returned unchanged; +- if the job finishes before the effective timeout deadline, the job error is + returned unchanged; - if the timeout expires first, the public error is `async: timeout`; - if the parent context is canceled first, including when the parent's own deadline expires first, the parent context error is returned; - a job that observes the local `x.async` timeout by returning `context deadline exceeded` is normalized to `async: timeout`. +Timeout ownership matters at the deadline boundary. If the job result is received +first but the job finished at or after an `x.async`-owned timeout deadline, +`with_timeout_context()` still returns `async: timeout`. That normalization is +not applied to a shorter parent-owned deadline; parent cancellation keeps the +parent context error path. + The result channel used internally is buffered so the spawned job can finish and publish its result even if the caller has already returned on timeout. @@ -352,8 +452,10 @@ publish its result even if the caller has already returned on timeout. - `Pool.close()` drains accepted jobs before returning and reports the first job error. - `every()` is blocking and serial, so periodic iterations cannot overlap. +- `PeriodicHandle.stop()` is cooperative, and `PeriodicHandle.wait()` is + one-shot so detached periodic lifecycle ownership is explicit. - It uses `sync.Mutex` to protect mutable lifecycle/result state in `Group`, - `Task[T]`, and `Pool`. + `Task[T]`, `Pool`, and `PeriodicHandle`. - It keeps `sync.WaitGroup.add()` and `sync.WaitGroup.wait()` separated by a lifecycle mutex for `Group` and `Pool` where accepted work can race with shutdown. @@ -365,8 +467,8 @@ validation and resource limits for that domain. This milestone does not include: -- blocking pool submission; -- ticker objects or detached periodic handles; +- unbounded blocking pool submission without a context or timeout; +- ticker objects; - multi-consumer futures or promise chaining; - panic recovery; - scheduler changes; @@ -379,17 +481,11 @@ replace them with a new runtime. Possible future additions, if accepted by the project maintainers and backed by tests, could still fit the current philosophy: -- blocking or timeout-based pool submission, built on existing channels; -- detached periodic handles with explicit `stop()` / `wait()` lifecycle; -- helpers that combine `Group`, `Pool`, `Task[T]`, and timeout for server-style - code; - more examples and integration tests for other V modules that already use concurrency; - careful rewrites of existing V modules that need concurrency, if maintainers decide `x.async` makes their lifecycle, cancellation, or backpressure simpler and safer without breaking compatibility; -- optional error collection helpers, as long as first-error behavior stays - simple and documented; - more benchmarks and stress tests that remain bounded and reproducible. Other ideas would require a separate design because they move beyond this @@ -413,6 +509,7 @@ Small runnable examples live in `vlib/x/async/examples/`: - `basic_group.v`: first error propagation and cooperative sibling cancellation. - `basic_task.v`: run one value-returning task and consume it with `wait()`. - `worker_pool.v`: fixed concurrency with explicit `try_submit()` backpressure. +- `bounded_pool_submit.v`: context/timeout-bounded `Pool` admission. - `periodic.v`: a blocking `every()` loop stopped by context cancellation. - `timeout.v`: run one cooperative job with a bounded timeout. - `net_http/`: synthetic `net.http` request/response work through `Pool`. @@ -446,13 +543,15 @@ V build artefact collisions; it is separate from the runtime guarantees of `x.async`. The tests cover successful groups, first-error propagation, empty waits, -rejected submissions after `wait()`, one-shot task waits, task values and -errors, pool worker limits, pool queue backpressure, pool close/wait behavior, -periodic execution, periodic cancellation, non-overlapping periodic iterations, +rejected submissions after `wait()`, optional bounded group error collection, +one-shot task waits, task values and errors, pool worker limits, pool queue +backpressure, pool close/wait behavior, bounded pool submission with context +cancellation and timeout, periodic execution, detached periodic handle stop/wait +lifecycle, periodic cancellation, non-overlapping periodic iterations, cooperative cancellation, timeout errors, parent cancellation, nil job rejection, parent deadline preservation, invalid intervals, zero/already-expired timeouts, -stress cases, and jobs that ignore cancellation. Internal tests also verify -that the derived `AsyncContext` closes `done()` and propagates parent +stress cases, and jobs that ignore cancellation. Internal tests also verify that +the derived `AsyncContext` closes `done()` and propagates parent cancellation. Module-oriented integration tests live in `vlib/x/async/tests/`. They are @@ -470,11 +569,13 @@ sh vlib/x/async/benchmarks/run_async_benchmark.sh The script uses the local `./v`, runs serially, and isolates `VTMP`, `VCACHE`, and the benchmark executable output. It measures short default runs for -`Group`, `Task[T]`, `Pool`, `with_timeout()`, and `every()`. +`Group`, `Task[T]`, `Pool`, bounded pool admission, `with_timeout()`, and +`every()`. The default sizes are intentionally modest. They can be changed with `XASYNC_BENCH_GROUP_ROUNDS`, `XASYNC_BENCH_GROUP_JOBS`, `XASYNC_BENCH_TASK_ROUNDS`, `XASYNC_BENCH_POOL_JOBS`, -`XASYNC_BENCH_POOL_WORKERS`, `XASYNC_BENCH_TIMEOUT_ROUNDS`, +`XASYNC_BENCH_POOL_WORKERS`, `XASYNC_BENCH_POOL_BOUNDED_ROUNDS`, +`XASYNC_BENCH_POOL_BOUNDED_TIMEOUT_MS`, `XASYNC_BENCH_TIMEOUT_ROUNDS`, `XASYNC_BENCH_EVERY_ITERATIONS`, and `XASYNC_BENCH_EVERY_INTERVAL_MS`. Benchmark output is local diagnostic data, not a portable performance claim. diff --git a/vlib/x/async/benchmarks/README.md b/vlib/x/async/benchmarks/README.md index 8a8507688..913d4e7d7 100644 --- a/vlib/x/async/benchmarks/README.md +++ b/vlib/x/async/benchmarks/README.md @@ -7,7 +7,7 @@ claims. ## Available benchmark - `async_benchmark.v`: measures short default runs for `Group`, `Task[T]`, - `Pool`, `with_timeout()`, and `every()`. + `Pool`, bounded pool admission, `with_timeout()`, and `every()`. - `run_async_benchmark.sh`: builds and runs the benchmark with the local `./v`, isolated `VTMP`/`VCACHE`, and a temporary output binary. @@ -32,9 +32,15 @@ The defaults are intentionally small. Tune them locally with: - `XASYNC_BENCH_TASK_ROUNDS` - `XASYNC_BENCH_POOL_JOBS` - `XASYNC_BENCH_POOL_WORKERS` +- `XASYNC_BENCH_POOL_BOUNDED_ROUNDS` +- `XASYNC_BENCH_POOL_BOUNDED_TIMEOUT_MS` - `XASYNC_BENCH_TIMEOUT_ROUNDS` - `XASYNC_BENCH_EVERY_ITERATIONS` - `XASYNC_BENCH_EVERY_INTERVAL_MS` +The bounded pool admission case exercises `submit_with_timeout()` while the +pool is full and `submit_with_context()` after capacity opens. The timeout only +bounds admission; the benchmark does not make performance assertions. + Do not commit machine-specific benchmark results as permanent truth. Tests remain the authority for functional and concurrency-safety validation. diff --git a/vlib/x/async/benchmarks/async_benchmark.v b/vlib/x/async/benchmarks/async_benchmark.v index 6d1ba991d..91d97db1c 100644 --- a/vlib/x/async/benchmarks/async_benchmark.v +++ b/vlib/x/async/benchmarks/async_benchmark.v @@ -9,6 +9,8 @@ const default_group_jobs = 16 const default_task_rounds = 32 const default_pool_jobs = 64 const default_pool_workers = 4 +const default_pool_bounded_rounds = 8 +const default_pool_bounded_timeout_ms = 2 const default_timeout_rounds = 32 const default_every_iterations = 5 const default_every_interval_ms = 1 @@ -26,6 +28,10 @@ fn run_benchmarks() ! { task_rounds := env_int('XASYNC_BENCH_TASK_ROUNDS', default_task_rounds, 1, 500) pool_jobs := env_int('XASYNC_BENCH_POOL_JOBS', default_pool_jobs, 1, 1000) pool_workers := env_int('XASYNC_BENCH_POOL_WORKERS', default_pool_workers, 1, 64) + pool_bounded_rounds := env_int('XASYNC_BENCH_POOL_BOUNDED_ROUNDS', default_pool_bounded_rounds, + 1, 100) + pool_bounded_timeout_ms := env_int('XASYNC_BENCH_POOL_BOUNDED_TIMEOUT_MS', + default_pool_bounded_timeout_ms, 1, 100) timeout_rounds := env_int('XASYNC_BENCH_TIMEOUT_ROUNDS', default_timeout_rounds, 1, 500) every_iterations := env_int('XASYNC_BENCH_EVERY_ITERATIONS', default_every_iterations, 1, 100) every_interval_ms := @@ -45,6 +51,9 @@ fn run_benchmarks() ! { checksum += bench_pool(pool_jobs, pool_workers)! b.measure('Pool: jobs=${pool_jobs}, workers=${pool_workers}, checksum=${checksum}') + checksum += bench_pool_bounded_admission(pool_bounded_rounds, pool_bounded_timeout_ms)! + b.measure('Pool bounded admission: rounds=${pool_bounded_rounds}, timeout_ms=${pool_bounded_timeout_ms}, checksum=${checksum}') + checksum += bench_timeout(timeout_rounds)! b.measure('with_timeout: rounds=${timeout_rounds}, checksum=${checksum}') @@ -115,6 +124,69 @@ fn bench_pool(jobs int, workers int) !int { return total } +fn bench_pool_bounded_admission(rounds int, timeout_ms int) !int { + mut total := 0 + admission_timeout := time.Duration(timeout_ms) * time.millisecond + for _ in 0 .. rounds { + mut pool := xasync.new_pool(workers: 1, queue_size: 1)! + started := chan bool{cap: 2} + release := chan bool{cap: 3} + finished := chan bool{cap: 2} + attempting := chan bool{cap: 1} + accepted := chan bool{cap: 1} + ran := chan int{cap: 1} + blocking_job := fn [started, release, finished] (mut ctx context.Context) ! { + _ = ctx + started <- true + _ := <-release + finished <- true + } + + pool.try_submit(blocking_job)! + wait_for_benchmark_bool(started, 'bounded admission pool job did not start')! + pool.try_submit(blocking_job)! + + mut timed_out := false + pool.submit_with_timeout(admission_timeout, fn (mut ctx context.Context) ! { + _ = ctx + }) or { + if err.msg() != 'async: timeout' { + return err + } + timed_out = true + } + if !timed_out { + release <- true + release <- true + pool.close()! + return error('submit_with_timeout accepted while the pool was full') + } + + submit_thread := spawn fn [mut pool, attempting, accepted, ran] () { + attempting <- true + pool.submit_with_context(context.background(), fn [ran] (mut ctx context.Context) ! { + _ = ctx + ran <- 1 + }) or { + accepted <- false + return + } + accepted <- true + }() + wait_for_benchmark_bool(attempting, 'bounded admission submitter did not start')! + + release <- true + wait_for_benchmark_bool(finished, 'bounded admission first job did not finish')! + wait_for_benchmark_bool(accepted, + 'submit_with_context did not accept after capacity opened')! + release <- true + total += read_benchmark_int(ran, 'bounded admission accepted job did not run')! + pool.close()! + submit_thread.wait() + } + return total +} + fn bench_timeout(rounds int) !int { mut total := 0 for _ in 0 .. rounds { @@ -171,3 +243,28 @@ fn bench_every(iterations int, interval_ms int) !int { worker.wait() return total } + +fn wait_for_benchmark_bool(signal chan bool, message string) ! { + select { + ok := <-signal { + if !ok { + return error(message) + } + } + 1 * time.second { + return error(message) + } + } +} + +fn read_benchmark_int(signal chan int, message string) !int { + select { + value := <-signal { + return value + } + 1 * time.second { + return error(message) + } + } + return 0 +} diff --git a/vlib/x/async/benchmarks/run_async_benchmark.sh b/vlib/x/async/benchmarks/run_async_benchmark.sh index 6803c5b7b..92c2e19a2 100755 --- a/vlib/x/async/benchmarks/run_async_benchmark.sh +++ b/vlib/x/async/benchmarks/run_async_benchmark.sh @@ -22,10 +22,8 @@ out="$tmp_root/out/async_benchmark" mkdir -p "$vtmp" "$vcache" "$(dirname "$out")" echo "Running x.async benchmark with isolated VTMP and VCACHE." -echo "Tune sizes with XASYNC_BENCH_GROUP_ROUNDS, XASYNC_BENCH_GROUP_JOBS," -echo "XASYNC_BENCH_TASK_ROUNDS, XASYNC_BENCH_POOL_JOBS," -echo "XASYNC_BENCH_POOL_WORKERS, XASYNC_BENCH_TIMEOUT_ROUNDS," -echo "XASYNC_BENCH_EVERY_ITERATIONS, and XASYNC_BENCH_EVERY_INTERVAL_MS." +echo "Tune sizes with XASYNC_BENCH_* environment variables." +echo "See vlib/x/async/benchmarks/README.md for the full list." if command -v timeout >/dev/null 2>&1; then timeout 180s env VTMP="$vtmp" VCACHE="$vcache" ./v -prod -o "$out" run vlib/x/async/benchmarks/async_benchmark.v diff --git a/vlib/x/async/errors.v b/vlib/x/async/errors.v index e0428c206..babd11a01 100644 --- a/vlib/x/async/errors.v +++ b/vlib/x/async/errors.v @@ -4,9 +4,11 @@ module async // callers and tests. Keep them short, stable, and explicit about whether the // failure came from x.async or from the parent context. const err_group_go_after_wait = 'async: group does not accept new tasks after wait starts' +const err_group_max_errors_invalid = 'async: group max_errors must be positive' const err_group_wait_called = 'async: group wait was already called' const err_interval_invalid = 'async: interval must be positive' const err_nil_job = 'async: job function is nil' +const err_periodic_wait_called = 'async: periodic wait was already called' const err_pool_closed = 'async: pool is closed' const err_pool_queue_full = 'async: pool queue is full' const err_pool_queue_size_invalid = 'async: pool queue size must be positive' diff --git a/vlib/x/async/examples/README.md b/vlib/x/async/examples/README.md index 254a80ec1..d37708ce9 100644 --- a/vlib/x/async/examples/README.md +++ b/vlib/x/async/examples/README.md @@ -23,6 +23,8 @@ validation script to avoid bit rot, but the regression guarantees live in the cancellation. - `basic_task.v`: one value-returning task consumed with `wait()`. - `worker_pool.v`: fixed concurrency and explicit `try_submit()` backpressure. +- `bounded_pool_submit.v`: wait for pool admission with a timeout or context + while jobs keep using the pool context. - `periodic.v`: a blocking `every()` loop stopped by context cancellation. - `timeout.v`: one cooperative job bounded by `with_timeout()`. - `net_http/`: synthetic `net.http` request/response processing with `Pool`. @@ -39,6 +41,7 @@ From the repository root: ./v run vlib/x/async/examples/basic_group.v ./v run vlib/x/async/examples/basic_task.v ./v run vlib/x/async/examples/worker_pool.v +./v run vlib/x/async/examples/bounded_pool_submit.v ./v run vlib/x/async/examples/periodic.v ./v run vlib/x/async/examples/timeout.v ./v run vlib/x/async/examples/net_http/request_batch.v diff --git a/vlib/x/async/examples/bounded_pool_submit.v b/vlib/x/async/examples/bounded_pool_submit.v new file mode 100644 index 000000000..5987c0af9 --- /dev/null +++ b/vlib/x/async/examples/bounded_pool_submit.v @@ -0,0 +1,77 @@ +import context +import time +import x.async as xasync + +fn main() { + mut pool := xasync.new_pool(workers: 1, queue_size: 1)! + started := chan bool{cap: 2} + release := chan bool{cap: 2} + accepted := chan bool{cap: 1} + ran := chan bool{cap: 1} + context_ran := chan bool{cap: 1} + blocking_job := fn [started, release] (mut ctx context.Context) ! { + _ = ctx + started <- true + _ := <-release + } + + pool.try_submit(blocking_job)! + if !wait_bool(started) { + eprintln('first pool job did not start') + exit(1) + } + pool.try_submit(blocking_job)! + + submitter := spawn fn [mut pool, accepted, ran] () { + pool.submit_with_timeout(1 * time.second, fn [ran] (mut ctx context.Context) ! { + _ = ctx + ran <- true + }) or { + eprintln('bounded submit failed: ${err.msg()}') + accepted <- false + return + } + accepted <- true + }() + + release <- true + if !wait_bool(accepted) { + eprintln('bounded submit was not accepted after capacity opened') + exit(1) + } + + release <- true + if !wait_bool(ran) { + eprintln('bounded submit job did not run') + exit(1) + } + + parent_ctx, cancel := xasync.with_cancel() + defer { + cancel() + } + pool.submit_with_context(parent_ctx, fn [context_ran] (mut ctx context.Context) ! { + _ = ctx + context_ran <- true + })! + if !wait_bool(context_ran) { + eprintln('context-bounded submit job did not run') + exit(1) + } + + pool.close()! + submitter.wait() + println('timeout and context bounded submits completed') +} + +fn wait_bool(ch chan bool) bool { + select { + ok := <-ch { + return ok + } + 1 * time.second { + return false + } + } + return false +} diff --git a/vlib/x/async/group.v b/vlib/x/async/group.v index 3cba8ab68..cdd480cb8 100644 --- a/vlib/x/async/group.v +++ b/vlib/x/async/group.v @@ -3,6 +3,14 @@ module async import context import sync +// GroupConfig enables optional Group behavior while preserving new_group() defaults. +@[params] +pub struct GroupConfig { +pub: + collect_errors bool + max_errors int +} + // Group coordinates a set of related concurrent jobs. // // Jobs share a derived context. The first job error is returned by wait() and @@ -19,6 +27,11 @@ mut: // Stored once. Later job errors never replace the first failure observed by // the group, even when several jobs fail concurrently after cancellation. first_err IError = none + // Optional bounded collection of observed job errors. It does not change + // wait(), which still returns only first_err. + collect_errors bool + max_errors int + collected_errors []IError // Set before wait() calls wg.wait(). Once true, no further jobs are accepted. waiting bool } @@ -29,12 +42,26 @@ mut: // derived context is owned by the group and canceled on first job error or when // wait() completes. pub fn new_group(parent context.Context) &Group { + return new_group_internal(parent, GroupConfig{}) +} + +// new_group_with_config creates a Group with opt-in bounded error collection. +pub fn new_group_with_config(parent context.Context, config GroupConfig) !&Group { + if config.collect_errors && config.max_errors <= 0 { + return error(err_group_max_errors_invalid) + } + return new_group_internal(parent, config) +} + +fn new_group_internal(parent context.Context, config GroupConfig) &Group { ctx, cancel := new_cancel_context(parent) return &Group{ - ctx: context.Context(ctx) - cancel: cancel - wg: sync.new_waitgroup() - mutex: sync.new_mutex() + ctx: context.Context(ctx) + cancel: cancel + wg: sync.new_waitgroup() + mutex: sync.new_mutex() + collect_errors: config.collect_errors + max_errors: config.max_errors } } @@ -81,6 +108,17 @@ pub fn (mut g Group) wait() ! { } } +// errors returns a snapshot of collected job errors. +// +// Collection is opt-in through new_group_with_config(). The order of concurrently +// observed errors is not guaranteed. +pub fn (mut g Group) errors() []IError { + g.mutex.lock() + errs := g.collected_errors.clone() + g.mutex.unlock() + return errs +} + fn run_group_job(mut g Group, f JobFn) { defer { g.wg.done() @@ -98,6 +136,9 @@ fn (mut g Group) set_first_error(err IError) { g.first_err = err should_cancel = true } + if g.collect_errors && g.collected_errors.len < g.max_errors { + g.collected_errors << err + } g.mutex.unlock() if should_cancel { // Cancel outside the group mutex. context cancellation can notify child diff --git a/vlib/x/async/group_test.v b/vlib/x/async/group_test.v index daf6297bb..07c882f09 100644 --- a/vlib/x/async/group_test.v +++ b/vlib/x/async/group_test.v @@ -46,6 +46,28 @@ fn test_group_error_returns_first_error() { assert false } +fn test_group_default_does_not_collect_errors() { + mut group := xasync.new_group(context.background()) + group.go(fn (mut ctx context.Context) ! { + _ = ctx + return error('group failed') + })! + group.wait() or { + assert err.msg() == 'group failed' + assert group.errors().len == 0 + return + } + assert false +} + +fn test_group_rejects_invalid_error_collection_config() { + xasync.new_group_with_config(context.background(), collect_errors: true, max_errors: 0) or { + assert err.msg() == 'async: group max_errors must be positive' + return + } + assert false +} + fn test_group_wait_without_tasks() { parent := context.background() mut group := xasync.new_group(parent) @@ -142,6 +164,102 @@ fn test_group_first_error_remains_stable_with_concurrent_secondary_errors() { assert false } +fn test_group_collects_errors_without_changing_wait_error() { + mut group := xasync.new_group_with_config(context.background(), + collect_errors: true + max_errors: 4 + )! + secondary_ready := chan bool{cap: 3} + for i in 0 .. 3 { + group.go(fn [secondary_ready, i] (mut ctx context.Context) ! { + secondary_ready <- true + done := ctx.done() + select { + _ := <-done { + return error('secondary collected error ${i}') + } + 1 * time.second { + return error('secondary timeout ${i}') + } + } + })! + } + for _ in 0 .. 3 { + assert <-secondary_ready + } + group.go(fn (mut ctx context.Context) ! { + _ = ctx + return error('primary collected error') + })! + group.wait() or { + assert err.msg() == 'primary collected error' + errs := group.errors() + assert errs.len == 4 + assert error_messages_contain(errs, 'primary collected error') + return + } + assert false +} + +fn test_group_error_collection_is_bounded_and_includes_first_error() { + mut group := xasync.new_group_with_config(context.background(), + collect_errors: true + max_errors: 2 + )! + secondary_ready := chan bool{cap: 4} + for i in 0 .. 4 { + group.go(fn [secondary_ready, i] (mut ctx context.Context) ! { + secondary_ready <- true + done := ctx.done() + select { + _ := <-done { + return error('bounded secondary error ${i}') + } + 1 * time.second { + return error('bounded secondary timeout ${i}') + } + } + })! + } + for _ in 0 .. 4 { + assert <-secondary_ready + } + group.go(fn (mut ctx context.Context) ! { + _ = ctx + return error('bounded primary error') + })! + group.wait() or { + assert err.msg() == 'bounded primary error' + errs := group.errors() + assert errs.len == 2 + assert error_messages_contain(errs, 'bounded primary error') + return + } + assert false +} + +fn test_group_errors_returns_snapshot_copy() { + mut group := xasync.new_group_with_config(context.background(), + collect_errors: true + max_errors: 1 + )! + group.go(fn (mut ctx context.Context) ! { + _ = ctx + return error('snapshot source error') + })! + group.wait() or { + assert err.msg() == 'snapshot source error' + mut snapshot := group.errors() + assert snapshot.len == 1 + snapshot[0] = error('mutated snapshot error') + second_snapshot := group.errors() + assert second_snapshot.len == 1 + assert second_snapshot[0].msg() == 'snapshot source error' + return + } + assert false +} + fn test_group_many_short_jobs_return_first_error() { jobs := 64 mut group := xasync.new_group(context.background()) @@ -209,3 +327,12 @@ fn test_group_parent_cancellation_is_observed_by_cooperative_job() { } } } + +fn error_messages_contain(errs []IError, msg string) bool { + for err in errs { + if err.msg() == msg { + return true + } + } + return false +} diff --git a/vlib/x/async/periodic.v b/vlib/x/async/periodic.v index 560cfce29..fbba69318 100644 --- a/vlib/x/async/periodic.v +++ b/vlib/x/async/periodic.v @@ -1,8 +1,37 @@ module async import context +import sync import time +enum PeriodicResultKind { + completed + job_error + context_error +} + +struct PeriodicResult { + kind PeriodicResultKind + err IError = none +} + +// PeriodicHandle owns one detached periodic loop started by start_every(). +// +// The handle exposes only explicit lifecycle operations: stop() requests +// cooperative shutdown, and wait() consumes the loop result once. +@[heap] +pub struct PeriodicHandle { +mut: + parent context.Context + ctx context.Context + cancel context.CancelFn = unsafe { nil } + result_ch chan PeriodicResult + mutex &sync.Mutex = sync.new_mutex() + stopped bool + stop_owns_cancel bool + waited bool +} + // every runs f repeatedly with interval spacing until ctx is canceled or f fails. // // This helper is intentionally blocking: it does not start a hidden scheduler or @@ -59,6 +88,81 @@ pub fn every(parent context.Context, interval time.Duration, f JobFn) ! { } } +// start_every starts f in one detached periodic loop and returns its lifecycle handle. +// +// The first iteration runs after one interval. Iterations never overlap; a slow +// job delays the next interval. The returned handle must be stopped and waited +// on by its owner, otherwise the detached loop can keep running. +pub fn start_every(parent context.Context, interval time.Duration, f JobFn) !&PeriodicHandle { + if interval.nanoseconds() <= 0 { + return error(err_interval_invalid) + } + if f == unsafe { nil } { + return error(err_nil_job) + } + + mut parent_ctx := parent + initial_err := parent_ctx.err() + if initial_err !is none { + return initial_err + } + + ctx, cancel := new_cancel_context(parent) + mut handle := &PeriodicHandle{ + parent: parent + ctx: context.Context(ctx) + cancel: cancel + result_ch: chan PeriodicResult{cap: 1} + mutex: sync.new_mutex() + } + spawn run_periodic_handle(mut handle, interval, f) + return handle +} + +// stop requests cooperative shutdown of the detached periodic loop. +// +// stop is idempotent and non-blocking. A running job still has to return +// naturally; wait() is responsible for observing the final result. +pub fn (mut h PeriodicHandle) stop() { + mut parent := h.parent + parent_err := parent.err() + h.mutex.lock() + if h.stopped { + h.mutex.unlock() + return + } + h.stopped = true + h.stop_owns_cancel = parent_err is none + should_cancel := h.stop_owns_cancel + cancel := h.cancel + h.mutex.unlock() + if should_cancel { + cancel() + } +} + +// wait blocks until the detached periodic loop exits. +// +// wait is one-shot. It returns job errors and parent cancellation errors. A +// normal stop() request is reported as successful completion. +pub fn (mut h PeriodicHandle) wait() ! { + h.mutex.lock() + if h.waited { + h.mutex.unlock() + return error(err_periodic_wait_called) + } + h.waited = true + h.mutex.unlock() + + result := <-h.result_ch + if result.err !is none { + if result.kind == .context_error && h.is_stop_result() { + return + } + return result.err + } +} + fn run_periodic_iteration(mut ctx context.Context, f JobFn) ! { f(mut ctx)! err := ctx.err() @@ -66,3 +170,80 @@ fn run_periodic_iteration(mut ctx context.Context, f JobFn) ! { return err } } + +fn run_periodic_handle(mut h PeriodicHandle, interval time.Duration, f JobFn) { + mut ctx := h.ctx + result := run_detached_periodic_loop(mut ctx, interval, f) + h.cancel() + h.result_ch <- result +} + +fn run_detached_periodic_loop(mut ctx context.Context, interval time.Duration, f JobFn) PeriodicResult { + done_ch := ctx.done() + mut watch_done := true + select { + _ := <-done_ch { + err := ctx.err() + if err !is none { + return PeriodicResult{ + kind: .context_error + err: err + } + } + watch_done = false + } + else {} + } + + for { + if watch_done { + select { + _ := <-done_ch { + err := ctx.err() + if err !is none { + return PeriodicResult{ + kind: .context_error + err: err + } + } + watch_done = false + } + interval { + result := run_detached_periodic_iteration(mut ctx, f) + if result.err !is none { + return result + } + } + } + } else { + time.sleep(interval) + result := run_detached_periodic_iteration(mut ctx, f) + if result.err !is none { + return result + } + } + } + return PeriodicResult{} +} + +fn run_detached_periodic_iteration(mut ctx context.Context, f JobFn) PeriodicResult { + f(mut ctx) or { return PeriodicResult{ + kind: .job_error + err: err + } } + err := ctx.err() + if err !is none { + return PeriodicResult{ + kind: .context_error + err: err + } + } + return PeriodicResult{} +} + +fn (mut h PeriodicHandle) is_stop_result() bool { + h.mutex.lock() + stop_owns_cancel := h.stop_owns_cancel + h.mutex.unlock() + return stop_owns_cancel +} diff --git a/vlib/x/async/periodic_test.v b/vlib/x/async/periodic_test.v index eaeef786d..a5008d529 100644 --- a/vlib/x/async/periodic_test.v +++ b/vlib/x/async/periodic_test.v @@ -202,6 +202,205 @@ fn test_every_does_not_overlap_iterations() { worker.wait() } +fn test_start_every_rejects_invalid_interval() { + xasync.start_every(context.background(), 0 * time.millisecond, fn (mut ctx context.Context) ! { + _ = ctx + }) or { + assert err.msg() == 'async: interval must be positive' + return + } + assert false +} + +fn test_start_every_rejects_negative_interval() { + xasync.start_every(context.background(), -1 * time.millisecond, fn (mut ctx context.Context) ! { + _ = ctx + }) or { + assert err.msg() == 'async: interval must be positive' + return + } + assert false +} + +fn test_start_every_rejects_nil_job() { + nil_job := unsafe { xasync.JobFn(nil) } + xasync.start_every(context.background(), 1 * time.second, nil_job) or { + assert err.msg() == 'async: job function is nil' + return + } + assert false +} + +fn test_start_every_returns_immediately_when_parent_is_already_canceled() { + parent_ctx, cancel := xasync.with_cancel() + cancel() + xasync.start_every(parent_ctx, 1 * time.second, fn (mut ctx context.Context) ! { + _ = ctx + }) or { + assert err.msg() == 'context canceled' + return + } + assert false +} + +fn test_periodic_handle_stop_before_first_iteration() { + ran := chan bool{cap: 1} + mut handle := xasync.start_every(context.background(), 1 * time.second, fn [ran] (mut ctx context.Context) ! { + _ = ctx + ran <- true + })! + handle.stop() + handle.wait()! + assert_no_periodic_signal(ran, 'periodic handle ran before first interval after stop') +} + +fn test_periodic_handle_stop_between_ticks() { + ticks := chan bool{cap: 2} + mut handle := xasync.start_every(context.background(), 5 * time.millisecond, fn [ticks] (mut ctx context.Context) ! { + _ = ctx + ticks <- true + })! + wait_for_periodic_entry(ticks) + handle.stop() + handle.wait()! + assert_no_periodic_signal(ticks, 'periodic handle ticked again after stop and wait') +} + +fn test_periodic_handle_stop_during_long_job_waits_for_job_to_return() { + entered := chan bool{cap: 1} + release := chan bool{cap: 1} + waited := chan bool{cap: 1} + mut handle := xasync.start_every(context.background(), 5 * time.millisecond, fn [entered, release] (mut ctx context.Context) ! { + _ = ctx + entered <- true + _ := <-release + })! + wait_for_periodic_entry(entered) + handle.stop() + + wait_thread := spawn fn [mut handle, waited] () { + handle.wait() or { + waited <- false + return + } + waited <- true + }() + assert_no_periodic_signal(waited, 'periodic handle wait returned before long job completed') + release <- true + wait_for_periodic_entry(waited) + wait_thread.wait() +} + +fn test_periodic_handle_wait_returns_job_error() { + mut handle := xasync.start_every(context.background(), 5 * time.millisecond, fn (mut ctx context.Context) ! { + _ = ctx + return error('periodic handle failed') + })! + handle.wait() or { + assert err.msg() == 'periodic handle failed' + return + } + assert false +} + +fn test_periodic_handle_wait_returns_job_context_canceled_error_after_stop() { + entered := chan bool{cap: 1} + release := chan bool{cap: 1} + mut handle := xasync.start_every(context.background(), 5 * time.millisecond, fn [entered, release] (mut ctx context.Context) ! { + _ = ctx + entered <- true + _ := <-release + return error('context canceled') + })! + wait_for_periodic_entry(entered) + handle.stop() + release <- true + handle.wait() or { + assert err.msg() == 'context canceled' + return + } + assert false, 'job-returned context canceled error was treated as normal stop' +} + +fn test_periodic_handle_wait_returns_parent_cancellation() { + parent_ctx, cancel := xasync.with_cancel() + mut handle := xasync.start_every(parent_ctx, 1 * time.second, fn (mut ctx context.Context) ! { + _ = ctx + })! + cancel() + handle.wait() or { + assert err.msg() == 'context canceled' + return + } + assert false +} + +fn test_periodic_handle_stop_after_parent_cancel_keeps_parent_error() { + parent_ctx, cancel := xasync.with_cancel() + mut handle := xasync.start_every(parent_ctx, 1 * time.second, fn (mut ctx context.Context) ! { + _ = ctx + })! + cancel() + handle.stop() + handle.wait() or { + assert err.msg() == 'context canceled' + return + } + assert false +} + +fn test_periodic_handle_does_not_overlap_iterations() { + active := chan bool{cap: 1} + active <- true + entered := chan bool{cap: 2} + release := chan bool{cap: 2} + overlap := chan bool{cap: 1} + mut handle := xasync.start_every(context.background(), 5 * time.millisecond, fn [active, entered, release, overlap] (mut ctx context.Context) ! { + select { + _ := <-active {} + else { + overlap <- true + return error('periodic handle overlap') + } + } + entered <- true + _ = ctx + _ := <-release + active <- true + })! + + wait_for_periodic_entry(entered) + assert_no_periodic_signal(entered, + 'periodic handle iterations overlapped while first job was still running') + select { + did_overlap := <-overlap { + assert !did_overlap + } + else {} + } + + release <- true + wait_for_periodic_entry(entered) + handle.stop() + release <- true + handle.wait()! + assert_no_periodic_signal(entered, 'periodic handle started another iteration after stop') +} + +fn test_periodic_handle_double_stop_and_wait() { + mut handle := xasync.start_every(context.background(), 1 * time.second, fn (mut ctx context.Context) ! { + _ = ctx + })! + handle.stop() + handle.stop() + handle.wait()! + handle.wait() or { + assert err.msg() == 'async: periodic wait was already called' + return + } + assert false +} + fn wait_for_periodic_entry(entered chan bool) { select { did_enter := <-entered { @@ -212,3 +411,12 @@ fn wait_for_periodic_entry(entered chan bool) { } } } + +fn assert_no_periodic_signal(signal chan bool, message string) { + select { + _ := <-signal { + assert false, message + } + 50 * time.millisecond {} + } +} diff --git a/vlib/x/async/pool.v b/vlib/x/async/pool.v index 9bed5f09d..28e6bcd09 100644 --- a/vlib/x/async/pool.v +++ b/vlib/x/async/pool.v @@ -2,6 +2,7 @@ module async import context import sync +import time // PoolConfig configures a fixed-size concurrency pool. // @@ -23,11 +24,12 @@ pub: @[heap] pub struct Pool { mut: - ctx context.Context - cancel context.CancelFn = unsafe { nil } - tokens chan bool - wg &sync.WaitGroup = sync.new_waitgroup() - max_jobs int + ctx context.Context + cancel context.CancelFn = unsafe { nil } + tokens chan bool + submit_ready chan int + wg &sync.WaitGroup = sync.new_waitgroup() + max_jobs int // Protects lifecycle flags, accepted job count, WaitGroup.add(), and // first_err. This lock is never held while a user JobFn runs. mutex &sync.Mutex = sync.new_mutex() @@ -55,12 +57,13 @@ pub fn new_pool_with_context(parent context.Context, config PoolConfig) !&Pool { } ctx, cancel := new_cancel_context(parent) mut pool := &Pool{ - ctx: context.Context(ctx) - cancel: cancel - tokens: chan bool{cap: config.workers} - wg: sync.new_waitgroup() - max_jobs: config.workers + config.queue_size - mutex: sync.new_mutex() + ctx: context.Context(ctx) + cancel: cancel + tokens: chan bool{cap: config.workers} + submit_ready: chan int{cap: 1} + wg: sync.new_waitgroup() + max_jobs: config.workers + config.queue_size + mutex: sync.new_mutex() } for _ in 0 .. config.workers { pool.tokens <- true @@ -85,10 +88,7 @@ pub fn (mut p Pool) try_submit(f JobFn) ! { p.mutex.unlock() return error(err_pool_queue_full) } - p.accepted++ - // add() is protected by the same mutex as wait(), so callers cannot race an - // accepted job against pool shutdown. - p.wg.add(1) + p.accept_job_locked() p.mutex.unlock() // The JobFn is passed directly to the spawned wrapper instead of being // stored in a channel. That keeps closure ownership with V's normal spawn @@ -96,6 +96,89 @@ pub fn (mut p Pool) try_submit(f JobFn) ! { spawn run_pool_job(mut p, f) } +// submit_with_context waits until f can be accepted, parent is canceled, or the pool closes. +// +// The parent context bounds admission only. Once f is accepted, the job runs with +// the pool's own context, matching try_submit() and preserving pool lifecycle ownership. +pub fn (mut p Pool) submit_with_context(parent context.Context, f JobFn) ! { + if f == unsafe { nil } { + return error(err_nil_job) + } + mut submit_ctx := parent + initial_err := submit_ctx.err() + if initial_err !is none { + return initial_err + } + done := submit_ctx.done() + mut watch_done := true + select { + _ := <-done { + err := submit_ctx.err() + if err !is none { + return err + } + watch_done = false + } + else {} + } + for { + p.mutex.lock() + if p.closed { + p.mutex.unlock() + return error(err_pool_closed) + } + if watch_done { + err := submit_ctx.err() + if err !is none { + p.mutex.unlock() + return err + } + } + if p.accepted < p.max_jobs { + p.accept_job_locked() + p.mutex.unlock() + spawn run_pool_job(mut p, f) + return + } + submit_ready := p.submit_ready + p.mutex.unlock() + if !watch_done { + _ := <-submit_ready + continue + } + select { + _ := <-submit_ready { + continue + } + _ := <-done { + err := submit_ctx.err() + if err !is none { + return err + } + watch_done = false + continue + } + } + } +} + +// submit_with_timeout waits up to timeout for f to be accepted by the pool. +// +// The timeout bounds admission only. Accepted jobs still receive the pool context. +pub fn (mut p Pool) submit_with_timeout(timeout time.Duration, f JobFn) ! { + timeout_ctx, cancel := new_timeout_context(context.background(), timeout) + mut submit_ctx := context.Context(timeout_ctx) + defer { + cancel() + } + p.submit_with_context(submit_ctx, f) or { + if err.msg() == context_deadline_exceeded && timeout_ctx.was_canceled_by_timeout() { + return error(err_timeout) + } + return err + } +} + // wait closes the pool to new submissions, drains accepted jobs, and waits for completion. // // wait is one-shot. It returns the first job error observed by any accepted job. @@ -107,6 +190,7 @@ pub fn (mut p Pool) wait() ! { } p.waited = true p.closed = true + p.wake_submitters_locked() p.mutex.unlock() p.wg.wait() @@ -142,9 +226,26 @@ fn run_pool_job(mut p Pool, f JobFn) { fn (mut p Pool) finish_accepted_job() { p.mutex.lock() p.accepted-- + p.wake_submitters_locked() p.mutex.unlock() } +fn (mut p Pool) accept_job_locked() { + p.accepted++ + // add() is protected by the same mutex as wait(), so callers cannot race an + // accepted job against pool shutdown. + p.wg.add(1) +} + +fn (mut p Pool) wake_submitters_locked() { + if !p.submit_ready.closed { + p.submit_ready.close() + } + if !p.closed { + p.submit_ready = chan int{cap: 1} + } +} + fn (mut p Pool) set_first_error(err IError) { p.mutex.lock() if p.first_err is none { diff --git a/vlib/x/async/pool_test.v b/vlib/x/async/pool_test.v index a9a62cf62..1c465f436 100644 --- a/vlib/x/async/pool_test.v +++ b/vlib/x/async/pool_test.v @@ -68,6 +68,210 @@ fn test_pool_queue_full_returns_backpressure_error() { assert false } +fn test_pool_submit_with_context_waits_until_capacity_is_available() { + mut pool := xasync.new_pool(workers: 1, queue_size: 1)! + started := chan bool{cap: 2} + release := chan bool{cap: 2} + attempting := chan bool{cap: 1} + accepted := chan bool{cap: 1} + ran := chan bool{cap: 1} + blocking_job := fn [started, release] (mut ctx context.Context) ! { + _ = ctx + started <- true + _ := <-release + } + + pool.try_submit(blocking_job)! + wait_for_bool_signal(started, 'first blocking pool job did not start') + pool.try_submit(blocking_job)! + + submit_thread := spawn fn [mut pool, attempting, accepted, ran] () { + attempting <- true + pool.submit_with_context(context.background(), fn [ran] (mut ctx context.Context) ! { + _ = ctx + ran <- true + }) or { + accepted <- false + return + } + accepted <- true + }() + wait_for_bool_signal(attempting, 'bounded submit thread did not start') + select { + _ := <-accepted { + assert false, 'bounded submit returned while pool was full' + } + 50 * time.millisecond {} + } + + release <- true + wait_for_bool_signal(accepted, 'bounded submit did not accept after capacity opened') + release <- true + wait_for_bool_signal(ran, 'accepted bounded submit job did not run') + pool.close()! + submit_thread.wait() +} + +fn test_pool_submit_with_timeout_returns_timeout_without_leaking_acceptance() { + mut pool := xasync.new_pool(workers: 1, queue_size: 1)! + started := chan bool{cap: 2} + release := chan bool{cap: 2} + finished := chan bool{cap: 2} + blocking_job := fn [started, release, finished] (mut ctx context.Context) ! { + _ = ctx + started <- true + _ := <-release + finished <- true + } + + pool.try_submit(blocking_job)! + wait_for_bool_signal(started, 'first blocking pool job did not start') + pool.try_submit(blocking_job)! + + pool.submit_with_timeout(20 * time.millisecond, fn (mut ctx context.Context) ! { + _ = ctx + }) or { + assert err.msg() == 'async: timeout' + release <- true + wait_for_bool_signal(finished, 'first blocking pool job did not finish') + pool.submit_with_timeout(1 * time.second, fn (mut ctx context.Context) ! { + _ = ctx + })! + release <- true + pool.close()! + return + } + assert false +} + +fn test_pool_submit_with_context_parent_cancel_does_not_accept_job() { + mut pool := xasync.new_pool(workers: 1, queue_size: 1)! + started := chan bool{cap: 1} + release := chan bool{cap: 2} + attempting := chan bool{cap: 1} + would_run := chan bool{cap: 1} + result := chan string{cap: 1} + parent_ctx, cancel := xasync.with_cancel() + blocking_job := fn [started, release] (mut ctx context.Context) ! { + _ = ctx + started <- true + _ := <-release + } + + pool.try_submit(blocking_job)! + wait_for_bool_signal(started, 'blocking pool job did not start') + pool.try_submit(blocking_job)! + + submit_thread := spawn fn [mut pool, parent_ctx, attempting, would_run, result] () { + attempting <- true + pool.submit_with_context(parent_ctx, fn [would_run] (mut ctx context.Context) ! { + _ = ctx + would_run <- true + }) or { + result <- err.msg() + return + } + result <- 'accepted' + }() + wait_for_bool_signal(attempting, 'bounded submit thread did not start') + select { + msg := <-result { + assert false, 'bounded submit returned before parent cancellation: ${msg}' + } + 50 * time.millisecond {} + } + + cancel() + select { + msg := <-result { + assert msg == 'context canceled' + } + 1 * time.second { + assert false, 'bounded submit did not return after parent cancellation' + } + } + release <- true + release <- true + pool.close()! + assert_no_bool_signal(would_run, 'job was accepted after parent cancellation') + submit_thread.wait() +} + +fn test_pool_close_wakes_waiting_submitter_without_accepting_job() { + mut pool := xasync.new_pool(workers: 1, queue_size: 1)! + started := chan bool{cap: 1} + release := chan bool{cap: 2} + attempting := chan bool{cap: 1} + would_run := chan bool{cap: 1} + result := chan string{cap: 1} + closed := chan bool{cap: 1} + blocking_job := fn [started, release] (mut ctx context.Context) ! { + _ = ctx + started <- true + _ := <-release + } + + pool.try_submit(blocking_job)! + wait_for_bool_signal(started, 'blocking pool job did not start') + pool.try_submit(blocking_job)! + + submit_thread := spawn fn [mut pool, attempting, would_run, result] () { + attempting <- true + pool.submit_with_context(context.background(), fn [would_run] (mut ctx context.Context) ! { + _ = ctx + would_run <- true + }) or { + result <- err.msg() + return + } + result <- 'accepted' + }() + wait_for_bool_signal(attempting, 'bounded submit thread did not start') + select { + msg := <-result { + assert false, 'bounded submit returned before close: ${msg}' + } + 50 * time.millisecond {} + } + + close_thread := spawn fn [mut pool, closed] () { + pool.close() or { + closed <- false + return + } + closed <- true + }() + select { + msg := <-result { + assert msg == 'async: pool is closed' + } + 1 * time.second { + assert false, 'bounded submit did not return after pool close started' + } + } + release <- true + release <- true + wait_for_bool_signal(closed, 'pool close did not finish after accepted jobs were released') + assert_no_bool_signal(would_run, 'job was accepted after pool close started') + submit_thread.wait() + close_thread.wait() +} + +fn test_pool_bounded_submit_rejects_nil_job() { + mut pool := xasync.new_pool(workers: 1, queue_size: 1)! + nil_job := unsafe { xasync.JobFn(nil) } + pool.submit_with_context(context.background(), nil_job) or { + assert err.msg() == 'async: job function is nil' + pool.submit_with_timeout(20 * time.millisecond, nil_job) or { + assert err.msg() == 'async: job function is nil' + pool.close()! + return + } + assert false + } + assert false +} + fn test_pool_submit_after_close_is_refused() { mut pool := xasync.new_pool(workers: 1, queue_size: 1)! pool.close()! @@ -329,6 +533,142 @@ fn test_pool_short_stress_many_jobs() { } } +fn test_pool_bounded_submit_stress_waiting_submitters_are_released() { + workers := 2 + queue_size := 2 + initial_jobs := workers + queue_size + submitters := 8 + mut pool := xasync.new_pool(workers: workers, queue_size: queue_size)! + started := chan bool{cap: initial_jobs} + release := chan bool{cap: initial_jobs} + attempting := chan bool{cap: submitters} + accepted := chan int{cap: submitters} + ran := chan int{cap: submitters} + blocking_job := fn [started, release] (mut ctx context.Context) ! { + _ = ctx + started <- true + _ := <-release + } + + for _ in 0 .. initial_jobs { + pool.try_submit(blocking_job)! + } + for _ in 0 .. workers { + wait_for_bool_signal(started, 'initial pool worker did not start') + } + + mut submit_threads := []thread{} + for i in 0 .. submitters { + submit_threads << spawn fn [mut pool, attempting, accepted, ran, i] () { + attempting <- true + pool.submit_with_context(context.background(), fn [ran, i] (mut ctx context.Context) ! { + _ = ctx + ran <- i + }) or { + failed_index := -1 - i + accepted <- failed_index + return + } + accepted <- i + }() + } + for _ in 0 .. submitters { + wait_for_bool_signal(attempting, 'bounded submitter did not start') + } + + for _ in 0 .. initial_jobs { + release <- true + } + + mut seen_accepted := []bool{len: submitters} + for _ in 0 .. submitters { + i := wait_for_int_signal(accepted, 'bounded submitter was not accepted') + assert i >= 0 + assert i < submitters + assert !seen_accepted[i] + seen_accepted[i] = true + } + mut seen_ran := []bool{len: submitters} + for _ in 0 .. submitters { + i := wait_for_int_signal(ran, 'accepted bounded submitter job did not run') + assert i >= 0 + assert i < submitters + assert !seen_ran[i] + seen_ran[i] = true + } + + pool.close()! + for t in submit_threads { + t.wait() + } +} + +fn test_pool_bounded_submit_stress_waiting_submitters_rejected_on_close() { + workers := 2 + queue_size := 2 + initial_jobs := workers + queue_size + submitters := 8 + mut pool := xasync.new_pool(workers: workers, queue_size: queue_size)! + started := chan bool{cap: initial_jobs} + release := chan bool{cap: initial_jobs} + attempting := chan bool{cap: submitters} + result := chan string{cap: submitters} + ran := chan bool{cap: submitters} + closed := chan bool{cap: 1} + blocking_job := fn [started, release] (mut ctx context.Context) ! { + _ = ctx + started <- true + _ := <-release + } + + for _ in 0 .. initial_jobs { + pool.try_submit(blocking_job)! + } + for _ in 0 .. workers { + wait_for_bool_signal(started, 'initial pool worker did not start') + } + + mut submit_threads := []thread{} + for i in 0 .. submitters { + submit_threads << spawn fn [mut pool, attempting, result, ran, i] () { + attempting <- true + pool.submit_with_context(context.background(), fn [ran] (mut ctx context.Context) ! { + _ = ctx + ran <- true + }) or { + result <- err.msg() + return + } + result <- 'accepted ${i}' + }() + } + for _ in 0 .. submitters { + wait_for_bool_signal(attempting, 'bounded submitter did not start') + } + + close_thread := spawn fn [mut pool, closed] () { + pool.close() or { + closed <- false + return + } + closed <- true + }() + for _ in 0 .. submitters { + msg := wait_for_string_signal(result, 'bounded submitter did not return after close') + assert msg == 'async: pool is closed' + } + assert_no_bool_signal(ran, 'bounded submitter job ran after close started') + + for _ in 0 .. initial_jobs { + release <- true + } + wait_for_bool_signal(closed, 'pool close did not finish after releasing accepted jobs') + for t in submit_threads { + t.wait() + } + close_thread.wait() +} + fn wait_for_bool_signal(signal chan bool, message string) { select { ok := <-signal { @@ -340,6 +680,30 @@ fn wait_for_bool_signal(signal chan bool, message string) { } } +fn wait_for_int_signal(signal chan int, message string) int { + select { + value := <-signal { + return value + } + 1 * time.second { + assert false, message + } + } + return 0 +} + +fn wait_for_string_signal(signal chan string, message string) string { + select { + value := <-signal { + return value + } + 1 * time.second { + assert false, message + } + } + return '' +} + fn assert_no_bool_signal(signal chan bool, message string) { select { _ := <-signal { -- 2.39.5