diff --git a/.github/actions/check-nvx-rs/action.yml b/.github/actions/check-nvx-rs/action.yml
index 02d37edb2..9ec2984f8 100644
--- a/.github/actions/check-nvx-rs/action.yml
+++ b/.github/actions/check-nvx-rs/action.yml
@@ -71,7 +71,7 @@ runs:
set -euo pipefail
cargo "+${{ steps.toolchains.outputs.toolchain }}" test --locked --all-features
cargo "+${{ steps.toolchains.outputs.toolchain }}" test --locked
- # Without features, the library still holds the protocol and the execution engine.
+ # Without features, the library still holds the protocol.
cargo "+${{ steps.toolchains.outputs.toolchain }}" build --locked --lib --no-default-features
cargo "+${{ steps.toolchains.outputs.toolchain }}" test --locked --lib --no-default-features
diff --git a/doc/acceptance-case-matrix.md b/doc/acceptance-case-matrix.md
index d04744f7d..bc85a7719 100644
--- a/doc/acceptance-case-matrix.md
+++ b/doc/acceptance-case-matrix.md
@@ -34,7 +34,7 @@ its own in turn.
| [The adapter's other tests](#the-adapters-other-tests) | 169 | 0 | 9 | 160 |
| [MicroVM scenarios](#microvm-scenarios) | 54 | 50 | 3 | 1 |
| [CI](#ci) | 69 | 43 | 3 | 23 |
-| [Docs](#docs) | 36 | 28 | 2 | 6 |
+| [Docs](#docs) | 36 | 26 | 3 | 7 |
## The adapter's scenarios
@@ -478,18 +478,18 @@ this matrix. The README's own code is no doctest.
| Code | Fate | Note |
| --- | --- | --- |
| `nvx-rs/src/activation/mod.rs - activation (line 130)` | kept | |
-| `nvx-rs/src/exec/mod.rs - exec (line 113)` | kept | Its module leaves with the exec engine (#1087), whose PR retires it |
+| `nvx-rs/src/exec/mod.rs - exec (line 113)` | retired | The exec engine leaves (SL6.11a) |
| `nvx-rs/src/gpt/mod.rs - gpt (line 43)` | kept | |
| `nvx-rs/src/guest/activation.rs - guest::activation (line 195)` | kept | Its capture preparation and admission fence leave (#1087), whose PR rewrites it on `guest::snapshot::capture` |
| `nvx-rs/src/guest/activation/state.rs - guest::activation::state::Activation (line 56)` | kept | |
| `nvx-rs/src/guest/cgroup.rs - guest::cgroup (line 36)` | kept | Its module leaves with the exec engine (#1087), whose PR retires it |
| `nvx-rs/src/guest/entry.rs - guest::entry (line 90)` | kept | Its module leaves with the exec engine (#1087), whose PR retires it |
-| `nvx-rs/src/guest/exec.rs - guest::exec (line 83)` | kept | Its module leaves with the exec engine (#1087), whose PR retires it |
-| `nvx-rs/src/guest/exec.rs - guest::exec::Execution::adopt (line 625)` | kept | Its module leaves with the exec engine (#1087), whose PR retires it |
+| `nvx-rs/src/guest/exec.rs - guest::exec (line 80)` | kept | Its module leaves with the exec engine (#1087), whose PR retires it |
+| `nvx-rs/src/guest/exec.rs - guest::exec::Execution::adopt (line 620)` | kept | Its module leaves with the exec engine (#1087), whose PR retires it |
| `nvx-rs/src/guest/gpt.rs - guest::gpt (line 18)` | kept | The guest's space becomes private (#1127), whose PR moves it to the space's unit tests |
| `nvx-rs/src/guest/mappings/mapper.rs - guest::mappings::mapper (line 107)` | kept | The guest's space becomes private (#1127), whose PR moves it to the space's unit tests |
| `nvx-rs/src/guest/mappings/table.rs - guest::mappings::table::MappingTable (line 47)` | kept | |
-| `nvx-rs/src/guest/mod.rs - guest (line 41)` | kept | The guest-agent guidelines quote it |
+| `nvx-rs/src/guest/mod.rs - guest (line 40)` | kept | The guest-agent guidelines quote it |
| `nvx-rs/src/guest/nvxs.rs - guest::nvxs (line 22)` | kept | |
| `nvx-rs/src/guest/process.rs - guest::process (line 12)` | kept | Its module leaves with the exec engine (#1087), whose PR retires it |
| `nvx-rs/src/guest/process.rs - guest::process (line 39)` | kept | Its module leaves with the exec engine (#1087), whose PR retires it |
@@ -497,7 +497,7 @@ this matrix. The README's own code is no doctest.
| `nvx-rs/src/guest/root.rs - guest::root (line 130)` | kept | The guest's space becomes private (#1127), whose PR moves it to the space's unit tests |
| `nvx-rs/src/guest/snapshot.rs - guest::snapshot (line 25)` | kept | The guest-agent guidelines quote it |
| `nvx-rs/src/guest/spawn.rs - guest::spawn (line 41)` | kept | Its module leaves with the exec engine (#1087), whose PR retires it |
-| `nvx-rs/src/guest/spawn.rs - guest::spawn::LaunchStage::from_code (line 201)` | kept | Its module leaves with the exec engine (#1087), whose PR retires it |
+| `nvx-rs/src/guest/spawn.rs - guest::spawn::LaunchStage::from_code (line 201)` | rewritten | `nvx-rs/src/guest/spawn/tests.rs::launch_stages_keep_their_numbers`. The exec engine's launch failure and outcome, which the example read, leave with the engine (SL6.11a) |
| `nvx-rs/src/guest/terminal.rs - guest::terminal (line 10)` | kept | |
| `nvx-rs/src/guest/time/mod.rs - guest::time (line 15)` | kept | |
| `nvx-rs/src/host/sandbox/prepared.rs - host::sandbox::prepared::PreparedLaunch (line 92)` | kept | Its launch type leaves with the old launch paths (#1087), whose PR retires it |
@@ -507,9 +507,9 @@ this matrix. The README's own code is no doctest.
| `nvx-rs/src/host/template/plan.rs - host::template::plan::TemplatePlan (line 137)` | retired | `TemplatePlan` leaves with the template store (SL6.2a, #1121) |
| `nvx-rs/src/host/template/store.rs - host::template::store::Store (line 473)` | retired | The template store leaves (SL6.2a, #1121) |
| `nvx-rs/src/host/transport.rs - host::transport (line 22)` | kept | |
-| `nvx-rs/src/lib.rs - (line 58)` | rewritten | Its `Instance` and sandbox store leave (SL6.1, #1121): `nvx-rs/src/lib.rs - (line 62)` compiles a cold boot and a stop through `host::vm` |
+| `nvx-rs/src/lib.rs - (line 58)` | rewritten | Its `Instance` and sandbox store leave (SL6.1, #1121): `nvx-rs/src/lib.rs - (line 57)` compiles a cold boot and a stop through `host::vm` |
| `nvx-rs/src/protocol/nvxs/mod.rs - protocol::nvxs (line 68)` | kept | |
| `nvx-rs/src/host/snapshot/capture.rs - host::snapshot::capture::finish_capture (line 48)` | kept | |
| `nvx-rs/src/host/store/mod.rs - host::store (line 104)` | retired | The sandbox store leaves (SL6.1, #1121) |
-| `nvx-rs/README.md`, its Example: an instance through `LaunchSpec` and the sandbox store | rewritten | Nothing compiles it. `nvx-rs/examples/host-workflows.rs` compiles the documented host workflows, and the library's guide (SL6.10a) replaces the example. SL6.1 (#1121) made it the cold boot that `nvx-rs/src/lib.rs - (line 62)` compiles |
+| `nvx-rs/README.md`, its Example: an instance through `LaunchSpec` and the sandbox store | rewritten | Nothing compiles it. `nvx-rs/examples/host-workflows.rs` compiles the documented host workflows, and the library's guide (SL6.10a) replaces the example. SL6.1 (#1121) made it the cold boot that `nvx-rs/src/lib.rs - (line 57)` compiles |
| `nvx-rs/examples/external-launch.rs` | kept | Compiled with the crate's targets, and run by hand on a host |
diff --git a/doc/guest-agent-guidelines.md b/doc/guest-agent-guidelines.md
index 189296142..76740bcaf 100644
--- a/doc/guest-agent-guidelines.md
+++ b/doc/guest-agent-guidelines.md
@@ -55,7 +55,7 @@ announcer, which is a program, `/sbin/nvx-ingress`: an init runs it itself,
or has `nvx-ingress start` run it from a socket that it binds for it. Until
the nvx-rs slimming
([microsoft/nvx#1076](https://github.com/microsoft/nvx/issues/1076)) removes
-them, the crate also keeps the older execution engine and its process
+them, the crate also keeps the older process
modules, `guest::{exec, spawn, entry, cgroup, process}`, which NVX's older
guest agents use. nvx-init uses none of these modules, and its time daemon
runs for the VM's life.
diff --git a/doc/modules/README.md b/doc/modules/README.md
index 781aedeaf..f29641690 100644
--- a/doc/modules/README.md
+++ b/doc/modules/README.md
@@ -51,13 +51,6 @@ to a module that it contains.

-### `nvx::exec`
-
-`nvx::exec`, in [`nvx-rs/src/exec/mod.rs`](../../nvx-rs/src/exec/mod.rs),
-contains 6 modules.
-
-
-
### `nvx::gpt`
`nvx::gpt`, in [`nvx-rs/src/gpt/mod.rs`](../../nvx-rs/src/gpt/mod.rs), contains
diff --git a/doc/modules/nvx-exec.svg b/doc/modules/nvx-exec.svg
deleted file mode 100644
index 6bc850016..000000000
--- a/doc/modules/nvx-exec.svg
+++ /dev/null
@@ -1,112 +0,0 @@
-
-
-
-
-
diff --git a/nvx-rs/README.md b/nvx-rs/README.md
index 513d02f27..678cd5bf6 100644
--- a/nvx-rs/README.md
+++ b/nvx-rs/README.md
@@ -43,8 +43,6 @@ nvx-rs/
│ │ by the host and the guest
│ ├── hypervisor.rs
│ ├── error.rs nvx::Error, its ErrorKind, and its stable ErrorCode
-│ ├── exec/ the execution engine: the streams, credit, and outcome of an execution's two
-│ │ sides, without I/O, threads, or features, for any framing and host
│ ├── host/ feature `host` (default): the runtime that every launch shares, VM
│ │ directories and configurations, artifacts, launch, network and egress policy
│ │ files, filesystem, NVXS connection, session, process, platform, templates'
@@ -55,12 +53,12 @@ nvx-rs/
│ ├── guest/ feature `guest`: the programs that run in Linux guests, the guest's NVXS
│ │ connection, the libraries through which a guest init or agent opens the
│ │ terminal on which it serves its session, starts a program, drives its
-│ │ execution, alone or through the execution engine, collects its children's
-│ │ statuses, settles its executions' cgroups, mounts host mappings beneath the
-│ │ descriptors of the roots that it names, assembles and tears down a layered
-│ │ root, reads a GPT disk's partition table through its block device, and
-│ │ prepares a template's guest for a capture before its root's assembly, and
-│ │ the typed calls through which it runs the time ABI's steps and the reseed
+│ │ execution, collects its children's statuses, settles its executions'
+│ │ cgroups, mounts host mappings beneath the descriptors of the roots that it
+│ │ names, assembles and tears down a layered root, reads a GPT disk's partition
+│ │ table through its block device, and prepares a template's guest for a
+│ │ capture before its root's assembly, and the typed calls through which it
+│ │ runs the time ABI's steps and the reseed
│ └── bin/ nvx-rs, nvx-fake-openvmm, and the guest binaries
├── tests/ fake-OpenVMM, CLI, NVXS session, process and spawn, checked-handoff,
│ host-module layering, and guest-binary tests, the CLI's scenario tests and
@@ -115,7 +113,6 @@ changes only when the module's purpose does.
| [`nvx::gpt`](#nvxgpt) | Strict reading of GPT disks whose first partition configures an adapter and whose others hold a root's read-only layers |
| [`nvx::uuid`](#nvxuuid) | UUIDs, such as those of GPT disks and filesystems, in their canonical text and in GPT's layout |
| [`nvx::Hypervisor`](#nvxhypervisor) | KVM, MSHV, or WHP, with its serde form behind the `serde` feature |
-| [`nvx::exec`](#nvxexec) | The execution engine: the streams, credit, and outcome of an execution's two sides, without I/O, for any framing |
| [`nvx::host::runtime`](#nvxhostruntime) | `Runtime`: what every launch shares, the runtime's files and their identity, the hypervisor and its probe, and the launch options |
| [`nvx::host::vm`](#nvxhostvm) | A private VM directory, cold, capture and restore preparation, owned launch commands, verified detached launch, confirmed stop, and caller-saved reattachment |
| [`nvx::host::snapshot`](#nvxhostsnapshot) | Strict private snapshots, post-exit capture finishing, verified metadata, VM compatibility, captured NIC addresses, and time eligibility |
@@ -142,7 +139,6 @@ changes only when the module's purpose does.
| [`nvx::guest::process`](#nvxguestprocess) | `Child`: the status of one child that the caller names, and of no other |
| [`nvx::guest::cgroup`](#nvxguestcgroup) | Cgroup v2 handles through which a guest init or agent places, accounts, kills, and settles its executions |
| [`nvx::guest::exec`](#nvxguestexec) | `Execution`: drives one program as an execution, for any guest init or agent |
-| [`nvx::guest::exec::binding`](#nvxguestexecbinding) | `Binding`: an `nvx::exec::Runner` bound to an `Execution`, the glue that a guest adapter would otherwise write |
| [`nvx::guest::time`](#nvxguesttime) | x86_64: the steps of NVX time ABI v1 as typed calls in the caller's process, for any guest init or agent |
| [`nvx::guest::snapshot`](#nvxguestsnapshot) | x86_64: bounded Fresh capture with inline restore repair and staged failure dispositions, without execution control |
| [`nvx::guest::activation`](#nvxguestactivation) | x86_64: fence-free activation, with the retained template capture and clone repair interface |
@@ -393,31 +389,6 @@ dependencies, so the host and the guest share it.
KVM, MSHV, or WHP, with its serde form behind the `serde` feature.
-### `nvx::exec`
-
-The execution engine: the stream state, credit, ordering, and outcome of one
-execution, for any framing that carries its messages, without I/O, threads,
-global state, or features, so that it builds with `--no-default-features`.
-
-A `Runner` is the side that runs the workload: it reads the workload's output
-under the peer's credit, holds the input that the peer sent within its window
-until the workload takes it, requests the terminations, discards, input close,
-and settlement that its adapter applies, and returns one terminal `Outcome`,
-which tells whether the execution is reusable or fenced.
-
-A `Peer` is the other side: it takes output into windows that its sink empties,
-returns credit as the sink takes bytes, sends input within the runner's window,
-and ends the execution on a cancellation, a timeout, a drop, a reset, or a
-disconnection.
-
-Both take the current `Instant` from the caller, report their next deadline, and
-never wait, so a blocked stream, sink, window, or transport stops no timer,
-cancellation, or other execution, and both bound their memory.
-
-The adapter chooses one byte window per stream, or frames charged against one
-pool that stdout and stderr share, and the maximum chunk, in `Limits`; a `Key`'s
-generation tells stale messages from foreign ones.
-
### `nvx::host::runtime`
`Runtime`: what every launch shares, the paths of the OpenVMM executable, guest
@@ -1465,29 +1436,6 @@ one.
NVX's managed agent is one caller.
-### `nvx::guest::exec::binding`
-
-`Binding`: an `nvx::exec::Runner` bound to an `Execution`, the glue that a guest
-adapter would otherwise write.
-
-The runner's read allowances become grants, its discards discards, the input
-that it accepted writes and then a close, its terminations a cancellation, a
-disconnection, or a termination, and its settlement a kill of the execution's
-cgroup and checks until no process occupies it; without a cgroup, nothing proves
-that the program's descendants are gone, so the execution is fenced.
-
-It feeds the execution's events back to the runner, statuses that a caller who
-reaps hands over with `observe` included, and never blocks: the caller polls its
-descriptors and its input writer until its deadline, and each call takes a
-bounded number of steps, so that no execution holds the caller's loop.
-
-The caller's own `signal` passes through to the execution, which it does not
-end: the runner sees only the program's end, once the program exits.
-
-A launch that failed reaches the runner with the number of the `LaunchStage`
-that failed, `LaunchStage::code`, or with 0, which numbers no stage, if the
-binding could not read the launch report.
-
### `nvx::guest::time`
x86_64: `TimeAbi`, the steps of NVX time ABI v1 as typed calls in the caller's
@@ -3111,12 +3059,7 @@ The process and spawn test spawns VMM stand-ins through the low-level spawn,
waits for them through their verified references, finds one again after the
process that spawned it crashed, and checks that a VMM inherits only its
standard streams and the descriptors or handles that it is allowed while
-another thread spawns processes. The tests of `nvx::exec` drive a runner and a
-peer through a
-simulated workload, transport, and sink on a clock of their own, and, on Linux,
-the tests of `nvx::guest::exec::binding` drive real programs through them, two
-at once in one loop, one of them with a stalled sink; run as root, one of them
-also settles a real cgroup.
+another thread spawns processes.
On Linux, with the `cli` feature, a loopback test drives
the host's private `ControlSession` through a broker shim into the real managed
agent's session, dispatch, and execution engine, and `nvx.py` also lints and
diff --git a/nvx-rs/public-api/linux.txt b/nvx-rs/public-api/linux.txt
index 2967ea292..33fdecb62 100644
--- a/nvx-rs/public-api/linux.txt
+++ b/nvx-rs/public-api/linux.txt
@@ -226,243 +226,6 @@ const nvx::contract::SNAPSHOT_FORMAT_VERSION
const nvx::contract::TIME_ABI_VERSION
fn nvx::contract::manifest_is_compatible
fn nvx::contract::manifest_mismatches
-mod nvx::exec
-enum nvx::exec::Cause
-variant nvx::exec::Cause::Cancelled
-variant nvx::exec::Cause::Disconnected
-variant nvx::exec::Cause::Dropped
-variant nvx::exec::Cause::Reset
-variant nvx::exec::Cause::TimedOut
-struct nvx::exec::Key
-method nvx::exec::Key::generation
-method nvx::exec::Key::id
-method nvx::exec::Key::new
-struct nvx::exec::LaunchFailure
-method nvx::exec::LaunchFailure::error
-method nvx::exec::LaunchFailure::new
-method nvx::exec::LaunchFailure::stage
-struct nvx::exec::Limits
-method nvx::exec::Limits::chunk
-method nvx::exec::Limits::input
-method nvx::exec::Limits::is_shared
-method nvx::exec::Limits::new
-method nvx::exec::Limits::output
-method nvx::exec::Limits::shared_output
-struct nvx::exec::Outcome
-method nvx::exec::Outcome::cause
-method nvx::exec::Outcome::is_reusable
-method nvx::exec::Outcome::is_truncated
-method nvx::exec::Outcome::launch_failure
-method nvx::exec::Outcome::new
-method nvx::exec::Outcome::status
-method nvx::exec::Outcome::with_cause
-method nvx::exec::Outcome::with_launch_failure
-method nvx::exec::Outcome::with_reusable
-method nvx::exec::Outcome::with_truncated
-struct nvx::exec::Peer
-method nvx::exec::Peer::buffered
-method nvx::exec::Peer::cancel
-method nvx::exec::Peer::consumed
-method nvx::exec::Peer::credit
-method nvx::exec::Peer::deadline
-method nvx::exec::Peer::disconnect
-method nvx::exec::Peer::dropped
-method nvx::exec::Peer::end_input
-method nvx::exec::Peer::end_output
-method nvx::exec::Peer::has_started
-method nvx::exec::Peer::high_water
-method nvx::exec::Peer::input
-method nvx::exec::Peer::input_closed
-method nvx::exec::Peer::is_detached
-method nvx::exec::Peer::is_finished
-method nvx::exec::Peer::key
-method nvx::exec::Peer::limits
-method nvx::exec::Peer::memory_bound
-method nvx::exec::Peer::new
-method nvx::exec::Peer::next_event
-method nvx::exec::Peer::next_message
-method nvx::exec::Peer::output
-method nvx::exec::Peer::pending_output
-method nvx::exec::Peer::reset
-method nvx::exec::Peer::started
-method nvx::exec::Peer::terminal
-method nvx::exec::Peer::tick
-method nvx::exec::Peer::write_allowance
-enum nvx::exec::Pool
-variant nvx::exec::Pool::Input
-variant nvx::exec::Pool::Output
-variant nvx::exec::Pool::Stderr
-variant nvx::exec::Pool::Stdout
-enum nvx::exec::Receipt
-variant nvx::exec::Receipt::Applied
-variant nvx::exec::Receipt::Late
-variant nvx::exec::Receipt::Stale
-struct nvx::exec::Runner
-method nvx::exec::Runner::awaits_settlement
-method nvx::exec::Runner::buffered
-method nvx::exec::Runner::cancel
-method nvx::exec::Runner::credit
-method nvx::exec::Runner::deadline
-method nvx::exec::Runner::discarded
-method nvx::exec::Runner::disconnect
-method nvx::exec::Runner::dropped
-method nvx::exec::Runner::end_input
-method nvx::exec::Runner::exited
-method nvx::exec::Runner::high_water
-method nvx::exec::Runner::input
-method nvx::exec::Runner::input_closed
-method nvx::exec::Runner::input_written
-method nvx::exec::Runner::is_detached
-method nvx::exec::Runner::is_finished
-method nvx::exec::Runner::key
-method nvx::exec::Runner::launched
-method nvx::exec::Runner::limits
-method nvx::exec::Runner::memory_bound
-method nvx::exec::Runner::new
-method nvx::exec::Runner::next_message
-method nvx::exec::Runner::next_request
-method nvx::exec::Runner::output
-method nvx::exec::Runner::output_closed
-method nvx::exec::Runner::output_failed
-method nvx::exec::Runner::output_hung_up
-method nvx::exec::Runner::pending_input
-method nvx::exec::Runner::read_allowance
-method nvx::exec::Runner::reset
-method nvx::exec::Runner::settled
-method nvx::exec::Runner::tick
-enum nvx::exec::Settlement
-variant nvx::exec::Settlement::Failed
-variant nvx::exec::Settlement::Settled
-enum nvx::exec::Status
-variant nvx::exec::Status::Exited
-field nvx::exec::Status::Exited::0
-variant nvx::exec::Status::Signaled
-field nvx::exec::Status::Signaled::0
-enum nvx::exec::Stream
-const nvx::exec::Stream::ALL
-variant nvx::exec::Stream::Stderr
-variant nvx::exec::Stream::Stdout
-struct nvx::exec::Timers
-method nvx::exec::Timers::grace
-method nvx::exec::Timers::new
-method nvx::exec::Timers::settlement
-method nvx::exec::Timers::timeout
-method nvx::exec::Timers::timeout_duration
-enum nvx::exec::Unit
-variant nvx::exec::Unit::Bytes
-variant nvx::exec::Unit::Frames
-enum nvx::exec::Violation
-variant nvx::exec::Violation::Chunk
-variant nvx::exec::Violation::Credit
-variant nvx::exec::Violation::Ended
-variant nvx::exec::Violation::Key
-variant nvx::exec::Violation::Order
-variant nvx::exec::Violation::Pool
-variant nvx::exec::Violation::Window
-struct nvx::exec::Window
-method nvx::exec::Window::bytes
-method nvx::exec::Window::frames
-method nvx::exec::Window::size
-method nvx::exec::Window::unit
-mod nvx::exec::peer
-enum nvx::exec::peer::Event
-variant nvx::exec::peer::Event::End
-field nvx::exec::peer::Event::End::0
-variant nvx::exec::peer::Event::Finished
-field nvx::exec::peer::Event::Finished::0
-variant nvx::exec::peer::Event::InputClosed
-variant nvx::exec::peer::Event::Started
-enum nvx::exec::peer::Message
-variant nvx::exec::peer::Message::Cancel
-field nvx::exec::peer::Message::Cancel::0
-variant nvx::exec::peer::Message::Credit
-field nvx::exec::peer::Message::Credit::pool
-field nvx::exec::peer::Message::Credit::units
-variant nvx::exec::peer::Message::Input
-field nvx::exec::peer::Message::Input::length
-variant nvx::exec::peer::Message::InputEnd
-struct nvx::exec::peer::Peer
-method nvx::exec::peer::Peer::buffered
-method nvx::exec::peer::Peer::cancel
-method nvx::exec::peer::Peer::consumed
-method nvx::exec::peer::Peer::credit
-method nvx::exec::peer::Peer::deadline
-method nvx::exec::peer::Peer::disconnect
-method nvx::exec::peer::Peer::dropped
-method nvx::exec::peer::Peer::end_input
-method nvx::exec::peer::Peer::end_output
-method nvx::exec::peer::Peer::has_started
-method nvx::exec::peer::Peer::high_water
-method nvx::exec::peer::Peer::input
-method nvx::exec::peer::Peer::input_closed
-method nvx::exec::peer::Peer::is_detached
-method nvx::exec::peer::Peer::is_finished
-method nvx::exec::peer::Peer::key
-method nvx::exec::peer::Peer::limits
-method nvx::exec::peer::Peer::memory_bound
-method nvx::exec::peer::Peer::new
-method nvx::exec::peer::Peer::next_event
-method nvx::exec::peer::Peer::next_message
-method nvx::exec::peer::Peer::output
-method nvx::exec::peer::Peer::pending_output
-method nvx::exec::peer::Peer::reset
-method nvx::exec::peer::Peer::started
-method nvx::exec::peer::Peer::terminal
-method nvx::exec::peer::Peer::tick
-method nvx::exec::peer::Peer::write_allowance
-mod nvx::exec::runner
-enum nvx::exec::runner::Message
-variant nvx::exec::runner::Message::Credit
-field nvx::exec::runner::Message::Credit::units
-variant nvx::exec::runner::Message::End
-field nvx::exec::runner::Message::End::0
-variant nvx::exec::runner::Message::InputClosed
-variant nvx::exec::runner::Message::Output
-field nvx::exec::runner::Message::Output::length
-field nvx::exec::runner::Message::Output::stream
-variant nvx::exec::runner::Message::Started
-variant nvx::exec::runner::Message::Terminal
-field nvx::exec::runner::Message::Terminal::0
-enum nvx::exec::runner::Request
-variant nvx::exec::runner::Request::CloseInput
-variant nvx::exec::runner::Request::Discard
-field nvx::exec::runner::Request::Discard::0
-variant nvx::exec::runner::Request::Settle
-variant nvx::exec::runner::Request::Terminate
-field nvx::exec::runner::Request::Terminate::0
-struct nvx::exec::runner::Runner
-method nvx::exec::runner::Runner::awaits_settlement
-method nvx::exec::runner::Runner::buffered
-method nvx::exec::runner::Runner::cancel
-method nvx::exec::runner::Runner::credit
-method nvx::exec::runner::Runner::deadline
-method nvx::exec::runner::Runner::discarded
-method nvx::exec::runner::Runner::disconnect
-method nvx::exec::runner::Runner::dropped
-method nvx::exec::runner::Runner::end_input
-method nvx::exec::runner::Runner::exited
-method nvx::exec::runner::Runner::high_water
-method nvx::exec::runner::Runner::input
-method nvx::exec::runner::Runner::input_closed
-method nvx::exec::runner::Runner::input_written
-method nvx::exec::runner::Runner::is_detached
-method nvx::exec::runner::Runner::is_finished
-method nvx::exec::runner::Runner::key
-method nvx::exec::runner::Runner::launched
-method nvx::exec::runner::Runner::limits
-method nvx::exec::runner::Runner::memory_bound
-method nvx::exec::runner::Runner::new
-method nvx::exec::runner::Runner::next_message
-method nvx::exec::runner::Runner::next_request
-method nvx::exec::runner::Runner::output
-method nvx::exec::runner::Runner::output_closed
-method nvx::exec::runner::Runner::output_failed
-method nvx::exec::runner::Runner::output_hung_up
-method nvx::exec::runner::Runner::pending_input
-method nvx::exec::runner::Runner::read_allowance
-method nvx::exec::runner::Runner::reset
-method nvx::exec::runner::Runner::settled
-method nvx::exec::runner::Runner::tick
mod nvx::gpt
struct nvx::gpt::Disk
method nvx::gpt::Disk::first_partition
@@ -1228,36 +991,6 @@ field nvx::guest::exec::Source::Output::0
enum nvx::guest::exec::Stream
variant nvx::guest::exec::Stream::Stderr
variant nvx::guest::exec::Stream::Stdout
-mod nvx::guest::exec::binding
-struct nvx::guest::exec::binding::Binding
-method nvx::guest::exec::binding::Binding::cancel
-method nvx::guest::exec::binding::Binding::credit
-method nvx::guest::exec::binding::Binding::deadline
-method nvx::guest::exec::binding::Binding::descriptors
-method nvx::guest::exec::binding::Binding::disconnect
-method nvx::guest::exec::binding::Binding::dropped
-method nvx::guest::exec::binding::Binding::end_input
-method nvx::guest::exec::binding::Binding::execution
-method nvx::guest::exec::binding::Binding::id
-method nvx::guest::exec::binding::Binding::input
-method nvx::guest::exec::binding::Binding::input_writer
-method nvx::guest::exec::binding::Binding::is_finished
-method nvx::guest::exec::binding::Binding::new
-method nvx::guest::exec::binding::Binding::next
-method nvx::guest::exec::binding::Binding::observe
-method nvx::guest::exec::binding::Binding::reset
-method nvx::guest::exec::binding::Binding::runner
-method nvx::guest::exec::binding::Binding::signal
-enum nvx::guest::exec::binding::Event
-variant nvx::guest::exec::binding::Event::InputFailed
-field nvx::guest::exec::binding::Event::InputFailed::0
-variant nvx::guest::exec::binding::Event::KillFailed
-field nvx::guest::exec::binding::Event::KillFailed::0
-field nvx::guest::exec::binding::Event::KillFailed::1
-variant nvx::guest::exec::binding::Event::Message
-field nvx::guest::exec::binding::Event::Message::0
-variant nvx::guest::exec::binding::Event::SettlementFailed
-field nvx::guest::exec::binding::Event::SettlementFailed::0
mod nvx::guest::exit (hidden)
fn nvx::guest::exit::main (hidden)
mod nvx::guest::gpt (hidden)
diff --git a/nvx-rs/public-api/windows.txt b/nvx-rs/public-api/windows.txt
index de5f1ecc5..526f2453f 100644
--- a/nvx-rs/public-api/windows.txt
+++ b/nvx-rs/public-api/windows.txt
@@ -226,243 +226,6 @@ const nvx::contract::SNAPSHOT_FORMAT_VERSION
const nvx::contract::TIME_ABI_VERSION
fn nvx::contract::manifest_is_compatible
fn nvx::contract::manifest_mismatches
-mod nvx::exec
-enum nvx::exec::Cause
-variant nvx::exec::Cause::Cancelled
-variant nvx::exec::Cause::Disconnected
-variant nvx::exec::Cause::Dropped
-variant nvx::exec::Cause::Reset
-variant nvx::exec::Cause::TimedOut
-struct nvx::exec::Key
-method nvx::exec::Key::generation
-method nvx::exec::Key::id
-method nvx::exec::Key::new
-struct nvx::exec::LaunchFailure
-method nvx::exec::LaunchFailure::error
-method nvx::exec::LaunchFailure::new
-method nvx::exec::LaunchFailure::stage
-struct nvx::exec::Limits
-method nvx::exec::Limits::chunk
-method nvx::exec::Limits::input
-method nvx::exec::Limits::is_shared
-method nvx::exec::Limits::new
-method nvx::exec::Limits::output
-method nvx::exec::Limits::shared_output
-struct nvx::exec::Outcome
-method nvx::exec::Outcome::cause
-method nvx::exec::Outcome::is_reusable
-method nvx::exec::Outcome::is_truncated
-method nvx::exec::Outcome::launch_failure
-method nvx::exec::Outcome::new
-method nvx::exec::Outcome::status
-method nvx::exec::Outcome::with_cause
-method nvx::exec::Outcome::with_launch_failure
-method nvx::exec::Outcome::with_reusable
-method nvx::exec::Outcome::with_truncated
-struct nvx::exec::Peer
-method nvx::exec::Peer::buffered
-method nvx::exec::Peer::cancel
-method nvx::exec::Peer::consumed
-method nvx::exec::Peer::credit
-method nvx::exec::Peer::deadline
-method nvx::exec::Peer::disconnect
-method nvx::exec::Peer::dropped
-method nvx::exec::Peer::end_input
-method nvx::exec::Peer::end_output
-method nvx::exec::Peer::has_started
-method nvx::exec::Peer::high_water
-method nvx::exec::Peer::input
-method nvx::exec::Peer::input_closed
-method nvx::exec::Peer::is_detached
-method nvx::exec::Peer::is_finished
-method nvx::exec::Peer::key
-method nvx::exec::Peer::limits
-method nvx::exec::Peer::memory_bound
-method nvx::exec::Peer::new
-method nvx::exec::Peer::next_event
-method nvx::exec::Peer::next_message
-method nvx::exec::Peer::output
-method nvx::exec::Peer::pending_output
-method nvx::exec::Peer::reset
-method nvx::exec::Peer::started
-method nvx::exec::Peer::terminal
-method nvx::exec::Peer::tick
-method nvx::exec::Peer::write_allowance
-enum nvx::exec::Pool
-variant nvx::exec::Pool::Input
-variant nvx::exec::Pool::Output
-variant nvx::exec::Pool::Stderr
-variant nvx::exec::Pool::Stdout
-enum nvx::exec::Receipt
-variant nvx::exec::Receipt::Applied
-variant nvx::exec::Receipt::Late
-variant nvx::exec::Receipt::Stale
-struct nvx::exec::Runner
-method nvx::exec::Runner::awaits_settlement
-method nvx::exec::Runner::buffered
-method nvx::exec::Runner::cancel
-method nvx::exec::Runner::credit
-method nvx::exec::Runner::deadline
-method nvx::exec::Runner::discarded
-method nvx::exec::Runner::disconnect
-method nvx::exec::Runner::dropped
-method nvx::exec::Runner::end_input
-method nvx::exec::Runner::exited
-method nvx::exec::Runner::high_water
-method nvx::exec::Runner::input
-method nvx::exec::Runner::input_closed
-method nvx::exec::Runner::input_written
-method nvx::exec::Runner::is_detached
-method nvx::exec::Runner::is_finished
-method nvx::exec::Runner::key
-method nvx::exec::Runner::launched
-method nvx::exec::Runner::limits
-method nvx::exec::Runner::memory_bound
-method nvx::exec::Runner::new
-method nvx::exec::Runner::next_message
-method nvx::exec::Runner::next_request
-method nvx::exec::Runner::output
-method nvx::exec::Runner::output_closed
-method nvx::exec::Runner::output_failed
-method nvx::exec::Runner::output_hung_up
-method nvx::exec::Runner::pending_input
-method nvx::exec::Runner::read_allowance
-method nvx::exec::Runner::reset
-method nvx::exec::Runner::settled
-method nvx::exec::Runner::tick
-enum nvx::exec::Settlement
-variant nvx::exec::Settlement::Failed
-variant nvx::exec::Settlement::Settled
-enum nvx::exec::Status
-variant nvx::exec::Status::Exited
-field nvx::exec::Status::Exited::0
-variant nvx::exec::Status::Signaled
-field nvx::exec::Status::Signaled::0
-enum nvx::exec::Stream
-const nvx::exec::Stream::ALL
-variant nvx::exec::Stream::Stderr
-variant nvx::exec::Stream::Stdout
-struct nvx::exec::Timers
-method nvx::exec::Timers::grace
-method nvx::exec::Timers::new
-method nvx::exec::Timers::settlement
-method nvx::exec::Timers::timeout
-method nvx::exec::Timers::timeout_duration
-enum nvx::exec::Unit
-variant nvx::exec::Unit::Bytes
-variant nvx::exec::Unit::Frames
-enum nvx::exec::Violation
-variant nvx::exec::Violation::Chunk
-variant nvx::exec::Violation::Credit
-variant nvx::exec::Violation::Ended
-variant nvx::exec::Violation::Key
-variant nvx::exec::Violation::Order
-variant nvx::exec::Violation::Pool
-variant nvx::exec::Violation::Window
-struct nvx::exec::Window
-method nvx::exec::Window::bytes
-method nvx::exec::Window::frames
-method nvx::exec::Window::size
-method nvx::exec::Window::unit
-mod nvx::exec::peer
-enum nvx::exec::peer::Event
-variant nvx::exec::peer::Event::End
-field nvx::exec::peer::Event::End::0
-variant nvx::exec::peer::Event::Finished
-field nvx::exec::peer::Event::Finished::0
-variant nvx::exec::peer::Event::InputClosed
-variant nvx::exec::peer::Event::Started
-enum nvx::exec::peer::Message
-variant nvx::exec::peer::Message::Cancel
-field nvx::exec::peer::Message::Cancel::0
-variant nvx::exec::peer::Message::Credit
-field nvx::exec::peer::Message::Credit::pool
-field nvx::exec::peer::Message::Credit::units
-variant nvx::exec::peer::Message::Input
-field nvx::exec::peer::Message::Input::length
-variant nvx::exec::peer::Message::InputEnd
-struct nvx::exec::peer::Peer
-method nvx::exec::peer::Peer::buffered
-method nvx::exec::peer::Peer::cancel
-method nvx::exec::peer::Peer::consumed
-method nvx::exec::peer::Peer::credit
-method nvx::exec::peer::Peer::deadline
-method nvx::exec::peer::Peer::disconnect
-method nvx::exec::peer::Peer::dropped
-method nvx::exec::peer::Peer::end_input
-method nvx::exec::peer::Peer::end_output
-method nvx::exec::peer::Peer::has_started
-method nvx::exec::peer::Peer::high_water
-method nvx::exec::peer::Peer::input
-method nvx::exec::peer::Peer::input_closed
-method nvx::exec::peer::Peer::is_detached
-method nvx::exec::peer::Peer::is_finished
-method nvx::exec::peer::Peer::key
-method nvx::exec::peer::Peer::limits
-method nvx::exec::peer::Peer::memory_bound
-method nvx::exec::peer::Peer::new
-method nvx::exec::peer::Peer::next_event
-method nvx::exec::peer::Peer::next_message
-method nvx::exec::peer::Peer::output
-method nvx::exec::peer::Peer::pending_output
-method nvx::exec::peer::Peer::reset
-method nvx::exec::peer::Peer::started
-method nvx::exec::peer::Peer::terminal
-method nvx::exec::peer::Peer::tick
-method nvx::exec::peer::Peer::write_allowance
-mod nvx::exec::runner
-enum nvx::exec::runner::Message
-variant nvx::exec::runner::Message::Credit
-field nvx::exec::runner::Message::Credit::units
-variant nvx::exec::runner::Message::End
-field nvx::exec::runner::Message::End::0
-variant nvx::exec::runner::Message::InputClosed
-variant nvx::exec::runner::Message::Output
-field nvx::exec::runner::Message::Output::length
-field nvx::exec::runner::Message::Output::stream
-variant nvx::exec::runner::Message::Started
-variant nvx::exec::runner::Message::Terminal
-field nvx::exec::runner::Message::Terminal::0
-enum nvx::exec::runner::Request
-variant nvx::exec::runner::Request::CloseInput
-variant nvx::exec::runner::Request::Discard
-field nvx::exec::runner::Request::Discard::0
-variant nvx::exec::runner::Request::Settle
-variant nvx::exec::runner::Request::Terminate
-field nvx::exec::runner::Request::Terminate::0
-struct nvx::exec::runner::Runner
-method nvx::exec::runner::Runner::awaits_settlement
-method nvx::exec::runner::Runner::buffered
-method nvx::exec::runner::Runner::cancel
-method nvx::exec::runner::Runner::credit
-method nvx::exec::runner::Runner::deadline
-method nvx::exec::runner::Runner::discarded
-method nvx::exec::runner::Runner::disconnect
-method nvx::exec::runner::Runner::dropped
-method nvx::exec::runner::Runner::end_input
-method nvx::exec::runner::Runner::exited
-method nvx::exec::runner::Runner::high_water
-method nvx::exec::runner::Runner::input
-method nvx::exec::runner::Runner::input_closed
-method nvx::exec::runner::Runner::input_written
-method nvx::exec::runner::Runner::is_detached
-method nvx::exec::runner::Runner::is_finished
-method nvx::exec::runner::Runner::key
-method nvx::exec::runner::Runner::launched
-method nvx::exec::runner::Runner::limits
-method nvx::exec::runner::Runner::memory_bound
-method nvx::exec::runner::Runner::new
-method nvx::exec::runner::Runner::next_message
-method nvx::exec::runner::Runner::next_request
-method nvx::exec::runner::Runner::output
-method nvx::exec::runner::Runner::output_closed
-method nvx::exec::runner::Runner::output_failed
-method nvx::exec::runner::Runner::output_hung_up
-method nvx::exec::runner::Runner::pending_input
-method nvx::exec::runner::Runner::read_allowance
-method nvx::exec::runner::Runner::reset
-method nvx::exec::runner::Runner::settled
-method nvx::exec::runner::Runner::tick
mod nvx::gpt
struct nvx::gpt::Disk
method nvx::gpt::Disk::first_partition
diff --git a/nvx-rs/src/exec/credit.rs b/nvx-rs/src/exec/credit.rs
deleted file mode 100644
index 623418253..000000000
--- a/nvx-rs/src/exec/credit.rs
+++ /dev/null
@@ -1,35 +0,0 @@
-//! The credit arithmetic of an execution's pools, and the bounded queues of its streams.
-//!
-//! A sender holds [`Credit`]: the units of a pool that the receiver granted it and that it has
-//! not used, which it spends on each data message it sends. A receiver holds [`Granted`]: the
-//! units that it granted and that the sender has not used yet, which bound each data message it
-//! accepts, and the units that its consumer freed, which it grants again. Both start from the
-//! pool's window: the receiver's first grant opens the whole window, and a grant can never open
-//! more than the window, so the units in flight never exceed it.
-
-mod inbox;
-mod receiver;
-mod sender;
-
-use std::collections::VecDeque;
-
-pub(super) use self::inbox::Inbox;
-pub(super) use self::receiver::Granted;
-pub(super) use self::sender::Credit;
-
-/// Returns `count` bytes as units of a byte pool, which a chunk's bound keeps within `u32`.
-pub(super) fn units(count: usize) -> u32 {
- u32::try_from(count).unwrap_or(u32::MAX)
-}
-
-/// Moves the first bytes of `queue` into `buffer`, at most `length` of them, and returns how
-/// many it moved.
-pub(super) fn take(queue: &mut VecDeque, buffer: &mut [u8], length: usize) -> usize {
- let length = length.min(buffer.len()).min(queue.len());
- let (first, second) = queue.as_slices();
- let head = length.min(first.len());
- buffer[..head].copy_from_slice(&first[..head]);
- buffer[head..length].copy_from_slice(&second[..length - head]);
- queue.drain(..length);
- length
-}
diff --git a/nvx-rs/src/exec/credit/inbox.rs b/nvx-rs/src/exec/credit/inbox.rs
deleted file mode 100644
index da8a44c07..000000000
--- a/nvx-rs/src/exec/credit/inbox.rs
+++ /dev/null
@@ -1,90 +0,0 @@
-//! The receiver's side of a stream: see the [module](super) documentation.
-
-use std::collections::VecDeque;
-
-use super::units;
-use crate::exec::Unit;
-
-/// The bytes that one stream's receiver holds for its consumer, in order, with the bounds of the
-/// frames that they arrived in when the pool counts frames.
-#[derive(Debug)]
-pub(in crate::exec) struct Inbox {
- /// What the pool counts.
- unit: Unit,
- /// The bytes, in order.
- bytes: VecDeque,
- /// The bytes left of each frame that the inbox holds, oldest first, when the pool counts
- /// frames.
- frames: VecDeque,
-}
-
-impl Inbox {
- /// Returns an empty inbox of a pool that counts `unit`.
- pub(in crate::exec) fn new(unit: Unit) -> Self {
- Self {
- unit,
- bytes: VecDeque::new(),
- frames: VecDeque::new(),
- }
- }
-
- /// Returns the number of bytes held.
- pub(in crate::exec) fn len(&self) -> usize {
- self.bytes.len()
- }
-
- /// Returns whether the inbox holds nothing.
- pub(in crate::exec) fn is_empty(&self) -> bool {
- self.bytes.is_empty()
- }
-
- /// Appends one data message, which its pool charged.
- pub(in crate::exec) fn push(&mut self, bytes: &[u8]) {
- self.bytes.extend(bytes);
- if self.unit == Unit::Frames {
- self.frames.push_back(bytes.len());
- }
- }
-
- /// Returns the bytes at the front, as many as are contiguous: a part of what the inbox holds
- /// while it holds anything.
- pub(in crate::exec) fn front(&self) -> &[u8] {
- self.bytes.as_slices().0
- }
-
- /// Removes the first `count` bytes, at most as many as the inbox holds, and returns the units
- /// that this frees: each byte, or each frame whose last byte it removes.
- pub(in crate::exec) fn pop(&mut self, count: usize) -> u32 {
- let count = count.min(self.bytes.len());
- self.bytes.drain(..count);
- match self.unit {
- Unit::Bytes => units(count),
- Unit::Frames => {
- let mut left = count;
- let mut freed = 0;
- while left > 0 {
- let Some(frame) = self.frames.front_mut() else {
- break;
- };
- let taken = left.min(*frame);
- *frame -= taken;
- left -= taken;
- if *frame == 0 {
- self.frames.pop_front();
- freed += 1;
- }
- }
- freed
- }
- }
- }
-
- /// Removes everything, releases the storage, and returns how many bytes it removed and the
- /// units that this frees.
- pub(in crate::exec) fn clear(&mut self) -> (usize, u32) {
- let length = self.bytes.len();
- let freed = self.pop(length);
- self.bytes = VecDeque::new();
- (length, freed)
- }
-}
diff --git a/nvx-rs/src/exec/credit/receiver.rs b/nvx-rs/src/exec/credit/receiver.rs
deleted file mode 100644
index 154269780..000000000
--- a/nvx-rs/src/exec/credit/receiver.rs
+++ /dev/null
@@ -1,49 +0,0 @@
-//! The receiver's side of a pool: see the [module](super) documentation.
-
-use crate::exec::{Violation, Window};
-
-/// What the receiver of one pool granted, and what its consumer freed.
-#[derive(Debug)]
-pub(in crate::exec) struct Granted {
- /// The pool's window.
- window: Window,
- /// The units that the sender may still send: granted, and not used yet.
- open: u32,
- /// The units that the consumer freed, or that the receiver never granted, which it grants
- /// next.
- returnable: u32,
-}
-
-impl Granted {
- /// Returns the window of a pool before its first grant, which opens the whole window.
- pub(in crate::exec) fn new(window: Window) -> Self {
- Self {
- window,
- open: 0,
- returnable: window.size(),
- }
- }
-
- /// Charges a data message of `length` bytes, which the caller checked against the maximum
- /// chunk. A message beyond the units that remain open is refused and changes nothing.
- pub(in crate::exec) fn charge(&mut self, length: usize) -> Result<(), Violation> {
- let cost = self.window.cost(length);
- if cost > self.open {
- return Err(Violation::Window);
- }
- self.open -= cost;
- Ok(())
- }
-
- /// Frees `units` that the consumer took, to grant them again.
- pub(in crate::exec) fn free(&mut self, units: u32) {
- self.returnable += units;
- }
-
- /// Returns the units to grant now, if any, and opens them.
- pub(in crate::exec) fn take_grant(&mut self) -> Option {
- let units = std::mem::take(&mut self.returnable);
- self.open += units;
- (units > 0).then_some(units)
- }
-}
diff --git a/nvx-rs/src/exec/credit/sender.rs b/nvx-rs/src/exec/credit/sender.rs
deleted file mode 100644
index c30e77864..000000000
--- a/nvx-rs/src/exec/credit/sender.rs
+++ /dev/null
@@ -1,50 +0,0 @@
-//! The sender's side of a pool: see the [module](super) documentation.
-
-use crate::exec::{Unit, Violation, Window};
-
-/// The credit that a receiver granted the sender of one pool.
-#[derive(Debug)]
-pub(in crate::exec) struct Credit {
- /// The pool's window.
- window: Window,
- /// The units that the sender may still send.
- available: u32,
-}
-
-impl Credit {
- /// Returns the credit of a pool with `window`, before the receiver's first grant.
- pub(in crate::exec) fn new(window: Window) -> Self {
- Self {
- window,
- available: 0,
- }
- }
-
- /// Adds a grant of `units`. A grant of no units, or of more than the units in flight, which
- /// would leave the sender more than the window, is refused and changes nothing.
- pub(in crate::exec) fn grant(&mut self, units: u32) -> Result<(), Violation> {
- if units == 0
- || u64::from(self.available) + u64::from(units) > u64::from(self.window.size())
- {
- return Err(Violation::Credit);
- }
- self.available += units;
- Ok(())
- }
-
- /// Returns how many of the first `length` queued bytes one data message may carry now: as
- /// many as the bytes of credit, or all of them while a frame of credit is left.
- pub(in crate::exec) fn sendable(&self, length: usize) -> usize {
- match self.window.unit() {
- Unit::Bytes => length.min(self.available as usize),
- Unit::Frames if self.available > 0 => length,
- Unit::Frames => 0,
- }
- }
-
- /// Spends the credit of one data message of `length` bytes, which
- /// [`sendable`](Self::sendable) allowed.
- pub(in crate::exec) fn spend(&mut self, length: usize) {
- self.available -= self.window.cost(length);
- }
-}
diff --git a/nvx-rs/src/exec/mod.rs b/nvx-rs/src/exec/mod.rs
deleted file mode 100644
index 739bf0ece..000000000
--- a/nvx-rs/src/exec/mod.rs
+++ /dev/null
@@ -1,722 +0,0 @@
-//! The credit and stdio engine of an execution, which every adapter shares, whatever its framing.
-//!
-//! An execution runs a workload on one side, the [`Runner`], for another side, the [`Peer`],
-//! which sends its standard input and takes its standard output and error. The engine owns what
-//! both sides must agree on: the state of each stream, the credit arithmetic that bounds the
-//! bytes in flight, the order of each stream's data and end, and the execution's outcome. The
-//! adapter owns everything else: its framing and transport, which messages it puts in one frame,
-//! its admission of executions, and the processes, which a guest's adapter drives through
-//! `guest::exec::binding` (Linux, feature `guest`).
-//!
-//! The engine needs only the standard library: no feature enables it, and it has no wire format,
-//! async runtime, threads, global state, admission, or reaper. Each side is a state machine that
-//! never blocks and never waits: the adapter hands it each input as it arrives, with the current
-//! [`Instant`] where time matters, retrieves what it must do, one bounded item at a time, and
-//! [ticks](Runner::tick) it once its [deadline](Runner::deadline) passes. So a stream, a sink, a
-//! window, or a transport that blocks holds back only the bytes that wait on it, never a timer, a
-//! cancellation, a reset, or another execution, and tests drive both sides deterministically.
-//!
-//! # Messages
-//!
-//! The two sides exchange these messages, each of which the adapter frames and tags with the
-//! execution's [`Key`]:
-//!
-//! | From | Message | Meaning |
-//! | --- | --- | --- |
-//! | Peer | [credit](peer::Message::Credit) for output | the peer's consumer freed output units |
-//! | Peer | [input](peer::Message::Input) | bytes of the workload's standard input |
-//! | Peer | [end of input](peer::Message::InputEnd) | the input ends after its last byte |
-//! | Peer | [cancel](peer::Message::Cancel) | end the workload |
-//! | Runner | [credit](runner::Message::Credit) for input | the workload took input units |
-//! | Runner | [started](runner::Message::Started) | the workload's program runs |
-//! | Runner | [output](runner::Message::Output) | bytes of standard output or error |
-//! | Runner | [end](runner::Message::End) | the stream ends after its last byte |
-//! | Runner | [input closed](runner::Message::InputClosed) | the workload's input closed early |
-//! | Runner | [terminal](runner::Message::Terminal) | the execution's [`Outcome`] |
-//!
-//! A message for an earlier generation of the execution's key is [stale](Receipt::Stale), and one
-//! that the receiving side no longer needs, such as a credit after the execution ended, or any
-//! message of the runner's once the peer detached from it or gave up waiting for it, is
-//! [late](Receipt::Late): neither changes anything. A message that breaks the rules, such as data
-//! beyond the window, is a [`Violation`], which changes nothing either, and after which the
-//! adapter decides what to do, such as reset its session.
-//!
-//! # Credit
-//!
-//! Each data message carries at most [`Limits::chunk`] bytes, and costs credit in its pool: as
-//! many units as its bytes, or one unit whatever its length, as the pool's [`Window`] counts. The
-//! adapter chooses the pools: the input's, and either one window for each output stream, or one
-//! pool that standard output and error [share](Limits::shared_output). Each receiver grants its
-//! whole window first, then each unit again once its consumer took the unit's bytes: the runner
-//! once the workload's input took them, the peer once its sink took them. A receiver refuses data
-//! beyond what it granted, and a sender refuses a grant that would leave it more than the window,
-//! so at most a window of units is ever in flight, and each side's buffers stay bounded, with no
-//! limit on a stream's total length: see [`Runner::memory_bound`] and [`Peer::memory_bound`].
-//!
-//! A sender queues at most one chunk of each stream ahead of the credit, which the runner reads
-//! from the workload as its [read allowance](Runner::read_allowance) lets it, and the peer takes
-//! from its source as its [write allowance](Peer::write_allowance) does. Data leaves the queue as
-//! the adapter retrieves it, as one message of whatever the queue holds, up to a chunk: an
-//! adapter whose transport is busy retrieves less often, and so sends fewer, fuller messages.
-//!
-//! # Completion
-//!
-//! - The workload's input ends only after every byte that the runner accepted: the runner
-//! requests to [close it](runner::Request::CloseInput) once the peer ended it and the workload
-//! took the rest. A workload that closes its input early, or ends, makes the runner tell the
-//! peer that the input [closed](runner::Message::InputClosed), and drop what the peer sends
-//! after, so that the peer's source never waits. Closing the input never closes the other
-//! direction.
-//! - Each stream keeps its order, and its end follows its last byte. A stream's end goes ahead
-//! of any later output of the other stream, which never holds it back; otherwise, the streams'
-//! messages have no order between them. The runner sends the terminal outcome after both
-//! streams' ends. An end does not report whether the runner lost bytes; only the terminal
-//! outcome does. A peer that detaches or gives up without that outcome conservatively marks
-//! both streams as truncated, even if their ends arrived.
-//! - A cancellation, a timeout, a drop, a reset, or a disconnection proceeds whatever the streams
-//! do. The runner requests to [end](runner::Request::Terminate) the workload, and pending output
-//! still flows for [`Timers`]'s grace, then the rest is discarded, and the terminal outcome
-//! marks each stream that lost bytes as [truncated](Outcome::is_truncated). A reset, a drop, or
-//! a disconnection of the runner leaves no peer to deliver to, so the runner discards at once.
-//! A workload that ends on its own, with no such cause, loses no byte, however slowly the peer
-//! takes them, if its execution settles, or if no process holds its streams any more, as the
-//! adapter tells the runner once they [hung up](Runner::output_hung_up).
-//! - The logical outcome, the delivery of the output, and the settlement of the execution are
-//! separate. Once the workload's status is known, the runner requests to
-//! [settle](runner::Request::Settle) its processes, whose result the adapter reports once its
-//! own check shows that the workload's descendants, and every process that holds one of its
-//! pipes, are gone; it may stop checking once the runner no longer
-//! [awaits](Runner::awaits_settlement) the result. The workload's exit, a signal to its process
-//! group, or a peer's acknowledgement never counts. The terminal outcome follows the status,
-//! the delivery, and the settlement, and says whether the execution is
-//! [reusable](Outcome::is_reusable) or fenced: a failed or late settlement fences it.
-//! - Each side bounds every wait: a termination gives the workload its grace and then
-//! [`Timers`]'s settlement time to end, close its streams, and settle, after which the runner
-//! gives up on an execution whose workload has not, and fences it, while one whose cleanup is
-//! complete waits only for the transport to take its last messages; a workload that ended on
-//! its own, but whose execution did not settle, leaves its streams the settlement time and then
-//! the grace after its end to close or hang up, after which the runner gives up on those still
-//! open that a process may hold, sends what it read of them, and marks them truncated; a launch
-//! result that has not arrived by then fences the execution, whether or not it settled, as a
-//! process may hold what reports it too, and the runner gives up on the result, and on those
-//! streams, and sends the workload's output and outcome without it; a peer that ends the
-//! execution itself waits as long for the terminal outcome, after which it gives up and fences
-//! it too; and a peer that receives a terminal outcome with a cause, without having ended the
-//! execution itself, gives its sink the grace from then on to take the output that it holds,
-//! and discards the rest.
-//!
-//! # Example
-//!
-//! A peer and a runner, connected directly, as an adapter's framing and transport would connect
-//! them:
-//!
-//! ```
-//! use std::time::{Duration, Instant};
-//!
-//! use nvx::exec::{
-//! peer, runner, Key, Limits, Peer, Runner, Settlement, Status, Stream, Timers, Window,
-//! };
-//!
-//! let now = Instant::now();
-//! let key = Key::new(1, 0);
-//! let window = Window::bytes(64 * 1024);
-//! let limits = Limits::new(16 * 1024, window, window).unwrap();
-//! let timers = Timers::new(Duration::from_secs(1), Duration::from_secs(5));
-//! let mut peer = Peer::new(key, limits, timers, now);
-//! let mut runner = Runner::new(key, limits, timers, now);
-//! let mut buffer = vec![0u8; limits.chunk()];
-//!
-//! // The workload runs, writes to its standard output, and exits; its processes settle.
-//! runner.launched(Ok(()));
-//! assert_eq!(runner.output(Stream::Stdout, b"hello"), 5);
-//! runner.output_closed(Stream::Stdout);
-//! runner.output_closed(Stream::Stderr);
-//! runner.exited(now, Some(Status::Exited(0)));
-//! assert_eq!(runner.next_request(), Some(runner::Request::Settle));
-//! runner.settled(Settlement::Settled);
-//!
-//! // The adapter carries each side's messages to the other until the peer finishes.
-//! let mut stdout = Vec::new();
-//! let outcome = loop {
-//! while let Some(message) = peer.next_message(&mut buffer) {
-//! match message {
-//! peer::Message::Credit { pool, units } => {
-//! runner.credit(key, pool, units).unwrap();
-//! }
-//! message => unreachable!("{message:?}"),
-//! }
-//! }
-//! while let Some(message) = runner.next_message(&mut buffer) {
-//! match message {
-//! runner::Message::Credit { units } => {
-//! peer.credit(key, units).unwrap();
-//! }
-//! runner::Message::Started => {
-//! peer.started(key).unwrap();
-//! }
-//! runner::Message::Output { stream, length } => {
-//! peer.output(key, stream, &buffer[..length]).unwrap();
-//! }
-//! runner::Message::End(stream) => {
-//! peer.end_output(key, stream).unwrap();
-//! }
-//! runner::Message::Terminal(outcome) => {
-//! peer.terminal(now, key, outcome).unwrap();
-//! }
-//! runner::Message::InputClosed => {
-//! peer.input_closed(key).unwrap();
-//! }
-//! message => unreachable!("{message:?}"),
-//! }
-//! }
-//! // The peer's sink takes what arrived.
-//! let pending = peer.pending_output(Stream::Stdout).to_vec();
-//! stdout.extend_from_slice(&pending);
-//! peer.consumed(Stream::Stdout, pending.len());
-//! let last = std::iter::from_fn(|| peer.next_event()).last();
-//! if let Some(peer::Event::Finished(outcome)) = last {
-//! break outcome;
-//! }
-//! };
-//! assert_eq!(stdout, b"hello");
-//! assert_eq!(outcome.status(), Some(Status::Exited(0)));
-//! assert!(outcome.is_reusable() && !outcome.is_truncated(Stream::Stdout));
-//! ```
-
-use std::fmt;
-use std::time::{Duration, Instant};
-
-mod credit;
-pub mod peer;
-pub mod runner;
-
-#[cfg(test)]
-mod tests;
-
-pub use peer::Peer;
-pub use runner::Runner;
-
-/// The identity of one execution: an identifier that the adapter assigns, such as a request ID or
-/// a stream slot, and the generation of that identifier, which tells a later execution with the
-/// same identifier, or the same execution in a later session, from an earlier one.
-#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
-pub struct Key {
- /// The execution's identifier.
- id: u64,
- /// The identifier's generation.
- generation: u64,
-}
-
-impl Key {
- /// Returns the key of execution `id` in `generation`.
- pub const fn new(id: u64, generation: u64) -> Self {
- Self { id, generation }
- }
-
- /// Returns the execution's identifier.
- pub const fn id(&self) -> u64 {
- self.id
- }
-
- /// Returns the identifier's generation.
- pub const fn generation(&self) -> u64 {
- self.generation
- }
-
- /// Checks the key of a message for this execution: `Ok(true)` for its own generation,
- /// `Ok(false)` for an earlier one, whose messages are stale, and [`Violation::Key`] for
- /// another identifier or a later generation.
- fn admits(self, message: Key) -> Result {
- if message.id != self.id {
- return Err(Violation::Key);
- }
- match message.generation.cmp(&self.generation) {
- std::cmp::Ordering::Less => Ok(false),
- std::cmp::Ordering::Equal => Ok(true),
- std::cmp::Ordering::Greater => Err(Violation::Key),
- }
- }
-}
-
-/// One of the workload's output streams.
-#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
-pub enum Stream {
- /// The workload's standard output.
- Stdout,
- /// The workload's standard error.
- Stderr,
-}
-
-impl Stream {
- /// Both streams, in the order in which the sides first serve them.
- pub const ALL: [Self; 2] = [Self::Stdout, Self::Stderr];
-
- /// Returns the stream's position in [`ALL`](Self::ALL).
- const fn index(self) -> usize {
- match self {
- Self::Stdout => 0,
- Self::Stderr => 1,
- }
- }
-}
-
-/// A pool of credit, which a [credit message](peer::Message::Credit) names.
-#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
-pub enum Pool {
- /// The workload's standard input.
- Input,
- /// The workload's standard output, which has a pool of its own unless the output streams
- /// [share](Limits::shared_output) one.
- Stdout,
- /// The workload's standard error, which has a pool of its own unless the output streams share
- /// one.
- Stderr,
- /// The one pool that standard output and error [share](Limits::shared_output).
- Output,
-}
-
-/// What a pool's credit counts.
-#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
-pub enum Unit {
- /// Data bytes: a data message costs as many units as its bytes.
- Bytes,
- /// Data messages: a data message costs one unit, whatever its length up to a chunk.
- Frames,
-}
-
-/// The size of a pool's window: the most units that may be in flight at once.
-#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
-pub struct Window {
- /// What the window counts.
- unit: Unit,
- /// The units of the window.
- size: u32,
-}
-
-impl Window {
- /// Returns a window of `size` bytes.
- pub const fn bytes(size: u32) -> Self {
- Self {
- unit: Unit::Bytes,
- size,
- }
- }
-
- /// Returns a window of `size` data messages.
- pub const fn frames(size: u32) -> Self {
- Self {
- unit: Unit::Frames,
- size,
- }
- }
-
- /// Returns what the window counts.
- pub const fn unit(&self) -> Unit {
- self.unit
- }
-
- /// Returns the units of the window.
- pub const fn size(&self) -> u32 {
- self.size
- }
-
- /// Returns the units that a data message of `length` bytes costs.
- fn cost(&self, length: usize) -> u32 {
- match self.unit {
- Unit::Bytes => credit::units(length),
- Unit::Frames => 1,
- }
- }
-
- /// Returns the most data bytes that the window holds with chunks of `chunk` bytes.
- fn bytes_with(&self, chunk: usize) -> usize {
- match self.unit {
- Unit::Bytes => self.size as usize,
- Unit::Frames => (self.size as usize).saturating_mul(chunk),
- }
- }
-}
-
-/// The limits on which both sides of an execution agree: the maximum chunk, the input's window,
-/// and the output's windows, one for each stream unless the streams share one pool.
-#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
-pub struct Limits {
- /// The most data bytes that one data message carries.
- chunk: u32,
- /// The window of the workload's standard input.
- input: Window,
- /// The window of each output pool.
- output: Window,
- /// Whether standard output and error share one pool.
- shared: bool,
-}
-
-impl Limits {
- /// Returns limits with data messages of at most `chunk` bytes, the input's `input` window,
- /// and one `output` window for each output stream, or `None` if the chunk or a window is
- /// empty.
- pub const fn new(chunk: u32, input: Window, output: Window) -> Option {
- if chunk == 0 || input.size == 0 || output.size == 0 {
- return None;
- }
- Some(Self {
- chunk,
- input,
- output,
- shared: false,
- })
- }
-
- /// Makes standard output and error share one pool with the output window, which both
- /// streams' data messages charge.
- pub const fn shared_output(mut self) -> Self {
- self.shared = true;
- self
- }
-
- /// Returns the most data bytes that one data message carries.
- pub const fn chunk(&self) -> usize {
- self.chunk as usize
- }
-
- /// Returns the window of the workload's standard input.
- pub const fn input(&self) -> Window {
- self.input
- }
-
- /// Returns the window of each output pool.
- pub const fn output(&self) -> Window {
- self.output
- }
-
- /// Returns whether standard output and error share one pool.
- pub const fn is_shared(&self) -> bool {
- self.shared
- }
-
- /// Returns the output pools, in the order of their positions.
- fn output_pools(&self) -> &'static [Pool] {
- if self.shared {
- &[Pool::Output]
- } else {
- &[Pool::Stdout, Pool::Stderr]
- }
- }
-
- /// Returns the position of `stream`'s output pool.
- fn pool_of(&self, stream: Stream) -> usize {
- if self.shared { 0 } else { stream.index() }
- }
-
- /// Returns the position of the output pool `pool`, or [`Violation::Pool`] if the limits have
- /// no such output pool.
- fn output_pool(&self, pool: Pool) -> Result {
- match (pool, self.shared) {
- (Pool::Output, true) => Ok(0),
- (Pool::Stdout, false) => Ok(0),
- (Pool::Stderr, false) => Ok(1),
- _ => Err(Violation::Pool),
- }
- }
-
- /// Returns the checked length of a data message of `length` bytes: [`Violation::Chunk`] if
- /// it is empty or longer than a chunk.
- fn check_chunk(&self, length: usize) -> Result<(), Violation> {
- if length == 0 || length > self.chunk() {
- return Err(Violation::Chunk);
- }
- Ok(())
- }
-
- /// Returns the most data bytes that the input's window holds.
- fn input_bytes(&self) -> usize {
- self.input.bytes_with(self.chunk())
- }
-
- /// Returns the most data bytes that the output's windows hold together.
- fn output_bytes(&self) -> usize {
- let pools = if self.shared { 1 } else { 2 };
- self.output.bytes_with(self.chunk()).saturating_mul(pools)
- }
-}
-
-/// The local timers of one side of an execution.
-#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
-pub struct Timers {
- /// How long the execution may run before the side ends it, if it ever does.
- timeout: Option,
- /// How long pending output still flows once a termination begins, and how long the output
- /// streams of an execution that did not settle may stay open after the settlement time.
- grace: Duration,
- /// How long the execution may then take to settle.
- settlement: Duration,
-}
-
-impl Timers {
- /// Returns timers with no timeout, a `grace` for pending output once a termination begins,
- /// and a `settlement` time after it, or after the workload's end, within which the runner
- /// expects the execution to settle, and the peer expects the terminal outcome of an execution
- /// that it ended. The output streams of a workload that ended on its own, but whose
- /// execution did not settle, may stay open for the grace after the settlement time.
- pub const fn new(grace: Duration, settlement: Duration) -> Self {
- Self {
- timeout: None,
- grace,
- settlement,
- }
- }
-
- /// Ends the execution for a [timeout](Cause::TimedOut) once `timeout` has passed since the
- /// side was created: the runner, unless the workload's end reaches it before a
- /// [tick](Runner::tick) fires the timeout, and the peer, if the execution has not finished.
- pub const fn timeout(mut self, timeout: Duration) -> Self {
- self.timeout = Some(timeout);
- self
- }
-
- /// Returns the timeout, if any.
- pub const fn timeout_duration(&self) -> Option {
- self.timeout
- }
-
- /// Returns the grace of pending output.
- pub const fn grace(&self) -> Duration {
- self.grace
- }
-
- /// Returns the settlement time.
- pub const fn settlement(&self) -> Duration {
- self.settlement
- }
-}
-
-/// Why a side ended an execution.
-#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
-#[non_exhaustive]
-pub enum Cause {
- /// The peer, or its caller, cancelled the execution.
- Cancelled,
- /// The execution's owner dropped it: the peer's caller dropped its handle, or the runner's
- /// adapter abandoned the execution.
- Dropped,
- /// The execution's timeout passed.
- TimedOut,
- /// The session between the sides was reset.
- Reset,
- /// The transport between the sides was lost.
- Disconnected,
-}
-
-impl fmt::Display for Cause {
- fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
- formatter.write_str(match self {
- Self::Cancelled => "cancelled",
- Self::Dropped => "dropped",
- Self::TimedOut => "timed out",
- Self::Reset => "reset",
- Self::Disconnected => "disconnected",
- })
- }
-}
-
-/// The status with which a workload's program ended, as its collector reaped it.
-#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
-pub enum Status {
- /// The program exited with this code.
- Exited(i32),
- /// A signal of this number ended the program.
- Signaled(i32),
-}
-
-/// Why a workload's program did not start: the adapter's number of the launch stage that failed,
-/// and the operating system's error number of the failure.
-///
-/// NVX's guest binding, `guest::exec::binding` (Linux, feature `guest`), numbers the stages as
-/// `guest::spawn::LaunchStage::code` does, which `LaunchStage::from_code` reverses, and reports 0
-/// for a launch whose report it could not read.
-#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
-pub struct LaunchFailure {
- /// The adapter's number of the stage that failed.
- stage: u32,
- /// The error number of the failure.
- error: i32,
-}
-
-impl LaunchFailure {
- /// Returns the failure of the stage that the adapter numbers `stage`, with the error number
- /// `error`.
- pub const fn new(stage: u32, error: i32) -> Self {
- Self { stage, error }
- }
-
- /// Returns the adapter's number of the stage that failed.
- pub const fn stage(&self) -> u32 {
- self.stage
- }
-
- /// Returns the error number of the failure.
- pub const fn error(&self) -> i32 {
- self.error
- }
-}
-
-/// The result of settling an execution's processes, which the runner's adapter reports.
-#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
-pub enum Settlement {
- /// No process of the execution is left: the workload, its descendants, and every process
- /// that held one of its pipes are gone.
- Settled,
- /// The adapter could not settle the execution, such as because a kill failed.
- Failed,
-}
-
-/// How an execution ended: the runner's terminal outcome, or the peer's own if none arrived.
-#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
-pub struct Outcome {
- /// The workload's status, if a collector reaped it.
- status: Option,
- /// Why a side ended the workload, if one did.
- cause: Option,
- /// Why the workload's program did not start, if it did not.
- launch_failure: Option,
- /// Whether each stream lost bytes, or the peer could not confirm that it ended whole.
- truncated: [bool; 2],
- /// Whether the execution settled, so that its resources may be reused.
- reusable: bool,
-}
-
-impl Outcome {
- /// Returns the outcome of a workload that ended with `status`, if known, with no cause, no
- /// launch failure, no truncation, and fenced: the form in which a peer's adapter decodes a
- /// terminal record, before it applies the record's other fields.
- pub const fn new(status: Option) -> Self {
- Self {
- status,
- cause: None,
- launch_failure: None,
- truncated: [false, false],
- reusable: false,
- }
- }
-
- /// Returns the outcome with `cause` as why a side ended the workload.
- pub const fn with_cause(mut self, cause: Cause) -> Self {
- self.cause = Some(cause);
- self
- }
-
- /// Returns the outcome with `failure` as why the workload's program did not start.
- pub const fn with_launch_failure(mut self, failure: LaunchFailure) -> Self {
- self.launch_failure = Some(failure);
- self
- }
-
- /// Returns the outcome with `stream` marked as truncated.
- pub const fn with_truncated(mut self, stream: Stream) -> Self {
- self.truncated[stream.index()] = true;
- self
- }
-
- /// Returns the outcome of an execution that settled, and so is reusable.
- pub const fn with_reusable(mut self) -> Self {
- self.reusable = true;
- self
- }
-
- /// Returns the workload's status, or `None` if no collector reaped it, as when its collector
- /// lost it or a side gave up on the execution.
- pub const fn status(&self) -> Option {
- self.status
- }
-
- /// Returns why a side ended the workload, or `None` if it ended on its own. A workload with a
- /// cause may still have exited on its own before the side's end reached it, which its status
- /// tells.
- pub const fn cause(&self) -> Option {
- self.cause
- }
-
- /// Returns why the workload's program did not start, or `None` if it started, or if no
- /// launch result arrived.
- pub const fn launch_failure(&self) -> Option {
- self.launch_failure
- }
-
- /// Returns whether `stream` lost bytes, or the peer could not confirm that it ended whole.
- /// A peer that finishes without the runner's terminal outcome conservatively marks both
- /// streams as truncated, even if their ends arrived.
- pub const fn is_truncated(&self, stream: Stream) -> bool {
- self.truncated[stream.index()]
- }
-
- /// Returns whether the execution settled, so that its resources may be reused, or `false` if
- /// it is fenced.
- pub const fn is_reusable(&self) -> bool {
- self.reusable
- }
-}
-
-/// What the side did with a message that it accepted.
-#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
-#[non_exhaustive]
-pub enum Receipt {
- /// The message changed the side's state.
- Applied,
- /// The message belongs to an earlier generation of the execution's key, and changed nothing.
- Stale,
- /// The side no longer needed the message, as for a credit after the execution ended, input
- /// after the workload's input closed, or any message of the runner's once the peer detached
- /// from it or gave up waiting for it: it changed nothing beyond the window it charged.
- Late,
-}
-
-/// A message that breaks the rules, which the side refuses without changing anything.
-#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
-#[non_exhaustive]
-pub enum Violation {
- /// The message's key names another execution, or a later generation of this one.
- Key,
- /// A credit names a pool that the execution's limits do not have.
- Pool,
- /// A credit grants no units, or more units than are in flight, which would leave its sender
- /// more than the window.
- Credit,
- /// A data message is empty, or longer than a chunk.
- Chunk,
- /// A data message exceeds the credit that its receiver granted.
- Window,
- /// Data or an end follows the stream's end, or a second terminal outcome follows the first.
- Ended,
- /// A message comes out of order, such as a second start, or a terminal outcome before both
- /// streams' ends.
- Order,
-}
-
-impl fmt::Display for Violation {
- fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
- formatter.write_str(match self {
- Self::Key => "the message names another execution or a later generation",
- Self::Pool => "the credit names a pool that the execution does not have",
- Self::Credit => "the credit grants no units or more units than are in flight",
- Self::Chunk => "the data message is empty or longer than a chunk",
- Self::Window => "the data message exceeds the granted credit",
- Self::Ended => "the message follows the end of its stream or of the execution",
- Self::Order => "the message comes out of order",
- })
- }
-}
-
-impl std::error::Error for Violation {}
-
-/// Returns the instant `duration` after `now`, or `None` if no instant that late exists.
-fn after(now: Instant, duration: Duration) -> Option {
- now.checked_add(duration)
-}
-
-/// Returns the earlier of two optional instants.
-fn earliest(first: Option, second: Option) -> Option {
- match (first, second) {
- (Some(first), Some(second)) => Some(first.min(second)),
- (first, second) => first.or(second),
- }
-}
diff --git a/nvx-rs/src/exec/peer.rs b/nvx-rs/src/exec/peer.rs
deleted file mode 100644
index b3391e93d..000000000
--- a/nvx-rs/src/exec/peer.rs
+++ /dev/null
@@ -1,679 +0,0 @@
-//! The side of an execution that requested it: see the [module](super) documentation.
-
-use std::collections::VecDeque;
-use std::time::Instant;
-
-use super::credit::{self, Credit, Granted, Inbox};
-use super::{
- Cause, Key, Limits, Outcome, Pool, Receipt, Stream, Timers, Violation, after, earliest,
-};
-
-/// A message from the peer to the runner, which the adapter frames and sends in order, as
-/// [`Peer::next_message`] returns them.
-#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
-#[non_exhaustive]
-pub enum Message {
- /// End the workload for `cause`: [`Cause::Cancelled`], [`Cause::Dropped`], or
- /// [`Cause::TimedOut`], as the peer's caller or timer ended the execution.
- Cancel(Cause),
- /// Credit of the output pool `pool`: `units` more of its window, which the runner may spend
- /// on output. The first one of each pool opens its whole window.
- Credit {
- /// The output pool: [`Pool::Output`] if the streams share one, otherwise
- /// [`Pool::Stdout`] or [`Pool::Stderr`].
- pool: Pool,
- /// The units of the credit.
- units: u32,
- },
- /// Bytes of the workload's input, in order: the first `length` bytes of the buffer that the
- /// adapter lent [`Peer::next_message`].
- Input {
- /// The number of bytes, at least one and at most a chunk.
- length: usize,
- },
- /// The end of the workload's input, after its last byte.
- InputEnd,
-}
-
-/// What the peer tells its caller, as [`Peer::next_event`] returns them.
-#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
-#[non_exhaustive]
-pub enum Event {
- /// The workload's program runs: the runner confirmed its launch.
- Started,
- /// `stream` ended: its sink took its last byte, or the peer discarded the rest.
- End(Stream),
- /// The workload's input closed before its end was sent, or the execution ended: the caller
- /// sends no more input, and the peer dropped what it held.
- InputClosed,
- /// The execution finished, with this outcome: the peer's last event. Its cause is the
- /// runner's, or, if the runner reported none or never reported, the peer's own.
- Finished(Outcome),
-}
-
-/// One output stream, as the peer receives and delivers it.
-#[derive(Debug)]
-struct Incoming {
- /// The bytes that the runner sent and the sink has not taken.
- inbox: Inbox,
- /// Whether the runner ended the stream.
- ended: bool,
- /// Whether the stream's end event was returned.
- end_event: bool,
- /// Whether the peer drops the stream's bytes, as no sink takes them any more.
- discarding: bool,
- /// Whether the peer dropped bytes of the stream.
- truncated: bool,
-}
-
-impl Incoming {
- /// Drops what the inbox holds, notes whether that lost bytes, and returns the units that
- /// this frees.
- fn drop_inbox(&mut self) -> u32 {
- let (length, freed) = self.inbox.clear();
- self.truncated |= length > 0;
- freed
- }
-}
-
-/// The workload's standard input, as the peer takes it from its caller and sends it.
-#[derive(Debug)]
-struct Outgoing {
- /// The bytes that the caller handed over and the peer has not sent: at most one chunk.
- queue: VecDeque,
- /// The credit that the runner granted.
- credit: Credit,
- /// Whether the caller ended the input.
- ended: bool,
- /// Whether the end of the input was sent.
- end_sent: bool,
- /// Whether the input closed: the runner closed it, or the execution ended.
- closed: bool,
- /// Whether the event that the input closed waits to be returned.
- closed_event: bool,
-}
-
-/// The side of an execution that requested it: see the [module](super) documentation.
-///
-/// The adapter creates a peer as it asks the runner's side to run a workload, and hands it the
-/// runner's messages and its caller's input. In return, it takes the peer's
-/// [messages](Self::next_message), which it sends the runner, and its
-/// [events](Self::next_event), which it tells its caller, until the execution
-/// [finishes](Event::Finished). Its sink takes each stream's [pending output](Self::pending_output)
-/// as it can, and acknowledges what it took with [`consumed`](Self::consumed); its source hands
-/// over input as the [write allowance](Self::write_allowance) lets it.
-#[derive(Debug)]
-pub struct Peer {
- /// The execution's key.
- key: Key,
- /// The limits that both sides agreed on.
- limits: Limits,
- /// The peer's timers.
- timers: Timers,
- /// Whether the runner confirmed the workload's launch.
- started: bool,
- /// Whether the event that the workload started waits to be returned.
- started_event: bool,
- /// The workload's standard input.
- input: Outgoing,
- /// The output streams.
- outputs: [Incoming; 2],
- /// The windows of the output pools: one per stream, or the first one shared.
- windows: [Granted; 2],
- /// The source whose message the peer returns first next: each output pool's credit, in the
- /// order of the pools, then the input. They take turns, so that none starves another.
- turn: usize,
- /// The runner's terminal outcome, once it arrived.
- terminal: Option,
- /// Why the peer ended the execution, if it did.
- cause: Option,
- /// Whether the cancellation waits to be sent.
- cancel_message: bool,
- /// When the execution's timeout passes, while it runs.
- timeout_at: Option,
- /// When the peer ended the execution, once it did.
- terminated_at: Option,
- /// When the grace of pending output ends, while the peer waits for it.
- grace_until: Option,
- /// When the peer gives up waiting for the terminal outcome of an execution that it ended.
- give_up_at: Option,
- /// Whether the peer gave up waiting for the terminal outcome.
- gave_up: bool,
- /// Whether the peer is detached from the runner, from which nothing more arrives.
- detached: bool,
- /// Whether the outcome was returned.
- finished: bool,
- /// The most bytes that the peer held at once.
- high_water: usize,
-}
-
-impl Peer {
- /// Returns the peer of the execution `key`, with the `limits` that both sides agreed on and
- /// its own `timers`, created at `now`, from which its timeout runs.
- pub fn new(key: Key, limits: Limits, timers: Timers, now: Instant) -> Self {
- let incoming = || Incoming {
- inbox: Inbox::new(limits.output.unit()),
- ended: false,
- end_event: false,
- discarding: false,
- truncated: false,
- };
- Self {
- key,
- limits,
- timers,
- started: false,
- started_event: false,
- input: Outgoing {
- queue: VecDeque::new(),
- credit: Credit::new(limits.input),
- ended: false,
- end_sent: false,
- closed: false,
- closed_event: false,
- },
- outputs: [incoming(), incoming()],
- windows: [Granted::new(limits.output), Granted::new(limits.output)],
- turn: 0,
- terminal: None,
- cause: None,
- cancel_message: false,
- timeout_at: timers.timeout.and_then(|timeout| after(now, timeout)),
- terminated_at: None,
- grace_until: None,
- give_up_at: None,
- gave_up: false,
- detached: false,
- finished: false,
- high_water: 0,
- }
- }
-
- /// Returns the execution's key.
- pub fn key(&self) -> Key {
- self.key
- }
-
- /// Returns the limits that both sides agreed on.
- pub fn limits(&self) -> Limits {
- self.limits
- }
-
- /// Hands the peer the runner's credit of `units` for the workload's input. A grant of no
- /// units, or of more than the units in flight, is a [`Violation::Credit`].
- pub fn credit(&mut self, key: Key, units: u32) -> Result {
- if !self.key.admits(key)? {
- return Ok(Receipt::Stale);
- }
- if self.ignores_runner() {
- return Ok(Receipt::Late);
- }
- self.input.credit.grant(units)?;
- if self.input.closed || self.input.end_sent {
- return Ok(Receipt::Late);
- }
- Ok(Receipt::Applied)
- }
-
- /// Hands the peer the runner's confirmation that the workload started. A second one, or one
- /// after the terminal outcome, is a [`Violation::Order`].
- pub fn started(&mut self, key: Key) -> Result {
- if !self.key.admits(key)? {
- return Ok(Receipt::Stale);
- }
- if self.ignores_runner() {
- return Ok(Receipt::Late);
- }
- if self.started || self.terminal.is_some() {
- return Err(Violation::Order);
- }
- self.started = true;
- self.started_event = true;
- Ok(Receipt::Applied)
- }
-
- /// Hands the peer bytes of `stream` from the runner, which the stream's pool must allow, and
- /// which the peer holds until its sink takes them. Output after the stream's end is a
- /// [`Violation::Ended`]. Once the peer discards the stream, it drops what arrives, as
- /// [`Receipt::Late`], and grants its units again at once.
- pub fn output(&mut self, key: Key, stream: Stream, bytes: &[u8]) -> Result {
- if !self.key.admits(key)? {
- return Ok(Receipt::Stale);
- }
- if self.ignores_runner() {
- return Ok(Receipt::Late);
- }
- if self.outputs[stream.index()].ended {
- return Err(Violation::Ended);
- }
- self.limits.check_chunk(bytes.len())?;
- let window = &mut self.windows[self.limits.pool_of(stream)];
- window.charge(bytes.len())?;
- let incoming = &mut self.outputs[stream.index()];
- if incoming.discarding {
- incoming.truncated = true;
- window.free(self.limits.output.cost(bytes.len()));
- return Ok(Receipt::Late);
- }
- incoming.inbox.push(bytes);
- self.note_buffered();
- Ok(Receipt::Applied)
- }
-
- /// Hands the peer the end of `stream` from the runner, after its last byte. A second end is
- /// a [`Violation::Ended`]. The end does not report whether the runner truncated the stream;
- /// only its terminal outcome does.
- pub fn end_output(&mut self, key: Key, stream: Stream) -> Result {
- if !self.key.admits(key)? {
- return Ok(Receipt::Stale);
- }
- if self.ignores_runner() {
- return Ok(Receipt::Late);
- }
- let incoming = &mut self.outputs[stream.index()];
- if incoming.ended {
- return Err(Violation::Ended);
- }
- incoming.ended = true;
- Ok(Receipt::Applied)
- }
-
- /// Hands the peer the runner's notice that the workload's input closed early. The peer drops
- /// the input that it holds, takes no more, and tells its caller.
- pub fn input_closed(&mut self, key: Key) -> Result {
- if !self.key.admits(key)? {
- return Ok(Receipt::Stale);
- }
- if self.ignores_runner() || self.input.closed || self.input.end_sent {
- return Ok(Receipt::Late);
- }
- self.close_input();
- Ok(Receipt::Applied)
- }
-
- /// Hands the peer the runner's terminal outcome at `now`, which follows both streams' ends:
- /// a terminal outcome before them is a [`Violation::Order`], and a second one a
- /// [`Violation::Ended`]. The peer finishes once its sink took the rest of the output, losing
- /// nothing, unless a side ended the execution: then the sink takes the rest only for the
- /// peer's grace, from when the peer ended the execution, or from `now` if the runner did.
- /// Once the peer detached from the runner or gave up waiting for it, the terminal outcome is
- /// [late](Receipt::Late), as every message of the runner's is then, and the peer's own
- /// outcome stands, whenever its caller retrieves it.
- pub fn terminal(
- &mut self,
- now: Instant,
- key: Key,
- outcome: Outcome,
- ) -> Result {
- if !self.key.admits(key)? {
- return Ok(Receipt::Stale);
- }
- self.tick(now);
- if self.ignores_runner() {
- return Ok(Receipt::Late);
- }
- if self.terminal.is_some() {
- return Err(Violation::Ended);
- }
- if !self.outputs.iter().all(|incoming| incoming.ended) {
- return Err(Violation::Order);
- }
- self.terminal = Some(outcome);
- self.give_up_at = None;
- self.timeout_at = None;
- self.cancel_message = false;
- self.close_input();
- if outcome.cause().is_some() && self.terminated_at.is_none() {
- self.terminated_at = Some(now);
- self.grace_until = after(now, self.timers.grace);
- if self.grace_until.is_some_and(|at| now >= at) {
- self.end_grace();
- }
- }
- Ok(Receipt::Applied)
- }
-
- /// Returns how many more bytes of input the caller may hand over with
- /// [`input`](Self::input): what the peer's queue has room for, at most one chunk ahead of the
- /// runner's credit, and none once the input ended or closed, or the peer ended the
- /// execution.
- pub fn write_allowance(&self) -> usize {
- if self.input.closed || self.input.ended || self.terminated_at.is_some() || self.finished {
- return 0;
- }
- self.limits.chunk() - self.input.queue.len()
- }
-
- /// Hands the peer bytes of the workload's input from its caller, in order, and returns how
- /// many it took: as many as the [write allowance](Self::write_allowance) lets it.
- pub fn input(&mut self, bytes: &[u8]) -> usize {
- let length = bytes.len().min(self.write_allowance());
- self.input.queue.extend(&bytes[..length]);
- self.note_buffered();
- length
- }
-
- /// Ends the workload's input after the bytes that the caller handed over.
- pub fn end_input(&mut self) {
- if !self.input.closed {
- self.input.ended = true;
- }
- }
-
- /// Returns the bytes of `stream` that the sink takes next, as many as are contiguous: a part
- /// of what the peer holds, while it holds any.
- pub fn pending_output(&self, stream: Stream) -> &[u8] {
- self.outputs[stream.index()].inbox.front()
- }
-
- /// Hands the peer the count of bytes of `stream`'s [pending output](Self::pending_output)
- /// that the sink took, which the peer grants the runner again.
- pub fn consumed(&mut self, stream: Stream, bytes: usize) {
- let freed = self.outputs[stream.index()].inbox.pop(bytes);
- self.windows[self.limits.pool_of(stream)].free(freed);
- }
-
- /// Ends the execution at `now` for [`Cause::Cancelled`]: the peer sends the runner its
- /// cancellation, takes no more input, delivers the pending output for its grace, and
- /// waits for the terminal outcome for the settlement time after that.
- pub fn cancel(&mut self, now: Instant) {
- self.tick(now);
- self.terminate(now, Cause::Cancelled);
- }
-
- /// Ends the execution at `now` for [`Cause::Dropped`], because the caller dropped its
- /// handle: as a [cancellation](Self::cancel), but the peer discards the output at once, as
- /// no sink takes it any more.
- pub fn dropped(&mut self, now: Instant) {
- self.tick(now);
- self.terminate(now, Cause::Dropped);
- if !self.finished {
- self.end_grace();
- }
- }
-
- /// Detaches the peer from the runner at `now` for [`Cause::Reset`], because the session was
- /// reset. The peer takes none of the runner's messages from then on.
- pub fn reset(&mut self, now: Instant) {
- self.detach(now, Cause::Reset);
- }
-
- /// Detaches the peer from the runner at `now` for [`Cause::Disconnected`], because the
- /// transport was lost. The peer takes none of the runner's messages from then on.
- pub fn disconnect(&mut self, now: Instant) {
- self.detach(now, Cause::Disconnected);
- }
-
- /// Fires the timers whose instants `now` has reached: the execution's timeout, the end of
- /// the grace of pending output, and the time at which the peer gives up waiting for the
- /// terminal outcome of an execution that it ended.
- pub fn tick(&mut self, now: Instant) {
- if self.finished {
- return;
- }
- if self.timeout_at.is_some_and(|at| now >= at) {
- self.timeout_at = None;
- self.terminate(now, Cause::TimedOut);
- }
- if self.grace_until.is_some_and(|at| now >= at) {
- self.end_grace();
- }
- if self.give_up_at.is_some_and(|at| now >= at) {
- self.give_up_at = None;
- self.gave_up = true;
- self.end_grace();
- }
- }
-
- /// Returns when the peer's next timer fires, for the adapter to [tick](Self::tick) it then,
- /// or `None` while no timer runs.
- pub fn deadline(&self) -> Option {
- if self.finished {
- return None;
- }
- [self.timeout_at, self.grace_until, self.give_up_at]
- .into_iter()
- .fold(None, earliest)
- }
-
- /// Returns the next message to send the runner, or `None` if none waits: the cancellation
- /// first, then credit of each output pool, the input, and its end once its bytes are sent.
- /// Each pool's credit and the input take turns: after one of them, the others that wait go
- /// first, so that none starves another, however much output the sink takes between two
- /// calls. An input message carries the most bytes that the queue, the credit, a chunk, and
- /// `buffer` allow, which it holds at the start of `buffer`; an empty buffer carries none.
- pub fn next_message(&mut self, buffer: &mut [u8]) -> Option {
- if self.finished || self.detached {
- return None;
- }
- if std::mem::take(&mut self.cancel_message) {
- return Some(Message::Cancel(self.cause.unwrap_or(Cause::Cancelled)));
- }
- let pools = self.limits.output_pools().len();
- let sources = pools + 1;
- for offset in 0..sources {
- let source = (self.turn + offset) % sources;
- let message = if source < pools {
- self.next_credit(source)
- } else {
- self.next_input(buffer)
- };
- if message.is_some() {
- self.turn = (source + 1) % sources;
- return message;
- }
- }
- None
- }
-
- /// Returns credit of the output pool at `index` to send the runner, if a stream of the pool
- /// is still open, no terminal outcome arrived, and the sink took bytes since the last grant.
- fn next_credit(&mut self, index: usize) -> Option {
- if self.terminal.is_some() {
- return None;
- }
- let open = Stream::ALL.into_iter().any(|stream| {
- self.limits.pool_of(stream) == index && !self.outputs[stream.index()].ended
- });
- if !open {
- return None;
- }
- let units = self.windows[index].take_grant()?;
- let pool = self.limits.output_pools()[index];
- Some(Message::Credit { pool, units })
- }
-
- /// Returns the next input message to send the runner: the input, then its end once its bytes
- /// are sent, unless the input closed.
- fn next_input(&mut self, buffer: &mut [u8]) -> Option {
- if self.input.closed {
- return None;
- }
- let wanted = self
- .input
- .queue
- .len()
- .min(self.limits.chunk())
- .min(buffer.len());
- let length = self.input.credit.sendable(wanted);
- if length > 0 {
- credit::take(&mut self.input.queue, buffer, length);
- self.input.credit.spend(length);
- return Some(Message::Input { length });
- }
- if self.input.ended && self.input.queue.is_empty() && !self.input.end_sent {
- self.input.end_sent = true;
- return Some(Message::InputEnd);
- }
- None
- }
-
- /// Returns the next event for the caller, or `None` if none waits: that the workload
- /// started, that its input closed, the end of each stream once its sink took its last byte,
- /// and the outcome last, once the runner's terminal outcome arrived and the sink took the
- /// rest of the output, or once the peer detached or gave up and delivered or discarded
- /// what it held.
- pub fn next_event(&mut self) -> Option {
- if self.finished {
- return None;
- }
- if std::mem::take(&mut self.started_event) {
- return Some(Event::Started);
- }
- if std::mem::take(&mut self.input.closed_event) {
- return Some(Event::InputClosed);
- }
- for stream in Stream::ALL {
- let incoming = &mut self.outputs[stream.index()];
- if incoming.ended && incoming.inbox.is_empty() && !incoming.end_event {
- incoming.end_event = true;
- return Some(Event::End(stream));
- }
- }
- let outcome = self.outcome()?;
- self.finished = true;
- Some(Event::Finished(outcome))
- }
-
- /// Returns whether the peer has returned the execution's outcome.
- pub fn is_finished(&self) -> bool {
- self.finished
- }
-
- /// Returns whether the runner confirmed the workload's launch.
- pub fn has_started(&self) -> bool {
- self.started
- }
-
- /// Returns whether the peer is detached from the runner.
- pub fn is_detached(&self) -> bool {
- self.detached
- }
-
- /// Returns the most bytes that the peer holds at once: one chunk of input, and the output's
- /// windows.
- pub fn memory_bound(&self) -> usize {
- self.limits
- .chunk()
- .saturating_add(self.limits.output_bytes())
- }
-
- /// Returns the bytes that the peer holds now.
- pub fn buffered(&self) -> usize {
- self.input.queue.len()
- + self
- .outputs
- .iter()
- .map(|incoming| incoming.inbox.len())
- .sum::()
- }
-
- /// Returns the most bytes that the peer has held at once.
- pub fn high_water(&self) -> usize {
- self.high_water
- }
-
- /// Returns whether the peer takes no more of the runner's messages: it finished, detached
- /// from the runner, or gave up waiting for it. So its outcome is fixed from then on, whenever
- /// its caller retrieves it.
- fn ignores_runner(&self) -> bool {
- self.finished || self.detached || self.gave_up
- }
-
- /// Ends the execution at `now` for `cause`, unless the peer ended it already: sends the
- /// runner the cancellation, unless the runner's terminal outcome arrived, closes the input,
- /// and begins the grace of pending output, after which the peer waits for the terminal
- /// outcome for the settlement time.
- fn terminate(&mut self, now: Instant, cause: Cause) {
- if self.finished || self.terminated_at.is_some() {
- return;
- }
- self.terminated_at = Some(now);
- self.timeout_at = None;
- self.cause = Some(cause);
- self.close_input();
- self.grace_until = after(now, self.timers.grace);
- if self.terminal.is_none() {
- self.cancel_message = !self.detached;
- self.give_up_at = self
- .grace_until
- .and_then(|grace| after(grace, self.timers.settlement));
- }
- if self.grace_until.is_some_and(|at| now >= at) {
- self.end_grace();
- }
- }
-
- /// Detaches the peer from the runner at `now` for `cause`: ends the execution if the peer
- /// has not, and stops waiting for the runner, from which nothing more arrives.
- fn detach(&mut self, now: Instant, cause: Cause) {
- self.tick(now);
- if self.finished || self.detached {
- return;
- }
- self.terminate(now, cause);
- self.detached = true;
- self.cancel_message = false;
- self.give_up_at = None;
- }
-
- /// Closes the input: drops what the peer holds of it, and tells the caller unless the caller
- /// ended it.
- fn close_input(&mut self) {
- if self.input.closed {
- return;
- }
- self.input.closed = true;
- self.input.queue = VecDeque::new();
- self.input.closed_event = !self.input.end_sent;
- }
-
- /// Ends the grace of pending output: drops what the inboxes hold, grants their units again,
- /// and discards what still arrives.
- fn end_grace(&mut self) {
- self.grace_until = None;
- for stream in Stream::ALL {
- let incoming = &mut self.outputs[stream.index()];
- incoming.discarding = true;
- let freed = incoming.drop_inbox();
- self.windows[self.limits.pool_of(stream)].free(freed);
- }
- }
-
- /// Returns the outcome once the peer can finish: the runner's terminal outcome, with the
- /// streams that the peer truncated, once the sink took or the peer discarded the rest of the
- /// output; or, once the peer detached or gave up, an outcome of its own, with no status, its
- /// cause, both streams marked truncated because their ends do not report loss, and fenced.
- fn outcome(&self) -> Option {
- if !self
- .outputs
- .iter()
- .all(|incoming| incoming.inbox.is_empty())
- {
- return None;
- }
- let mut outcome = match self.terminal {
- Some(terminal) => terminal,
- None if self.detached || self.gave_up => Outcome::new(None)
- .with_truncated(Stream::Stdout)
- .with_truncated(Stream::Stderr),
- None => return None,
- };
- if outcome.cause().is_none()
- && let Some(cause) = self.cause
- {
- outcome = outcome.with_cause(cause);
- }
- for stream in Stream::ALL {
- if self.outputs[stream.index()].truncated {
- outcome = outcome.with_truncated(stream);
- }
- }
- Some(outcome)
- }
-
- /// Notes the bytes that the peer holds, for its high-water mark.
- fn note_buffered(&mut self) {
- self.high_water = self.high_water.max(self.buffered());
- }
-}
diff --git a/nvx-rs/src/exec/runner.rs b/nvx-rs/src/exec/runner.rs
deleted file mode 100644
index f43a2fd09..000000000
--- a/nvx-rs/src/exec/runner.rs
+++ /dev/null
@@ -1,880 +0,0 @@
-//! The side of an execution that runs its workload: see the [module](super) documentation.
-
-use std::collections::VecDeque;
-use std::time::Instant;
-
-use super::credit::{self, Credit, Granted, Inbox};
-use super::{
- Cause, Key, LaunchFailure, Limits, Outcome, Pool, Receipt, Settlement, Status, Stream, Timers,
- Violation, after, earliest,
-};
-
-/// A message from the runner to the peer, which the adapter frames and sends in order, as
-/// [`Runner::next_message`] returns them.
-#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
-#[non_exhaustive]
-pub enum Message {
- /// Credit of the workload's input: `units` more of the input's window, which the peer may
- /// spend on input. The first one opens the whole window.
- Credit {
- /// The units of the credit.
- units: u32,
- },
- /// The workload's program runs: its launch succeeded.
- Started,
- /// Bytes of `stream`, in order: the first `length` bytes of the buffer that the adapter lent
- /// [`Runner::next_message`].
- Output {
- /// The stream that the bytes belong to.
- stream: Stream,
- /// The number of bytes, at least one and at most a chunk.
- length: usize,
- },
- /// The end of `stream`, after its last byte.
- End(Stream),
- /// The workload's input closed before the peer ended it, because the workload closed it,
- /// ended, or was ended: the peer sends no more input, and the runner drops what arrives.
- InputClosed,
- /// The execution's terminal outcome, after both streams' ends: the runner's last message.
- /// Once the runner is detached from the peer, by a [reset](Runner::reset), a
- /// [disconnection](Runner::disconnect), or a [drop](Runner::dropped), it is the only message
- /// that the runner still returns, and the adapter keeps it rather than send it.
- Terminal(Outcome),
-}
-
-/// A request to the adapter, which applies it to the workload, as [`Runner::next_request`]
-/// returns them. The runner makes each one once.
-#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
-#[non_exhaustive]
-pub enum Request {
- /// End the workload, and its processes, for `cause`, closing its input.
- Terminate(Cause),
- /// Read and drop the rest of `stream`, reporting the bytes with [`Runner::discarded`], until
- /// it closes.
- Discard(Stream),
- /// Close the workload's input: the peer ended it, and the workload took every byte that the
- /// runner accepted.
- CloseInput,
- /// Settle the execution's processes: end what is left of them, and report with
- /// [`Runner::settled`] once the adapter's own check shows that none is left.
- Settle,
-}
-
-/// One output stream, as the runner reads and sends it.
-#[derive(Debug, Default)]
-struct Output {
- /// The bytes read from the workload and not sent yet: at most one chunk.
- queue: VecDeque,
- /// Whether the workload's stream ended: every writer closed it, or reading it failed.
- closed: bool,
- /// Whether every writer closed the workload's stream, though the adapter has not read it to
- /// its end: it ends once the adapter reads what it still holds.
- hung_up: bool,
- /// Whether the stream's end message was returned.
- ended: bool,
- /// Whether the runner drops the stream's bytes.
- discarding: bool,
- /// Whether the request to discard the stream waits to be returned.
- discard_requested: bool,
- /// Whether the stream lost bytes.
- truncated: bool,
-}
-
-impl Output {
- /// Returns whether the stream is done: it closed, and its queue is empty.
- fn is_done(&self) -> bool {
- self.closed && self.queue.is_empty()
- }
-
- /// Drops the queue, and notes whether that lost bytes.
- fn drop_queue(&mut self) {
- if !self.queue.is_empty() {
- self.truncated = true;
- }
- self.queue = VecDeque::new();
- }
-}
-
-/// The workload's standard input, as the runner receives and writes it.
-#[derive(Debug)]
-struct Input {
- /// The bytes that the peer sent and the workload has not taken.
- inbox: Inbox,
- /// The input's window.
- window: Granted,
- /// Whether the peer ended the input.
- peer_ended: bool,
- /// Whether the workload's input is closed, or the runner requested to close it.
- closed: bool,
- /// Whether the request to close the input waits to be returned.
- close_requested: bool,
- /// Whether the message that the input closed early waits to be returned.
- closed_message: bool,
-}
-
-/// Where the settlement of the execution stands.
-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
-enum Settling {
- /// The runner has not requested it.
- Idle,
- /// The runner requested it, and expects its result by this instant, if any.
- Requested(Option),
- /// The execution settled.
- Settled,
- /// The execution is fenced: its settlement failed, or came too late.
- Fenced,
-}
-
-/// The side of an execution that runs its workload: see the [module](super) documentation.
-///
-/// The adapter creates a runner when it accepts the peer's request to run a workload, and hands
-/// it the peer's messages and the workload's events. In return, it takes the runner's
-/// [requests](Self::next_request), which it applies to the workload, and its
-/// [messages](Self::next_message), which it sends the peer, until the
-/// [terminal outcome](Message::Terminal). It reads each output stream of the workload only as far
-/// as the [read allowance](Self::read_allowance) lets it, and writes the
-/// [pending input](Self::pending_input) to the workload as the workload takes it.
-#[derive(Debug)]
-pub struct Runner {
- /// The execution's key.
- key: Key,
- /// The limits that both sides agreed on.
- limits: Limits,
- /// The runner's timers.
- timers: Timers,
- /// The workload's launch result, once it arrived.
- launch: Option>,
- /// Whether the message that the workload started waits to be returned.
- started_message: bool,
- /// The workload's status, once its end is known: `Some(None)` if its status is lost.
- exit: Option