diff --git a/docs/backlogs/000176-quantify-rollback-saturation-alongside-mandatory-ddl.md b/docs/backlogs/000176-quantify-rollback-saturation-alongside-mandatory-ddl.md new file mode 100644 index 00000000..eae32bc0 --- /dev/null +++ b/docs/backlogs/000176-quantify-rollback-saturation-alongside-mandatory-ddl.md @@ -0,0 +1,44 @@ +# Backlog: Quantify rollback saturation alongside mandatory DDL + +## Summary + +doradb-bench lacks product-level scenarios that measure large transaction rollback and heterogeneous overlap among caller DDL or maintenance and transaction cleanup on the shared mandatory runtime. + +## Reference + +Task docs/tasks/000252-mandatory-runtime-lifecycle-fairness-evolution-readiness.md; RFC docs/rfcs/0026-engine-owned-mandatory-background-runtime.md Phase 5; docs/benchmark-tool.md; backlog 000147 covers checkpoint lifecycle only. + +## Deferred From (Optional) + +docs/tasks/000252-mandatory-runtime-lifecycle-fairness-evolution-readiness.md; docs/rfcs/0026-engine-owned-mandatory-background-runtime.md Phase 5 + +## Deferral Context (Optional) + +- Defer Reason: Task 000252 closes fixed-runtime lifecycle and fairness evidence using existing benchmark workloads; the approved scope forbids approximating product workloads that do not yet exist. +- Findings: Existing doradb-bench commands cover table and index DDL plus homogeneous no-op, insert, lookup, scan, and stream controls, but not large rollback, general maintenance, or heterogeneous caller and cleanup overlap. Meaningful measurements require explicit preparation, reset, role, configuration, artifact, and process-lifecycle choices. +- Direction Hint: Begin with a product benchmark design. Define caller and cleanup roles, transaction size and state preparation, one-runner and multi-runner matrices, caller limits, reuse and cleanup rules, and comparable output artifacts. Reuse the existing statistics boundary and avoid private test harness substitutes; preserve backlog 000147 for checkpoint and freeze lifecycle scenarios. + +## Scope Hint + +Design benchmark workload roles, data preparation, reset and reuse rules, runtime sizing, operation counts, result artifacts, and process lifecycle for large rollback plus mixed caller and internal mandatory work. Keep correctness stress and scheduling-policy design outside benchmark implementation. + +## Acceptance Hint + +doradb-bench can run repeatable isolated large-rollback and heterogeneous mandatory-runtime workloads, reports runtime statistics and timing artifacts with explicit runner and caller-limit configuration, and documents fair baseline and candidate setup without a test-only surrogate. + +## Notes (Optional) + + +## Close Reason (Added When Closed) + +When a backlog item is moved to `docs/backlogs/closed/`, append: + +```md +## Close Reason + +- Type: +- Detail: +- Closed By: +- Reference: +- Closed At: +``` diff --git a/docs/backlogs/next-id b/docs/backlogs/next-id index b235d700..6505223c 100644 --- a/docs/backlogs/next-id +++ b/docs/backlogs/next-id @@ -1 +1 @@ -000176 +000177 diff --git a/docs/benchmark-tool.md b/docs/benchmark-tool.md index 84a05d60..9280db45 100644 --- a/docs/benchmark-tool.md +++ b/docs/benchmark-tool.md @@ -164,7 +164,7 @@ workloads accepts `--seed`. | `--batch-size`, `-b` | `prepare`, insert and non-stream read workloads | `prepare`: `1`; `run`: manifest default | Operations per transaction. For inserts this means rows per commit; for reads this means lookup/index-scan requests or table-scan iterations per read transaction. | | `--seed` | `run insert-seq`, `insert-rand`, `lookup-rand`, `index-scan`, `index-stream` | `0` | `u64` reproducibility input for payload bytes, randomized insert order, randomized read key selection, or randomized scan bounds. | | `--log-sync` | `run ...` | `fsync` | Redo-log durability sync method. `fsync` and `fdatasync` submit the matching native file-sync operation; `none` skips durable sync and is crash-unsafe. | -| `--include-stats` | `run ...` | `false` | Captures and prints internal transaction-system, storage-IO, and buffer-pool stats. Omit this for prerequisite runs such as data loading before a measured read workload. | +| `--include-stats` | `run ...` | `false` | Captures and prints internal transaction-system, storage-IO, buffer-pool, and engine-global mandatory-runtime stats. Omit this for prerequisite runs such as data loading before a measured read workload. | Run defaults resolve as follows: @@ -221,7 +221,13 @@ errors are written to stderr. value size, batch size, seed, prepared index mode, loaded key range, threads, sessions, log sync mode, and table id. - `Internal Stats`, only with `--include-stats`: public transaction-system, - storage-IO, and buffer-pool stats deltas when available. + storage-IO, buffer-pool, and mandatory-runtime stats when available. The + mandatory snapshot is captured once per engine, not summed once per session; + its fixed names are `mandatory.operation.*` and + `mandatory.transaction_cleanup.*`. Monotonic fields are deltas and active + counts are the independently sampled ending values. Caller terminal counters + are published before their result waiters wake, so the ending snapshot + includes the final observed mandatory operation. - `Final Result`: operation count, inserted rows, found count, not-found count, returned rows, elapsed time, throughput, average nanoseconds per operation, and failures. @@ -282,7 +288,7 @@ RFC-0025: `trx-noop`. - Phase 2's no-per-item stream budget uses `index-stream`. - RFC-0026 Phase 2's runtime-owned table-DDL path uses `table-ddl`. -- Phase 5's successful index-DDL path uses `index-ddl`. +- RFC-0026 Phase 3's runtime-owned index-DDL path uses `index-ddl`. - Existing insert, lookup, table-scan, and index-scan workloads remain the row/index/page-loop evidence. @@ -316,14 +322,17 @@ baseline/candidate DDL trials should therefore use equivalently fresh prepared roots and normally one cycle per invocation: ```bash -rtk cargo run --release -p doradb-bench -- --root target/doradb-bench/rfc0025-table-ddl prepare --index none -rtk cargo run --release -p doradb-bench -- --root target/doradb-bench/rfc0025-table-ddl run table-ddl --log-sync none -rtk cargo run --release -p doradb-bench -- --root target/doradb-bench/rfc0025-index-ddl prepare --index none -rtk cargo run --release -p doradb-bench -- --root target/doradb-bench/rfc0025-index-ddl run index-ddl --log-sync none +rtk cargo run --release -p doradb-bench -- --root target/doradb-bench/rfc0026-table-ddl prepare --index none +rtk cargo run --release -p doradb-bench -- --root target/doradb-bench/rfc0026-table-ddl run table-ddl --log-sync none --include-stats +rtk cargo run --release -p doradb-bench -- --root target/doradb-bench/rfc0026-index-ddl prepare --index none +rtk cargo run --release -p doradb-bench -- --root target/doradb-bench/rfc0026-index-ddl run index-ddl --log-sync none --include-stats ``` The tool supplies workload shapes and fixed result artifacts, not repetition or aggregation. Users remain responsible for repeated paired baseline/candidate runs on the same host and configuration, then reporting median and dispersion. -Checkpoint and persisted/cold measurements remain deferred to the backlogs -linked at the start of this document. +Checkpoint, freeze, shutdown/reopen, and persisted/cold measurements remain +deferred to backlog 000147. Large rollback and heterogeneous DDL, maintenance, +and internal-cleanup measurements require their separately designed +`doradb-bench` backlog; the current homogeneous session runner and fixed engine +configuration are not a substitute performance harness for those roles. diff --git a/docs/engine-component-lifetime.md b/docs/engine-component-lifetime.md index 9d5615c1..fbb1583e 100644 --- a/docs/engine-component-lifetime.md +++ b/docs/engine-component-lifetime.md @@ -175,6 +175,23 @@ submitted synchronously without a lossy channel. Independent transactions can therefore clean up concurrently, while each transaction's rollback remains sequential. +`MandatoryRuntimeConfig::worker_threads` controls OS runners, not the accepted +caller count. `concurrency_limit` bounds accepted caller obligations, not +caller-side preparation futures or internal cleanup. Increasing caller +capacity can retain more logical locks, memory, and publication work without +increasing runner throughput. Increasing runners can increase storage and +metadata contention and cannot make blocking code cooperative. Configuration +is validated once during startup, rejects zero sizes, and cannot resize a +running engine. + +One runner provides concurrency only when accepted work reaches an await or +explicit yield that returns scheduler control. Multiple runners allow true +overlap, but neither configuration promises executor ordering, a queue-latency +bound, or a general fairness SLA. Internal admission is non-lossy and separate +from caller backpressure; it intentionally does not create a bounded cleanup +backlog because correctness obligations cannot be rejected after ownership is +claimed. + Mandatory results reuse the common completion cell through a move-once take path. The single observer owns no task, permit, engine reference, session authority, or prepared resource. Dropping it cannot cancel execution. A @@ -218,6 +235,64 @@ and completion waiters are published, and the permit is released exactly once. If the domain panic policy itself unwinds, the panic-minimal fallback retains the whole armed owner instead of dropping raw-reference-sensitive undo. +### Fixed-Class Statistics And Task Events + +`Session::mandatory_runtime_stats()` returns one engine-global snapshot with +fixed `operation` and `transaction_cleanup` classes. Each class publishes +monotonic `submitted_count`, `started_count`, `completed_count`, +`error_count`, `panic_count`, `detached_observer_count`, +`admission_wait_nanos`, `queue_wait_nanos`, and `execution_nanos` fields plus +the current authoritative `active_count`. Fields are independently sampled; +concurrent snapshots do not promise a transactionally consistent equation. +Caller terminal counts, outcomes, and execution time are recorded before +completion publication wakes the observer, so a snapshot taken immediately +after an observed result includes that result. `active_count` remains +independently sampled until the supervisor releases its permit. +The inspection remains available after poison while engine/session lifecycle +inspection is admitted and creates no runtime work. + +Accepted caller task labels are `create_table`, `drop_table`, `create_index`, +`drop_index`, `freeze_table`, `checkpoint_table`, `checkpoint_catalog`, +`truncate_redo_log`, `checkpoint_catalog_and_truncate_redo_log`, and +`cleanup_secondary_mem_indexes`. Internal cleanup labels are +`terminal_rollback`, `abandoned_transaction`, and `failed_precommit`. These +labels and the two class names are diagnostic vocabulary, not scheduling +policy or a per-label registry. + +Every accepted task emits debug records with +`event=mandatory_task component=mandatory_runtime`: `action=start result=ok` +includes immutable class, task, optional session-operation/table identities, +successful admission wait, and executor queue wait; `action=finish` reports +`result=ok|error|panic`, the same identity, execution time, and +`observer=attached|detached|none`. An unobserved ordinary error retains its +error-level `action=discard_unobserved` record, and task panic retains the +engine-poison error record. The storage crate does not install a logger. + +### Cooperative Poll Audit + +Accepted execution acquires no logical operation lock or metadata gate after +the synchronous `PreparedExecution::accept` edge. The bounded-poll audit found: + +- CREATE/DROP TABLE and DROP INDEX perform bounded state transitions around + awaited storage, transaction, lifecycle, or publication boundaries. +- CREATE INDEX hot-row collection and construction yield after their named + 128-row batches; cold input proceeds through awaited storage batches. Its + larger bounded-memory/parallel redesign remains backlog 000104. +- freeze/checkpoint, catalog checkpoint, redo retention/truncation, and + secondary `MemIndex` cleanup proceed through operation-specific awaited IO, + retry, scan-batch, or transaction boundaries. Synchronous filesystem regions + remain the runtime-independent blocking-work scope of backlog 000137. +- terminal rollback, abandoned cleanup, and failed-precommit cleanup use the + same row/index undo paths. Those paths explicitly yield after 128 completed + undo entries, after the current entry is unlinked and popped and before the + next entry is borrowed. +- normal finish and panic preservation perform fixed ownership publication or + move residual payloads into fatal retention; they do not reacquire operation + authority or loop on scheduler state. + +These boundaries provide cooperative progress evidence for the fixed runtime; +they do not establish preemption or a general starvation-free scheduler. + ## Admission, Shutdown, And Drop The engine lifecycle has three states: @@ -234,6 +309,12 @@ shutdown can proceed. work remains. The infallible `Engine::shutdown()` waits for the same work to drain and returns only after final teardown completes. +Lifecycle records distinguish `mode=try origin=explicit` from blocking +`mode=wait origin=explicit|owner_drop`. A busy try-shutdown record and its +returned attachment use the same `strong_refs`, `operation_blocked`, +`operation_state`, `voluntary_blocked`, `mandatory_session_blocked`, +`cleanup_queued`, `mandatory_callers`, and `mandatory_internal` fields. + Normal shutdown is: 1. close engine and mandatory caller admission and flip `Running -> ShuttingDown` @@ -309,9 +390,11 @@ registry-owned component owners start their final `QuiescentBox` drains. An unintended owner drop can therefore block indefinitely while caller-retained foreground work, runtime references, or engine-owned background work remains live. Callers should finish foreground work and invoke -explicit shutdown at a controlled point when blocking there is operationally -important. Drop does not cancel accepted work or tear down components before -that work reaches terminal state. +`try_shutdown` or explicit shutdown at a controlled point when blocker +diagnostics and blocking location are operationally important. Drop does not +cancel accepted work or tear down components before that work reaches terminal +state. Future priority or reserved-runner lanes, adaptive sizing, task groups, +and a separate blocking/CPU pool require workload evidence and separate design. ## Quiescent Ownership diff --git a/docs/public-error-audit.csv b/docs/public-error-audit.csv index 0ed79065..f0dc5e42 100644 --- a/docs/public-error-audit.csv +++ b/docs/public-error-audit.csv @@ -23,6 +23,7 @@ doradb-storage/src/session.rs,Session::drop_table,4 doradb-storage/src/session.rs,Session::freeze_table,5 doradb-storage/src/session.rs,Session::list_table_ids,1 doradb-storage/src/session.rs,Session::lock_table,2 +doradb-storage/src/session.rs,Session::mandatory_runtime_stats,1 doradb-storage/src/session.rs,Session::storage_io_stats,1 doradb-storage/src/session.rs,Session::total_row_pages,3 doradb-storage/src/session.rs,Session::transaction_system_stats,1 diff --git a/docs/rfcs/0026-engine-owned-mandatory-background-runtime.md b/docs/rfcs/0026-engine-owned-mandatory-background-runtime.md index d2d195aa..4c4e8361 100644 --- a/docs/rfcs/0026-engine-owned-mandatory-background-runtime.md +++ b/docs/rfcs/0026-engine-owned-mandatory-background-runtime.md @@ -1248,9 +1248,12 @@ focused validation. finalize engine/session/runtime shutdown diagnostics, task/result observability, blocking owner-drop drain, bounded-poll audits, configuration documentation, cross-operation stress tests, and paired - performance measurements. Synchronize RFC-0025 Phases 3 through 7 as - superseded by this RFC and preserve the fixed-runtime implementation - evidence that closed backlog 000123. [D3] [D7] [D9] [B1] + performance measurements through workloads already implemented in + `doradb-bench`. Checkpoint/maintenance and large rollback/mixed-runtime + performance shapes require dedicated benchmark design and are not + approximated by test-only harnesses in this phase. Synchronize RFC-0025 + Phases 3 through 7 as superseded by this RFC and preserve the fixed-runtime + implementation evidence that closed backlog 000123. [D3] [D7] [D9] [B1] - Goals: Demonstrate one execution owner, no dropped accepted payload, lossless shutdown wakeups, no transaction/statement hot-path overhead, bounded caller-operation backlog, progress for cleanup under @@ -1261,18 +1264,24 @@ focused validation. caller preparation plus atomic prepared-runtime submission; no legacy foreground handoff or runtime-side operation-lock acquisition remains. - Phase-local Choices: Finalize stable diagnostic labels/counters, select - focused stress repetition counts and benchmark thresholds, and determine - whether new workload evidence justifies a follow-up scheduling-policy RFC - or separate work on backlog 000167. + focused stress repetition counts and benchmark thresholds for existing + table/index DDL, no-op, insert, lookup, scan, and stream commands, and + determine whether deterministic correctness evidence justifies a follow-up + scheduling-policy RFC or separate work on backlog 000167. Missing + checkpoint/freeze/shutdown-reopen performance coverage remains backlog + 000147; missing large rollback and heterogeneous mandatory-runtime coverage + is recorded in backlog 000176 as separate deferred `doradb-bench` design. - Non-goals: Do not implement adaptive resizing, priority lanes, parallel recovery/checkpoint/index algorithms, forced shutdown, or explicit operation cancellation. - - Task Doc: `docs/tasks/TBD.md` - - Task Issue: `#0` - - Phase Status: `pending` - - Implementation Summary: `pending` + - Task Doc: `docs/tasks/000252-mandatory-runtime-lifecycle-fairness-evolution-readiness.md` + - Task Issue: `#931` + - Phase Status: done + - Implementation Summary: Implemented fixed-class observability, cooperative cleanup progress, coherent shutdown diagnostics, and the final RFC-0026 runtime evidence without changing accepted-operation ownership or adding scheduler policy. [Task Resolve Sync: docs/tasks/000252-mandatory-runtime-lifecycle-fairness-evolution-readiness.md @ 2026-08-03] - Related Backlogs: - `docs/backlogs/closed/000123-adaptive-background-worker-runtime.md` + - `docs/backlogs/000147-doradb-bench-checkpoint-lifecycle-scenarios.md` + - `docs/backlogs/000176-quantify-rollback-saturation-alongside-mandatory-ddl.md` ## Test Strategy diff --git a/docs/tasks/000252-mandatory-runtime-lifecycle-fairness-evolution-readiness.md b/docs/tasks/000252-mandatory-runtime-lifecycle-fairness-evolution-readiness.md new file mode 100644 index 00000000..9a47b275 --- /dev/null +++ b/docs/tasks/000252-mandatory-runtime-lifecycle-fairness-evolution-readiness.md @@ -0,0 +1,300 @@ +--- +id: 000252 +title: Lifecycle, Fairness, And Evolution Readiness for Mandatory Runtime +status: implemented # proposal | implemented | superseded +created: 2026-08-03 +github_issue: 931 +--- + +# Task: Lifecycle, Fairness, And Evolution Readiness for Mandatory Runtime + +## Summary + +Completed RFC-0026 Phase 5 by closing the fixed mandatory runtime's lifecycle, +fairness-evidence, observability, and evolution-readiness work. The shipped +runtime exposes fixed-class operation and transaction-cleanup statistics, +stable task lifecycle events, coherent explicit-shutdown and owner-Drop +diagnostics, and cooperative progress for long undo and CREATE INDEX work. + +Every production table/index DDL and effectful maintenance root remains +caller-prepared and atomically transferred to one engine-owned executor. +Deterministic one-runner and multi-runner tests prove cleanup progress and +accepted-task overlap without adding priority, reserved-runner, or adaptive +scheduling policy. + +Active lifecycle, transaction, benchmark, and RFC documentation now describes +the implemented ownership model. Paired measurements used only existing +`doradb-bench` workloads; checkpoint lifecycle and large-rollback or +heterogeneous-runtime workloads remain separately designed follow-ups. + +## Context + +`Issue Labels:` +`- type:task` +`- priority:medium` +`- codex` + +`Parent RFC:` +`- docs/rfcs/0026-engine-owned-mandatory-background-runtime.md` + +`Related Backlogs:` +`- docs/backlogs/closed/000123-adaptive-background-worker-runtime.md` +`- docs/backlogs/000104-stream-parallel-create-index-cold-build.md` +`- docs/backlogs/000137-runtime-agnostic-blocking-work-abstraction.md` +`- docs/backlogs/000147-doradb-bench-checkpoint-lifecycle-scenarios.md` +`- docs/backlogs/000167-logical-lock-deadlock-handling.md` +`- docs/backlogs/000176-quantify-rollback-saturation-alongside-mandatory-ddl.md` + +Tasks 000248 through 000251 established the engine-owned fixed executor, +concurrent transaction cleanup, caller-prepared DDL, and caller-prepared +maintenance. This phase retained their central contract: preparation acquires +the complete session operation, logical locks, metadata gates, and workflow +authority before capacity admission; acceptance synchronously consumes that +carrier; observer Drop never changes execution ownership. + +The runtime has two immutable classes. Bounded `operation` admission protects +caller obligations, while non-lossy `transaction_cleanup` admission bypasses +the caller limit. Both classes share fixed OS runners on +`async_executor::Executor`. Admission independence alone did not prove +progress because cooperative async execution can be monopolized by an +unbounded poll, so accepted paths required a bounded-poll audit and +deterministic scheduler evidence. + +The existing benchmark tool supported table/index DDL and homogeneous no-op, +insert, lookup, scan, and stream workloads. It did not define checkpoint, +general maintenance, large rollback, or heterogeneous mandatory-runtime +workloads. Product-level benchmark design for those shapes requires explicit +state preparation, reset/reuse, roles, runtime configuration, result artifacts, +and process lifecycle; test-only substitutes would not answer those choices. + +## Goals + +1. Preserve one engine execution owner for every accepted mandatory operation. +2. Keep caller admission bounded and internal cleanup submission non-lossy, + with separate authoritative active counts. +3. Publish low-overhead fixed-class task, outcome, observer, and timing + statistics without a live-task registry. +4. Emit stable structured start and finish events for success, error, panic, + and detached-observer outcomes. +5. Make try-shutdown, explicit blocking shutdown, and owner Drop report + coherent lifecycle diagnostics while retaining the lossless drain. +6. Prove accepted execution acquires no operation lock or metadata gate. +7. Bound synchronous polls where work scales with rows or undo records. +8. Prove cleanup progress under saturated caller admission and active DDL with + one runner, and true task overlap with multiple runners. +9. Document immutable runtime sizing, cooperative fairness, backpressure, + shutdown, statistics, and future scheduling boundaries. +10. Integrate runtime statistics into existing benchmark output without + touching ordinary transaction, statement, DML, read, scan, or stream paths. +11. Validate performance through existing workloads and preserve missing + workload design as explicit follow-up work. + +## Non-Goals + +1. No adaptive sizing, priority queue, reserved cleanup lane, task group, + work-stealing policy, or distinct blocking/CPU pool. +2. No wall-clock fairness SLA, maximum queue latency, executor ordering, or + general starvation-free scheduling promise. +3. No runtime watchdog, preemption, forced shutdown, or implicit/explicit + operation cancellation. +4. No operation-lock or metadata-gate acquisition inside accepted execution. +5. No change to operation results, persistence formats, redo, recovery, + publication, or storage algorithms. +6. No parallel recovery, checkpoint, rollback, or index construction. +7. No per-label maps, histograms, quantiles, central live-task registry, or + general metrics subsystem. +8. No runtime work or counter access on ordinary statement, transaction, DML, + lookup, scan, stream, row, index, or buffer hot paths. +9. No new benchmark command, lifecycle phase, parameter, or private surrogate + for checkpoint, maintenance, rollback, or mixed mandatory work. +10. No rewrite of RFC-0025 history, reopening of backlog 000123, or expansion + of logical-lock deadlock policy from backlog 000167. + +## Plan + +### Statistics and supervision + +`MandatoryRuntimeStats` contains fixed `operation` and +`transaction_cleanup` snapshots. Each `MandatoryTaskStats` reports +submitted, started, completed, error, panic, detached-observer, active, and +admission/queue/execution timing values. + +Relaxed saturating atomics hold monotonic diagnostics; existing admission state +remains authoritative for current active counts. Successful operation +submission is counted immediately before ownership acceptance. Successful +internal submission is counted before its detached executor task can publish +later lifecycle state. Rejected or abandoned pre-acceptance work is not +counted. Start is the first supervised poll; completion and outcome accounting +precede caller result publication so an awakened observer sees its terminal +metrics. Snapshots are independently sampled and do not promise transactional +equations among fields. + +`Session::mandatory_runtime_stats()` uses the existing read-only inspection +boundary. It remains available after engine poison while lifecycle inspection +is admitted, creates no session operation or runtime work, and closes with the +normal session/engine boundary. + +Immutable task metadata supplies the fixed class, task label, optional session +operation, and optional table id. One debug start event records admission and +queue wait; one debug finish event records result, execution time, and +attached/detached/none observer state. Existing error-level reporting remains +authoritative for unobserved ordinary errors and runtime panics. + +### Shutdown diagnostics + +A private shutdown origin distinguishes `explicit` from `owner_drop` while +both paths retain the same blocking drain. Try-shutdown busy logs and +`ShutdownBusy` attachments share origin, strong references, operation state, +voluntary/mandatory session blockers, queued cleanup, and active caller and +internal task counts. Blocking shutdown adds no polling, cancellation, timeout, +or periodic warnings, and component teardown ordering is unchanged. + +### Cooperative progress + +The accepted-path audit classified fixed work, natural awaited IO/event +boundaries, explicit logical batches, and intentionally unchanged synchronous +regions. Table DDL and DROP INDEX are structurally bounded around existing +awaits. CREATE INDEX cold work awaits disk-index construction; hot-row build +retains the shared 128-item yield boundary. + +Table and catalog checkpoint, freeze, redo truncation, and secondary MemIndex +cleanup return control at existing IO, event, retry, scan, or operation +boundaries. Terminal rollback, abandoned cleanup, and failed-precommit cleanup +yield after each 128 completed row-undo or index-undo records, after short-lived +guards are released. Undo order, failure precedence, retained-failure policy, +and terminal publication are unchanged. + +Synchronous filesystem work remains backlog 000137, and bounded-memory or +parallel large-index construction remains backlog 000104. These findings do +not weaken the cooperative contract or imply a scheduling policy. + +### Benchmark and documentation boundary + +The existing `--include-stats` model captures one engine-global snapshot +before and after a workload and emits deltas under +`mandatory.operation.*` and `mandatory.transaction_cleanup.*`. It does not +sum identical snapshots across sessions or add benchmark runtime controls. + +Release comparisons used equivalent fresh roots, `log-sync=none`, one +unreported warm-up, and seven alternating baseline/candidate pairs. Results +report median, median absolute deviation, and range. A repeatable regression +above 5% on statement/transaction no-op controls would block completion. +Missing checkpoint/freeze lifecycle work remains backlog 000147; large +rollback and mixed caller/cleanup design is backlog 000176. + +## Implementation Notes + +Implemented fixed-class observability, cooperative cleanup progress, coherent shutdown diagnostics, and the final RFC-0026 runtime evidence without changing accepted-operation ownership or adding scheduler policy. + +- The production label catalog is `create_table`, `drop_table`, + `create_index`, `drop_index`, `freeze_table`, `checkpoint_table`, + `checkpoint_catalog`, `truncate_redo_log`, + `checkpoint_catalog_and_truncate_redo_log`, and + `cleanup_secondary_mem_indexes` for caller operations; internal labels are + `terminal_rollback`, `abandoned_transaction`, and + `failed_precommit`. +- The bounded-poll audit found no operation-lock or metadata-gate acquisition + below a synchronous acceptance edge. Undo loops gained shared 128-item + cooperative budgets; no other production loop required a new yield. +- One-runner/caller-limit-one admission and cleanup proofs passed 32 + repetitions. Two independent accepted tasks overlapped for 32 repetitions + with two runners. A real 129-hot-row CREATE INDEX plus terminal cleanup + passed eight one-runner repetitions. No stale task, permit, observer, + listener, retained lock, or duplicate completion remained. +- Review moved caller completion metrics before observer wakeup, made internal + submission accounting observable before detached execution, retained active + accounting until permit release, and replaced a runner-blocking overlap + barrier with an async rendezvous. It also made the real index/cleanup test + wait for actual internal admission and completed try-shutdown blocker + reporting. +- No deterministic scheduling-policy failure or logical-lock cycle was found, + so no scheduling RFC or backlog-000167 update was required. + +Paired release measurements used elapsed nanoseconds. The four no-op deltas +were +1.34%, -0.27%, -0.74%, and +1.91%, all below the 5% repeat gate. DDL, +insert, lookup, scan, and stream ranges overlapped; no non-overlapping candidate +slowdown was found. Candidate statistics smoke runs reported zero mandatory +submissions for no-op, insert, lookup, scan, and stream workloads. One +successful table/index DDL cycle reported operation +`submitted/started/completed = 2/2/2`, zero active tasks, and zero cleanup +submissions. + +Final verification passed: + +- branch-diff style audit across 12 Rust files; +- strict workspace clippy and formatting checks; +- workspace build and 1,633 standard nextest cases; +- alternate `doradb-storage` `libaio` nextest validation; +- 94 `doradb-bench` nextest cases; +- normal dependency-tree verification for `async-executor v1.13.3`. + +`.config/nextest.toml` remained unchanged. Documentation was synchronized +across the benchmark tool, component lifecycle, transaction system, public +error audit, task, and RFC. Deferred benchmark design is recorded in backlog +000176. + +## Impacts + +- `doradb-storage` exposes additive `MandatoryRuntimeStats` and + `MandatoryTaskStats` types plus one read-only session accessor. +- Mandatory admission, first poll, and finish perform relaxed counter updates + and timestamp reads; active counts continue to come from admission state. +- Transaction cleanup and hot CREATE INDEX construction yield at resource-safe + logical boundaries, which can extend isolated completion slightly while + allowing other ready mandatory work to progress. +- Explicit shutdown and owner Drop retain identical lossless drain behavior but + now have distinct diagnostic origins and complete blocker attachments. +- `doradb-bench` emits one engine-global fixed-class statistics delta through + existing result artifacts. +- Public DDL, maintenance, transaction, statement, DML, scan, and stream + results are unchanged. +- Storage, redo, recovery, catalog, table, index, and manifest formats are + unchanged; no dependency or configuration compatibility was changed. + +## Test Cases + +1. New engines report zero class counters and authoritative zero active counts. +2. Accepted success, error, panic, observer detach, and internal cleanup update + only their specified class and outcome counters. +3. Capacity wait, cancelled preparation, and closed internal admission create + no submitted task. +4. Completion statistics are visible when an observer wakes; active count + reaches zero only after owner and permit release. +5. Statistics remain inspectable after poison and close at the lifecycle + boundary without creating runtime work. +6. Start/finish events retain fixed identity, result, observer, and timing + labels; unobserved errors and panics retain error-level reports. +7. Every DDL and maintenance path holds operation locks, metadata gates, and + workflow authority before acceptance and acquires none afterward. +8. Dropping an accepted observer does not release, cancel, or resubmit work. +9. One-runner saturated admission allows internal cleanup to complete at a + cooperative caller boundary before the next caller receives capacity. +10. Long hot-index and undo work returns scheduler control between logical + batches without changing ordering or error behavior. +11. Two-runner tasks overlap, and internal admission never consumes or wakes a + caller permit. +12. Try-shutdown attachments agree with logs for voluntary, mandatory-session, + queued-cleanup, caller-task, and internal-task blockers. +13. Explicit shutdown and owner Drop wait for exact blockers and complete + without lost listener wakeups. +14. Legacy and default configurations retain two runners and four caller + permits; zero values remain rejected. +15. Benchmark statistics are global, flattened, and not multiplied by session + count; ordinary workloads report zero runtime submissions. +16. DDL benchmark cycles report coherent accepted and completed deltas. +17. Standard, alternate-`libaio`, benchmark, formatting, clippy, build, + dependency-tree, and style validations pass. + +## Open Questions + +No architecture-blocking question remains. Future improvements are explicitly +separate: + +- checkpoint/freeze/shutdown-reopen benchmark design: + `docs/backlogs/000147-doradb-bench-checkpoint-lifecycle-scenarios.md`; +- large rollback and heterogeneous caller/cleanup benchmark design: + `docs/backlogs/000176-quantify-rollback-saturation-alongside-mandatory-ddl.md`; +- logical-lock deadlock policy, if future evidence requires it: + `docs/backlogs/000167-logical-lock-deadlock-handling.md`. + +Scheduling priority or reserved-runner work requires new deterministic evidence; this task found none. diff --git a/docs/tasks/next-id b/docs/tasks/next-id index 5d649a9e..c5d97093 100644 --- a/docs/tasks/next-id +++ b/docs/tasks/next-id @@ -1 +1 @@ -000252 +000253 diff --git a/docs/transaction-system.md b/docs/transaction-system.md index b49ecd88..505317b6 100644 --- a/docs/transaction-system.md +++ b/docs/transaction-system.md @@ -241,7 +241,8 @@ statement-lock state while `Transaction::exec` is active. It lends one `Statement` facade with direct disjoint borrows of the checked-out `TrxInner`, operation attachment, effects, and locks; DML methods therefore do not resolve the entry or unwrap the carrier. Normal statement finish releases -statement locks and returns the core to `ForegroundAvailable`. This check-in +statement locks and returns the core to the available payload position inside +outer `Voluntary` ownership. This check-in ends only the operation-local lease, not the semantic transaction lifetime; the weak public `Transaction` remains reusable for its next call. @@ -250,8 +251,8 @@ checkout succeeds, dropping the future is terminal for that public transaction. The callback and any pending acquisition guard are destroyed first. `StmtState` then discards statement redo, appends residual row and index undo after prior transaction undo, releases statement locks, and returns the -complete core directly as `CleanupReady`. It never exposes an intervening -`ForegroundAvailable` state. The existing identity cleanup job claims the +complete core directly as outer `CleanupReady`. It never exposes an intervening +available payload position. The exact-identity cleanup job claims the core and performs whole-transaction rollback; later calls through the stale public facade return `TransactionDiscarded`. An ordinary callback error is different: statement-local rollback completes before ordinary check-in, so @@ -294,20 +295,25 @@ child completely back to `Mandatory(None)` before a retry installs a fresh publishing the outer terminal state; supervised unwind retains unsafe child state in `FailedRetained`. -After explicit rollback claims terminal ownership and publishes `RollingBack`, +After explicit rollback claims terminal ownership and publishes `Completing`, the claimed transaction core, undo buffers, locks, and session cleanup -attachment are handed to the transaction-system cleanup worker before rollback -awaits row or index storage work. The public `rollback().await` future only -waits on the worker-owned completion cell; dropping that waiter does not cancel +attachment are synchronously submitted as a `terminal_rollback` task to the +engine-owned mandatory runtime before rollback awaits row or index storage +work. Abandoned transactions use `abandoned_transaction` tasks, and redo groups +that fail after precommit use `failed_precommit` tasks. These independently +accounted cleanup tasks bypass caller-operation capacity; cleanup within one +transaction remains sequential. The public `rollback().await` future only +waits on the task-owned completion cell; dropping that waiter does not cancel rollback cleanup or release rollback-capable undo without making ownership explicit. -The coherent outer labels are `ForegroundAvailable`, `ForegroundRunning`, -`CleanupReady`, `CleanupRunning`, `BackgroundQueued`, `BackgroundRunning`, -`CompletionOwned`, `Terminal`, and `FailedRetained`. The background labels are -reserved for later RFC-0025 phases; Phase 1 does not store or transfer whole -DDL/maintenance futures. Handle-drop intent is orthogonal while a transaction -core is checked out, so checkout return publishes `CleanupReady` exactly once. +The coherent outer labels are `Voluntary`, `Mandatory`, `CleanupReady`, +`Completing`, `Terminal`, and `FailedRetained`. `Voluntary` and `Mandatory` +optionally contain the nested private-transaction positions `Available`, +`Running`, `CleanupReady`, and `Completing`; public transaction checkout is +represented by payload ownership within `Voluntary(None)`. Handle-drop intent +is orthogonal while a transaction core is checked out, so checkout return +publishes outer `CleanupReady` exactly once. Cleanup messages carry `(SessionOperationKey, TrxID)` and stale, replaced, or duplicate hints are neutral. Registry resolution uses only the operation key; the cleanup claim atomically validates the message's `TrxID`, claimable state, diff --git a/doradb-bench/src/output.rs b/doradb-bench/src/output.rs index 77e3aaa1..088c47f8 100644 --- a/doradb-bench/src/output.rs +++ b/doradb-bench/src/output.rs @@ -2,8 +2,8 @@ use crate::cli::{IndexMode, LogSyncMode, Workload}; use crate::error::{BenchError, Result}; use crate::manifest::{internal_stats_csv_path, result_csv_path, result_markdown_path}; use doradb_storage::{ - BufferPoolCounters, BufferPoolRuntimeStats, BufferPoolStats, Session, StorageIoStats, - TransactionSystemStats, + BufferPoolCounters, BufferPoolRuntimeStats, BufferPoolStats, MandatoryRuntimeStats, + MandatoryTaskStats, Session, StorageIoStats, TransactionSystemStats, }; use std::fs; use std::io::ErrorKind; @@ -15,6 +15,7 @@ pub(super) struct InternalStatsSnapshot { trx: TransactionSystemStats, storage: StorageIoStats, buffer: BufferPoolStats, + mandatory: MandatoryRuntimeStats, } impl InternalStatsSnapshot { @@ -23,6 +24,7 @@ impl InternalStatsSnapshot { trx: session.transaction_system_stats()?, storage: session.storage_io_stats()?, buffer: session.buffer_pool_stats()?, + mandatory: session.mandatory_runtime_stats()?, }) } } @@ -126,6 +128,7 @@ pub(super) fn internal_metrics( push_transaction_metrics(&mut metrics, before.trx, after.trx); push_storage_metrics(&mut metrics, before.storage, after.storage); push_buffer_metrics(&mut metrics, &before.buffer, &after.buffer); + push_mandatory_metrics(&mut metrics, before.mandatory, after.mandatory); metrics } @@ -538,6 +541,86 @@ fn push_buffer_metrics( push_one_buffer_pool(metrics, "buffer.disk", before.disk, after.disk); } +fn push_mandatory_metrics( + metrics: &mut Vec, + before: MandatoryRuntimeStats, + after: MandatoryRuntimeStats, +) { + push_mandatory_task_metrics( + metrics, + "mandatory.operation", + before.operation, + after.operation, + ); + push_mandatory_task_metrics( + metrics, + "mandatory.transaction_cleanup", + before.transaction_cleanup, + after.transaction_cleanup, + ); +} + +fn push_mandatory_task_metrics( + metrics: &mut Vec, + prefix: &str, + before: MandatoryTaskStats, + after: MandatoryTaskStats, +) { + push_metric( + metrics, + &format!("{prefix}.submitted_count"), + delta(after.submitted_count, before.submitted_count), + ); + push_metric( + metrics, + &format!("{prefix}.started_count"), + delta(after.started_count, before.started_count), + ); + push_metric( + metrics, + &format!("{prefix}.completed_count"), + delta(after.completed_count, before.completed_count), + ); + push_metric( + metrics, + &format!("{prefix}.error_count"), + delta(after.error_count, before.error_count), + ); + push_metric( + metrics, + &format!("{prefix}.panic_count"), + delta(after.panic_count, before.panic_count), + ); + push_metric( + metrics, + &format!("{prefix}.detached_observer_count"), + delta( + after.detached_observer_count, + before.detached_observer_count, + ), + ); + push_metric( + metrics, + &format!("{prefix}.active_count"), + after.active_count as u128, + ); + push_metric( + metrics, + &format!("{prefix}.admission_wait_nanos"), + delta(after.admission_wait_nanos, before.admission_wait_nanos), + ); + push_metric( + metrics, + &format!("{prefix}.queue_wait_nanos"), + delta(after.queue_wait_nanos, before.queue_wait_nanos), + ); + push_metric( + metrics, + &format!("{prefix}.execution_nanos"), + delta(after.execution_nanos, before.execution_nanos), + ); +} + fn push_one_buffer_pool( metrics: &mut Vec, prefix: &str, @@ -728,6 +811,15 @@ mod tests { }, ..BufferPoolStats::default() }, + mandatory: MandatoryRuntimeStats { + operation: MandatoryTaskStats { + submitted_count: 4, + active_count: 1, + queue_wait_nanos: 10, + ..MandatoryTaskStats::default() + }, + ..MandatoryRuntimeStats::default() + }, }; let after = InternalStatsSnapshot { trx: TransactionSystemStats { @@ -763,10 +855,26 @@ mod tests { }, ..BufferPoolStats::default() }, + mandatory: MandatoryRuntimeStats { + operation: MandatoryTaskStats { + submitted_count: 7, + completed_count: 2, + active_count: 3, + queue_wait_nanos: 16, + ..MandatoryTaskStats::default() + }, + transaction_cleanup: MandatoryTaskStats { + submitted_count: 2, + started_count: 2, + completed_count: 2, + execution_nanos: 20, + ..MandatoryTaskStats::default() + }, + }, }; let metrics = internal_metrics(&before, &after); - assert_eq!(metrics.len(), 73); + assert_eq!(metrics.len(), 93); assert_eq!(metric_value(&metrics, "transaction.commit_count"), 5); assert_eq!(metric_value(&metrics, "transaction.trx_count"), 0); assert_eq!(metric_value(&metrics, "transaction.log_bytes"), 5); @@ -778,6 +886,30 @@ mod tests { assert_eq!(metric_value(&metrics, "buffer.meta.write_errors"), 0); assert_eq!(metric_value(&metrics, "buffer.mem.completed_reads"), 3); assert_eq!(metric_value(&metrics, "buffer.disk.write_errors"), 0); + assert_eq!( + metric_value(&metrics, "mandatory.operation.submitted_count"), + 3 + ); + assert_eq!( + metric_value(&metrics, "mandatory.operation.completed_count"), + 2 + ); + assert_eq!( + metric_value(&metrics, "mandatory.operation.active_count"), + 3 + ); + assert_eq!( + metric_value(&metrics, "mandatory.operation.queue_wait_nanos"), + 6 + ); + assert_eq!( + metric_value(&metrics, "mandatory.transaction_cleanup.submitted_count"), + 2 + ); + assert_eq!( + metric_value(&metrics, "mandatory.transaction_cleanup.execution_nanos"), + 20 + ); } #[test] diff --git a/doradb-storage/src/catalog/index.rs b/doradb-storage/src/catalog/index.rs index 10dafb19..2daa75a0 100644 --- a/doradb-storage/src/catalog/index.rs +++ b/doradb-storage/src/catalog/index.rs @@ -22,7 +22,7 @@ use crate::obs; use crate::quiescent::QuiescentGuard; use crate::row::RowRead; use crate::runtime::mandatory::{AcceptedExecution, MandatoryTaskMetadata, PreparedExecution}; -use crate::runtime::yield_now; +use crate::runtime::{POLL_BUDGET, yield_now}; use crate::session::{AcceptedDdlScope, PreparedDdlScope}; use crate::table::{DeleteMarker, Table, TableRuntimeLayout, secondary_disk_tree_encoder}; use crate::trx::{PreparedCatalogWriteAuthority, Transaction, trx_is_committed}; @@ -483,13 +483,20 @@ impl<'a> CreateIndexRuntimeBuilder<'a> { index_spec, build_ts, #[cfg(test)] - test: _, + test, } = self; let ty_infer = |col_no| metadata.col.col_type(col_no); let mem = UniqueMemIndex::new(index_pool, index_guard, index_spec, ty_infer, build_ts).await?; - let insert_res = - insert_create_index_unique_hot_rows(&mem, index_guard, &hot_rows, build_ts).await; + let insert_res = insert_create_index_unique_hot_rows( + &mem, + index_guard, + &hot_rows, + build_ts, + #[cfg(test)] + &test, + ) + .await; if let Err(err) = insert_res { if let Err(report) = mem.destroy(index_guard).await { let report = report.attach(format!( @@ -535,8 +542,15 @@ impl<'a> CreateIndexRuntimeBuilder<'a> { let forced_population_failure: RuntimeResult<()> = Ok(()); let insert_res = match forced_population_failure { Ok(()) => { - insert_create_index_non_unique_hot_rows(&mem, index_guard, &hot_rows, build_ts) - .await + insert_create_index_non_unique_hot_rows( + &mem, + index_guard, + &hot_rows, + build_ts, + #[cfg(test)] + &test, + ) + .await } Err(err) => Err(err), }; @@ -1813,6 +1827,7 @@ async fn insert_create_index_unique_hot_rows( index_guard: &PoolGuard, hot_rows: &[CreateIndexRowEntry], build_ts: TrxID, + #[cfg(test)] test: &tests::IndexDdlTestController, ) -> OperationOrRuntimeResult<()> { for (row_no, row) in hot_rows.iter().enumerate() { match mem @@ -1829,7 +1844,10 @@ async fn insert_create_index_unique_hot_rows( .into()); } } - if row_no % 64 == 63 { + if row_no % POLL_BUDGET == POLL_BUDGET - 1 { + #[cfg(test)] + test.reach_phase(tests::IndexDdlTestPhase::CreateHotBuildBatchComplete) + .await; yield_now().await; } } @@ -1841,6 +1859,7 @@ async fn insert_create_index_non_unique_hot_rows( index_guard: &PoolGuard, hot_rows: &[CreateIndexRowEntry], build_ts: TrxID, + #[cfg(test)] test: &tests::IndexDdlTestController, ) -> RuntimeResult<()> { for (row_no, row) in hot_rows.iter().enumerate() { match mem @@ -1856,7 +1875,10 @@ async fn insert_create_index_non_unique_hot_rows( ); } } - if row_no % 64 == 63 { + if row_no % POLL_BUDGET == POLL_BUDGET - 1 { + #[cfg(test)] + test.reach_phase(tests::IndexDdlTestPhase::CreateHotBuildBatchComplete) + .await; yield_now().await; } } @@ -2107,7 +2129,10 @@ pub(crate) mod tests { ActiveIndexSpec, ColumnAttributes, ColumnSpec, CurrentTableState, IndexAttributes, IndexKey, IndexSpec, ResolvedVisibleTableMetadata, TableMetadata, tests::table2, }; - use crate::conf::{EngineConfig, EvictableBufferPoolConfig, FileSystemConfig, TrxSysConfig}; + use crate::conf::{ + EngineConfig, EvictableBufferPoolConfig, FileSystemConfig, MandatoryRuntimeConfig, + TrxSysConfig, + }; use crate::engine::Engine; use crate::error::LifecycleError; use crate::file::cow_file::tests::old_root_drop_count; @@ -2122,11 +2147,12 @@ pub(crate) mod tests { use crate::table::tests::assert_freeze_created; use crate::trx::{MAX_SNAPSHOT_TS, Transaction}; use crate::value::{Val, ValKind}; - use smol::Timer; + use smol::{Timer, future::race}; use std::panic::{AssertUnwindSafe, catch_unwind}; use std::path::PathBuf; use std::sync::Arc; use std::sync::mpsc::sync_channel; + use std::task::Poll; use std::thread::spawn; use std::time::Duration; use tempfile::TempDir; @@ -2148,6 +2174,7 @@ pub(crate) mod tests { CreateColdCollectionComplete, CreateDiskTreeBuilt, CreateHotCollectionComplete, + CreateHotBuildBatchComplete, CreateRuntimeStaged, CreateCatalogStaged, CreateCatalogCommitted, @@ -2785,6 +2812,130 @@ pub(crate) mod tests { }); } + #[test] + fn test_terminal_cleanup_progresses_during_accepted_index_ddl_on_one_runner() { + smol::block_on(async { + let temp_dir = TempDir::new().unwrap(); + let engine = Engine::bootstrap( + lightweight_test_engine_config( + temp_dir.path().to_path_buf(), + "index_ddl_cleanup_progress", + ) + .mandatory_runtime( + MandatoryRuntimeConfig::default() + .worker_threads(1) + .concurrency_limit(1), + ), + ) + .await + .unwrap(); + let table_id = table2(&engine).await; + let table = table_for_internal_assertion(&engine, table_id); + let mut ddl_session = engine.new_session().unwrap(); + insert_rows(&table, &mut ddl_session, 0, 129, "fairness").await; + let before = ddl_session.mandatory_runtime_stats().unwrap(); + + for iteration in 0..8 { + let expected_cleanup_submitted = + before.transaction_cleanup.submitted_count + iteration + 1; + race( + async { + let (entered, release) = engine + .inner() + .index_ddl_test + .install_gate(IndexDdlTestPhase::CreateHotBuildBatchComplete); + let mut create = Box::pin(ddl_session.create_index( + table_id, + IndexSpec::new( + vec![IndexKey::new(1)], + IndexAttributes::empty(), + ), + )); + assert!(matches!( + futures::poll!(create.as_mut()), + Poll::Pending + )); + entered.recv_async().await.unwrap(); + + let mut cleanup_session = engine.new_session().unwrap(); + let mut rollback = + Box::pin(cleanup_session.begin_trx().unwrap().rollback()); + let mut rollback_result = None; + loop { + if rollback_result.is_none() + && let Poll::Ready(result) = futures::poll!(rollback.as_mut()) + { + rollback_result = Some(result); + } + let cleanup_submitted = engine + .inner() + .mandatory_runtime + .stats() + .transaction_cleanup + .submitted_count; + if cleanup_submitted >= expected_cleanup_submitted { + break; + } + assert!( + rollback_result.is_none(), + "rollback completed before mandatory cleanup submission" + ); + Timer::after(Duration::from_millis(1)).await; + } + + release.send_async(()).await.unwrap(); + match rollback_result { + Some(result) => result, + None => rollback.await, + } + .unwrap(); + cleanup_session.close().await.unwrap(); + + let index_no = create.await.unwrap(); + ddl_session.drop_index(table_id, index_no).await.unwrap(); + }, + async { + Timer::after(Duration::from_secs(5)).await; + panic!( + "one-runner index DDL and cleanup scheduling timed out: iteration={iteration}" + ); + }, + ) + .await; + } + + let after = ddl_session.mandatory_runtime_stats().unwrap(); + assert_eq!( + after + .operation + .submitted_count + .saturating_sub(before.operation.submitted_count), + 16 + ); + assert_eq!( + after + .operation + .completed_count + .saturating_sub(before.operation.completed_count), + 16 + ); + assert_eq!( + after + .transaction_cleanup + .submitted_count + .saturating_sub(before.transaction_cleanup.submitted_count), + 8 + ); + assert_eq!( + after + .transaction_cleanup + .completed_count + .saturating_sub(before.transaction_cleanup.completed_count), + 8 + ); + }); + } + #[test] fn test_create_index_execution_panic_before_first_effect_is_supervised() { smol::block_on(async { diff --git a/doradb-storage/src/conf/engine.rs b/doradb-storage/src/conf/engine.rs index 1ab79451..a0d1998c 100644 --- a/doradb-storage/src/conf/engine.rs +++ b/doradb-storage/src/conf/engine.rs @@ -11,12 +11,26 @@ use super::consts::{ }; use super::{EvictableBufferPoolConfig, FileSystemConfig, TrxSysConfig}; -/// Fixed engine-owned mandatory runtime configuration. +/// Immutable sizing for the engine-owned mandatory runtime. +/// +/// The default is two fixed operating-system runner threads and four accepted +/// caller-operation permits. One runner provides concurrency only when tasks +/// reach cooperative await or yield points; multiple runners allow overlap but +/// do not provide a fairness or queue-latency guarantee. +/// +/// `concurrency_limit` does not count caller-side preparation futures or +/// engine-internal transaction cleanup. Internal cleanup bypasses caller quota +/// so correctness obligations are not lost, while shutdown still waits for +/// both classes. Raising the caller limit may retain more locks, memory, and +/// publication work without raising executor throughput. Raising the runner +/// count may increase storage and metadata contention and cannot repair +/// blocking task code. Sizing is validated once during engine startup; a +/// running engine cannot be resized. #[derive(Clone, Debug, Serialize, Deserialize)] pub struct MandatoryRuntimeConfig { - /// Number of operating-system threads driving the mandatory executor. + /// Number of fixed operating-system threads driving the shared executor. pub worker_threads: usize, - /// Maximum number of accepted caller operations. + /// Maximum accepted caller obligations, excluding internal cleanup. pub concurrency_limit: usize, } @@ -31,14 +45,14 @@ impl Default for MandatoryRuntimeConfig { } impl MandatoryRuntimeConfig { - /// Set the number of mandatory-runtime runner threads. + /// Set the fixed number of mandatory-runtime runner threads. #[inline] pub fn worker_threads(mut self, worker_threads: usize) -> Self { self.worker_threads = worker_threads; self } - /// Set the maximum number of accepted caller operations. + /// Set accepted caller capacity without limiting internal cleanup. #[inline] pub fn concurrency_limit(mut self, concurrency_limit: usize) -> Self { self.concurrency_limit = concurrency_limit; diff --git a/doradb-storage/src/engine.rs b/doradb-storage/src/engine.rs index 7831f5a6..da7169cb 100644 --- a/doradb-storage/src/engine.rs +++ b/doradb-storage/src/engine.rs @@ -85,6 +85,22 @@ impl TryFrom for EngineLifecycleState { } } +#[derive(Clone, Copy)] +enum ShutdownOrigin { + Explicit, + OwnerDrop, +} + +impl ShutdownOrigin { + #[inline] + const fn label(self) -> &'static str { + match self { + Self::Explicit => "explicit", + Self::OwnerDrop => "owner_drop", + } + } +} + struct EngineLifecycle { /// Packed lifecycle state and active admission count. /// @@ -381,7 +397,7 @@ impl Engine { return Ok(()); } obs::info!( - "event=engine_lifecycle component=engine action=shutdown_start result=ok mode=try" + "event=engine_lifecycle component=engine action=shutdown_start result=ok mode=try origin=explicit" ); inner.lifecycle.close_admission(); inner.mandatory_runtime.close_admission(); @@ -390,13 +406,16 @@ impl Engine { let _shutdown = inner.lifecycle.shutdown_lock.lock(); if inner.lifecycle.inspect_state() == EngineLifecycleState::Shutdown { obs::info!( - "event=engine_lifecycle component=engine action=shutdown_finish result=ok mode=try already_shutdown=true" + "event=engine_lifecycle component=engine action=shutdown_finish result=ok mode=try origin=explicit already_shutdown=true" ); return Ok(()); } let blocker = inner.session_registry.first_shutdown_blocker(); let operation_blocked = blocker.is_some(); + let operation_state = blocker + .as_ref() + .map_or("none", |blocker| blocker.state.label()); let voluntary_blocked = blocker.as_ref().is_some_and(|blocker| { matches!( blocker.state, @@ -421,12 +440,16 @@ impl Engine { || mandatory_internal != 0 { let strong_refs = strong_count - 1; - let busy = strong_refs.max(usize::from(operation_blocked)); + let busy = strong_refs + .max(usize::from(operation_blocked)) + .max(mandatory_callers) + .max(mandatory_internal); obs::warn!( - "event=engine_lifecycle component=engine action=shutdown_finish result=busy mode=try busy={} strong_refs={} operation_blocked={} voluntary_blocked={} mandatory_session_blocked={} cleanup_queued={} mandatory_callers={} mandatory_internal={}", + "event=engine_lifecycle component=engine action=shutdown_finish result=busy mode=try origin=explicit busy={} strong_refs={} operation_blocked={} operation_state={} voluntary_blocked={} mandatory_session_blocked={} cleanup_queued={} mandatory_callers={} mandatory_internal={}", busy, strong_refs, operation_blocked, + operation_state, voluntary_blocked, mandatory_session_blocked, cleanup_queued, @@ -434,12 +457,12 @@ impl Engine { mandatory_internal ); return Err(Report::new(LifecycleError::ShutdownBusy).attach(format!( - "strong_refs={strong_refs}, operation_blocked={operation_blocked}, voluntary_blocked={voluntary_blocked}, mandatory_session_blocked={mandatory_session_blocked}, mandatory_callers={mandatory_callers}, mandatory_internal={mandatory_internal}" + "origin=explicit, strong_refs={strong_refs}, operation_blocked={operation_blocked}, operation_state={operation_state}, voluntary_blocked={voluntary_blocked}, mandatory_session_blocked={mandatory_session_blocked}, cleanup_queued={cleanup_queued}, mandatory_callers={mandatory_callers}, mandatory_internal={mandatory_internal}" ))); } self.finish_shutdown_locked(inner); obs::info!( - "event=engine_lifecycle component=engine action=shutdown_finish result=ok mode=try" + "event=engine_lifecycle component=engine action=shutdown_finish result=ok mode=try origin=explicit" ); Ok(()) } @@ -452,17 +475,18 @@ impl Engine { /// component shutdown in reverse registration order. #[inline] pub fn shutdown(&self) { - self.shutdown_inner(); + self.shutdown_inner(ShutdownOrigin::Explicit); } #[inline] - fn shutdown_inner(&self) { + fn shutdown_inner(&self, origin: ShutdownOrigin) { let inner = self.inner(); if inner.lifecycle.inspect_state() == EngineLifecycleState::Shutdown { return; } obs::info!( - "event=engine_lifecycle component=engine action=shutdown_start result=ok mode=wait" + "event=engine_lifecycle component=engine action=shutdown_start result=ok mode=wait origin={}", + origin.label(), ); inner.lifecycle.close_admission(); inner.mandatory_runtime.close_admission(); @@ -475,7 +499,8 @@ impl Engine { let _shutdown = inner.lifecycle.shutdown_lock.lock(); if inner.lifecycle.inspect_state() == EngineLifecycleState::Shutdown { obs::info!( - "event=engine_lifecycle component=engine action=shutdown_finish result=ok mode=wait already_shutdown=true" + "event=engine_lifecycle component=engine action=shutdown_finish result=ok mode=wait origin={} already_shutdown=true", + origin.label(), ); return; } @@ -485,7 +510,8 @@ impl Engine { if strong_count == 1 && shutdown_wait.is_none() { self.finish_shutdown_locked(inner); obs::info!( - "event=engine_lifecycle component=engine action=shutdown_finish result=ok mode=wait" + "event=engine_lifecycle component=engine action=shutdown_finish result=ok mode=wait origin={}", + origin.label(), ); return; } @@ -523,8 +549,8 @@ impl Engine { /// id identifies the exact public or private transaction to claim under the /// entry mutex. The second identity prevents a stale hint from claiming a /// newer private transaction installed under the same operation key. Other - /// active operation states only block shutdown; a later whole-operation - /// handoff transfers its owned task rather than using this identity pair. + /// active operation states only block shutdown; accepted mandatory work + /// already owns its cleanup authority through the stable operation entry. #[inline] fn queue_shutdown_operation_cleanup( &self, @@ -548,7 +574,7 @@ impl Drop for Engine { // Implicit owner drop runs the same synchronous drain as explicit // shutdown. An unintended drop may therefore block until every // foreground operation and engine-owned background task completes. - self.shutdown_inner(); + self.shutdown_inner(ShutdownOrigin::OwnerDrop); // Field order releases the owner runtime ref before registry-owned // component owners. @@ -1963,7 +1989,7 @@ mod tests { assert_eq!( err.report().downcast_ref::().map(String::as_str), Some( - "strong_refs=1, operation_blocked=false, voluntary_blocked=false, mandatory_session_blocked=false, mandatory_callers=0, mandatory_internal=0" + "origin=explicit, strong_refs=1, operation_blocked=false, operation_state=none, voluntary_blocked=false, mandatory_session_blocked=false, cleanup_queued=false, mandatory_callers=0, mandatory_internal=0" ) ); @@ -2046,7 +2072,7 @@ mod tests { assert_eq!( err.report().downcast_ref::().map(String::as_str), Some( - "strong_refs=1, operation_blocked=false, voluntary_blocked=false, mandatory_session_blocked=false, mandatory_callers=0, mandatory_internal=0" + "origin=explicit, strong_refs=1, operation_blocked=false, operation_state=none, voluntary_blocked=false, mandatory_session_blocked=false, cleanup_queued=false, mandatory_callers=0, mandatory_internal=0" ) ); assert_eq!(session_registry_len(&engine.inner().session_registry), 1); @@ -2079,7 +2105,7 @@ mod tests { assert_eq!( err.report().downcast_ref::().map(String::as_str), Some( - "strong_refs=0, operation_blocked=true, voluntary_blocked=true, mandatory_session_blocked=false, mandatory_callers=0, mandatory_internal=0" + "origin=explicit, strong_refs=0, operation_blocked=true, operation_state=voluntary, voluntary_blocked=true, mandatory_session_blocked=false, cleanup_queued=false, mandatory_callers=0, mandatory_internal=0" ) ); @@ -2112,7 +2138,7 @@ mod tests { assert_eq!( err.report().downcast_ref::().map(String::as_str), Some( - "strong_refs=0, operation_blocked=true, voluntary_blocked=true, mandatory_session_blocked=false, mandatory_callers=0, mandatory_internal=0" + "origin=explicit, strong_refs=0, operation_blocked=true, operation_state=voluntary, voluntary_blocked=true, mandatory_session_blocked=false, cleanup_queued=false, mandatory_callers=0, mandatory_internal=0" ) ); diff --git a/doradb-storage/src/lib.rs b/doradb-storage/src/lib.rs index 6532c7a5..325a20f5 100644 --- a/doradb-storage/src/lib.rs +++ b/doradb-storage/src/lib.rs @@ -58,8 +58,8 @@ pub use session::{ CatalogRedoMaintenanceOutcome, RedoTruncationBlockerInfo, RedoTruncationOutcome, Session, }; pub use stats::{ - BufferPoolCounters, BufferPoolRuntimeStats, BufferPoolStats, IoBackendStats, StorageIoStats, - TransactionSystemStats, + BufferPoolCounters, BufferPoolRuntimeStats, BufferPoolStats, IoBackendStats, + MandatoryRuntimeStats, MandatoryTaskStats, StorageIoStats, TransactionSystemStats, }; pub use table::{ CheckpointCancelReason, CheckpointDelayReason, CheckpointOutcome, FreezeOutcome, diff --git a/doradb-storage/src/runtime/mandatory.rs b/doradb-storage/src/runtime/mandatory.rs index 234b1c67..1ccd72ca 100644 --- a/doradb-storage/src/runtime/mandatory.rs +++ b/doradb-storage/src/runtime/mandatory.rs @@ -9,6 +9,7 @@ use crate::id::{SessionOperationKey, TableID}; use crate::obs; use crate::poison::EnginePoisoner; use crate::quiescent::{QuiescentBox, QuiescentGuard}; +use crate::stats::{MandatoryRuntimeStats, MandatoryTaskStats}; use crate::{runtime, thread}; use error_stack::{Report, ResultExt}; use event_listener::{Event, listener}; @@ -16,13 +17,15 @@ use futures::FutureExt; use futures::future::{Either, select}; use parking_lot::Mutex; use std::any::Any; +use std::fmt::{self, Display, Formatter}; use std::future::Future; use std::mem::take; use std::panic::{AssertUnwindSafe, resume_unwind}; use std::result::Result as StdResult; use std::sync::Arc; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::thread::JoinHandle; +use std::time::{Duration, Instant}; /// Immutable classification of one mandatory task. #[derive(Clone, Copy, Debug, PartialEq, Eq)] @@ -108,6 +111,141 @@ impl MandatoryTaskMetadata { .map_or_else(|| "none".to_owned(), |table_id| table_id.to_string()) ) } + + #[inline] + const fn task_class(&self) -> &'static str { + self.class.label() + } + + #[inline] + const fn task_label(&self) -> &'static str { + self.label + } + + #[inline] + const fn session_operation(&self) -> Option { + self.session_operation + } + + #[inline] + const fn table_id(&self) -> Option { + self.table_id + } +} + +struct OptionalValue(Option); + +impl Display for OptionalValue +where + T: Display, +{ + #[inline] + fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result { + match self.0.as_ref() { + Some(value) => value.fmt(formatter), + None => formatter.write_str("none"), + } + } +} + +#[derive(Default)] +struct MandatoryTaskCounters { + submitted_count: AtomicUsize, + started_count: AtomicUsize, + completed_count: AtomicUsize, + error_count: AtomicUsize, + panic_count: AtomicUsize, + detached_observer_count: AtomicUsize, + admission_wait_nanos: AtomicUsize, + queue_wait_nanos: AtomicUsize, + execution_nanos: AtomicUsize, +} + +impl MandatoryTaskCounters { + #[inline] + fn record_submitted(&self, admission_wait_nanos: usize) { + self.submitted_count.fetch_add(1, Ordering::Relaxed); + self.admission_wait_nanos + .fetch_add(admission_wait_nanos, Ordering::Relaxed); + } + + #[inline] + fn record_started(&self, queue_wait_nanos: usize) { + self.started_count.fetch_add(1, Ordering::Relaxed); + self.queue_wait_nanos + .fetch_add(queue_wait_nanos, Ordering::Relaxed); + } + + #[inline] + fn record_completed(&self, result: MandatoryTaskResult, execution_nanos: usize) { + match result { + MandatoryTaskResult::Ok => {} + MandatoryTaskResult::Error => { + self.error_count.fetch_add(1, Ordering::Relaxed); + } + MandatoryTaskResult::Panic => { + self.panic_count.fetch_add(1, Ordering::Relaxed); + } + } + self.execution_nanos + .fetch_add(execution_nanos, Ordering::Relaxed); + self.completed_count.fetch_add(1, Ordering::Relaxed); + } + + #[inline] + fn record_observer_detached(&self) { + self.detached_observer_count.fetch_add(1, Ordering::Relaxed); + } + + #[inline] + fn snapshot(&self, active_count: usize) -> MandatoryTaskStats { + MandatoryTaskStats { + submitted_count: self.submitted_count.load(Ordering::Relaxed), + started_count: self.started_count.load(Ordering::Relaxed), + completed_count: self.completed_count.load(Ordering::Relaxed), + error_count: self.error_count.load(Ordering::Relaxed), + panic_count: self.panic_count.load(Ordering::Relaxed), + detached_observer_count: self.detached_observer_count.load(Ordering::Relaxed), + active_count, + admission_wait_nanos: self.admission_wait_nanos.load(Ordering::Relaxed), + queue_wait_nanos: self.queue_wait_nanos.load(Ordering::Relaxed), + execution_nanos: self.execution_nanos.load(Ordering::Relaxed), + } + } +} + +#[derive(Clone, Copy)] +enum MandatoryTaskResult { + Ok, + Error, + Panic, +} + +impl MandatoryTaskResult { + #[inline] + const fn label(self) -> &'static str { + match self { + Self::Ok => "ok", + Self::Error => "error", + Self::Panic => "panic", + } + } +} + +#[derive(Clone, Copy)] +enum PublishedObserver { + Attached, + Detached, +} + +impl PublishedObserver { + #[inline] + const fn label(self) -> &'static str { + match self { + Self::Attached => "attached", + Self::Detached => "detached", + } + } } struct MandatoryAdmissionState { @@ -370,17 +508,20 @@ struct MandatoryCompletion { completion: Completion, observation: Mutex, metadata: MandatoryTaskMetadata, + counters: Arc, } impl MandatoryCompletion { #[inline] fn endpoints( metadata: MandatoryTaskMetadata, + counters: Arc, ) -> (CompletionProducer, CompletionObserver) { let inner = Arc::new(Self { completion: Completion::new(), observation: Mutex::new(ObservationState::Attached), metadata, + counters, }); ( CompletionProducer { @@ -396,8 +537,11 @@ impl MandatoryCompletion { Ok(value) => drop(value), Err(error) => { obs::error!( - "event=mandatory_completion component=mandatory_runtime action=discard_unobserved result=error {} error={error:?}", - self.metadata.diagnostic() + "event=mandatory_completion component=mandatory_runtime action=discard_unobserved result=error task_class={} task_label={} session_operation={} table_id={} error={error:?}", + self.metadata.task_class(), + self.metadata.task_label(), + OptionalValue(self.metadata.session_operation()), + OptionalValue(self.metadata.table_id()), ); } } @@ -415,11 +559,17 @@ impl CompletionProducer { } #[inline] - fn complete(self, result: CompletionResult) { + fn counters(&self) -> &Arc { + &self.inner.counters + } + + #[inline] + fn complete(self, result: CompletionResult) -> PublishedObserver { let observation = self.inner.observation.lock(); match *observation { ObservationState::Attached => { self.inner.completion.complete(result); + PublishedObserver::Attached } ObservationState::Detached => { self.inner.completion.complete(result); @@ -431,6 +581,7 @@ impl CompletionProducer { }; drop(observation); self.inner.handle_unobserved(result); + PublishedObserver::Detached } ObservationState::Consumed => { unreachable!("observer cannot consume before producer completion") @@ -469,6 +620,7 @@ impl Drop for CompletionObserver { return; } self.armed = false; + self.inner.counters.record_observer_detached(); let mut observation = self.inner.observation.lock(); assert!( matches!(*observation, ObservationState::Attached), @@ -507,6 +659,10 @@ pub(crate) struct MandatoryRuntime { /// while redo can produce final cleanup, then closes and drains before the /// executor runners stop. internal_admission: MandatoryInternalAdmission, + /// Monotonic diagnostics for accepted caller operations. + operation_counters: Arc, + /// Monotonic diagnostics for internal transaction cleanup. + transaction_cleanup_counters: Arc, /// One-way stop state shared by all executor runners. stopping: AtomicBool, /// Wakeup used to stop every runner after both admissions drain. @@ -522,6 +678,8 @@ impl MandatoryRuntime { executor: async_executor::Executor::new(), admission: MandatoryAdmission::new(config.concurrency_limit), internal_admission: MandatoryInternalAdmission::new(), + operation_counters: Arc::new(MandatoryTaskCounters::default()), + transaction_cleanup_counters: Arc::new(MandatoryTaskCounters::default()), stopping: AtomicBool::new(false), stop_event: Event::new(), poisoner, @@ -558,6 +716,17 @@ impl MandatoryRuntime { (callers, internal) } + /// Return an independently sampled fixed-class statistics snapshot. + #[inline] + pub(crate) fn stats(&self) -> MandatoryRuntimeStats { + let (_, operation_active) = self.admission.inspect(); + let (_, cleanup_active) = self.internal_admission.inspect(); + MandatoryRuntimeStats { + operation: self.operation_counters.snapshot(operation_active), + transaction_cleanup: self.transaction_cleanup_counters.snapshot(cleanup_active), + } + } + async fn run(&self) { self.executor.run(self.wait_for_stop()).await; } @@ -778,6 +947,7 @@ impl QuiescentGuard { .change_context(LifecycleError::RuntimeUnavailable) .attach("phase=mandatory_admission_health_check")); } + let admission_started_at = Instant::now(); let acquire = self.admission.acquire(self.clone()); futures::pin_mut!(acquire); futures::pin_mut!(poison_listener); @@ -795,14 +965,25 @@ impl QuiescentGuard { }; // Winning admission is the poison-race linearization point. A later // poison does not cancel work that is already admitted and accounted. + let admission_wait_nanos = elapsed_nanos(admission_started_at); let metadata = prepared.metadata(); - let (producer, observer) = MandatoryCompletion::endpoints(metadata); + let (producer, observer) = + MandatoryCompletion::endpoints(metadata, Arc::clone(&self.operation_counters)); // No await or expected rejection exists below this ownership edge. + let queued_at = Instant::now(); + self.operation_counters + .record_submitted(admission_wait_nanos); let accepted = prepared.accept(); let task_runtime = self.clone(); self.executor - .spawn(task_runtime.supervise_accepted(accepted, producer, permit)) + .spawn(task_runtime.supervise_accepted( + accepted, + producer, + permit, + queued_at, + admission_wait_nanos, + )) .detach(); Ok(observer) } @@ -817,9 +998,11 @@ impl QuiescentGuard { return Err(job); }; let metadata = job.metadata(); + let queued_at = Instant::now(); let task_runtime = self.clone(); + self.transaction_cleanup_counters.record_submitted(0); self.executor - .spawn(task_runtime.supervise_internal(job, metadata, permit)) + .spawn(task_runtime.supervise_internal(job, metadata, permit, queued_at)) .detach(); Ok(()) } @@ -834,23 +1017,59 @@ impl QuiescentGuard { mut accepted: A, producer: CompletionProducer, permit: MandatoryPermit, + queued_at: Instant, + admission_wait_nanos: usize, ) where A: AcceptedExecution, { + let started_at = Instant::now(); + let queue_wait_nanos = duration_nanos(started_at.duration_since(queued_at)); + let metadata = producer.metadata().clone(); + let counters = Arc::clone(producer.counters()); + counters.record_started(queue_wait_nanos); + obs::debug!( + "event=mandatory_task component=mandatory_runtime action=start result=ok task_class={} task_label={} session_operation={} table_id={} admission_wait_nanos={} queue_wait_nanos={}", + metadata.task_class(), + metadata.task_label(), + OptionalValue(metadata.session_operation()), + OptionalValue(metadata.table_id()), + admission_wait_nanos, + queue_wait_nanos, + ); let outcome = AssertUnwindSafe(async { accepted.execute().await }) .catch_unwind() .await; - match outcome { - Ok(result) => { + let (result, completion_result) = match outcome { + Ok(completion_result) => { + let task_result = if completion_result.is_ok() { + MandatoryTaskResult::Ok + } else { + MandatoryTaskResult::Error + }; accepted.finish(); - producer.complete(result); + (task_result, completion_result) } Err(panic) => { let error = accepted.handle_panic(panic).await; - self.poison_mandatory_panic(producer.metadata()); - producer.complete(Err::(error)); + self.poison_mandatory_panic(&metadata); + (MandatoryTaskResult::Panic, Err::(error)) } - } + }; + let execution_nanos = elapsed_nanos(started_at); + counters.record_completed(result, execution_nanos); + // Publish terminal metrics before waking the observer so an immediate + // statistics snapshot includes the completion that it just consumed. + let observer = producer.complete(completion_result); + obs::debug!( + "event=mandatory_task component=mandatory_runtime action=finish result={} task_class={} task_label={} session_operation={} table_id={} execution_nanos={} observer={}", + result.label(), + metadata.task_class(), + metadata.task_label(), + OptionalValue(metadata.session_operation()), + OptionalValue(metadata.table_id()), + execution_nanos, + observer.label(), + ); drop(accepted); drop(permit); } @@ -865,10 +1084,23 @@ impl QuiescentGuard { mut job: J, metadata: MandatoryTaskMetadata, permit: MandatoryInternalPermit, + queued_at: Instant, ) where J: MandatoryInternalTask, { - if AssertUnwindSafe(async { job.run().await }) + let started_at = Instant::now(); + let queue_wait_nanos = duration_nanos(started_at.duration_since(queued_at)); + self.transaction_cleanup_counters + .record_started(queue_wait_nanos); + obs::debug!( + "event=mandatory_task component=mandatory_runtime action=start result=ok task_class={} task_label={} session_operation={} table_id={} admission_wait_nanos=0 queue_wait_nanos={}", + metadata.task_class(), + metadata.task_label(), + OptionalValue(metadata.session_operation()), + OptionalValue(metadata.table_id()), + queue_wait_nanos, + ); + let result = if AssertUnwindSafe(async { job.run().await }) .catch_unwind() .await .is_err() @@ -876,12 +1108,37 @@ impl QuiescentGuard { job.preserve_after_panic(); let fatal = self.poison_mandatory_panic(&metadata); job.publish_panic(fatal.into_completion_bridge()); - } + MandatoryTaskResult::Panic + } else { + MandatoryTaskResult::Ok + }; + let execution_nanos = elapsed_nanos(started_at); + self.transaction_cleanup_counters + .record_completed(result, execution_nanos); + obs::debug!( + "event=mandatory_task component=mandatory_runtime action=finish result={} task_class={} task_label={} session_operation={} table_id={} execution_nanos={} observer=none", + result.label(), + metadata.task_class(), + metadata.task_label(), + OptionalValue(metadata.session_operation()), + OptionalValue(metadata.table_id()), + execution_nanos, + ); drop(job); drop(permit); } } +#[inline] +fn elapsed_nanos(started_at: Instant) -> usize { + duration_nanos(started_at.elapsed()) +} + +#[inline] +fn duration_nanos(duration: Duration) -> usize { + duration.as_nanos() as usize +} + #[cfg(test)] mod tests { use super::*; @@ -910,8 +1167,11 @@ mod tests { fn mandatory_observer_discloses_operation_error() { runtime::block_on(async { let metadata = MandatoryTaskMetadata::operation("test", None); - let (producer, observer) = MandatoryCompletion::<()>::endpoints(metadata); - producer.complete(Err::<(), _>(CompletionErrorBridge::capture( + let (producer, observer) = MandatoryCompletion::<()>::endpoints( + metadata, + Arc::new(MandatoryTaskCounters::default()), + ); + let _ = producer.complete(Err::<(), _>(CompletionErrorBridge::capture( Report::new(OperationError::TableNotFound).attach("operation=test"), ))); let error = observer.wait().await.unwrap_err(); @@ -956,24 +1216,42 @@ mod tests { } let drops = Arc::new(AtomicUsize::new(0)); + let counters = Arc::new(MandatoryTaskCounters::default()); let (producer, observer) = MandatoryCompletion::::endpoints( MandatoryTaskMetadata::operation("test", None), + Arc::clone(&counters), ); drop(observer); - producer.complete(Ok(DropCount(Arc::clone(&drops)))); + let _ = producer.complete(Ok(DropCount(Arc::clone(&drops)))); assert_eq!(drops.load(Ordering::Relaxed), 1); + assert_eq!(counters.detached_observer_count.load(Ordering::Relaxed), 1); + + let (producer, observer) = MandatoryCompletion::::endpoints( + MandatoryTaskMetadata::operation("test", None), + Arc::clone(&counters), + ); + assert!(matches!( + producer.complete(Ok(DropCount(Arc::clone(&drops)))), + PublishedObserver::Attached + )); + drop(observer); + assert_eq!(drops.load(Ordering::Relaxed), 2); + assert_eq!(counters.detached_observer_count.load(Ordering::Relaxed), 2); } struct SyntheticPrepared { moves: Arc, finishes: Arc, + fail: bool, } struct SyntheticAccepted { moves: Arc, finishes: Arc, + fail: bool, } + #[derive(Debug)] struct SyntheticOutput(usize); struct ExecutePanicPrepared { @@ -1046,6 +1324,153 @@ mod tests { completion: Arc>, } + struct GatedPrepared { + started: Arc>, + release: Arc>, + } + + struct GatedAccepted { + started: Arc>, + release: Arc>, + } + + impl PreparedExecution for GatedPrepared { + type Output = (); + type Accepted = GatedAccepted; + + const LABEL: &'static str = "gated_operation"; + + #[inline] + fn metadata(&self) -> MandatoryTaskMetadata { + MandatoryTaskMetadata::operation(Self::LABEL, None) + } + + #[inline] + fn accept(self) -> Self::Accepted { + GatedAccepted { + started: self.started, + release: self.release, + } + } + } + + impl AcceptedExecution for GatedAccepted { + type Output = (); + + #[inline] + async fn execute(&mut self) -> CompletionResult { + self.started.complete(Ok(())); + self.release.wait_result().await + } + + #[inline] + fn finish(&mut self) {} + + #[inline] + async fn handle_panic(&mut self, _panic: Box) -> CompletionErrorBridge { + CompletionErrorBridge::capture( + Report::new(FatalError::MandatoryTaskPanic).attach("gated operation panic"), + ) + } + } + + struct SignalInternal { + completed: Arc>, + } + + struct OverlapRendezvous { + registrations: AtomicUsize, + ready: Completion<()>, + } + + impl OverlapRendezvous { + #[inline] + fn new() -> Self { + Self { + registrations: AtomicUsize::new(0), + ready: Completion::new(), + } + } + + #[inline] + async fn arrive(&self) -> CompletionResult<()> { + let registrations = self.registrations.fetch_add(1, Ordering::AcqRel) + 1; + assert!(registrations <= 2, "overlap task registered more than once"); + if registrations == 2 { + self.ready.complete(Ok(())); + } + self.ready.wait_result().await + } + } + + struct OverlapPrepared { + rendezvous: Arc, + } + + struct OverlapAccepted { + rendezvous: Arc, + } + + impl PreparedExecution for OverlapPrepared { + type Output = (); + type Accepted = OverlapAccepted; + + const LABEL: &'static str = "overlap_operation"; + + #[inline] + fn metadata(&self) -> MandatoryTaskMetadata { + MandatoryTaskMetadata::operation(Self::LABEL, None) + } + + #[inline] + fn accept(self) -> Self::Accepted { + OverlapAccepted { + rendezvous: self.rendezvous, + } + } + } + + impl AcceptedExecution for OverlapAccepted { + type Output = (); + + #[inline] + async fn execute(&mut self) -> CompletionResult { + self.rendezvous.arrive().await + } + + #[inline] + fn finish(&mut self) {} + + #[inline] + async fn handle_panic(&mut self, _panic: Box) -> CompletionErrorBridge { + CompletionErrorBridge::capture( + Report::new(FatalError::MandatoryTaskPanic).attach("overlap operation panic"), + ) + } + } + + impl MandatoryInternalTask for SignalInternal { + const LABEL: &'static str = "signal_internal"; + + #[inline] + fn metadata(&self) -> MandatoryTaskMetadata { + MandatoryTaskMetadata::transaction_cleanup(Self::LABEL, None) + } + + #[inline] + async fn run(&mut self) { + self.completed.complete(Ok(())); + } + + #[inline] + fn preserve_after_panic(&mut self) {} + + #[inline] + fn publish_panic(&mut self, error: CompletionErrorBridge) { + self.completed.complete(Err(error)); + } + } + impl Drop for SyntheticPanicInternal { #[inline] fn drop(&mut self) { @@ -1095,6 +1520,7 @@ mod tests { SyntheticAccepted { moves: self.moves, finishes: self.finishes, + fail: self.fail, } } } @@ -1104,7 +1530,13 @@ mod tests { #[inline] async fn execute(&mut self) -> CompletionResult { - Ok(SyntheticOutput(self.moves.load(Ordering::Relaxed))) + if self.fail { + Err(CompletionErrorBridge::capture( + Report::new(OperationError::TableNotFound).attach("synthetic error"), + )) + } else { + Ok(SyntheticOutput(self.moves.load(Ordering::Relaxed))) + } } #[inline] @@ -1138,18 +1570,222 @@ mod tests { let mandatory = registry.dependency::(); let moves = Arc::new(AtomicUsize::new(0)); let finishes = Arc::new(AtomicUsize::new(0)); + assert_eq!(mandatory.stats(), MandatoryRuntimeStats::default()); let observer = mandatory .submit(SyntheticPrepared { moves: Arc::clone(&moves), finishes: Arc::clone(&finishes), + fail: false, }) .await .unwrap(); assert_eq!(observer.wait().await.unwrap().0, 1); assert_eq!(moves.load(Ordering::Relaxed), 1); assert_eq!(finishes.load(Ordering::Relaxed), 1); + let stats = mandatory.stats(); + assert_eq!(stats.operation.submitted_count, 1); + assert_eq!(stats.operation.started_count, 1); + assert_eq!(stats.operation.completed_count, 1); + assert_eq!(stats.operation.error_count, 0); + assert_eq!(stats.operation.panic_count, 0); + assert_eq!(stats.operation.detached_observer_count, 0); + assert_eq!(stats.transaction_cleanup, MandatoryTaskStats::default()); + mandatory.drain_callers().await; + assert_eq!(mandatory.stats().operation.active_count, 0); + + (registry, mandatory) + }); + mandatory.close_admission(); + runtime::block_on(mandatory.drain_callers()); + registry.shutdown_all(); + } + + #[test] + fn ordinary_error_and_observer_detach_are_counted_by_outcome() { + let (registry, mandatory) = runtime::block_on(async { + let mut builder = RegistryBuilder::new(); + builder.build::(()).await.unwrap(); + builder + .build::( + MandatoryRuntimeConfig::default() + .worker_threads(1) + .concurrency_limit(1), + ) + .await + .unwrap(); + builder.build::(()).await.unwrap(); + let registry = builder.finish(); + let mandatory = registry.dependency::(); + let moves = Arc::new(AtomicUsize::new(0)); + let finishes = Arc::new(AtomicUsize::new(0)); + + let error = mandatory + .submit(SyntheticPrepared { + moves: Arc::clone(&moves), + finishes: Arc::clone(&finishes), + fail: true, + }) + .await + .unwrap() + .wait() + .await + .unwrap_err(); + assert_eq!(error.kind(), ErrorKind::Operation); + let stats = mandatory.stats().operation; + assert_eq!(stats.submitted_count, 1); + assert_eq!(stats.started_count, 1); + assert_eq!(stats.completed_count, 1); + assert_eq!(stats.error_count, 1); + assert_eq!(stats.panic_count, 0); + mandatory.drain_callers().await; + let observer = mandatory + .submit(SyntheticPrepared { + moves, + finishes, + fail: false, + }) + .await + .unwrap(); + drop(observer); + mandatory.drain_callers().await; + + let stats = mandatory.stats().operation; + assert_eq!(stats.submitted_count, 2); + assert_eq!(stats.started_count, 2); + assert_eq!(stats.completed_count, 2); + assert_eq!(stats.error_count, 1); + assert_eq!(stats.panic_count, 0); + assert_eq!(stats.detached_observer_count, 1); + assert_eq!(stats.active_count, 0); + (registry, mandatory) + }); + mandatory.close_admission(); + runtime::block_on(mandatory.drain_callers()); + registry.shutdown_all(); + } + + #[test] + fn internal_work_progresses_with_one_runner_and_saturated_caller_capacity() { + let (registry, mandatory) = runtime::block_on(async { + let mut builder = RegistryBuilder::new(); + builder.build::(()).await.unwrap(); + builder + .build::( + MandatoryRuntimeConfig::default() + .worker_threads(1) + .concurrency_limit(1), + ) + .await + .unwrap(); + builder.build::(()).await.unwrap(); + let registry = builder.finish(); + let mandatory = registry.dependency::(); + for _ in 0..32 { + let first_started = Arc::new(Completion::new()); + let first_release = Arc::new(Completion::new()); + let first = mandatory + .submit(GatedPrepared { + started: Arc::clone(&first_started), + release: Arc::clone(&first_release), + }) + .await + .unwrap(); + first_started.wait_result().await.unwrap(); + + let moves = Arc::new(AtomicUsize::new(0)); + let finishes = Arc::new(AtomicUsize::new(0)); + let waiting = mandatory.submit(SyntheticPrepared { + moves: Arc::clone(&moves), + finishes: Arc::clone(&finishes), + fail: false, + }); + futures::pin_mut!(waiting); + assert!(matches!( + futures::poll!(waiting.as_mut()), + std::task::Poll::Pending + )); + assert_eq!(moves.load(Ordering::Relaxed), 0); + + let internal_completed = Arc::new(Completion::new()); + assert!( + mandatory + .submit_internal(SignalInternal { + completed: Arc::clone(&internal_completed), + }) + .is_ok() + ); + internal_completed.wait_result().await.unwrap(); + mandatory.internal_admission.drain().await; + assert!(matches!( + futures::poll!(waiting.as_mut()), + std::task::Poll::Pending + )); + assert_eq!(moves.load(Ordering::Relaxed), 0); + + first_release.complete(Ok(())); + first.wait().await.unwrap(); + let second = waiting.await.unwrap(); + assert_eq!(second.wait().await.unwrap().0, 1); + mandatory.drain_callers().await; + } + + let stats = mandatory.stats(); + assert_eq!(stats.operation.submitted_count, 64); + assert_eq!(stats.operation.completed_count, 64); + assert_eq!(stats.operation.active_count, 0); + assert_eq!(stats.transaction_cleanup.submitted_count, 32); + assert_eq!(stats.transaction_cleanup.completed_count, 32); + assert_eq!(stats.transaction_cleanup.active_count, 0); + (registry, mandatory) + }); + mandatory.close_admission(); + runtime::block_on(mandatory.drain_callers()); + registry.shutdown_all(); + } + + #[test] + fn independent_accepted_tasks_overlap_without_blocking_runners() { + let (registry, mandatory) = runtime::block_on(async { + let mut builder = RegistryBuilder::new(); + builder.build::(()).await.unwrap(); + builder + .build::( + MandatoryRuntimeConfig::default() + .worker_threads(2) + .concurrency_limit(2), + ) + .await + .unwrap(); + builder.build::(()).await.unwrap(); + let registry = builder.finish(); + let mandatory = registry.dependency::(); + for _ in 0..32 { + let rendezvous = Arc::new(OverlapRendezvous::new()); + let first = mandatory + .submit(OverlapPrepared { + rendezvous: Arc::clone(&rendezvous), + }) + .await + .unwrap(); + let second = mandatory + .submit(OverlapPrepared { + rendezvous: Arc::clone(&rendezvous), + }) + .await + .unwrap(); + + first.wait().await.unwrap(); + second.wait().await.unwrap(); + assert_eq!(rendezvous.registrations.load(Ordering::Acquire), 2); + mandatory.drain_callers().await; + } + let stats = mandatory.stats().operation; + assert_eq!(stats.submitted_count, 64); + assert_eq!(stats.started_count, 64); + assert_eq!(stats.completed_count, 64); + assert_eq!(stats.active_count, 0); (registry, mandatory) }); mandatory.close_admission(); @@ -1196,6 +1832,16 @@ mod tests { ); assert_eq!(preserved.load(Ordering::Relaxed), 1); assert_eq!(published.load(Ordering::Relaxed), 1); + mandatory.internal_admission.drain().await; + let stats = mandatory.stats(); + assert_eq!(stats.operation, MandatoryTaskStats::default()); + assert_eq!(stats.transaction_cleanup.submitted_count, 1); + assert_eq!(stats.transaction_cleanup.started_count, 1); + assert_eq!(stats.transaction_cleanup.completed_count, 1); + assert_eq!(stats.transaction_cleanup.error_count, 0); + assert_eq!(stats.transaction_cleanup.panic_count, 1); + assert_eq!(stats.transaction_cleanup.detached_observer_count, 0); + assert_eq!(stats.transaction_cleanup.active_count, 0); (registry, mandatory, dropped) }); @@ -1241,6 +1887,15 @@ mod tests { let poison = format!("{poison:?}"); assert!(poison.contains("task_class=operation"), "{poison}"); assert!(poison.contains("task_label=execute_panic"), "{poison}"); + mandatory.drain_callers().await; + let stats = mandatory.stats().operation; + assert_eq!(stats.submitted_count, 1); + assert_eq!(stats.started_count, 1); + assert_eq!(stats.completed_count, 1); + assert_eq!(stats.error_count, 0); + assert_eq!(stats.panic_count, 1); + assert_eq!(stats.detached_observer_count, 0); + assert_eq!(stats.active_count, 0); (registry, mandatory) }); diff --git a/doradb-storage/src/runtime/mod.rs b/doradb-storage/src/runtime/mod.rs index ae16dc4e..610d05c5 100644 --- a/doradb-storage/src/runtime/mod.rs +++ b/doradb-storage/src/runtime/mod.rs @@ -5,6 +5,9 @@ use std::task::{Context, Poll}; pub(crate) mod mandatory; +/// Logical work items completed between cooperative executor yields. +pub(crate) const POLL_BUDGET: usize = 128; + /// One-shot cooperative yield future. pub(crate) struct YieldNow { yielded: bool, diff --git a/doradb-storage/src/session.rs b/doradb-storage/src/session.rs index 91738c09..765026a0 100644 --- a/doradb-storage/src/session.rs +++ b/doradb-storage/src/session.rs @@ -23,8 +23,9 @@ use crate::notify::EventNotifyOnDrop; use crate::quiescent::QuiescentGuard; use crate::runtime::mandatory::{AcceptedExecution, MandatoryTaskMetadata, PreparedExecution}; use crate::stats::{ - BufferPoolStats, StorageIoStats, TransactionSystemStats, buffer_pool_runtime_stats_snapshot, - storage_io_stats_snapshot, transaction_system_stats_snapshot, + BufferPoolStats, MandatoryRuntimeStats, StorageIoStats, TransactionSystemStats, + buffer_pool_runtime_stats_snapshot, storage_io_stats_snapshot, + transaction_system_stats_snapshot, }; use crate::table::{ CheckpointDelayReason, CheckpointOutcome, CheckpointRetryObservation, FreezeOutcome, @@ -1317,6 +1318,21 @@ impl Session { }) } + /// Return mandatory-runtime task and timing statistics by fixed task class. + /// + /// This read-only diagnostic remains observable after storage poison while + /// the engine lifecycle is still running. It is not available after engine + /// shutdown, session close, or registry removal. Monotonic fields and + /// current active counts are independently sampled. + #[inline] + pub fn mandatory_runtime_stats(&self) -> Result { + let session = self + .pin_inspection() + .attach("operation=query_mandatory_runtime_stats") + .disclose()?; + Ok(session.engine.mandatory_runtime.stats()) + } + /// Freeze a row-page prefix or report the existing table-owned batch. #[inline] pub async fn freeze_table( @@ -3001,7 +3017,10 @@ pub(crate) mod tests { use crate::lock::tests::LockDebugEntryState; use crate::log::LogSync; use crate::log::format::REDO_DEFAULT_DATA_START_OFFSET; - use crate::stats::{BufferPoolCounters, BufferPoolRuntimeStats, TransactionSystemStats}; + use crate::stats::{ + BufferPoolCounters, BufferPoolRuntimeStats, MandatoryRuntimeStats, MandatoryTaskStats, + TransactionSystemStats, + }; use crate::table::tests::{ FailingFirstWriteHook, assert_freeze_created, has_lock_entry, lightweight_test_engine_config, lock_entry_count, maintenance_lock_owner, @@ -3637,6 +3656,7 @@ pub(crate) mod tests { assert!(session.transaction_system_stats().is_ok()); assert!(session.storage_io_stats().is_ok()); assert!(session.buffer_pool_stats().is_ok()); + assert!(session.mandatory_runtime_stats().is_ok()); assert!( session .wait_for_checkpoint_retry(CheckpointDelayReason::ActiveRoot { @@ -5688,6 +5708,7 @@ pub(crate) mod tests { let trx0 = session.transaction_system_stats().unwrap(); let storage0 = session.storage_io_stats().unwrap(); let pools0 = session.buffer_pool_stats().unwrap(); + let mandatory0 = session.mandatory_runtime_stats().unwrap(); assert_eq!(trx0.commit_count, 0); assert_eq!(trx0.trx_count, 0); assert_eq!(trx0.log_bytes, 0); @@ -5695,11 +5716,14 @@ pub(crate) mod tests { assert!(pools0.mem.capacity > 0); assert!(pools0.index.capacity > 0); assert!(pools0.disk.capacity > 0); + assert_eq!(mandatory0, MandatoryRuntimeStats::default()); let _table_id = table1(&engine).await; let trx1 = session.transaction_system_stats().unwrap(); let storage1 = session.storage_io_stats().unwrap(); let pools1 = session.buffer_pool_stats().unwrap(); + engine.inner().mandatory_runtime.drain_callers().await; + let mandatory1 = session.mandatory_runtime_stats().unwrap(); // Commit waiters can complete before the redo thread publishes // aggregate stats, so this test verifies monotonic snapshots // rather than immediate progress from the preceding operation. @@ -5709,6 +5733,17 @@ pub(crate) mod tests { assert!(storage1.pool_read_requests >= storage0.pool_read_requests); assert!(storage1.background_write_requests >= storage0.background_write_requests); assert_buffer_pool_stats_monotonic(&pools0, &pools1); + assert_eq!(mandatory1.operation.submitted_count, 1); + assert_eq!(mandatory1.operation.started_count, 1); + assert_eq!(mandatory1.operation.completed_count, 1); + assert_eq!(mandatory1.operation.error_count, 0); + assert_eq!(mandatory1.operation.panic_count, 0); + assert_eq!(mandatory1.operation.detached_observer_count, 0); + assert_eq!(mandatory1.operation.active_count, 0); + assert_eq!( + mandatory1.transaction_cleanup, + MandatoryTaskStats::default() + ); }); } @@ -5733,6 +5768,7 @@ pub(crate) mod tests { session.transaction_system_stats().unwrap_err(), session.storage_io_stats().unwrap_err(), session.buffer_pool_stats().unwrap_err(), + session.mandatory_runtime_stats().unwrap_err(), ] { assert_eq!(err.kind(), ErrorKind::Lifecycle); assert_eq!( @@ -5760,6 +5796,9 @@ pub(crate) mod tests { ); assert_runtime_unavailable_after_shutdown(session.storage_io_stats().unwrap_err()); assert_runtime_unavailable_after_shutdown(session.buffer_pool_stats().unwrap_err()); + assert_runtime_unavailable_after_shutdown( + session.mandatory_runtime_stats().unwrap_err(), + ); }); } @@ -5788,6 +5827,7 @@ pub(crate) mod tests { assert!(session.transaction_system_stats().is_ok()); assert!(session.storage_io_stats().is_ok()); assert!(session.buffer_pool_stats().is_ok()); + assert!(session.mandatory_runtime_stats().is_ok()); let err = session.truncate_redo_log().await.unwrap_err(); assert_runtime_unavailable_after_fatal(err, FatalError::RedoWrite); diff --git a/doradb-storage/src/stats.rs b/doradb-storage/src/stats.rs index 933001fe..fbd553e6 100644 --- a/doradb-storage/src/stats.rs +++ b/doradb-storage/src/stats.rs @@ -4,6 +4,44 @@ use crate::file::fs::StorageServiceStats as InternalStorageServiceStats; use crate::io::BackendStats as InternalIoBackendStats; use crate::trx::sys::TrxSysStats as InternalTrxSysStats; +/// Snapshot of the engine-owned mandatory runtime's fixed task classes. +/// +/// Count and duration fields are monotonic diagnostics. Active counts are +/// independently sampled current state, so concurrent snapshots do not promise +/// equations between submitted, started, completed, and active work. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct MandatoryRuntimeStats { + /// Accepted caller DDL and maintenance task statistics. + pub operation: MandatoryTaskStats, + /// Engine-internal transaction-cleanup task statistics. + pub transaction_cleanup: MandatoryTaskStats, +} + +/// Snapshot of one mandatory runtime task class. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct MandatoryTaskStats { + /// Number of tasks successfully accepted and detached for execution. + pub submitted_count: usize, + /// Number of tasks that received their first executor poll. + pub started_count: usize, + /// Number of tasks that published terminal supervisor handling. + pub completed_count: usize, + /// Number of accepted caller tasks that returned an ordinary error. + pub error_count: usize, + /// Number of tasks whose supervised execution panicked. + pub panic_count: usize, + /// Number of caller observers dropped without consuming their result. + pub detached_observer_count: usize, + /// Current tasks retained by the authoritative class admission accounting. + pub active_count: usize, + /// Total successful caller-admission wait time in nanoseconds. + pub admission_wait_nanos: usize, + /// Total accepted-to-first-poll queue time in nanoseconds. + pub queue_wait_nanos: usize, + /// Total first-poll-to-terminal-publication execution time in nanoseconds. + pub execution_nanos: usize, +} + /// Monotonic transaction-system, redo, and purge statistics. #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] pub struct TransactionSystemStats { diff --git a/doradb-storage/src/trx/mod.rs b/doradb-storage/src/trx/mod.rs index 7acc95a9..a4383554 100644 --- a/doradb-storage/src/trx/mod.rs +++ b/doradb-storage/src/trx/mod.rs @@ -1139,7 +1139,7 @@ struct SessionOperationEntryInner { /// Public transactions have no separate outer authority and always use /// `false`. For private transactions this remains separate from `state` /// because dropping the outer authority while the core is checked out - /// leaves `ForegroundRunning(Some(Running))` unchanged until that core + /// leaves `Voluntary(Some(Running))` unchanged until that core /// returns. The flag then determines whether return/completion resumes the /// attached outer operation or publishes outer cleanup/terminal state. outer_foreground_alive: bool, diff --git a/doradb-storage/src/trx/undo/index.rs b/doradb-storage/src/trx/undo/index.rs index 87b0b033..18e06441 100644 --- a/doradb-storage/src/trx/undo/index.rs +++ b/doradb-storage/src/trx/undo/index.rs @@ -3,6 +3,7 @@ use crate::catalog::{TableCache, is_catalog_table}; use crate::error::RuntimeResult as Result; use crate::id::{RowID, TableID, TrxID}; use crate::row::ops::SelectKey; +use crate::runtime::{POLL_BUDGET, yield_now}; use crate::table::IndexRollback; /// Buffer of index undo entries accumulated for rollback and GC handoff. @@ -47,6 +48,7 @@ impl IndexUndoLogs { guards: &PoolGuards, ts: TrxID, ) -> Result<()> { + let mut budget = POLL_BUDGET; while !self.0.is_empty() { { // Keep the current entry vector-owned across every await. If @@ -71,6 +73,11 @@ impl IndexUndoLogs { } } self.0.pop(); + budget -= 1; + if budget == 0 && !self.0.is_empty() { + yield_now().await; + budget = POLL_BUDGET; + } } Ok(()) } diff --git a/doradb-storage/src/trx/undo/row.rs b/doradb-storage/src/trx/undo/row.rs index 90c72b49..11fbd748 100644 --- a/doradb-storage/src/trx/undo/row.rs +++ b/doradb-storage/src/trx/undo/row.rs @@ -4,6 +4,7 @@ use crate::catalog::{TableCache, is_catalog_table}; use crate::error::RuntimeResult as Result; use crate::id::{RowID, TableID, TrxID}; use crate::row::ops::{SelectKey, UndoCol, UpdateCol}; +use crate::runtime::{POLL_BUDGET, yield_now}; use crate::trx::{MIN_SNAPSHOT_TS, SharedTrxStatus, trx_is_committed}; use crate::value::Val; use event_listener::EventListener; @@ -128,6 +129,7 @@ impl RowUndoLogs { table_cache: &mut TableCache<'_>, guards: &PoolGuards, ) -> Result<()> { + let mut budget = POLL_BUDGET; while !self.0.is_empty() { { // Keep the current entry vector-owned across every await. Its @@ -157,6 +159,11 @@ impl RowUndoLogs { } } self.0.pop(); + budget -= 1; + if budget == 0 && !self.0.is_empty() { + yield_now().await; + budget = POLL_BUDGET; + } } Ok(()) }