From e80386b6af8a071a2e13fd654ebcdb2105fb9202 Mon Sep 17 00:00:00 2001 From: GGRei Date: Tue, 23 Jun 2026 10:17:44 +0200 Subject: [PATCH] executor: add owner-thread executor module (#27540) --- vlib/x/executor/README.md | 251 +++++++++++ vlib/x/executor/admission_test.v | 388 ++++++++++++++++++ vlib/x/executor/benchmarks/README.md | 40 ++ .../executor/benchmarks/executor_benchmark.v | 190 +++++++++ .../benchmarks/run_executor_benchmark.sh | 32 ++ vlib/x/executor/config_test.v | 28 ++ vlib/x/executor/errors.v | 17 + vlib/x/executor/errors_test.v | 133 ++++++ vlib/x/executor/examples/README.md | 45 ++ vlib/x/executor/examples/basic_post.v | 24 ++ vlib/x/executor/examples/bounded_admission.v | 30 ++ vlib/x/executor/examples/owner_loop_pump.v | 38 ++ vlib/x/executor/examples/run_sync.v | 38 ++ vlib/x/executor/examples/shutdown.v | 35 ++ .../examples/synthetic_ffi_owner_state.v | 36 ++ .../examples/synthetic_render_upload.v | 53 +++ vlib/x/executor/executor.v | 170 ++++++++ vlib/x/executor/lifecycle_test.v | 259 ++++++++++++ vlib/x/executor/owner.v | 44 ++ vlib/x/executor/owner_test.v | 64 +++ vlib/x/executor/pump_test.v | 88 ++++ vlib/x/executor/run.v | 178 ++++++++ vlib/x/executor/stress_test.v | 112 +++++ vlib/x/executor/submit.v | 170 ++++++++ vlib/x/executor/sync_call.v | 44 ++ vlib/x/executor/sync_call_test.v | 319 ++++++++++++++ vlib/x/executor/tools/README.md | 44 ++ .../tools/check_no_async_dependency.sh | 39 ++ vlib/x/executor/tools/validate.sh | 46 +++ 29 files changed, 2955 insertions(+) create mode 100644 vlib/x/executor/README.md create mode 100644 vlib/x/executor/admission_test.v create mode 100644 vlib/x/executor/benchmarks/README.md create mode 100644 vlib/x/executor/benchmarks/executor_benchmark.v create mode 100755 vlib/x/executor/benchmarks/run_executor_benchmark.sh create mode 100644 vlib/x/executor/config_test.v create mode 100644 vlib/x/executor/errors.v create mode 100644 vlib/x/executor/errors_test.v create mode 100644 vlib/x/executor/examples/README.md create mode 100644 vlib/x/executor/examples/basic_post.v create mode 100644 vlib/x/executor/examples/bounded_admission.v create mode 100644 vlib/x/executor/examples/owner_loop_pump.v create mode 100644 vlib/x/executor/examples/run_sync.v create mode 100644 vlib/x/executor/examples/shutdown.v create mode 100644 vlib/x/executor/examples/synthetic_ffi_owner_state.v create mode 100644 vlib/x/executor/examples/synthetic_render_upload.v create mode 100644 vlib/x/executor/executor.v create mode 100644 vlib/x/executor/lifecycle_test.v create mode 100644 vlib/x/executor/owner.v create mode 100644 vlib/x/executor/owner_test.v create mode 100644 vlib/x/executor/pump_test.v create mode 100644 vlib/x/executor/run.v create mode 100644 vlib/x/executor/stress_test.v create mode 100644 vlib/x/executor/submit.v create mode 100644 vlib/x/executor/sync_call.v create mode 100644 vlib/x/executor/sync_call_test.v create mode 100644 vlib/x/executor/tools/README.md create mode 100755 vlib/x/executor/tools/check_no_async_dependency.sh create mode 100755 vlib/x/executor/tools/validate.sh diff --git a/vlib/x/executor/README.md b/vlib/x/executor/README.md new file mode 100644 index 000000000..e6933445f --- /dev/null +++ b/vlib/x/executor/README.md @@ -0,0 +1,251 @@ +# x.executor + +`x.executor` is a small owner-loop executor for V programs. +It lets callers submit short callbacks to the thread or loop that owns a +resource, without exposing scheduler internals or starting a hidden runtime. + +The module is intentionally narrow: + +- one explicit executor; +- one explicit bounded queue; +- explicit admission and backpressure; +- explicit owner pumping through `run()`, `run_one()` or `drain_pending()`; +- explicit shutdown through `stop()` and `wait()`; +- no GUI, render, audio, network, or scheduler dependency. + +`x.executor` is a sibling of `x.async`, not a layer on top of it. This module +does not import `x.async`, and the examples in this directory use only +synthetic local work, `spawn`, channels, `context`, and `time`. + +## Why + +Some resources must be touched only by one owner thread or loop: + +- UI state owned by a main thread; +- render-context state owned by a render loop; +- FFI handles with same-thread requirements; +- single-owner game, plugin, scripting, cache, or test state. + +The safe abstraction is not "move scheduler work to another processor". The +safe abstraction is a bounded queue that the owner loop drains deliberately. +If a callback is slow, it still blocks the owner loop while it runs. Applications +remain responsible for keeping callbacks short. + +## Quick Start + +```v +import x.executor + +fn main() { + mut ex := executor.new(queue_size: 8)! + updates := chan string{cap: 1} + + ex.try_post(fn [updates] () ! { + updates <- 'owner mutation ran' + })! + + ran := ex.run_one()! + assert ran + assert (<-updates) == 'owner mutation ran' + + ex.stop() + assert !ex.run_one()! + ex.wait()! +} +``` + +## API + +### JobFn + +```v ignore +pub type JobFn = fn () ! +``` + +Jobs are ordinary V functions. They receive no hidden context. If a caller needs +to stop waiting for admission, it should use `post_with_context()` or +`post_with_timeout()`. Once accepted, a job is not preemptively killed. + +### Construction + +```v ignore +mut ex := executor.new(queue_size: 128)! +``` + +`queue_size` must be positive and is fixed for the executor lifetime. The queue +is bounded; a full queue returns an explicit backpressure error. + +`new()` does not capture owner identity. Owner identity exists only while a +thread is actively pumping callbacks through `run()`, `run_one()`, or +`drain_pending()`. + +### Submission + +```v ignore +ex.try_post(fn () ! { + // owner-thread work +})! + +ex.post_with_context(parent_ctx, fn () ! { + // accepted only while parent_ctx remains active +})! + +ex.post_with_timeout(50 * time.millisecond, fn () ! { + // accepted only before the timeout expires +})! +``` + +`try_post()` never waits for queue capacity. If capacity is unavailable, it +returns `executor: queue is full`. + +`post_with_context()` and `post_with_timeout()` bound admission only. If the job +is accepted and later waits in the executor queue, context cancellation or +timeout does not cancel that accepted job. +Active owner-thread submissions are accepted while capacity is available. If +they would need to wait for capacity, they fail with +`executor: owner thread cannot wait for queue capacity`. + +There is deliberately no unbounded blocking `post()`. + +### Owner Pumping + +```v ignore +ex.run()! +ran := ex.run_one()! +count := ex.drain_pending(16)! +``` + +`run()` blocks and drains accepted jobs until shutdown reaches a terminal state. + +`run_one()` executes at most one pending job and returns whether a job ran. It is +useful for tests or host loops that already own their own frame/event pump. It +does not drain the whole accepted queue in one call. + +`drain_pending(max_jobs)` executes up to `max_jobs` pending jobs and returns the +number of jobs executed. `max_jobs` must be positive. If `max_jobs` is reached, +accepted jobs may remain queued for a later owner pump. + +FIFO ordering is guaranteed for jobs submitted by one producer. For concurrent +producers, order is the order in which submissions are accepted by the executor, +not wall-clock order. + +### Shutdown + +```v ignore +ex.stop() +// If this executor is driven by run_one() or drain_pending(), pump once more. +assert !ex.run_one()! +ex.wait()! +``` + +`stop()` is idempotent and non-blocking. It closes admission, wakes callers that +are waiting to submit, and lets already accepted jobs drain. When the executor is +driven by `run_one()` or `drain_pending()`, the owner loop should pump once more +after `stop()` so the executor can observe the closed admission state and publish +its terminal result before `wait()`. + +`wait()` returns only after the executor has reached a terminal state. Calling +`wait()` before any owner pump can reach a terminal state returns a stable error +instead of blocking forever. That precondition error, and the owner-callback +precondition error, do not consume the one valid wait. A valid `wait()` remains +one-shot, and a second valid call returns a stable error. + +### Job Errors + +The first job error closes admission and is stored once. `run()` keeps pumping +until already accepted jobs have drained, then returns the first job error. +Manual pump APIs are bounded: `run_one()` can return a job error after one job, +and `drain_pending(max_jobs)` can return the stored error after reaching +`max_jobs` even when accepted jobs remain queued. Callers that use manual pumps +must keep pumping until the terminal state is reached, or use `run()` when they +want automatic draining. `wait()` returns the stored first job error after the +terminal state is reached. Job errors are not silently discarded. + +### Synchronous Calls + +```v ignore +ex.run_sync(fn () ! { + // owner-thread work; caller waits for completion +})! +``` + +`run_sync()` from a foreign thread submits a wrapper job and waits for that job +to complete. Its job error is returned to the caller. + +`run_sync()` from the active owner thread runs inline to avoid deadlock. This +inline execution still requires open admission and is outside executor FIFO +ordering. If the inline call returns an error, that error becomes the executor's +first job error even when the outer callback catches it. + +Context-bound or timeout-bound synchronous calls are not part of this first API. + +### Owner Checks + +```v ignore +if ex.is_owner_thread() { + // currently pumping callbacks on the owner thread +} + +ex.assert_owner_thread()! +``` + +`is_owner_thread()` is true only while the current thread is actively pumping +callbacks for that executor. Owner identity is cleared before the pump returns. + +## Safety Notes + +`x.executor` is about owner-affinity control flow, not sandboxing. + +- It does not recover panics. +- It does not kill a running callback. +- It does not make slow owner callbacks safe for GUI or render loops. +- It does not provide priorities, work stealing, promises, or a scheduler. +- It does not depend on real GUI, OpenGL, audio, or network resources. + +All public module-owned errors use the `executor:` prefix. User job errors are +returned unchanged. + +## Examples + +Small runnable examples live in `vlib/x/executor/examples/`: + +- `basic_post.v`: submit one owner-loop mutation and pump it with `run_one()`. +- `bounded_admission.v`: show queue backpressure and timeout-bounded admission. +- `owner_loop_pump.v`: drain a synthetic frame loop with a per-frame job limit. +- `run_sync.v`: wait for owner-loop execution from a foreign caller. +- `shutdown.v`: stop from an owner callback and reject later submissions. +- `synthetic_render_upload.v`: synthetic render-resource mutation. +- `synthetic_ffi_owner_state.v`: synthetic same-thread FFI handle mutation. + +Each example is local and synthetic. None starts a server, opens a window, +touches graphics/audio APIs, or imports `x.async`. + +## Tests + +The targeted test suite lives next to the module: + +```sh +v test vlib/x/executor +v -prod test vlib/x/executor +``` + +For guarded local validation, prefer: + +```sh +sh vlib/x/executor/tools/validate.sh +``` + +The validation script runs the dependency guard, formatting verification, +examples, dev tests, and `-prod` tests serially with isolated `VTMP` and +`VCACHE`. + +## Benchmarks + +Small local benchmarks live in `vlib/x/executor/benchmarks/`: + +```sh +sh vlib/x/executor/benchmarks/run_executor_benchmark.sh +``` + +Benchmarks are observation tools, not correctness tests or portable performance +claims. Defaults are intentionally modest and environment overrides are clamped. diff --git a/vlib/x/executor/admission_test.v b/vlib/x/executor/admission_test.v new file mode 100644 index 000000000..5430e9c3e --- /dev/null +++ b/vlib/x/executor/admission_test.v @@ -0,0 +1,388 @@ +module executor + +import context +import time + +fn test_try_post_returns_backpressure_when_queue_is_full() { + mut ex := new(queue_size: 1)! + ex.try_post(fn () ! {})! + ex.try_post(fn () ! {}) or { + assert err.msg() == 'executor: queue is full' + ex.stop() + return + } + assert false, 'try_post accepted a job beyond queue capacity' +} + +fn test_post_with_timeout_returns_timeout_when_queue_stays_full() { + mut ex := new(queue_size: 1)! + ex.try_post(fn () ! {})! + ex.post_with_timeout(20 * time.millisecond, fn () ! {}) or { + assert err.msg() == 'executor: timeout' + ex.stop() + return + } + assert false, 'post_with_timeout accepted a job while the queue stayed full' +} + +fn test_post_with_timeout_zero_timeout_rejects_available_capacity() { + mut ex := new(queue_size: 1)! + ran := chan bool{cap: 1} + ex.post_with_timeout(time.Duration(0), fn [ran] () ! { + ran <- true + return + }) or { + assert err.msg() == 'executor: timeout' + assert ex.drain_pending(1)! == 0 + assert_no_bool(ran, 'zero-timeout admission job ran') + ex.stop() + ex.wait()! + return + } + assert false, 'post_with_timeout accepted a job with zero timeout' +} + +fn test_post_with_timeout_negative_timeout_rejects_available_capacity() { + mut ex := new(queue_size: 1)! + ran := chan bool{cap: 1} + ex.post_with_timeout(-1 * time.nanosecond, fn [ran] () ! { + ran <- true + return + }) or { + assert err.msg() == 'executor: timeout' + assert ex.drain_pending(1)! == 0 + assert_no_bool(ran, 'negative-timeout admission job ran') + ex.stop() + ex.wait()! + return + } + assert false, 'post_with_timeout accepted a job with negative timeout' +} + +fn test_post_with_timeout_long_timeout_accepts_available_capacity() { + jobs := 4 + mut ex := new(queue_size: jobs)! + ran := chan int{cap: jobs} + + for i in 0 .. jobs { + ex.post_with_timeout(1 * time.hour, fn [ran, i] () ! { + value := i + 1 + ran <- value + return + })! + } + + assert ex.drain_pending(jobs)! == jobs + mut total := 0 + for _ in 0 .. jobs { + total += <-ran + } + assert total == 10 + + ex.stop() + assert !ex.run_one()! + ex.wait()! +} + +fn test_post_with_context_returns_parent_cancellation_without_accepting_job() { + mut ex := new(queue_size: 1)! + ex.try_post(fn () ! {})! + mut background := context.background() + mut parent, cancel := context.with_cancel(mut background) + cancel() + ran := chan bool{cap: 1} + + ex.post_with_context(parent, fn [ran] () ! { + ran <- true + }) or { + assert err.msg() == 'context canceled' + ex.stop() + assert_no_bool(ran, 'canceled admission job ran') + return + } + assert false, 'post_with_context accepted after parent cancellation' +} + +fn test_post_with_context_rejects_nil_job() { + mut ex := new(queue_size: 1)! + nil_job := unsafe { JobFn(nil) } + ex.post_with_context(context.background(), nil_job) or { + assert err.msg() == 'executor: job function is nil' + ex.stop() + return + } + assert false, 'post_with_context accepted a nil job' +} + +fn test_post_with_timeout_rejects_nil_job() { + mut ex := new(queue_size: 1)! + nil_job := unsafe { JobFn(nil) } + ex.post_with_timeout(1 * time.second, nil_job) or { + assert err.msg() == 'executor: job function is nil' + ex.stop() + return + } + assert false, 'post_with_timeout accepted a nil job' +} + +fn test_owner_post_with_context_full_queue_returns_owner_capacity_error() { + mut ex := new(queue_size: 1)! + owner_started := chan bool{cap: 1} + filler_accepted := chan bool{cap: 1} + owner_result := chan string{cap: 1} + filler_ran := chan bool{cap: 1} + nested_ran := chan bool{cap: 1} + + ex.try_post(fn [mut ex, owner_started, filler_accepted, owner_result, nested_ran] () ! { + owner_started <- true + assert wait_for_bool(filler_accepted, 'filler job was not accepted') + ex.post_with_context(context.background(), fn [nested_ran] () ! { + nested_ran <- true + }) or { + owner_result <- err.msg() + return + } + owner_result <- 'accepted' + })! + + filler := spawn fn [mut ex, owner_started, filler_accepted, filler_ran] () { + assert wait_for_bool(owner_started, 'owner callback did not start') + ex.try_post(fn [filler_ran] () ! { + filler_ran <- true + }) or { + filler_accepted <- false + return + } + filler_accepted <- true + }() + + runner := spawn fn [mut ex, owner_result] () { + ex.run_one() or { + owner_result <- err.msg() + return + } + }() + + msg := wait_for_string(owner_result, 'owner post_with_context did not return') + assert msg == 'executor: owner thread cannot wait for queue capacity' + + ex.stop() + runner.wait() + filler.wait() + assert ex.run_one()! + assert wait_for_bool(filler_ran, 'filler job did not run after owner rejection') + assert_no_bool(nested_ran, 'owner post_with_context job ran after rejection') + ex.wait()! +} + +fn test_owner_post_with_timeout_full_queue_returns_owner_capacity_error() { + mut ex := new(queue_size: 1)! + owner_started := chan bool{cap: 1} + filler_accepted := chan bool{cap: 1} + owner_result := chan string{cap: 1} + filler_ran := chan bool{cap: 1} + nested_ran := chan bool{cap: 1} + + ex.try_post(fn [mut ex, owner_started, filler_accepted, owner_result, nested_ran] () ! { + owner_started <- true + assert wait_for_bool(filler_accepted, 'filler job was not accepted') + ex.post_with_timeout(1 * time.hour, fn [nested_ran] () ! { + nested_ran <- true + }) or { + owner_result <- err.msg() + return + } + owner_result <- 'accepted' + })! + + filler := spawn fn [mut ex, owner_started, filler_accepted, filler_ran] () { + assert wait_for_bool(owner_started, 'owner callback did not start') + ex.try_post(fn [filler_ran] () ! { + filler_ran <- true + }) or { + filler_accepted <- false + return + } + filler_accepted <- true + }() + + runner := spawn fn [mut ex, owner_result] () { + ex.run_one() or { + owner_result <- err.msg() + return + } + }() + + msg := wait_for_string(owner_result, 'owner post_with_timeout did not return') + assert msg == 'executor: owner thread cannot wait for queue capacity' + + ex.stop() + runner.wait() + filler.wait() + assert ex.run_one()! + assert wait_for_bool(filler_ran, 'filler job did not run after owner rejection') + assert_no_bool(nested_ran, 'owner post_with_timeout job ran after rejection') + ex.wait()! +} + +fn test_owner_try_post_full_queue_returns_queue_full_and_pump_continues() { + mut ex := new(queue_size: 1)! + owner_started := chan bool{cap: 1} + filler_accepted := chan bool{cap: 1} + owner_result := chan string{cap: 1} + filler_ran := chan bool{cap: 1} + nested_ran := chan bool{cap: 1} + + ex.try_post(fn [mut ex, owner_started, filler_accepted, owner_result, nested_ran] () ! { + owner_started <- true + assert wait_for_bool(filler_accepted, 'filler job was not accepted') + ex.try_post(fn [nested_ran] () ! { + nested_ran <- true + }) or { + owner_result <- err.msg() + return + } + owner_result <- 'accepted' + })! + + filler := spawn fn [mut ex, owner_started, filler_accepted, filler_ran] () { + assert wait_for_bool(owner_started, 'owner callback did not start') + ex.try_post(fn [filler_ran] () ! { + filler_ran <- true + }) or { + filler_accepted <- false + return + } + filler_accepted <- true + }() + + assert ex.run_one()! + assert wait_for_string(owner_result, 'owner try_post did not return') == 'executor: queue is full' + filler.wait() + assert ex.run_one()! + assert wait_for_bool(filler_ran, 'pump did not continue to filler job') + assert_no_bool(nested_ran, 'owner try_post job ran after queue-full rejection') + + ex.stop() + assert !ex.run_one()! + ex.wait()! +} + +fn test_owner_bounded_posts_accept_when_capacity_is_available() { + mut ex := new(queue_size: 2)! + owner_result := chan string{cap: 1} + context_ran := chan bool{cap: 1} + timeout_ran := chan bool{cap: 1} + + ex.try_post(fn [mut ex, owner_result, context_ran, timeout_ran] () ! { + ex.post_with_context(context.background(), fn [context_ran] () ! { + context_ran <- true + }) or { + owner_result <- err.msg() + return + } + ex.post_with_timeout(1 * time.hour, fn [timeout_ran] () ! { + timeout_ran <- true + }) or { + owner_result <- err.msg() + return + } + owner_result <- 'accepted' + })! + + assert ex.run_one()! + assert wait_for_string(owner_result, 'owner bounded posts did not return') == 'accepted' + assert ex.run_one()! + assert ex.run_one()! + assert wait_for_bool(context_ran, 'owner post_with_context job did not run') + assert wait_for_bool(timeout_ran, 'owner post_with_timeout job did not run') + + ex.stop() + assert !ex.run_one()! + ex.wait()! +} + +fn test_post_with_context_waits_until_capacity_opens() { + mut ex := new(queue_size: 1)! + release := chan bool{cap: 1} + started := chan bool{cap: 1} + attempting := chan bool{cap: 1} + accepted := chan bool{cap: 1} + ran := chan bool{cap: 1} + owner_result := chan string{cap: 1} + + ex.try_post(fn [started, release] () ! { + started <- true + _ := <-release + })! + + submitter := spawn fn [mut ex, attempting, accepted, ran] () { + attempting <- true + ex.post_with_context(context.background(), fn [ran] () ! { + ran <- true + }) or { + accepted <- false + return + } + accepted <- true + }() + assert wait_for_bool(attempting, 'bounded submitter did not start') + assert_no_bool(accepted, 'post_with_context returned before capacity opened') + + owner := spawn fn [mut ex, owner_result] () { + ran_one := ex.run_one() or { + owner_result <- err.msg() + return + } + if ran_one { + owner_result <- 'ran' + } else { + owner_result <- 'empty' + } + }() + assert wait_for_bool(started, 'blocking job did not start') + assert wait_for_bool(accepted, 'post_with_context did not accept after capacity opened') + release <- true + assert wait_for_string(owner_result, 'owner run_one did not return') == 'ran' + assert ex.run_one()! + assert wait_for_bool(ran, 'bounded admission job did not run') + + ex.stop() + assert !ex.run_one()! + ex.wait()! + owner.wait() + submitter.wait() +} + +fn wait_for_bool(signal chan bool, message string) bool { + select { + value := <-signal { + return value + } + 1 * time.second { + assert false, message + } + } + return false +} + +fn wait_for_string(signal chan string, message string) string { + select { + value := <-signal { + return value + } + 1 * time.second { + assert false, message + } + } + return '' +} + +fn assert_no_bool(signal chan bool, message string) { + select { + _ := <-signal { + assert false, message + } + 50 * time.millisecond {} + } +} diff --git a/vlib/x/executor/benchmarks/README.md b/vlib/x/executor/benchmarks/README.md new file mode 100644 index 000000000..25c8df6f5 --- /dev/null +++ b/vlib/x/executor/benchmarks/README.md @@ -0,0 +1,40 @@ +# x.executor benchmarks + +This folder contains cautious local benchmarks for the public `x.executor` API. +They are observation tools, not correctness tests and not portable performance +claims. + +## Available benchmark + +- `executor_benchmark.v`: measures short default runs for `try_post()` plus + `drain_pending()`, `run_sync()`, bounded admission timeout, and small + multi-producer submission. +- `run_executor_benchmark.sh`: builds and runs the benchmark with the local + `./v`, isolated `VTMP`/`VCACHE`, and a temporary output binary. + +## Run + +From the repository root: + +```sh +sh vlib/x/executor/benchmarks/run_executor_benchmark.sh +``` + +The script is serialized by design. Do not run multiple V benchmark or test +runners against the same checkout/cache without isolating `VTMP`, `VCACHE`, and +output paths. + +## Parameters + +The defaults are intentionally small. Tune them locally with: + +- `XEXECUTOR_BENCH_DRAIN_ROUNDS` +- `XEXECUTOR_BENCH_DRAIN_JOBS` +- `XEXECUTOR_BENCH_SYNC_ROUNDS` +- `XEXECUTOR_BENCH_TIMEOUT_ROUNDS` +- `XEXECUTOR_BENCH_TIMEOUT_MS` +- `XEXECUTOR_BENCH_PRODUCERS` + +All environment values are clamped. 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/executor/benchmarks/executor_benchmark.v b/vlib/x/executor/benchmarks/executor_benchmark.v new file mode 100644 index 000000000..4916ee950 --- /dev/null +++ b/vlib/x/executor/benchmarks/executor_benchmark.v @@ -0,0 +1,190 @@ +import benchmark +import context +import os +import time +import x.executor + +const default_drain_rounds = 16 +const default_drain_jobs = 32 +const default_sync_rounds = 32 +const default_timeout_rounds = 8 +const default_timeout_ms = 2 +const default_producers = 16 + +fn main() { + run_benchmarks() or { + eprintln('x.executor benchmark failed: ${err.msg()}') + exit(1) + } +} + +fn run_benchmarks() ! { + drain_rounds := env_int('XEXECUTOR_BENCH_DRAIN_ROUNDS', default_drain_rounds, 1, 200) + drain_jobs := env_int('XEXECUTOR_BENCH_DRAIN_JOBS', default_drain_jobs, 1, 512) + sync_rounds := env_int('XEXECUTOR_BENCH_SYNC_ROUNDS', default_sync_rounds, 1, 500) + timeout_rounds := env_int('XEXECUTOR_BENCH_TIMEOUT_ROUNDS', default_timeout_rounds, 1, 100) + timeout_ms := env_int('XEXECUTOR_BENCH_TIMEOUT_MS', default_timeout_ms, 1, 100) + producers := env_int('XEXECUTOR_BENCH_PRODUCERS', default_producers, 1, 256) + + println('x.executor cautious benchmark') + println('Override sizes with XEXECUTOR_BENCH_* environment variables.') + + mut checksum := 0 + mut b := benchmark.start() + checksum += bench_try_post_drain(drain_rounds, drain_jobs)! + b.measure('try_post + drain_pending: rounds=${drain_rounds}, jobs=${drain_jobs}, checksum=${checksum}') + + checksum += bench_run_sync(sync_rounds)! + b.measure('run_sync: rounds=${sync_rounds}, checksum=${checksum}') + + checksum += bench_bounded_timeout(timeout_rounds, timeout_ms)! + b.measure('bounded admission timeout: rounds=${timeout_rounds}, timeout_ms=${timeout_ms}, checksum=${checksum}') + + checksum += bench_many_producers(producers)! + b.measure('multi-producer post_with_context: producers=${producers}, checksum=${checksum}') +} + +fn env_int(name string, default_value int, min_value int, max_value int) int { + raw := os.getenv(name) + if raw == '' { + return default_value + } + mut value := raw.int() + if value < min_value { + value = min_value + } + if value > max_value { + value = max_value + } + return value +} + +fn bench_try_post_drain(rounds int, jobs int) !int { + mut total := 0 + for _ in 0 .. rounds { + mut ex := executor.new(queue_size: jobs)! + done := chan int{cap: jobs} + for _ in 0 .. jobs { + ex.try_post(fn [done] () ! { + done <- 1 + })! + } + drained := ex.drain_pending(jobs)! + total += drained + for _ in 0 .. jobs { + total += <-done + } + ex.stop() + ran_after_stop := ex.run_one()! + if ran_after_stop { + return error('try_post drain benchmark ran an unexpected job after stop') + } + ex.wait()! + } + return total +} + +fn bench_run_sync(rounds int) !int { + mut ex := executor.new(queue_size: rounds)! + run_result := chan string{cap: 1} + runner := spawn fn [mut ex, run_result] () { + ex.run() or { + run_result <- err.msg() + return + } + run_result <- 'ok' + }() + + done := chan int{cap: rounds} + for _ in 0 .. rounds { + ex.run_sync(fn [done] () ! { + done <- 1 + })! + } + + mut total := 0 + for _ in 0 .. rounds { + total += <-done + } + ex.stop() + run_msg := <-run_result + if run_msg != 'ok' { + runner.wait() + return error('run_sync benchmark owner loop returned an error') + } + ex.wait()! + runner.wait() + return total +} + +fn bench_bounded_timeout(rounds int, timeout_ms int) !int { + timeout := time.Duration(timeout_ms) * time.millisecond + mut total := 0 + for _ in 0 .. rounds { + mut ex := executor.new(queue_size: 1)! + ex.try_post(fn () ! {})! + ex.post_with_timeout(timeout, fn () ! {}) or { + if err.msg() != 'executor: timeout' { + return err + } + total++ + } + ran := ex.run_one()! + if !ran { + return error('bounded timeout benchmark did not run the accepted job') + } + ex.stop() + ran_after_stop := ex.run_one()! + if ran_after_stop { + return error('bounded timeout benchmark ran an unexpected job after stop') + } + ex.wait()! + } + return total +} + +fn bench_many_producers(producers int) !int { + mut ex := executor.new(queue_size: producers)! + accepted := chan bool{cap: producers} + ran := chan int{cap: producers} + mut threads := []thread{} + + for i in 0 .. producers { + threads << spawn fn [mut ex, accepted, ran, i] () { + ex.post_with_context(context.background(), fn [ran, i] () ! { + value := i + 1 + ran <- value + return + }) or { + accepted <- false + return + } + accepted <- true + }() + } + for _ in 0 .. producers { + accepted_ok := <-accepted + if !accepted_ok { + return error('producer admission failed') + } + } + drained := ex.drain_pending(producers)! + if drained != producers { + return error('multi-producer benchmark drained ${drained} jobs instead of ${producers}') + } + + mut total := 0 + for _ in 0 .. producers { + total += <-ran + } + for producer_thread in threads { + producer_thread.wait() + } + ex.stop() + ran_after_stop := ex.run_one()! + if ran_after_stop { + return error('multi-producer benchmark ran an unexpected job after stop') + } + ex.wait()! + return total +} diff --git a/vlib/x/executor/benchmarks/run_executor_benchmark.sh b/vlib/x/executor/benchmarks/run_executor_benchmark.sh new file mode 100755 index 000000000..d2c741104 --- /dev/null +++ b/vlib/x/executor/benchmarks/run_executor_benchmark.sh @@ -0,0 +1,32 @@ +#!/usr/bin/env sh +set -eu + +script_dir=$(CDPATH= cd "$(dirname "$0")" && pwd) +repo_root=$(CDPATH= cd "$script_dir/../../../.." && pwd) +cd "$repo_root" + +if [ ! -x ./v ]; then + echo "x.executor benchmarks must be run from a V checkout with local ./v" >&2 + exit 1 +fi + +tmp_root=$(mktemp -d "${TMPDIR:-/tmp}/xexecutor-benchmark.XXXXXX") +cleanup() { + rm -rf "$tmp_root" +} +trap cleanup EXIT INT TERM + +vtmp="$tmp_root/vtmp" +vcache="$tmp_root/vcache" +out="$tmp_root/out/executor_benchmark" +mkdir -p "$vtmp" "$vcache" "$(dirname "$out")" + +echo "Running x.executor benchmark with isolated VTMP and VCACHE." +echo "Tune sizes with XEXECUTOR_BENCH_* environment variables." +echo "See vlib/x/executor/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/executor/benchmarks/executor_benchmark.v +else + env VTMP="$vtmp" VCACHE="$vcache" ./v -prod -o "$out" run vlib/x/executor/benchmarks/executor_benchmark.v +fi diff --git a/vlib/x/executor/config_test.v b/vlib/x/executor/config_test.v new file mode 100644 index 000000000..e9dca6760 --- /dev/null +++ b/vlib/x/executor/config_test.v @@ -0,0 +1,28 @@ +module executor + +fn test_new_rejects_zero_queue_size() { + new(queue_size: 0) or { + assert err.msg() == 'executor: queue size must be positive' + return + } + assert false, 'new accepted a zero queue size' +} + +fn test_new_rejects_negative_queue_size() { + new(queue_size: -1) or { + assert err.msg() == 'executor: queue size must be positive' + return + } + assert false, 'new accepted a negative queue size' +} + +fn test_try_post_rejects_nil_job() { + mut ex := new(queue_size: 1)! + nil_job := unsafe { JobFn(nil) } + ex.try_post(nil_job) or { + assert err.msg() == 'executor: job function is nil' + ex.stop() + return + } + assert false, 'try_post accepted a nil job' +} diff --git a/vlib/x/executor/errors.v b/vlib/x/executor/errors.v new file mode 100644 index 000000000..1c3e4f7c9 --- /dev/null +++ b/vlib/x/executor/errors.v @@ -0,0 +1,17 @@ +module executor + +// Error strings are centralized because they are part of the public contract. +// Keep them stable, short, and explicitly scoped to x.executor. +const err_drain_limit_invalid = 'executor: drain limit must be positive' +const err_executor_closed = 'executor: executor is closed' +const err_executor_stopped = 'executor: executor is stopped' +const err_nil_job = 'executor: job function is nil' +const err_not_owner_thread = 'executor: current thread is not the executor owner' +const err_owner_pump_running = 'executor: owner pump is already running' +const err_owner_submit_wait = 'executor: owner thread cannot wait for queue capacity' +const err_queue_full = 'executor: queue is full' +const err_queue_size_invalid = 'executor: queue size must be positive' +const err_timeout = 'executor: timeout' +const err_wait_before_terminal = 'executor: wait requires a running pump or stopped executor' +const err_wait_called = 'executor: wait was already called' +const err_wait_owner_thread = 'executor: wait cannot be called from executor owner thread' diff --git a/vlib/x/executor/errors_test.v b/vlib/x/executor/errors_test.v new file mode 100644 index 000000000..d22d41218 --- /dev/null +++ b/vlib/x/executor/errors_test.v @@ -0,0 +1,133 @@ +module executor + +import time + +fn test_first_job_error_is_returned_after_accepted_jobs_drain() { + mut ex := new(queue_size: 4)! + ran := chan int{cap: 2} + result := chan string{cap: 1} + + ex.try_post(fn [ran] () ! { + ran <- 1 + return error('first owner failure') + })! + ex.try_post(fn [ran] () ! { + ran <- 2 + })! + + runner := spawn fn [mut ex, result] () { + ex.run() or { + result <- err.msg() + return + } + result <- 'ok' + }() + + assert wait_for_int(ran, 'first job did not run') == 1 + assert wait_for_int(ran, 'accepted job after error did not drain') == 2 + assert wait_for_string(result, 'run did not return first job error') == 'first owner failure' + ex.wait() or { + assert err.msg() == 'first owner failure' + runner.wait() + return + } + assert false, 'wait did not return first job error' +} + +fn test_admission_closes_after_first_job_error() { + mut ex := new(queue_size: 1)! + ex.try_post(fn () ! { + return error('owner failure') + })! + mut saw_job_error := false + ex.run_one() or { + assert err.msg() == 'owner failure' + saw_job_error = true + } + assert saw_job_error + + ex.try_post(fn () ! {}) or { + assert err.msg() == 'executor: executor is closed' + ex.wait() or { assert err.msg() == 'owner failure' } + return + } + assert false, 'try_post accepted after first job error' +} + +fn test_run_one_after_job_error_requires_continued_pumping_to_drain() { + mut ex := new(queue_size: 3)! + ran := chan int{cap: 2} + + ex.try_post(fn [ran] () ! { + ran <- 1 + return error('owner failure') + })! + ex.try_post(fn [ran] () ! { + ran <- 2 + })! + + ex.run_one() or { assert err.msg() == 'owner failure' } + assert wait_for_int(ran, 'erroring job did not run') == 1 + + ex.run_one() or { assert err.msg() == 'owner failure' } + assert wait_for_int(ran, 'accepted job after error was not drained') == 2 + + ex.wait() or { + assert err.msg() == 'owner failure' + return + } + assert false, 'wait did not return the stored owner failure' +} + +fn test_drain_pending_after_job_error_requires_continued_pumping_when_limit_is_reached() { + mut ex := new(queue_size: 4)! + ran := chan int{cap: 3} + + ex.try_post(fn [ran] () ! { + ran <- 1 + return error('owner failure') + })! + ex.try_post(fn [ran] () ! { + ran <- 2 + })! + ex.try_post(fn [ran] () ! { + ran <- 3 + })! + + ex.drain_pending(1) or { assert err.msg() == 'owner failure' } + assert wait_for_int(ran, 'erroring drain job did not run') == 1 + + ex.drain_pending(8) or { assert err.msg() == 'owner failure' } + assert wait_for_int(ran, 'second accepted job was not drained') == 2 + assert wait_for_int(ran, 'third accepted job was not drained') == 3 + + ex.wait() or { + assert err.msg() == 'owner failure' + return + } + assert false, 'wait did not return the stored owner failure' +} + +fn wait_for_int(signal chan int, message string) int { + select { + value := <-signal { + return value + } + 1 * time.second { + assert false, message + } + } + return 0 +} + +fn wait_for_string(signal chan string, message string) string { + select { + value := <-signal { + return value + } + 1 * time.second { + assert false, message + } + } + return '' +} diff --git a/vlib/x/executor/examples/README.md b/vlib/x/executor/examples/README.md new file mode 100644 index 000000000..21495d4d6 --- /dev/null +++ b/vlib/x/executor/examples/README.md @@ -0,0 +1,45 @@ +# x.executor examples + +This folder contains short public examples for the current `x.executor` API. +Each file focuses on one owner-loop idea and uses only local synthetic work. + +Examples are documentation-first programs. They are executed by the validation +script to avoid bit rot, but the regression guarantees live in the module tests. + +## Rules + +- No dependency on the sibling async module. +- No real GUI, render, audio, FFI, or network dependency. +- No fixed local path is used. +- No network listener is opened. +- Examples avoid `panic()` as control flow. +- Owner-loop callbacks are deliberately short. + +## Examples + +- `basic_post.v`: submit one mutation and pump it with `run_one()`. +- `bounded_admission.v`: explicit queue backpressure and timeout-bounded + admission. +- `owner_loop_pump.v`: synthetic frame loop using `drain_pending(max_jobs)`. +- `run_sync.v`: foreign caller waits for owner-loop execution. +- `shutdown.v`: owner callback calls `stop()` without deadlock. +- `synthetic_render_upload.v`: synthetic render-resource upload on an owner + loop. +- `synthetic_ffi_owner_state.v`: synthetic same-thread FFI handle mutation. + +## Run + +From the repository root: + +```sh +./v run vlib/x/executor/examples/basic_post.v +./v run vlib/x/executor/examples/bounded_admission.v +./v run vlib/x/executor/examples/owner_loop_pump.v +./v run vlib/x/executor/examples/run_sync.v +./v run vlib/x/executor/examples/shutdown.v +./v run vlib/x/executor/examples/synthetic_render_upload.v +./v run vlib/x/executor/examples/synthetic_ffi_owner_state.v +``` + +Use `tools/validate.sh` for the guarded module validation path. It isolates V +temporary/cache directories and keeps validation runners serialized. diff --git a/vlib/x/executor/examples/basic_post.v b/vlib/x/executor/examples/basic_post.v new file mode 100644 index 000000000..a497bb343 --- /dev/null +++ b/vlib/x/executor/examples/basic_post.v @@ -0,0 +1,24 @@ +import x.executor + +fn main() { + mut ex := executor.new(queue_size: 4)! + updates := chan string{cap: 1} + + ex.try_post(fn [updates] () ! { + updates <- 'owner state updated' + })! + + ran := ex.run_one()! + if !ran { + eprintln('executor did not run the queued job') + exit(1) + } + + println(<-updates) + ex.stop() + if ex.run_one()! { + eprintln('unexpected job after stop') + exit(1) + } + ex.wait()! +} diff --git a/vlib/x/executor/examples/bounded_admission.v b/vlib/x/executor/examples/bounded_admission.v new file mode 100644 index 000000000..64f7b348e --- /dev/null +++ b/vlib/x/executor/examples/bounded_admission.v @@ -0,0 +1,30 @@ +import time +import x.executor + +fn main() { + mut ex := executor.new(queue_size: 1)! + ran := chan string{cap: 2} + + ex.try_post(fn [ran] () ! { + ran <- 'first' + })! + + ex.try_post(fn () ! {}) or { println('try_post backpressure: ${err.msg()}') } + + ex.post_with_timeout(20 * time.millisecond, fn () ! {}) or { + println('timeout bounded admission: ${err.msg()}') + } + + if !ex.run_one()! { + eprintln('queued job did not run') + exit(1) + } + println(<-ran) + + ex.stop() + if ex.run_one()! { + eprintln('unexpected job after stop') + exit(1) + } + ex.wait()! +} diff --git a/vlib/x/executor/examples/owner_loop_pump.v b/vlib/x/executor/examples/owner_loop_pump.v new file mode 100644 index 000000000..b8247b32d --- /dev/null +++ b/vlib/x/executor/examples/owner_loop_pump.v @@ -0,0 +1,38 @@ +import x.executor + +fn main() { + mut ex := executor.new(queue_size: 8)! + events := chan int{cap: 6} + + for i in 0 .. 6 { + ex.try_post(fn [events, i] () ! { + events <- i + })! + } + + mut frame := 0 + mut drained_total := 0 + for drained_total < 6 { + frame++ + drained := ex.drain_pending(2)! + if drained == 0 { + eprintln('frame ${frame} did not drain any work') + exit(1) + } + drained_total += drained + println('frame ${frame}: drained ${drained} owner jobs') + } + + mut checksum := 0 + for _ in 0 .. 6 { + checksum += <-events + } + println('checksum=${checksum}') + + ex.stop() + if ex.drain_pending(1)! != 0 { + eprintln('unexpected job after stop') + exit(1) + } + ex.wait()! +} diff --git a/vlib/x/executor/examples/run_sync.v b/vlib/x/executor/examples/run_sync.v new file mode 100644 index 000000000..12e1a37e4 --- /dev/null +++ b/vlib/x/executor/examples/run_sync.v @@ -0,0 +1,38 @@ +import x.executor + +fn main() { + mut ex := executor.new(queue_size: 4)! + result := chan string{cap: 1} + run_result := chan string{cap: 1} + + runner := spawn fn [mut ex, run_result] () { + ex.run() or { + run_result <- err.msg() + return + } + run_result <- 'ok' + }() + + caller := spawn fn [mut ex, result] () { + ex.run_sync(fn [result] () ! { + result <- 'ran on owner loop' + }) or { result <- 'run_sync failed: ${err.msg()}' } + }() + + msg := <-result + if msg != 'ran on owner loop' { + eprintln(msg) + exit(1) + } + println(msg) + + ex.stop() + run_msg := <-run_result + if run_msg != 'ok' { + eprintln('run failed: ${run_msg}') + exit(1) + } + ex.wait()! + caller.wait() + runner.wait() +} diff --git a/vlib/x/executor/examples/shutdown.v b/vlib/x/executor/examples/shutdown.v new file mode 100644 index 000000000..005af05ab --- /dev/null +++ b/vlib/x/executor/examples/shutdown.v @@ -0,0 +1,35 @@ +import x.executor + +fn main() { + mut ex := executor.new(queue_size: 2)! + ran := chan bool{cap: 1} + run_result := chan string{cap: 1} + + ex.try_post(fn [mut ex, ran] () ! { + ex.stop() + ran <- true + })! + + runner := spawn fn [mut ex, run_result] () { + ex.run() or { + run_result <- err.msg() + return + } + run_result <- 'ok' + }() + + did_run := <-ran + if !did_run { + eprintln('owner stop job did not run') + exit(1) + } + msg := <-run_result + if msg != 'ok' { + eprintln('run returned error: ${msg}') + exit(1) + } + + ex.try_post(fn () ! {}) or { println('submit after stop: ${err.msg()}') } + ex.wait()! + runner.wait() +} diff --git a/vlib/x/executor/examples/synthetic_ffi_owner_state.v b/vlib/x/executor/examples/synthetic_ffi_owner_state.v new file mode 100644 index 000000000..fa5229c2b --- /dev/null +++ b/vlib/x/executor/examples/synthetic_ffi_owner_state.v @@ -0,0 +1,36 @@ +import x.executor + +struct FakeHandle { +mut: + calls int +} + +fn main() { + mut owner := executor.new(queue_size: 4)! + result := chan int{cap: 1} + mut handle := FakeHandle{} + + owner.try_post(fn [mut handle, result] () ! { + handle.calls++ + handle.calls++ + result <- handle.calls + })! + + if !owner.run_one()! { + eprintln('owner job did not run') + exit(1) + } + calls := <-result + if calls != 2 { + eprintln('unexpected handle call count: ${calls}') + exit(1) + } + + println('synthetic FFI handle calls=${calls}') + owner.stop() + if owner.run_one()! { + eprintln('unexpected owner job after stop') + exit(1) + } + owner.wait()! +} diff --git a/vlib/x/executor/examples/synthetic_render_upload.v b/vlib/x/executor/examples/synthetic_render_upload.v new file mode 100644 index 000000000..7b8440c75 --- /dev/null +++ b/vlib/x/executor/examples/synthetic_render_upload.v @@ -0,0 +1,53 @@ +import x.executor + +struct TextureUpload { + id int + bytes int +} + +fn main() { + mut render := executor.new(queue_size: 8)! + uploaded := chan TextureUpload{cap: 3} + producer_done := chan bool{cap: 1} + + producer := spawn fn [mut render, producer_done, uploaded] () { + for id in 0 .. 3 { + bytes := 128 + id + render.try_post(fn [uploaded, id, bytes] () ! { + uploaded <- TextureUpload{ + id: id + bytes: bytes + } + }) or { + producer_done <- false + return + } + } + producer_done <- true + }() + + producer_ok := <-producer_done + if !producer_ok { + eprintln('producer could not submit render uploads') + exit(1) + } + drained := render.drain_pending(8)! + if drained != 3 { + eprintln('expected 3 render uploads, got ${drained}') + exit(1) + } + + mut total_bytes := 0 + for _ in 0 .. 3 { + total_bytes += (<-uploaded).bytes + } + println('uploaded ${drained} synthetic textures, bytes=${total_bytes}') + + render.stop() + if render.drain_pending(1)! != 0 { + eprintln('unexpected render job after stop') + exit(1) + } + render.wait()! + producer.wait() +} diff --git a/vlib/x/executor/executor.v b/vlib/x/executor/executor.v new file mode 100644 index 000000000..ba8ffef66 --- /dev/null +++ b/vlib/x/executor/executor.v @@ -0,0 +1,170 @@ +module executor + +import sync + +// JobFn is the function signature executed by an Executor on the owner thread. +pub type JobFn = fn () ! + +// ExecutorConfig configures a bounded owner-thread executor. +@[params] +pub struct ExecutorConfig { +pub: + queue_size int +} + +struct ExecutorJobResult { +mut: + err IError = none +} + +struct ExecutorJob { + f JobFn @[required] + result_ch chan ExecutorJobResult + has_result bool + report_error bool +} + +// Executor serializes accepted jobs on whichever thread actively pumps it. +// +// The executor does not own a thread. The owner identity is recorded only while +// run(), run_one(), or drain_pending() is actively pumping callbacks. +@[heap] +pub struct Executor { +mut: + queue_size int + queue []ExecutorJob + active int + + mutex &sync.Mutex = sync.new_mutex() + job_ready chan int + submit_ready chan int + done chan int + + accepting bool = true + stopped bool + waited bool + + pumping bool + run_active bool + + owner_active bool + owner_thread_id u64 + + first_err IError = none +} + +// new creates an Executor with a fixed bounded queue. +// +// The owner thread is not captured at construction time. It is recorded only +// while an owner pump method is actively running. +pub fn new(config ExecutorConfig) !&Executor { + if config.queue_size <= 0 { + return error(err_queue_size_invalid) + } + return &Executor{ + queue_size: config.queue_size + queue: []ExecutorJob{cap: config.queue_size} + mutex: sync.new_mutex() + job_ready: chan int{cap: 1} + submit_ready: chan int{cap: 1} + done: chan int{cap: 1} + accepting: true + } +} + +fn (mut e Executor) pop_job_locked() ?ExecutorJob { + if e.queue.len == 0 { + return none + } + job := e.queue[0] + e.queue.delete(0) + e.active++ + e.wake_submitters_locked() + return job +} + +fn (mut e Executor) wake_job_ready_locked() { + if !e.job_ready.closed { + e.job_ready.close() + } + e.job_ready = chan int{cap: 1} +} + +fn (mut e Executor) wake_submitters_locked() { + if !e.submit_ready.closed { + e.submit_ready.close() + } + if e.accepting { + e.submit_ready = chan int{cap: 1} + } +} + +fn (mut e Executor) close_admission_locked() { + if e.accepting { + e.accepting = false + e.wake_submitters_locked() + e.wake_job_ready_locked() + } +} + +fn (mut e Executor) finish_terminal_locked() IError { + if !e.stopped { + e.stopped = true + e.close_admission_locked() + if !e.done.closed { + e.done.close() + } + } + return e.first_err +} + +fn (mut e Executor) get_first_error() IError { + e.mutex.lock() + err := e.first_err + e.mutex.unlock() + return err +} + +fn (mut e Executor) set_first_error(err IError) { + e.mutex.lock() + e.set_first_error_locked(err) + e.mutex.unlock() +} + +fn (mut e Executor) set_first_error_locked(err IError) { + if e.first_err is none { + e.first_err = err + } + e.close_admission_locked() +} + +fn (mut e Executor) execute_job(job ExecutorJob) IError { + defer { + e.finish_job() + } + first_err_before := e.get_first_error() + had_first_err := first_err_before !is none + mut result := ExecutorJobResult{} + job.f() or { result.err = err } + if result.err is none && !had_first_err { + err := e.get_first_error() + if err !is none { + result.err = err + } + } + if result.err !is none && job.report_error { + e.set_first_error(result.err) + } + if job.has_result { + job.result_ch <- result + } + return result.err +} + +fn (mut e Executor) finish_job() { + e.mutex.lock() + if e.active > 0 { + e.active-- + } + e.mutex.unlock() +} diff --git a/vlib/x/executor/lifecycle_test.v b/vlib/x/executor/lifecycle_test.v new file mode 100644 index 000000000..5d9c39fb4 --- /dev/null +++ b/vlib/x/executor/lifecycle_test.v @@ -0,0 +1,259 @@ +module executor + +import time + +fn test_run_drains_accepted_jobs_then_wait_returns() { + mut ex := new(queue_size: 4)! + ran := chan int{cap: 3} + result := chan string{cap: 1} + + for i in 0 .. 3 { + ex.try_post(fn [ran, i] () ! { + ran <- i + })! + } + + runner := spawn fn [mut ex, result] () { + ex.run() or { + result <- err.msg() + return + } + result <- 'ok' + }() + + assert wait_for_int(ran, 'first run job missing') == 0 + assert wait_for_int(ran, 'second run job missing') == 1 + assert wait_for_int(ran, 'third run job missing') == 2 + ex.stop() + assert wait_for_string(result, 'run did not return after stop') == 'ok' + ex.wait()! + runner.wait() +} + +fn test_stop_from_owner_callback_does_not_deadlock() { + mut ex := new(queue_size: 2)! + ran := chan bool{cap: 1} + result := chan string{cap: 1} + + ex.try_post(fn [mut ex, ran] () ! { + ex.stop() + ran <- true + })! + + runner := spawn fn [mut ex, result] () { + ex.run() or { + result <- err.msg() + return + } + result <- 'ok' + }() + + assert wait_for_bool(ran, 'owner callback did not run') + assert wait_for_string(result, 'run did not return after owner stop') == 'ok' + ex.wait()! + runner.wait() +} + +fn test_wait_from_owner_callback_after_stop_returns_error_without_consuming_wait() { + mut ex := new(queue_size: 2)! + wait_result := chan string{cap: 1} + run_result := chan string{cap: 1} + + ex.try_post(fn [mut ex, wait_result] () ! { + ex.stop() + ex.wait() or { + wait_result <- err.msg() + return + } + wait_result <- 'ok' + })! + + runner := spawn fn [mut ex, run_result] () { + ex.run() or { + run_result <- err.msg() + return + } + run_result <- 'ok' + }() + + assert wait_for_string(wait_result, 'owner wait did not return') == 'executor: wait cannot be called from executor owner thread' + assert wait_for_string(run_result, 'run did not return after owner stop') == 'ok' + ex.wait()! + runner.wait() +} + +fn test_wait_before_terminal_pump_returns_error() { + mut ex := new(queue_size: 1)! + ex.wait() or { + assert err.msg() == 'executor: wait requires a running pump or stopped executor' + ex.stop() + return + } + assert false, 'wait returned successfully before a terminal owner pump' +} + +fn test_wait_before_runner_run_does_not_consume_wait() { + mut ex := new(queue_size: 2)! + start_run := chan bool{cap: 1} + started := chan bool{cap: 1} + result := chan string{cap: 1} + + ex.try_post(fn [started] () ! { + started <- true + })! + + runner := spawn fn [mut ex, start_run, result] () { + _ := <-start_run + ex.run() or { + result <- err.msg() + return + } + result <- 'ok' + }() + + mut saw_precondition := false + ex.wait() or { + assert err.msg() == 'executor: wait requires a running pump or stopped executor' + saw_precondition = true + } + assert saw_precondition + + start_run <- true + assert wait_for_bool(started, 'runner did not start owner pump') + ex.stop() + assert wait_for_string(result, 'run did not return after stop') == 'ok' + ex.wait()! + mut saw_second_wait := false + ex.wait() or { + assert err.msg() == 'executor: wait was already called' + saw_second_wait = true + } + assert saw_second_wait + runner.wait() +} + +fn test_run_after_stop_closes_admission_and_second_run_is_stopped() { + mut ex := new(queue_size: 1)! + ex.stop() + + mut rejected_after_stop := false + ex.try_post(fn () ! {}) or { + assert err.msg() == 'executor: executor is closed' + rejected_after_stop = true + } + assert rejected_after_stop + + ex.run() or { + assert err.msg() == 'executor: executor is stopped' + return + } + assert false, 'run after stop returned successfully' +} + +fn test_second_wait_returns_error() { + mut ex := new(queue_size: 1)! + started := chan bool{cap: 1} + release := chan bool{cap: 1} + result := chan string{cap: 1} + + ex.try_post(fn [started, release] () ! { + started <- true + _ := <-release + })! + + runner := spawn fn [mut ex, result] () { + ex.run() or { + result <- err.msg() + return + } + result <- 'ok' + }() + + assert wait_for_bool(started, 'run did not start before wait test') + release <- true + ex.stop() + assert wait_for_string(result, 'run did not stop') == 'ok' + ex.wait()! + ex.wait() or { + assert err.msg() == 'executor: wait was already called' + runner.wait() + return + } + assert false, 'second wait returned successfully' +} + +fn test_second_concurrent_run_returns_error() { + mut ex := new(queue_size: 2)! + started := chan bool{cap: 1} + release := chan bool{cap: 1} + first_result := chan string{cap: 1} + second_result := chan string{cap: 1} + + ex.try_post(fn [started, release] () ! { + started <- true + _ := <-release + })! + + first := spawn fn [mut ex, first_result] () { + ex.run() or { + first_result <- err.msg() + return + } + first_result <- 'ok' + }() + assert wait_for_bool(started, 'first run did not start owner job') + + second := spawn fn [mut ex, second_result] () { + ex.run() or { + second_result <- err.msg() + return + } + second_result <- 'ok' + }() + + msg := wait_for_string(second_result, 'second run did not return') + assert msg == 'executor: owner pump is already running' + + release <- true + ex.stop() + assert wait_for_string(first_result, 'first run did not stop') == 'ok' + ex.wait()! + first.wait() + second.wait() +} + +fn wait_for_bool(signal chan bool, message string) bool { + select { + value := <-signal { + return value + } + 1 * time.second { + assert false, message + } + } + return false +} + +fn wait_for_int(signal chan int, message string) int { + select { + value := <-signal { + return value + } + 1 * time.second { + assert false, message + } + } + return 0 +} + +fn wait_for_string(signal chan string, message string) string { + select { + value := <-signal { + return value + } + 1 * time.second { + assert false, message + } + } + return '' +} diff --git a/vlib/x/executor/owner.v b/vlib/x/executor/owner.v new file mode 100644 index 000000000..9b0ebee1c --- /dev/null +++ b/vlib/x/executor/owner.v @@ -0,0 +1,44 @@ +module executor + +import sync + +// is_owner_thread reports whether the caller is currently inside an active owner pump. +pub fn (e &Executor) is_owner_thread() bool { + e.mutex.lock() + is_owner := e.owner_active && e.owner_thread_id == sync.thread_id() + e.mutex.unlock() + return is_owner +} + +// assert_owner_thread returns an error when called outside the active owner pump. +pub fn (e &Executor) assert_owner_thread() ! { + if !e.is_owner_thread() { + return error(err_not_owner_thread) + } +} + +fn (mut e Executor) begin_owner_pump(long_running bool) ! { + e.mutex.lock() + if e.stopped && long_running { + e.mutex.unlock() + return error(err_executor_stopped) + } + if e.pumping { + e.mutex.unlock() + return error(err_owner_pump_running) + } + e.pumping = true + e.run_active = long_running + e.owner_active = true + e.owner_thread_id = sync.thread_id() + e.mutex.unlock() +} + +fn (mut e Executor) end_owner_pump() { + e.mutex.lock() + e.pumping = false + e.run_active = false + e.owner_active = false + e.owner_thread_id = 0 + e.mutex.unlock() +} diff --git a/vlib/x/executor/owner_test.v b/vlib/x/executor/owner_test.v new file mode 100644 index 000000000..afeaae89b --- /dev/null +++ b/vlib/x/executor/owner_test.v @@ -0,0 +1,64 @@ +module executor + +import time + +fn test_owner_checks_are_true_only_while_pumping() { + mut ex := new(queue_size: 2)! + observed := chan bool{cap: 1} + + assert !ex.is_owner_thread() + mut rejected_before_pump := false + ex.assert_owner_thread() or { + assert err.msg() == 'executor: current thread is not the executor owner' + rejected_before_pump = true + } + assert rejected_before_pump + + ex.try_post(fn [mut ex, observed] () ! { + observed <- ex.is_owner_thread() + ex.assert_owner_thread()! + })! + + assert ex.run_one()! + assert wait_for_bool(observed, 'owner check was false inside owner callback') + assert !ex.is_owner_thread() + ex.assert_owner_thread() or { + assert err.msg() == 'executor: current thread is not the executor owner' + ex.stop() + assert !ex.run_one()! + ex.wait()! + return + } + assert false, 'assert_owner_thread succeeded after owner pump returned' +} + +fn test_owner_identity_is_cleared_after_job_error() { + mut ex := new(queue_size: 1)! + observed := chan bool{cap: 1} + + ex.try_post(fn [mut ex, observed] () ! { + observed <- ex.is_owner_thread() + return error('owner error') + })! + + mut saw_job_error := false + ex.run_one() or { + assert err.msg() == 'owner error' + saw_job_error = true + } + assert saw_job_error + assert wait_for_bool(observed, 'owner identity was not set during erroring job') + assert !ex.is_owner_thread() +} + +fn wait_for_bool(signal chan bool, message string) bool { + select { + value := <-signal { + return value + } + 1 * time.second { + assert false, message + } + } + return false +} diff --git a/vlib/x/executor/pump_test.v b/vlib/x/executor/pump_test.v new file mode 100644 index 000000000..36b87567e --- /dev/null +++ b/vlib/x/executor/pump_test.v @@ -0,0 +1,88 @@ +module executor + +import time + +fn test_run_one_executes_one_job_and_reports_true() { + mut ex := new(queue_size: 4)! + values := chan int{cap: 2} + + ex.try_post(fn [values] () ! { + values <- 1 + })! + ex.try_post(fn [values] () ! { + values <- 2 + })! + + assert ex.run_one()! + assert wait_for_int(values, 'first job did not run') == 1 + assert ex.run_one()! + assert wait_for_int(values, 'second job did not run') == 2 + assert !ex.run_one()! + + ex.stop() + assert !ex.run_one()! + ex.wait()! +} + +fn test_drain_pending_runs_at_most_limit() { + mut ex := new(queue_size: 4)! + values := chan int{cap: 4} + for i in 0 .. 4 { + ex.try_post(fn [values, i] () ! { + values <- i + })! + } + + assert ex.drain_pending(2)! == 2 + assert wait_for_int(values, 'first drained job missing') == 0 + assert wait_for_int(values, 'second drained job missing') == 1 + + assert ex.drain_pending(8)! == 2 + assert wait_for_int(values, 'third drained job missing') == 2 + assert wait_for_int(values, 'fourth drained job missing') == 3 + + ex.stop() + assert ex.drain_pending(1)! == 0 + ex.wait()! +} + +fn test_drain_pending_rejects_non_positive_limit() { + mut ex := new(queue_size: 1)! + ex.drain_pending(0) or { + assert err.msg() == 'executor: drain limit must be positive' + ex.stop() + return + } + assert false, 'drain_pending accepted a zero limit' +} + +fn test_single_producer_fifo_order_is_preserved() { + mut ex := new(queue_size: 8)! + values := chan int{cap: 5} + for i in 0 .. 5 { + ex.try_post(fn [values, i] () ! { + values <- i + })! + } + + assert ex.drain_pending(5)! == 5 + for expected in 0 .. 5 { + assert wait_for_int(values, 'FIFO job missing') == expected + } + + ex.stop() + assert !ex.run_one()! + ex.wait()! +} + +fn wait_for_int(signal chan int, message string) int { + select { + value := <-signal { + return value + } + 1 * time.second { + assert false, message + } + } + return 0 +} diff --git a/vlib/x/executor/run.v b/vlib/x/executor/run.v new file mode 100644 index 000000000..df232cf0a --- /dev/null +++ b/vlib/x/executor/run.v @@ -0,0 +1,178 @@ +module executor + +import sync + +// run blocks on the caller thread and executes accepted jobs serially. +// +// It exits after stop() or the first job error closes admission and all accepted +// jobs have been drained. +pub fn (mut e Executor) run() ! { + e.begin_owner_pump(true)! + defer { + e.end_owner_pump() + } + for { + e.mutex.lock() + job := e.pop_job_locked() or { + if !e.accepting { + err := e.finish_terminal_locked() + e.mutex.unlock() + if err !is none { + return err + } + return + } + job_ready := e.job_ready + e.mutex.unlock() + _ := <-job_ready + continue + } + e.mutex.unlock() + e.execute_job(job) + } +} + +// run_one executes at most one accepted job and reports whether a job ran. +pub fn (mut e Executor) run_one() !bool { + e.begin_owner_pump(false)! + defer { + e.end_owner_pump() + } + e.mutex.lock() + job := e.pop_job_locked() or { + if !e.accepting { + err := e.finish_terminal_locked() + e.mutex.unlock() + if err !is none { + return err + } + return false + } + e.mutex.unlock() + return false + } + e.mutex.unlock() + + job_err := e.execute_job(job) + terminal_err := e.finish_if_drained() + if job_err !is none { + return job_err + } + if terminal_err !is none { + return terminal_err + } + return true +} + +// drain_pending executes up to max_jobs jobs that are already accepted. +pub fn (mut e Executor) drain_pending(max_jobs int) !int { + if max_jobs <= 0 { + return error(err_drain_limit_invalid) + } + e.begin_owner_pump(false)! + defer { + e.end_owner_pump() + } + mut ran := 0 + mut first_err := IError(none) + for ran < max_jobs { + e.mutex.lock() + job := e.pop_job_locked() or { + if !e.accepting { + err := e.finish_terminal_locked() + e.mutex.unlock() + if err !is none { + return err + } + if first_err !is none { + return first_err + } + return ran + } + e.mutex.unlock() + if first_err !is none { + return first_err + } + return ran + } + e.mutex.unlock() + + job_err := e.execute_job(job) + ran++ + if job_err !is none && first_err is none { + first_err = job_err + } + } + terminal_err := e.finish_if_drained() + if terminal_err !is none { + return terminal_err + } + if first_err !is none { + return first_err + } + return ran +} + +// stop closes admission and wakes any blocked submitters or owner run loop. +// +// It is idempotent and non-blocking. Accepted jobs are drained by the owner +// pump; stop() itself never executes user callbacks. +pub fn (mut e Executor) stop() { + e.mutex.lock() + e.close_admission_locked() + if e.queue.len == 0 && e.active == 0 { + e.finish_terminal_locked() + } + e.mutex.unlock() +} + +// wait waits for a running owner pump to reach terminal state. +// +// A valid wait is one-shot. Calling it from the owner callback, or before a +// terminal pump is active, returns a stable precondition error without consuming +// that one allowed wait. +pub fn (mut e Executor) wait() ! { + thread_id := sync.thread_id() + e.mutex.lock() + if e.owner_active && e.owner_thread_id == thread_id { + e.mutex.unlock() + return error(err_wait_owner_thread) + } + if e.waited { + e.mutex.unlock() + return error(err_wait_called) + } + if e.stopped { + e.waited = true + err := e.first_err + e.mutex.unlock() + if err !is none { + return err + } + return + } + if !e.run_active { + e.mutex.unlock() + return error(err_wait_before_terminal) + } + e.waited = true + done := e.done + e.mutex.unlock() + + _ := <-done + err := e.get_first_error() + if err !is none { + return err + } +} + +fn (mut e Executor) finish_if_drained() IError { + e.mutex.lock() + if !e.accepting && e.queue.len == 0 && e.active == 0 { + err := e.finish_terminal_locked() + e.mutex.unlock() + return err + } + e.mutex.unlock() + return none +} diff --git a/vlib/x/executor/stress_test.v b/vlib/x/executor/stress_test.v new file mode 100644 index 000000000..e78692c84 --- /dev/null +++ b/vlib/x/executor/stress_test.v @@ -0,0 +1,112 @@ +module executor + +import context +import time + +fn test_many_producers_submit_bounded_work() { + mut ex := new(queue_size: 16)! + accepted := chan bool{cap: 16} + ran := chan int{cap: 16} + mut producers := []thread{} + + for i in 0 .. 16 { + producers << spawn fn [mut ex, accepted, ran, i] () { + ex.post_with_context(context.background(), fn [ran, i] () ! { + ran <- i + }) or { + accepted <- false + return + } + accepted <- true + }() + } + + for _ in 0 .. 16 { + assert wait_for_bool(accepted, 'producer did not finish admission') + } + assert ex.drain_pending(16)! == 16 + + mut total := 0 + for _ in 0 .. 16 { + total += wait_for_int(ran, 'producer job did not run') + } + assert total == 120 + + for producer in producers { + producer.wait() + } + ex.stop() + assert !ex.run_one()! + ex.wait()! +} + +fn test_stop_wakes_waiting_submitter_without_accepting_job() { + mut ex := new(queue_size: 1)! + attempting := chan bool{cap: 1} + result := chan string{cap: 1} + ran := chan bool{cap: 1} + + ex.try_post(fn () ! {})! + submitter := spawn fn [mut ex, attempting, result, ran] () { + attempting <- true + ex.post_with_context(context.background(), fn [ran] () ! { + ran <- true + }) or { + result <- err.msg() + return + } + result <- 'accepted' + }() + + assert wait_for_bool(attempting, 'waiting submitter did not start') + ex.stop() + msg := wait_for_string(result, 'waiting submitter was not woken by stop') + assert msg == 'executor: executor is closed' + assert_no_bool(ran, 'job was accepted after stop') + submitter.wait() +} + +fn wait_for_bool(signal chan bool, message string) bool { + select { + value := <-signal { + return value + } + 1 * time.second { + assert false, message + } + } + return false +} + +fn wait_for_int(signal chan int, message string) int { + select { + value := <-signal { + return value + } + 1 * time.second { + assert false, message + } + } + return 0 +} + +fn wait_for_string(signal chan string, message string) string { + select { + value := <-signal { + return value + } + 1 * time.second { + assert false, message + } + } + return '' +} + +fn assert_no_bool(signal chan bool, message string) { + select { + _ := <-signal { + assert false, message + } + 50 * time.millisecond {} + } +} diff --git a/vlib/x/executor/submit.v b/vlib/x/executor/submit.v new file mode 100644 index 000000000..d1cfdb72e --- /dev/null +++ b/vlib/x/executor/submit.v @@ -0,0 +1,170 @@ +module executor + +import context +import sync +import time + +// try_post accepts f if the executor is open and queue space is available. +// +// It never waits for queue capacity. A full queue returns +// `executor: queue is full`, making backpressure explicit. +pub fn (mut e Executor) try_post(f JobFn) ! { + if f == unsafe { nil } { + return error(err_nil_job) + } + e.enqueue_job_now(ExecutorJob{ + f: f + report_error: true + })! +} + +// post_with_context waits until f can be admitted, parent is canceled, or the executor closes. +// +// The parent context bounds admission only. Once accepted, f will run on a +// future owner pump unless the application never pumps the executor. +// Active-owner submissions are accepted while capacity is available. If they +// would need to wait for capacity, they fail with +// `executor: owner thread cannot wait for queue capacity`. +pub fn (mut e Executor) post_with_context(parent context.Context, f JobFn) ! { + if f == unsafe { nil } { + return error(err_nil_job) + } + e.enqueue_job_with_context(parent, ExecutorJob{ + f: f + report_error: true + })! +} + +// post_with_timeout waits up to timeout for f to be admitted. +// +// The timeout bounds admission only and does not preempt a running callback. +// Active-owner submissions are accepted while capacity is available. If they +// would need to wait for capacity, they fail with +// `executor: owner thread cannot wait for queue capacity`. +pub fn (mut e Executor) post_with_timeout(timeout time.Duration, f JobFn) ! { + if f == unsafe { nil } { + return error(err_nil_job) + } + if timeout <= 0 { + return error(err_timeout) + } + e.enqueue_job_with_timeout(time.now().add(timeout), ExecutorJob{ + f: f + report_error: true + })! +} + +fn (mut e Executor) enqueue_job_now(job ExecutorJob) ! { + e.mutex.lock() + if !e.accepting { + e.mutex.unlock() + return error(err_executor_closed) + } + if e.queue.len >= e.queue_size { + e.mutex.unlock() + return error(err_queue_full) + } + e.queue << job + e.wake_job_ready_locked() + e.mutex.unlock() +} + +fn (mut e Executor) enqueue_job_with_context(parent context.Context, job ExecutorJob) ! { + thread_id := sync.thread_id() + 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 { + e.mutex.lock() + if !e.accepting { + e.mutex.unlock() + return error(err_executor_closed) + } + if watch_done { + err := submit_ctx.err() + if err !is none { + e.mutex.unlock() + return err + } + } + if e.queue.len < e.queue_size { + e.queue << job + e.wake_job_ready_locked() + e.mutex.unlock() + return + } + if e.owner_active && e.owner_thread_id == thread_id { + e.mutex.unlock() + return error(err_owner_submit_wait) + } + submit_ready := e.submit_ready + e.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 + } + } + } +} + +fn (mut e Executor) enqueue_job_with_timeout(deadline time.Time, job ExecutorJob) ! { + thread_id := sync.thread_id() + for { + e.mutex.lock() + if !e.accepting { + e.mutex.unlock() + return error(err_executor_closed) + } + remaining := deadline - time.now() + if remaining <= 0 { + e.mutex.unlock() + return error(err_timeout) + } + if e.queue.len < e.queue_size { + e.queue << job + e.wake_job_ready_locked() + e.mutex.unlock() + return + } + if e.owner_active && e.owner_thread_id == thread_id { + e.mutex.unlock() + return error(err_owner_submit_wait) + } + submit_ready := e.submit_ready + e.mutex.unlock() + select { + _ := <-submit_ready { + continue + } + remaining { + return error(err_timeout) + } + } + } +} diff --git a/vlib/x/executor/sync_call.v b/vlib/x/executor/sync_call.v new file mode 100644 index 000000000..ca1e70be2 --- /dev/null +++ b/vlib/x/executor/sync_call.v @@ -0,0 +1,44 @@ +module executor + +import context +import sync + +// run_sync runs f on the owner thread and waits until it finishes. +// +// Calls from the active owner pump execute inline, outside executor mutexes, to +// avoid deadlock. Inline calls still require open admission; an inline error is +// stored as the executor's first job error even if the outer callback catches it. +// That inline execution is outside FIFO queue ordering. +pub fn (mut e Executor) run_sync(f JobFn) ! { + if f == unsafe { nil } { + return error(err_nil_job) + } + thread_id := sync.thread_id() + e.mutex.lock() + is_owner := e.owner_active && e.owner_thread_id == thread_id + accepting := e.accepting + e.mutex.unlock() + if is_owner { + if !accepting { + return error(err_executor_closed) + } + f() or { + e.set_first_error(err) + return err + } + return + } + + result_ch := chan ExecutorJobResult{cap: 1} + job := ExecutorJob{ + f: f + result_ch: result_ch + has_result: true + report_error: true + } + e.enqueue_job_with_context(context.background(), job)! + result := <-result_ch + if result.err !is none { + return result.err + } +} diff --git a/vlib/x/executor/sync_call_test.v b/vlib/x/executor/sync_call_test.v new file mode 100644 index 000000000..172f90f50 --- /dev/null +++ b/vlib/x/executor/sync_call_test.v @@ -0,0 +1,319 @@ +module executor + +import time + +fn test_run_sync_from_foreign_thread_waits_until_owner_executes_job() { + mut ex := new(queue_size: 4)! + sync_started := chan bool{cap: 1} + sync_ran := chan bool{cap: 1} + sync_done := chan bool{cap: 1} + run_result := chan string{cap: 1} + + runner := spawn fn [mut ex, run_result] () { + ex.run() or { + run_result <- err.msg() + return + } + run_result <- 'ok' + }() + + caller := spawn fn [mut ex, sync_started, sync_ran, sync_done] () { + sync_started <- true + ex.run_sync(fn [sync_ran] () ! { + sync_ran <- true + }) or { + sync_done <- false + return + } + sync_done <- true + }() + + assert wait_for_bool(sync_started, 'run_sync caller did not start') + assert wait_for_bool(sync_ran, 'run_sync job did not run on owner pump') + assert wait_for_bool(sync_done, 'run_sync caller did not finish') + ex.stop() + assert wait_for_string(run_result, 'run did not stop') == 'ok' + ex.wait()! + caller.wait() + runner.wait() +} + +fn test_run_sync_returns_job_error_to_foreign_caller() { + mut ex := new(queue_size: 2)! + run_result := chan string{cap: 1} + sync_result := chan string{cap: 1} + + runner := spawn fn [mut ex, run_result] () { + ex.run() or { + run_result <- err.msg() + return + } + run_result <- 'ok' + }() + + caller := spawn fn [mut ex, sync_result] () { + ex.run_sync(fn () ! { + return error('sync job failed') + }) or { + sync_result <- err.msg() + return + } + sync_result <- 'ok' + }() + + assert wait_for_string(sync_result, 'run_sync error was not returned') == 'sync job failed' + assert wait_for_string(run_result, 'run did not return after sync error') == 'sync job failed' + ex.wait() or { assert err.msg() == 'sync job failed' } + caller.wait() + runner.wait() +} + +fn test_run_sync_from_owner_runs_inline_without_deadlock() { + mut ex := new(queue_size: 4)! + events := chan int{cap: 4} + + ex.try_post(fn [mut ex, events] () ! { + events <- 1 + ex.run_sync(fn [events] () ! { + events <- 100 + })! + events <- 2 + })! + ex.try_post(fn [events] () ! { + events <- 3 + })! + + assert ex.drain_pending(2)! == 2 + assert wait_for_int(events, 'outer job did not start') == 1 + assert wait_for_int(events, 'inline run_sync job did not run') == 100 + assert wait_for_int(events, 'outer job did not finish') == 2 + assert wait_for_int(events, 'second queued job did not run') == 3 + + ex.stop() + assert !ex.run_one()! + ex.wait()! +} + +fn test_run_sync_owner_inline_error_caught_closes_admission_and_wait_returns_error() { + mut ex := new(queue_size: 4)! + events := chan int{cap: 3} + post_result := chan string{cap: 1} + run_result := chan string{cap: 1} + + ex.try_post(fn [mut ex, events, post_result] () ! { + events <- 1 + ex.run_sync(fn () ! { + return error('inline sync failed') + }) or { events <- 2 } + ex.try_post(fn () ! {}) or { + post_result <- err.msg() + return + } + post_result <- 'accepted' + ex.stop() + })! + ex.try_post(fn [events] () ! { + events <- 3 + })! + + runner := spawn fn [mut ex, run_result] () { + ex.run() or { + run_result <- err.msg() + return + } + run_result <- 'ok' + }() + + assert wait_for_int(events, 'outer job did not start') == 1 + assert wait_for_int(events, 'inline run_sync error was not caught') == 2 + assert wait_for_string(post_result, 'try_post after inline error did not return') == 'executor: executor is closed' + assert wait_for_int(events, 'accepted job after inline error was not drained') == 3 + assert wait_for_string(run_result, 'run did not return inline first error') == 'inline sync failed' + ex.wait() or { + assert err.msg() == 'inline sync failed' + runner.wait() + return + } + assert false, 'wait returned successfully after inline run_sync error' +} + +fn test_run_one_returns_owner_inline_run_sync_error_caught_by_outer_job() { + mut ex := new(queue_size: 4)! + events := chan int{cap: 2} + + ex.try_post(fn [mut ex, events] () ! { + ex.run_sync(fn () ! { + return error('inline sync failed') + }) or { events <- 1 } + })! + ex.try_post(fn [events] () ! { + events <- 2 + })! + + mut saw_inline_error := false + ex.run_one() or { + assert err.msg() == 'inline sync failed' + saw_inline_error = true + } + assert saw_inline_error + assert wait_for_int(events, 'outer job did not catch inline run_sync error') == 1 + + mut saw_closed := false + ex.try_post(fn () ! {}) or { + assert err.msg() == 'executor: executor is closed' + saw_closed = true + } + assert saw_closed + + ex.run_one() or { assert err.msg() == 'inline sync failed' } + assert wait_for_int(events, 'accepted job after inline error was not drained') == 2 + ex.wait() or { + assert err.msg() == 'inline sync failed' + return + } + assert false, 'wait returned successfully after inline run_sync error' +} + +fn test_drain_pending_limit_returns_owner_inline_run_sync_error_caught_by_outer_job() { + mut ex := new(queue_size: 4)! + events := chan int{cap: 2} + + ex.try_post(fn [mut ex, events] () ! { + ex.run_sync(fn () ! { + return error('inline sync failed') + }) or { events <- 1 } + })! + ex.try_post(fn [events] () ! { + events <- 2 + })! + + ex.drain_pending(1) or { assert err.msg() == 'inline sync failed' } + assert wait_for_int(events, 'outer job did not catch inline run_sync error') == 1 + + ex.drain_pending(8) or { assert err.msg() == 'inline sync failed' } + assert wait_for_int(events, 'accepted job after inline error was not drained') == 2 + ex.wait() or { + assert err.msg() == 'inline sync failed' + return + } + assert false, 'wait returned successfully after inline run_sync error' +} + +fn test_run_sync_from_owner_after_stop_returns_closed_without_running_nested_job() { + mut ex := new(queue_size: 2)! + result := chan string{cap: 1} + nested_ran := chan bool{cap: 1} + + ex.try_post(fn [mut ex, result, nested_ran] () ! { + ex.stop() + ex.run_sync(fn [nested_ran] () ! { + nested_ran <- true + }) or { + result <- err.msg() + return + } + result <- 'ok' + })! + + assert ex.run_one()! + assert wait_for_string(result, 'owner run_sync after stop did not return') == 'executor: executor is closed' + select { + _ := <-nested_ran { + assert false, 'nested run_sync job ran after stop' + } + else {} + } + ex.wait()! +} + +fn test_run_sync_from_owner_after_prior_job_error_is_refused_without_nested_execution() { + mut ex := new(queue_size: 3)! + result := chan string{cap: 1} + nested_ran := chan bool{cap: 1} + + ex.try_post(fn () ! { + return error('first job failed') + })! + ex.try_post(fn [mut ex, result, nested_ran] () ! { + ex.run_sync(fn [nested_ran] () ! { + nested_ran <- true + }) or { + result <- err.msg() + return + } + result <- 'ok' + })! + + mut run_result := 'ok' + ex.run() or { run_result = err.msg() } + assert run_result == 'first job failed' + assert wait_for_string(result, 'owner run_sync after first error did not return') == 'executor: executor is closed' + select { + _ := <-nested_ran { + assert false, 'nested run_sync job ran after first error' + } + else {} + } + ex.wait() or { + assert err.msg() == 'first job failed' + return + } + assert false, 'wait returned successfully after first job error' +} + +fn test_run_sync_after_stop_returns_error() { + mut ex := new(queue_size: 1)! + ex.stop() + ex.run_sync(fn () ! {}) or { + assert err.msg() == 'executor: executor is closed' + return + } + assert false, 'run_sync accepted after stop' +} + +fn test_run_sync_rejects_nil_job() { + mut ex := new(queue_size: 1)! + nil_job := unsafe { JobFn(nil) } + ex.run_sync(nil_job) or { + assert err.msg() == 'executor: job function is nil' + ex.stop() + return + } + assert false, 'run_sync accepted a nil job' +} + +fn wait_for_bool(signal chan bool, message string) bool { + select { + value := <-signal { + return value + } + 1 * time.second { + assert false, message + } + } + return false +} + +fn wait_for_int(signal chan int, message string) int { + select { + value := <-signal { + return value + } + 1 * time.second { + assert false, message + } + } + return 0 +} + +fn wait_for_string(signal chan string, message string) string { + select { + value := <-signal { + return value + } + 1 * time.second { + assert false, message + } + } + return '' +} diff --git a/vlib/x/executor/tools/README.md b/vlib/x/executor/tools/README.md new file mode 100644 index 000000000..bcc64ac75 --- /dev/null +++ b/vlib/x/executor/tools/README.md @@ -0,0 +1,44 @@ +# x.executor tools + +This folder contains local helper scripts for safe validation of `x.executor`. +Scripts must use repository-relative paths and must not depend on local machine +paths, secrets, or external services. + +## `check_no_async_dependency.sh` + +Checks the standalone dependency rule: + +- no V source under `vlib/x/executor` may contain `import x.async`; +- examples may not mention or use `x.async`; +- no V source file in the module may be named as an async bridge. + +From the repository root: + +```sh +sh vlib/x/executor/tools/check_no_async_dependency.sh +``` + +## `validate.sh` + +Runs the guarded validation path: + +- `check_no_async_dependency.sh`. +- `./v fmt -verify` for module tests, examples, and benchmarks V files. +- each public example under `vlib/x/executor/examples/`, run serially. +- `./v test vlib/x/executor`. +- `./v -prod test vlib/x/executor`. + +The script creates a fresh temporary root, sets isolated `VTMP` and `VCACHE`, +uses the repository-local `./v`, and runs commands serially. This avoids V +runner artefact/cache collisions when multiple external runners share the same +checkout/cache. + +From the repository root: + +```sh +sh vlib/x/executor/tools/validate.sh +``` + +If a crash appears through this serialized and isolated path, treat it as a +blocking runtime/test signal. Do not classify it as tooling noise without a new +investigation. diff --git a/vlib/x/executor/tools/check_no_async_dependency.sh b/vlib/x/executor/tools/check_no_async_dependency.sh new file mode 100755 index 000000000..2c1811082 --- /dev/null +++ b/vlib/x/executor/tools/check_no_async_dependency.sh @@ -0,0 +1,39 @@ +#!/usr/bin/env sh +set -eu + +script_dir=$(CDPATH= cd "$(dirname "$0")" && pwd) +repo_root=$(CDPATH= cd "$script_dir/../../../.." && pwd) +cd "$repo_root" + +target="vlib/x/executor" + +if [ ! -d "$target" ]; then + echo "$target does not exist" >&2 + exit 1 +fi + +if find "$target" -type f -name '*.v' -print | xargs grep -n 'import x\.async' >/tmp/xexecutor_async_imports.$$ 2>/dev/null; then + echo "x.executor must not import x.async:" >&2 + cat /tmp/xexecutor_async_imports.$$ >&2 + rm -f /tmp/xexecutor_async_imports.$$ + exit 1 +fi +rm -f /tmp/xexecutor_async_imports.$$ + +if [ -d "$target/examples" ] && find "$target/examples" -type f -print | xargs grep -n 'x\.async' >/tmp/xexecutor_async_examples.$$ 2>/dev/null; then + echo "x.executor examples must not mention or use x.async:" >&2 + cat /tmp/xexecutor_async_examples.$$ >&2 + rm -f /tmp/xexecutor_async_examples.$$ + exit 1 +fi +rm -f /tmp/xexecutor_async_examples.$$ + +if find "$target" -type f -name '*.v' | grep -Ei '(^|/)(async|.*bridge.*)\.v$' >/tmp/xexecutor_async_bridge_files.$$ 2>/dev/null; then + echo "x.executor must not add async or bridge V files:" >&2 + cat /tmp/xexecutor_async_bridge_files.$$ >&2 + rm -f /tmp/xexecutor_async_bridge_files.$$ + exit 1 +fi +rm -f /tmp/xexecutor_async_bridge_files.$$ + +echo "x.executor dependency check passed" diff --git a/vlib/x/executor/tools/validate.sh b/vlib/x/executor/tools/validate.sh new file mode 100755 index 000000000..3840f857e --- /dev/null +++ b/vlib/x/executor/tools/validate.sh @@ -0,0 +1,46 @@ +#!/usr/bin/env sh +set -eu + +script_dir=$(CDPATH= cd "$(dirname "$0")" && pwd) +repo_root=$(CDPATH= cd "$script_dir/../../../.." && pwd) +cd "$repo_root" + +if [ ! -x ./v ]; then + echo "x.executor validation must be run from a V checkout with local ./v" >&2 + exit 1 +fi + +tmp_root=$(mktemp -d "${TMPDIR:-/tmp}/xexecutor-validate.XXXXXX") +cleanup() { + rm -rf "$tmp_root" +} +trap cleanup EXIT INT TERM + +vtmp="$tmp_root/vtmp" +vcache="$tmp_root/vcache" +mkdir -p "$vtmp" "$vcache" + +run_v() { + seconds=$1 + shift + echo "+ ./v $*" + if command -v timeout >/dev/null 2>&1; then + timeout "$seconds" env VTMP="$vtmp" VCACHE="$vcache" ./v "$@" + else + env VTMP="$vtmp" VCACHE="$vcache" ./v "$@" + fi +} + +sh vlib/x/executor/tools/check_no_async_dependency.sh + +v_files=$(find vlib/x/executor -type f -name '*.v' | sort) +example_files=$(find vlib/x/executor/examples -type f -name '*.v' | sort) + +run_v 60 fmt -verify $v_files + +for example in $example_files; do + run_v 60 run "$example" +done + +run_v 120 test vlib/x/executor +run_v 180 -prod test vlib/x/executor -- 2.39.5