refactor: data movement handled by an independant crate
Promote the data-plane into its own crate with a swactor-facing DataPlaneNodeActor that owns wire-edge lifecycle, and move the edge/ring/egress/ingress machinery out of mvp-system into crates/data-plane. - data-plane/src/actor.rs: add the 769-line DataPlaneNodeActor implementing ActorInterface, owning WireEdgeEndpoint provisioning and the EdgeEstablisher lifecycle and draining commands to arena/worker/transport actors (DataPlaneArenaMsg/WorkerMsg/TransportMsg) with DataPlaneReportMsg back to a report sink - data-plane/src/lib.rs: expand the crate surface to expose actor, arena, edge_actor, edge_lifecycle, egress, and ingress alongside object_record/ring, and reframe it as actor-oriented wire-edge / ring / arena / object-movement contracts - data-plane: move edge_actor and edge_lifecycle (from node/), egress (from worker/), and ingress (from node_data/) into the crate, and add ArenaSample (serde Record, ARENA_SAMPLE_CHANNEL/INTERVAL) to arena.rs - data-plane: add DATA_PLANE_ACTOR_SPEC.md and the data_plane_actor_guarantees/edge_lifecycle_guarantees/egress_guarantees tests - mvp-system/node: add data_plane_bridge.rs wiring the node runtime to the DataPlaneNodeActor, drop the old node_data/arena.rs and node_data/mod.rs (now in data-plane), and remove the arena_manager_guarantees test - mvp-system: drop node_data from the lib.rs pub surface and repoint node/mod.rs to consume the data-plane crate Signed-off-by: Zachery Aaron Shores-Chmielewski <zacheryasc@gmail.com>
This commit is contained in:
parent
8d1588aaf7
commit
e323cd5e86
32 changed files with 1772 additions and 617 deletions
3
Cargo.lock
generated
3
Cargo.lock
generated
|
|
@ -952,7 +952,10 @@ dependencies = [
|
|||
name = "data-plane"
|
||||
version = "0.1.0"
|
||||
dependencies = [
|
||||
"datastream",
|
||||
"libc",
|
||||
"serde",
|
||||
"swactor",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
|
|
|||
|
|
@ -5,6 +5,9 @@ edition = "2024"
|
|||
publish = false
|
||||
|
||||
[dependencies]
|
||||
datastream = { path = "../datastream" }
|
||||
serde = { version = "1", features = ["derive"] }
|
||||
swactor = { path = "../.." }
|
||||
|
||||
[target.'cfg(target_os = "linux")'.dependencies]
|
||||
libc = "0.2"
|
||||
|
|
|
|||
523
crates/data-plane/DATA_PLANE_ACTOR_SPEC.md
Normal file
523
crates/data-plane/DATA_PLANE_ACTOR_SPEC.md
Normal file
|
|
@ -0,0 +1,523 @@
|
|||
# Data Plane Actor Architecture Specification
|
||||
|
||||
**Status:** implemented architecture contract for the `data-plane` crate.
|
||||
|
||||
This document describes the behavioral boundary between reusable data-plane actors
|
||||
and MVP-specific orchestration/runtime code. It is intentionally architectural: it
|
||||
names responsibilities, actor roles, message families, and ownership boundaries
|
||||
without prescribing file layout or migration steps.
|
||||
|
||||
---
|
||||
|
||||
## 1. Purpose
|
||||
|
||||
`data-plane` owns the behavior required to move model objects between stages and
|
||||
between a node process and its local GPU worker process.
|
||||
|
||||
The crate defines the actor protocol for:
|
||||
|
||||
- provisioning logical data edges;
|
||||
- distinguishing network/wire edges from node-local IPC rings;
|
||||
- leasing, installing, readying, faulting, stopping, quiescing, and releasing
|
||||
local rings;
|
||||
- binding logical edges to transport endpoints and local worker rings;
|
||||
- parsing, validating, sequencing, loading, producing, and reporting objects;
|
||||
- gating readiness and object visibility on the correct lifecycle transitions;
|
||||
- translating low-level arena, transport, and worker observations into coarse
|
||||
data-plane outcomes.
|
||||
|
||||
`mvp-system` uses `data-plane` as a reusable actor subsystem. It provides run
|
||||
intent, concrete runtime actor addresses, and MVP-specific report sinks. It does
|
||||
not own the fine-grained data-plane state machine.
|
||||
|
||||
---
|
||||
|
||||
## 2. Core boundary
|
||||
|
||||
Swactor actors carry control, lifecycle, and identity messages. Payload bytes do
|
||||
not move through actor mailboxes.
|
||||
|
||||
Payload bytes move through:
|
||||
|
||||
- arena-backed shared-memory rings for node-local IPC;
|
||||
- transport byte streams for node-to-node data edges;
|
||||
- GPU-worker-owned device allocations for compute-ready objects.
|
||||
|
||||
The data-plane actors decide when these byte paths are established, readable,
|
||||
writable, faulted, stopped, and safe to release. Runtime adapter actors execute
|
||||
concrete effects and report observations back.
|
||||
|
||||
---
|
||||
|
||||
## 3. Edge and ring model
|
||||
|
||||
### 3.1 Wire edge
|
||||
|
||||
A wire edge is a logical run-plan connection between a producer endpoint and a
|
||||
consumer endpoint.
|
||||
|
||||
It carries:
|
||||
|
||||
- run identity;
|
||||
- edge identity;
|
||||
- producer and consumer node identity;
|
||||
- edge kind, such as token input, activation, or token output;
|
||||
- object contract;
|
||||
- transport contract;
|
||||
- optional remote endpoint and remote actor identity.
|
||||
|
||||
A wire edge answers: "which logical data stream connects these stage endpoints?"
|
||||
|
||||
It does not answer: "which local arena offset or worker-process ring is being
|
||||
used on this node?"
|
||||
|
||||
### 3.2 Local IPC ring
|
||||
|
||||
A local IPC ring is a node-local buffer used by a Rust node process and its owned
|
||||
GPU worker process.
|
||||
|
||||
It carries:
|
||||
|
||||
- ring identity allocated by the local arena manager;
|
||||
- arena layout and capacity;
|
||||
- worker-process generation;
|
||||
- local role port, such as input or output;
|
||||
- direction relative to the worker process;
|
||||
- object contract installed into the worker;
|
||||
- quiescence and release state.
|
||||
|
||||
A local IPC ring answers: "how does this node exchange bytes with its local
|
||||
worker process for a specific edge endpoint?"
|
||||
|
||||
It does not answer: "which remote node or distributed route owns the other side
|
||||
of the logical edge?"
|
||||
|
||||
### 3.3 Binding
|
||||
|
||||
A wire edge endpoint may bind to zero or more local resources depending on its
|
||||
role:
|
||||
|
||||
```text
|
||||
inbound wire edge endpoint
|
||||
-> recv transport endpoint
|
||||
-> local worker ingress ring
|
||||
-> device object handle
|
||||
|
||||
outbound wire edge endpoint
|
||||
-> local worker egress ring
|
||||
-> send transport endpoint
|
||||
|
||||
local-only edge endpoint
|
||||
-> local producer/consumer binding
|
||||
-> optional worker ring
|
||||
```
|
||||
|
||||
The binding is owned by data-plane state. MVP code may observe the binding only
|
||||
through coarse reports such as edge ready, object loaded, object produced, edge
|
||||
faulted, and edge stopped.
|
||||
|
||||
---
|
||||
|
||||
## 4. Actor topology
|
||||
|
||||
The target topology is actor-oriented.
|
||||
|
||||
### 4.1 Data-plane node actor
|
||||
|
||||
One data-plane node actor owns the data-plane state for one local node within one
|
||||
active run.
|
||||
|
||||
It owns:
|
||||
|
||||
- local node identity;
|
||||
- run-scoped edge table;
|
||||
- mapping from wire edge endpoints to local rings;
|
||||
- mapping from ring ids to edge endpoints;
|
||||
- object sequence state;
|
||||
- device-handle visibility state;
|
||||
- data-plane child actor addresses;
|
||||
- MVP report sink addresses.
|
||||
|
||||
It receives provisioning intent from MVP code and observations from runtime
|
||||
adapters. It emits actor messages to arena, worker, transport, and MVP report
|
||||
sinks.
|
||||
|
||||
### 4.2 Wire edge actor
|
||||
|
||||
A wire edge actor owns the lifecycle of one logical edge endpoint on the local
|
||||
node.
|
||||
|
||||
It owns:
|
||||
|
||||
- provisioning state;
|
||||
- transport establishment state;
|
||||
- send/receive pump readiness;
|
||||
- stream faults;
|
||||
- logical edge readiness;
|
||||
- stop and fault propagation for that edge endpoint.
|
||||
|
||||
It does not own worker-process state or arena layout details except through a
|
||||
binding supplied by the data-plane node actor or ring actor.
|
||||
|
||||
### 4.3 Local worker ring actor
|
||||
|
||||
A local worker ring actor owns one local IPC ring lifecycle.
|
||||
|
||||
It owns:
|
||||
|
||||
- arena lease request and result;
|
||||
- worker ring installation;
|
||||
- worker ring readable/writable notifications;
|
||||
- ring fault and quiescence observations;
|
||||
- release proof collection;
|
||||
- arena lease release.
|
||||
|
||||
It does not own remote endpoint routing. It can be bound to a wire edge endpoint
|
||||
by edge id, but the ring lifecycle remains local.
|
||||
|
||||
### 4.4 GPU worker control adapter
|
||||
|
||||
The GPU worker control adapter is the actor-facing boundary to the owned Python
|
||||
worker process.
|
||||
|
||||
It owns or fronts:
|
||||
|
||||
- worker process generation;
|
||||
- command serialization to the worker;
|
||||
- stdout/stderr event parsing;
|
||||
- device handle generation checks;
|
||||
- worker stop/crash/restart observations.
|
||||
|
||||
The data plane treats this as an actor endpoint. Worker-process JSON and Python
|
||||
helper details are not exposed to MVP stage logic.
|
||||
|
||||
### 4.5 Transport adapter actors
|
||||
|
||||
Transport adapter actors own concrete wire byte movement.
|
||||
|
||||
They own or front:
|
||||
|
||||
- accepted edge streams;
|
||||
- outbound edge streams;
|
||||
- edge preamble validation;
|
||||
- byte read/write readiness;
|
||||
- transport-specific stream faults;
|
||||
- pump stop observations.
|
||||
|
||||
The data plane treats transport events as observations on a wire edge. Transport
|
||||
actors do not decide stage readiness or object admission.
|
||||
|
||||
### 4.6 MVP report sink
|
||||
|
||||
The MVP report sink receives coarse data-plane outcomes and maps them to
|
||||
MVP-specific control messages.
|
||||
|
||||
Examples:
|
||||
|
||||
- inbound edge ready;
|
||||
- outbound edge ready;
|
||||
- object loaded for stage execution;
|
||||
- object produced for downstream transport;
|
||||
- edge faulted;
|
||||
- local edges stopped.
|
||||
|
||||
The sink does not inspect ring cursors, arena leases, worker generations, or
|
||||
transport pump internals.
|
||||
|
||||
---
|
||||
|
||||
## 5. Actor API surface
|
||||
|
||||
Concrete Rust names are schematic. The contract is the message shape and
|
||||
ownership boundary.
|
||||
|
||||
### 5.1 Provisioning input
|
||||
|
||||
MVP sends provisioning intent to the data-plane node actor:
|
||||
|
||||
```text
|
||||
ProvisionDataPlaneRun {
|
||||
run_id,
|
||||
local_node_id,
|
||||
arena_actor,
|
||||
worker_actor,
|
||||
transport_actor,
|
||||
report_sink,
|
||||
}
|
||||
|
||||
ProvisionWireEdgeEndpoint {
|
||||
run_id,
|
||||
edge_id,
|
||||
direction,
|
||||
edge_kind,
|
||||
local_role_port,
|
||||
local_node_id,
|
||||
peer_node_id,
|
||||
peer_endpoint,
|
||||
object_spec,
|
||||
transport_spec,
|
||||
local_ring_spec,
|
||||
}
|
||||
```
|
||||
|
||||
`direction` is relative to the local node's stage role: inbound means the local
|
||||
stage consumes objects from the edge; outbound means the local stage produces
|
||||
objects to the edge.
|
||||
|
||||
Provisioning is declarative. MVP describes the intended edge endpoint and the
|
||||
actors available to execute effects. It does not prescribe lease/install/driver
|
||||
ordering.
|
||||
|
||||
### 5.2 Runtime observations
|
||||
|
||||
Runtime adapters report observations back to data-plane actors:
|
||||
|
||||
```text
|
||||
ArenaRingLeased
|
||||
ArenaRingLeaseRejected
|
||||
ArenaRingReleased
|
||||
ArenaRingReleaseRejected
|
||||
|
||||
WorkerReady
|
||||
WorkerRingInstalled
|
||||
WorkerRingFaulted
|
||||
WorkerRingQuiesced
|
||||
WorkerRingReadable
|
||||
WorkerRingWritable
|
||||
WorkerObjectLoaded
|
||||
WorkerObjectProduced
|
||||
WorkerObjectFailed
|
||||
WorkerStopped
|
||||
WorkerFaulted
|
||||
|
||||
TransportEdgeReady
|
||||
TransportBytesReceived
|
||||
TransportBytesSent
|
||||
TransportStreamClosed
|
||||
TransportStreamFaulted
|
||||
TransportPumpStopped
|
||||
```
|
||||
|
||||
Observations are facts, not commands. The data plane decides the next state and
|
||||
any follow-up messages.
|
||||
|
||||
### 5.3 Data-plane effects
|
||||
|
||||
The data plane sends effect requests to runtime adapter actors:
|
||||
|
||||
```text
|
||||
LeaseArenaRing
|
||||
CancelArenaRingLease
|
||||
ReleaseArenaRingLease
|
||||
|
||||
InstallWorkerRing
|
||||
UninstallWorkerRing
|
||||
NotifyWorkerRingReadable
|
||||
NotifyWorkerRingWritable
|
||||
LoadObjectFromWorkerRing
|
||||
ExecuteWorkerStep
|
||||
ReleaseWorkerDeviceObject
|
||||
|
||||
EstablishWireSend
|
||||
EstablishWireRecv
|
||||
WriteWireObject
|
||||
StopWirePump
|
||||
```
|
||||
|
||||
Effects are actor messages. The receiving adapter owns the concrete mechanism:
|
||||
memfd/mmap, JSON stdin/stdout, process supervision, iroh streams, or test doubles.
|
||||
|
||||
### 5.4 Data-plane reports
|
||||
|
||||
The data plane reports only stable semantic outcomes to MVP:
|
||||
|
||||
```text
|
||||
InboundEdgeReady { edge_id }
|
||||
OutboundEdgeReady { edge_id }
|
||||
ObjectLoaded { edge_id, object_id, sequence, device_handle }
|
||||
ObjectProduced { edge_id, object_id, sequence, extent }
|
||||
EdgeFaulted { edge_id, reason }
|
||||
EdgeStopped { edge_id }
|
||||
LocalEdgesStopped { run_id }
|
||||
WorkerDataPlaneFaulted { reason }
|
||||
```
|
||||
|
||||
Reports are the only data-plane messages MVP stage/orchestrator actors should
|
||||
need for normal stage progression.
|
||||
|
||||
---
|
||||
|
||||
## 6. Behavior owned by data-plane
|
||||
|
||||
### 6.1 Edge establishment
|
||||
|
||||
For each provisioned edge endpoint, data-plane actors own the establishment
|
||||
sequence.
|
||||
|
||||
Inbound endpoint:
|
||||
|
||||
```text
|
||||
provision endpoint
|
||||
-> lease local ingress ring
|
||||
-> install ring into worker input port
|
||||
-> establish receive transport if the edge is remote
|
||||
-> report inbound edge ready
|
||||
```
|
||||
|
||||
Outbound endpoint:
|
||||
|
||||
```text
|
||||
provision endpoint
|
||||
-> lease local egress ring when worker output is required
|
||||
-> install ring into worker output port
|
||||
-> establish send transport if the edge is remote
|
||||
-> report outbound edge ready
|
||||
```
|
||||
|
||||
Readiness is reported only after every required local and wire resource for that
|
||||
endpoint is ready. A local-only endpoint may omit transport establishment. A
|
||||
wire-only endpoint may omit worker-ring establishment when it terminates outside
|
||||
the local GPU worker.
|
||||
|
||||
### 6.2 Object ingress
|
||||
|
||||
For inbound data, data-plane actors own object admission.
|
||||
|
||||
The data plane:
|
||||
|
||||
- associates incoming bytes with the correct wire edge and stream;
|
||||
- validates object framing and object spec constraints;
|
||||
- preserves sequence ordering required by the edge contract;
|
||||
- writes or exposes the object through the local ingress ring;
|
||||
- asks the worker to load the object to device;
|
||||
- waits for a valid worker object-loaded observation;
|
||||
- reports object loaded to MVP only after the device handle is current and the
|
||||
logical object is complete.
|
||||
|
||||
MVP does not parse object headers, track ingress buffers, reload cursors, or gate
|
||||
object-loaded visibility.
|
||||
|
||||
### 6.3 Object egress
|
||||
|
||||
For outbound data, data-plane actors own object production and forwarding.
|
||||
|
||||
The data plane:
|
||||
|
||||
- receives compute/output observations from the worker;
|
||||
- binds produced objects to the correct outbound edge and sequence;
|
||||
- validates object extent and object contract;
|
||||
- publishes readable/writable state to the worker and transport actors;
|
||||
- forwards complete object records on the wire when the edge is remote;
|
||||
- reports object produced or step-visible outcomes to MVP at semantic
|
||||
boundaries, not cursor boundaries.
|
||||
|
||||
MVP does not decide when a local output ring is readable, when a transport stream
|
||||
should consume it, or when a produced object is safe to expose downstream.
|
||||
|
||||
### 6.4 Faults
|
||||
|
||||
The data plane owns data movement fault classification and propagation.
|
||||
|
||||
Fault sources include:
|
||||
|
||||
- arena lease rejection or release rejection;
|
||||
- worker ring installation failure;
|
||||
- worker ring fault;
|
||||
- malformed object framing;
|
||||
- sequence violation;
|
||||
- worker object load/produce failure;
|
||||
- transport stream read/write/protocol failure;
|
||||
- pump stop before quiescence;
|
||||
- stale worker generation or device handle.
|
||||
|
||||
A fault on one edge endpoint must not silently corrupt another endpoint. The data
|
||||
plane maps local faults to edge-scoped or worker-scoped reports, starts the
|
||||
required stop/quiescence path, and emits the appropriate MVP report.
|
||||
|
||||
### 6.5 Stop, quiescence, and release
|
||||
|
||||
The data plane owns teardown ordering for local resources.
|
||||
|
||||
For a bound edge/ring pair, stop requires:
|
||||
|
||||
```text
|
||||
stop transport pump if present
|
||||
-> uninstall or quiesce worker ring if installed
|
||||
-> prove no local reader/writer still uses the ring
|
||||
-> release arena lease
|
||||
-> report edge stopped
|
||||
```
|
||||
|
||||
Arena release must be gated by quiescence proof. MVP may request run or edge
|
||||
stop, but it does not supply low-level release proof or decide when a ring is
|
||||
safe to release.
|
||||
|
||||
---
|
||||
|
||||
## 7. MVP-system utilization
|
||||
|
||||
`mvp-system` remains responsible for MVP orchestration and stage semantics.
|
||||
|
||||
It owns:
|
||||
|
||||
- run planning and edge assignment;
|
||||
- stage provisioning authority;
|
||||
- membership/readiness gates outside the data plane;
|
||||
- weight loading and role configuration intent;
|
||||
- stage controller behavior;
|
||||
- prompt injection and token consumption;
|
||||
- mapping data-plane reports to MVP lifecycle messages;
|
||||
- selecting concrete runtime adapters for arena, worker process, and transport.
|
||||
|
||||
For data movement, MVP code acts as a client:
|
||||
|
||||
1. Spawn or obtain actor addresses for the data-plane node actor and required
|
||||
runtime adapters.
|
||||
2. Send run and edge provisioning intent to the data-plane node actor.
|
||||
3. Forward runtime observations from concrete adapter actors when those adapters
|
||||
are MVP-owned.
|
||||
4. Receive coarse data-plane reports.
|
||||
5. Translate those reports into stage-controller or orchestrator messages.
|
||||
|
||||
MVP must not rely on private edge states such as waiting-for-lease,
|
||||
waiting-for-worker-ring, waiting-for-driver, pump-stopped, ring-quiesced, or
|
||||
release-ready. Those are data-plane implementation states.
|
||||
|
||||
---
|
||||
|
||||
## 8. Required invariants
|
||||
|
||||
- `EdgeId` names a logical run-plan edge, not a local ring allocation.
|
||||
- `RingId` names a local arena-backed IPC ring, not a distributed edge.
|
||||
- A ring may be bound to an edge endpoint, but the identifiers are not
|
||||
interchangeable.
|
||||
- Actor messages carry control and identities, not tensor payload bytes.
|
||||
- Object-loaded reports are emitted only for complete, validated objects with
|
||||
current-generation device handles.
|
||||
- Edge-ready reports are emitted only after required wire and local IPC resources
|
||||
are ready.
|
||||
- Arena lease release is gated by local quiescence proof.
|
||||
- Worker process generation is part of every device-handle validity decision.
|
||||
- Transport faults and worker faults are classified by data-plane before they
|
||||
become MVP reports.
|
||||
- MVP stage logic observes semantic outcomes, never ring cursor mechanics.
|
||||
|
||||
---
|
||||
|
||||
## 9. Non-goals
|
||||
|
||||
`data-plane` does not own:
|
||||
|
||||
- global run planning;
|
||||
- placement optimization;
|
||||
- model layer assignment;
|
||||
- weight download or weight loading semantics;
|
||||
- prompt tokenization or output token policy;
|
||||
- membership convergence;
|
||||
- provider provisioning;
|
||||
- concrete iroh endpoint construction;
|
||||
- concrete Python helper implementation.
|
||||
|
||||
The crate defines reusable actor protocols and data-movement behavior. Concrete
|
||||
runtime adapters may live beside MVP code, inside reusable support crates, or in
|
||||
tests, as long as they satisfy the actor contracts above.
|
||||
769
crates/data-plane/src/actor.rs
Normal file
769
crates/data-plane/src/actor.rs
Normal file
|
|
@ -0,0 +1,769 @@
|
|||
//! Swactor-facing data-plane actor API.
|
||||
//!
|
||||
//! The actor owns lifecycle decisions for logical wire edge endpoints and their
|
||||
//! bound node-local worker rings. Runtime-specific actors execute the effect
|
||||
//! messages and report observations back.
|
||||
|
||||
use std::collections::{BTreeMap, BTreeSet};
|
||||
|
||||
use swactor::actor::{ActorAddress, ActorInterface};
|
||||
use swactor::runtime::Ctx;
|
||||
|
||||
use crate::edge_lifecycle as lifecycle;
|
||||
use crate::object_record;
|
||||
|
||||
pub use lifecycle::{
|
||||
DType, EdgeFaultReason, EdgeId, LeaseRequestId, NodeId, ObjectKind, ObjectSpec,
|
||||
QuiescenceProof, RingDirection, RingId, RingLayout, RingLeaseRejection, RingSpec, RunId,
|
||||
};
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub struct PortId(pub String);
|
||||
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
pub enum EdgeKind {
|
||||
TokenIn,
|
||||
Activation,
|
||||
TokenOut,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
pub enum EdgeEndpointDirection {
|
||||
Inbound,
|
||||
Outbound,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub struct PeerEndpoint(pub String);
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub enum TransportBinding {
|
||||
Remote { endpoint: Option<PeerEndpoint> },
|
||||
LocalOnly,
|
||||
}
|
||||
|
||||
impl TransportBinding {
|
||||
fn requires_transport(&self) -> bool {
|
||||
matches!(self, Self::Remote { .. })
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
pub enum WorkerRingBinding {
|
||||
Required,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub struct DataPlaneRunConfig {
|
||||
pub run_id: RunId,
|
||||
pub local_node_id: NodeId,
|
||||
pub arena_actor: ActorAddress,
|
||||
pub worker_actor: ActorAddress,
|
||||
pub transport_actor: ActorAddress,
|
||||
pub report_sink: ActorAddress,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub struct WireEdgeEndpoint {
|
||||
pub run_id: RunId,
|
||||
pub edge_id: EdgeId,
|
||||
pub direction: EdgeEndpointDirection,
|
||||
pub kind: EdgeKind,
|
||||
pub local_node_id: NodeId,
|
||||
pub peer_node_id: Option<NodeId>,
|
||||
pub peer_endpoint: Option<PeerEndpoint>,
|
||||
pub local_role_port: PortId,
|
||||
pub object_spec: ObjectSpec,
|
||||
pub ring_spec: RingSpec,
|
||||
pub object_record_spec: object_record::ObjectSpec,
|
||||
pub transport: TransportBinding,
|
||||
pub worker_ring: WorkerRingBinding,
|
||||
}
|
||||
|
||||
impl WireEdgeEndpoint {
|
||||
fn provision_event(&self) -> Option<lifecycle::EdgeEvent> {
|
||||
match self.direction {
|
||||
EdgeEndpointDirection::Inbound => {
|
||||
Some(lifecycle::EdgeEvent::ProvisionRx(lifecycle::ProvisionRx {
|
||||
run_id: self.run_id,
|
||||
edge_id: self.edge_id,
|
||||
local_node_id: self.local_node_id,
|
||||
object_spec: self.object_spec,
|
||||
ring_spec: self.ring_spec,
|
||||
}))
|
||||
}
|
||||
EdgeEndpointDirection::Outbound => {
|
||||
Some(lifecycle::EdgeEvent::ProvisionTx(lifecycle::ProvisionTx {
|
||||
run_id: self.run_id,
|
||||
edge_id: self.edge_id,
|
||||
local_node_id: self.local_node_id,
|
||||
consumer_node_id: self.peer_node_id?,
|
||||
object_spec: self.object_spec,
|
||||
ring_spec: self.ring_spec,
|
||||
}))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub enum DataPlaneNodeMsg {
|
||||
ConfigureRun(DataPlaneRunConfig),
|
||||
ProvisionWireEdgeEndpoint(WireEdgeEndpoint),
|
||||
Arena(ArenaObservation),
|
||||
Worker(WorkerObservation),
|
||||
Transport(TransportObservation),
|
||||
StopEdge { edge_id: EdgeId },
|
||||
StopRun { run_id: RunId },
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub enum ArenaObservation {
|
||||
RingLeased {
|
||||
request_id: LeaseRequestId,
|
||||
ring_id: RingId,
|
||||
layout: RingLayout,
|
||||
},
|
||||
RingLeaseRejected {
|
||||
request_id: LeaseRequestId,
|
||||
reason: RingLeaseRejection,
|
||||
},
|
||||
RingReleased {
|
||||
ring_id: RingId,
|
||||
},
|
||||
RingReleaseRejected {
|
||||
ring_id: RingId,
|
||||
},
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub enum WorkerObservation {
|
||||
WorkerReady,
|
||||
RingInstalled {
|
||||
edge_id: EdgeId,
|
||||
ring_id: RingId,
|
||||
},
|
||||
RingFaulted {
|
||||
edge_id: EdgeId,
|
||||
ring_id: RingId,
|
||||
reason: lifecycle::RingFaultReason,
|
||||
},
|
||||
RingQuiesced {
|
||||
ring_id: RingId,
|
||||
},
|
||||
QuiescenceProven {
|
||||
ring_id: RingId,
|
||||
},
|
||||
ObjectLoaded {
|
||||
edge_id: EdgeId,
|
||||
ring_id: RingId,
|
||||
object_id: object_record::ObjectId,
|
||||
sequence: u64,
|
||||
extent: u64,
|
||||
handle: DeviceHandle,
|
||||
},
|
||||
ObjectProduced {
|
||||
edge_id: EdgeId,
|
||||
ring_id: RingId,
|
||||
object_id: object_record::ObjectId,
|
||||
sequence: u64,
|
||||
extent: u64,
|
||||
},
|
||||
ObjectFailed {
|
||||
edge_id: EdgeId,
|
||||
ring_id: RingId,
|
||||
reason: object_record::ObjectFailureReason,
|
||||
},
|
||||
WorkerFaulted,
|
||||
WorkerStopped,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub enum TransportObservation {
|
||||
EdgeReady {
|
||||
edge_id: EdgeId,
|
||||
},
|
||||
BytesAvailable {
|
||||
edge_id: EdgeId,
|
||||
stream_id: StreamId,
|
||||
byte_count: u64,
|
||||
},
|
||||
StreamClosed {
|
||||
edge_id: EdgeId,
|
||||
},
|
||||
StreamFaulted {
|
||||
edge_id: EdgeId,
|
||||
reason: lifecycle::StreamFaultReason,
|
||||
},
|
||||
PumpStopped {
|
||||
edge_id: EdgeId,
|
||||
ring_id: RingId,
|
||||
},
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub enum DataPlaneArenaMsg {
|
||||
LeaseRing {
|
||||
request_id: LeaseRequestId,
|
||||
edge_id: EdgeId,
|
||||
direction: RingDirection,
|
||||
ring_spec: RingSpec,
|
||||
},
|
||||
CancelQueuedLease {
|
||||
request_id: LeaseRequestId,
|
||||
edge_id: EdgeId,
|
||||
},
|
||||
ReleaseRing {
|
||||
ring_id: RingId,
|
||||
proof: QuiescenceProof,
|
||||
},
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub enum DataPlaneWorkerMsg {
|
||||
InstallRing {
|
||||
edge_id: EdgeId,
|
||||
ring_id: RingId,
|
||||
direction: RingDirection,
|
||||
layout: RingLayout,
|
||||
object_spec: ObjectSpec,
|
||||
ring_spec: RingSpec,
|
||||
role_port: PortId,
|
||||
},
|
||||
UninstallRing {
|
||||
edge_id: EdgeId,
|
||||
ring_id: RingId,
|
||||
},
|
||||
NotifyRingReadable {
|
||||
ring_id: RingId,
|
||||
},
|
||||
NotifyRingWritable {
|
||||
ring_id: RingId,
|
||||
},
|
||||
LoadObjectFromRing {
|
||||
edge_id: EdgeId,
|
||||
ring_id: RingId,
|
||||
object_id: object_record::ObjectId,
|
||||
sequence: u64,
|
||||
extent: u64,
|
||||
},
|
||||
ExecuteStep {
|
||||
edge_id: EdgeId,
|
||||
sequence: u64,
|
||||
},
|
||||
ReleaseDeviceObject {
|
||||
handle: DeviceHandle,
|
||||
},
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub enum DataPlaneTransportMsg {
|
||||
EstablishSend {
|
||||
edge_id: EdgeId,
|
||||
consumer_node_id: NodeId,
|
||||
ring_id: RingId,
|
||||
layout: RingLayout,
|
||||
},
|
||||
EstablishRecv {
|
||||
edge_id: EdgeId,
|
||||
ring_id: RingId,
|
||||
layout: RingLayout,
|
||||
},
|
||||
WriteWireObject {
|
||||
edge_id: EdgeId,
|
||||
object_id: object_record::ObjectId,
|
||||
sequence: u64,
|
||||
extent: u64,
|
||||
},
|
||||
StopWirePump {
|
||||
edge_id: EdgeId,
|
||||
ring_id: RingId,
|
||||
},
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub enum DataPlaneReportMsg {
|
||||
InboundEdgeReady {
|
||||
edge_id: EdgeId,
|
||||
},
|
||||
OutboundEdgeReady {
|
||||
edge_id: EdgeId,
|
||||
},
|
||||
ObjectLoaded {
|
||||
edge_id: EdgeId,
|
||||
object_id: object_record::ObjectId,
|
||||
sequence: u64,
|
||||
extent: u64,
|
||||
handle: DeviceHandle,
|
||||
},
|
||||
ObjectProduced {
|
||||
edge_id: EdgeId,
|
||||
object_id: object_record::ObjectId,
|
||||
sequence: u64,
|
||||
extent: u64,
|
||||
},
|
||||
EdgeFaulted {
|
||||
edge_id: EdgeId,
|
||||
reason: EdgeFaultReason,
|
||||
},
|
||||
EdgeStopped {
|
||||
edge_id: EdgeId,
|
||||
},
|
||||
LocalEdgesStopped {
|
||||
run_id: RunId,
|
||||
},
|
||||
WorkerDataPlaneFaulted {
|
||||
reason: WorkerDataPlaneFaultReason,
|
||||
},
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
pub enum WorkerDataPlaneFaultReason {
|
||||
WorkerFaulted,
|
||||
WorkerStopped,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
|
||||
pub struct WorkerGeneration(pub u64);
|
||||
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
pub struct DeviceHandle {
|
||||
pub generation: WorkerGeneration,
|
||||
pub id: u64,
|
||||
}
|
||||
|
||||
impl DeviceHandle {
|
||||
pub const fn new(generation: WorkerGeneration, id: u64) -> Self {
|
||||
Self { generation, id }
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
|
||||
pub struct StreamId(pub u64);
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
struct EdgeBinding {
|
||||
direction: EdgeEndpointDirection,
|
||||
role_port_index: usize,
|
||||
transport: TransportBinding,
|
||||
}
|
||||
|
||||
pub struct DataPlaneNodeActor {
|
||||
run: Option<DataPlaneRunConfig>,
|
||||
lifecycle: lifecycle::EdgeEstablisher,
|
||||
command_cursor: usize,
|
||||
event_cursor: usize,
|
||||
bindings: BTreeMap<EdgeId, EdgeBinding>,
|
||||
stopped_edges: BTreeSet<EdgeId>,
|
||||
role_ports: Vec<PortId>,
|
||||
rings: BTreeMap<RingId, EdgeId>,
|
||||
pending_reports: Vec<DataPlaneReportMsg>,
|
||||
stopping_run: Option<RunId>,
|
||||
local_edges_stopped_reported: bool,
|
||||
}
|
||||
|
||||
impl DataPlaneNodeActor {
|
||||
pub fn new(local_node_id: NodeId) -> Self {
|
||||
Self {
|
||||
run: None,
|
||||
lifecycle: lifecycle::EdgeEstablisher::new(local_node_id),
|
||||
command_cursor: 0,
|
||||
event_cursor: 0,
|
||||
bindings: BTreeMap::new(),
|
||||
stopped_edges: BTreeSet::new(),
|
||||
role_ports: Vec::new(),
|
||||
rings: BTreeMap::new(),
|
||||
pending_reports: Vec::new(),
|
||||
stopping_run: None,
|
||||
local_edges_stopped_reported: false,
|
||||
}
|
||||
}
|
||||
|
||||
fn configure(&mut self, config: DataPlaneRunConfig) {
|
||||
self.lifecycle = lifecycle::EdgeEstablisher::new(config.local_node_id);
|
||||
self.command_cursor = 0;
|
||||
self.event_cursor = 0;
|
||||
self.bindings.clear();
|
||||
self.role_ports.clear();
|
||||
self.stopped_edges.clear();
|
||||
self.rings.clear();
|
||||
self.pending_reports.clear();
|
||||
self.stopping_run = None;
|
||||
self.local_edges_stopped_reported = false;
|
||||
self.run = Some(config);
|
||||
}
|
||||
|
||||
fn observe(&mut self, msg: DataPlaneNodeMsg) {
|
||||
match msg {
|
||||
DataPlaneNodeMsg::ConfigureRun(config) => self.configure(config),
|
||||
DataPlaneNodeMsg::ProvisionWireEdgeEndpoint(endpoint) => {
|
||||
if let Some(event) = endpoint.provision_event() {
|
||||
let role_port_index = self.role_ports.len();
|
||||
self.role_ports.push(endpoint.local_role_port.clone());
|
||||
self.bindings.insert(
|
||||
endpoint.edge_id,
|
||||
EdgeBinding {
|
||||
direction: endpoint.direction,
|
||||
role_port_index,
|
||||
transport: endpoint.transport.clone(),
|
||||
},
|
||||
);
|
||||
self.lifecycle.observe(event);
|
||||
}
|
||||
}
|
||||
DataPlaneNodeMsg::Arena(observation) => match observation {
|
||||
ArenaObservation::RingLeased {
|
||||
request_id,
|
||||
ring_id,
|
||||
layout,
|
||||
} => self.lifecycle.observe(lifecycle::EdgeEvent::RingLeased {
|
||||
request_id,
|
||||
ring_id,
|
||||
layout,
|
||||
}),
|
||||
ArenaObservation::RingLeaseRejected { request_id, reason } => self
|
||||
.lifecycle
|
||||
.observe(lifecycle::EdgeEvent::RingLeaseRejected { request_id, reason }),
|
||||
ArenaObservation::RingReleased { .. }
|
||||
| ArenaObservation::RingReleaseRejected { .. } => {}
|
||||
},
|
||||
DataPlaneNodeMsg::Worker(observation) => match observation {
|
||||
WorkerObservation::WorkerReady => {}
|
||||
WorkerObservation::RingInstalled { edge_id, ring_id } => {
|
||||
self.rings.insert(ring_id, edge_id);
|
||||
self.lifecycle
|
||||
.observe(lifecycle::EdgeEvent::RingInstalled { edge_id, ring_id });
|
||||
}
|
||||
WorkerObservation::RingFaulted {
|
||||
edge_id,
|
||||
ring_id,
|
||||
reason,
|
||||
} => self.lifecycle.observe(lifecycle::EdgeEvent::RingFault {
|
||||
edge_id,
|
||||
ring_id,
|
||||
reason,
|
||||
}),
|
||||
WorkerObservation::RingQuiesced { ring_id } => self
|
||||
.lifecycle
|
||||
.observe(lifecycle::EdgeEvent::RingQuiesced { ring_id }),
|
||||
WorkerObservation::QuiescenceProven { ring_id } => self
|
||||
.lifecycle
|
||||
.observe(lifecycle::EdgeEvent::QuiescenceProven { ring_id }),
|
||||
WorkerObservation::ObjectLoaded {
|
||||
edge_id,
|
||||
object_id,
|
||||
sequence,
|
||||
extent,
|
||||
handle,
|
||||
..
|
||||
} => self.report_object_loaded(edge_id, object_id, sequence, extent, handle),
|
||||
WorkerObservation::ObjectProduced {
|
||||
edge_id,
|
||||
object_id,
|
||||
sequence,
|
||||
extent,
|
||||
..
|
||||
} => self.report_object_produced(edge_id, object_id, sequence, extent),
|
||||
WorkerObservation::ObjectFailed {
|
||||
edge_id, ring_id, ..
|
||||
} => self.lifecycle.observe(lifecycle::EdgeEvent::RingFault {
|
||||
edge_id,
|
||||
ring_id,
|
||||
reason: lifecycle::RingFaultReason::WorkerRejectedRing,
|
||||
}),
|
||||
WorkerObservation::WorkerFaulted => {
|
||||
self.report_worker_fault(WorkerDataPlaneFaultReason::WorkerFaulted)
|
||||
}
|
||||
WorkerObservation::WorkerStopped => {
|
||||
self.report_worker_fault(WorkerDataPlaneFaultReason::WorkerStopped)
|
||||
}
|
||||
},
|
||||
DataPlaneNodeMsg::Transport(observation) => match observation {
|
||||
TransportObservation::EdgeReady { edge_id } => self
|
||||
.lifecycle
|
||||
.observe(lifecycle::EdgeEvent::DriverEdgeReady { edge_id }),
|
||||
TransportObservation::BytesAvailable { .. }
|
||||
| TransportObservation::StreamClosed { .. } => {}
|
||||
TransportObservation::StreamFaulted { edge_id, reason } => self
|
||||
.lifecycle
|
||||
.observe(lifecycle::EdgeEvent::StreamFault { edge_id, reason }),
|
||||
TransportObservation::PumpStopped { edge_id, ring_id } => self
|
||||
.lifecycle
|
||||
.observe(lifecycle::EdgeEvent::PumpStopped { edge_id, ring_id }),
|
||||
},
|
||||
DataPlaneNodeMsg::StopEdge { edge_id } => self
|
||||
.lifecycle
|
||||
.observe(lifecycle::EdgeEvent::StopEdge { edge_id }),
|
||||
DataPlaneNodeMsg::StopRun { run_id } => {
|
||||
self.stopping_run = Some(run_id);
|
||||
self.local_edges_stopped_reported = false;
|
||||
let edge_ids = self.bindings.keys().copied().collect::<Vec<_>>();
|
||||
for edge_id in edge_ids {
|
||||
self.lifecycle
|
||||
.observe(lifecycle::EdgeEvent::StopEdge { edge_id });
|
||||
}
|
||||
self.maybe_report_local_edges_stopped();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn ring_for_edge(&self, edge_id: EdgeId) -> Option<RingId> {
|
||||
self.rings
|
||||
.iter()
|
||||
.find_map(|(ring_id, seen_edge)| (*seen_edge == edge_id).then_some(*ring_id))
|
||||
}
|
||||
|
||||
fn role_port_for(&self, edge_id: EdgeId) -> PortId {
|
||||
self.bindings
|
||||
.get(&edge_id)
|
||||
.and_then(|binding| self.role_ports.get(binding.role_port_index))
|
||||
.cloned()
|
||||
.unwrap_or_else(|| PortId(String::new()))
|
||||
}
|
||||
|
||||
fn report_object_loaded(
|
||||
&mut self,
|
||||
edge_id: EdgeId,
|
||||
object_id: object_record::ObjectId,
|
||||
sequence: u64,
|
||||
extent: u64,
|
||||
handle: DeviceHandle,
|
||||
) {
|
||||
self.pending_reports.push(DataPlaneReportMsg::ObjectLoaded {
|
||||
edge_id,
|
||||
object_id,
|
||||
sequence,
|
||||
extent,
|
||||
handle,
|
||||
});
|
||||
}
|
||||
|
||||
fn report_object_produced(
|
||||
&mut self,
|
||||
edge_id: EdgeId,
|
||||
object_id: object_record::ObjectId,
|
||||
sequence: u64,
|
||||
extent: u64,
|
||||
) {
|
||||
self.pending_reports
|
||||
.push(DataPlaneReportMsg::ObjectProduced {
|
||||
edge_id,
|
||||
object_id,
|
||||
sequence,
|
||||
extent,
|
||||
});
|
||||
}
|
||||
|
||||
fn report_worker_fault(&mut self, reason: WorkerDataPlaneFaultReason) {
|
||||
self.pending_reports
|
||||
.push(DataPlaneReportMsg::WorkerDataPlaneFaulted { reason });
|
||||
}
|
||||
|
||||
fn report_local_edges_stopped(&mut self, run_id: RunId) {
|
||||
self.pending_reports
|
||||
.push(DataPlaneReportMsg::LocalEdgesStopped { run_id });
|
||||
}
|
||||
|
||||
fn maybe_report_local_edges_stopped(&mut self) {
|
||||
let Some(run_id) = self.stopping_run else {
|
||||
return;
|
||||
};
|
||||
if self.local_edges_stopped_reported {
|
||||
return;
|
||||
}
|
||||
if self
|
||||
.bindings
|
||||
.keys()
|
||||
.all(|edge_id| self.stopped_edges.contains(edge_id))
|
||||
{
|
||||
self.local_edges_stopped_reported = true;
|
||||
self.report_local_edges_stopped(run_id);
|
||||
}
|
||||
}
|
||||
|
||||
fn drain(&mut self, ctx: &Ctx) {
|
||||
let Some(run) = self.run.clone() else {
|
||||
return;
|
||||
};
|
||||
|
||||
loop {
|
||||
let mut progressed = false;
|
||||
let mut immediate_ready = Vec::new();
|
||||
|
||||
while self.command_cursor < self.lifecycle.commands().len() {
|
||||
let command = self.lifecycle.commands()[self.command_cursor].clone();
|
||||
self.command_cursor += 1;
|
||||
progressed = true;
|
||||
match command {
|
||||
lifecycle::EdgeCommand::LeaseRing {
|
||||
request_id,
|
||||
edge_id,
|
||||
direction,
|
||||
ring_spec,
|
||||
} => {
|
||||
let _ = ctx.send(
|
||||
run.arena_actor,
|
||||
DataPlaneArenaMsg::LeaseRing {
|
||||
request_id,
|
||||
edge_id,
|
||||
direction,
|
||||
ring_spec,
|
||||
},
|
||||
);
|
||||
}
|
||||
lifecycle::EdgeCommand::CancelQueuedLease {
|
||||
request_id,
|
||||
edge_id,
|
||||
} => {
|
||||
let _ = ctx.send(
|
||||
run.arena_actor,
|
||||
DataPlaneArenaMsg::CancelQueuedLease {
|
||||
request_id,
|
||||
edge_id,
|
||||
},
|
||||
);
|
||||
}
|
||||
lifecycle::EdgeCommand::InstallWorkerRing {
|
||||
edge_id,
|
||||
ring_id,
|
||||
direction,
|
||||
layout,
|
||||
object_spec,
|
||||
ring_spec,
|
||||
} => {
|
||||
let _ = ctx.send(
|
||||
run.worker_actor,
|
||||
DataPlaneWorkerMsg::InstallRing {
|
||||
edge_id,
|
||||
ring_id,
|
||||
direction,
|
||||
layout,
|
||||
object_spec,
|
||||
ring_spec,
|
||||
role_port: self.role_port_for(edge_id),
|
||||
},
|
||||
);
|
||||
}
|
||||
lifecycle::EdgeCommand::UninstallWorkerRing { edge_id, ring_id } => {
|
||||
let _ = ctx.send(
|
||||
run.worker_actor,
|
||||
DataPlaneWorkerMsg::UninstallRing { edge_id, ring_id },
|
||||
);
|
||||
}
|
||||
lifecycle::EdgeCommand::EstablishSend {
|
||||
edge_id,
|
||||
consumer_node_id,
|
||||
layout,
|
||||
} => {
|
||||
let ring_id = self.ring_for_edge(edge_id).unwrap_or(RingId(0));
|
||||
if self
|
||||
.bindings
|
||||
.get(&edge_id)
|
||||
.is_some_and(|binding| binding.transport.requires_transport())
|
||||
{
|
||||
let _ = ctx.send(
|
||||
run.transport_actor,
|
||||
DataPlaneTransportMsg::EstablishSend {
|
||||
edge_id,
|
||||
consumer_node_id,
|
||||
ring_id,
|
||||
layout,
|
||||
},
|
||||
);
|
||||
} else {
|
||||
immediate_ready.push(edge_id);
|
||||
}
|
||||
}
|
||||
lifecycle::EdgeCommand::EstablishRecv { edge_id, layout } => {
|
||||
let ring_id = self.ring_for_edge(edge_id).unwrap_or(RingId(0));
|
||||
if self
|
||||
.bindings
|
||||
.get(&edge_id)
|
||||
.is_some_and(|binding| binding.transport.requires_transport())
|
||||
{
|
||||
let _ = ctx.send(
|
||||
run.transport_actor,
|
||||
DataPlaneTransportMsg::EstablishRecv {
|
||||
edge_id,
|
||||
ring_id,
|
||||
layout,
|
||||
},
|
||||
);
|
||||
} else {
|
||||
immediate_ready.push(edge_id);
|
||||
}
|
||||
}
|
||||
lifecycle::EdgeCommand::StopPump { edge_id, ring_id } => {
|
||||
let _ = ctx.send(
|
||||
run.transport_actor,
|
||||
DataPlaneTransportMsg::StopWirePump { edge_id, ring_id },
|
||||
);
|
||||
}
|
||||
lifecycle::EdgeCommand::ReleaseArenaLease { ring_id, proof } => {
|
||||
let _ = ctx.send(
|
||||
run.arena_actor,
|
||||
DataPlaneArenaMsg::ReleaseRing { ring_id, proof },
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for edge_id in immediate_ready {
|
||||
self.lifecycle
|
||||
.observe(lifecycle::EdgeEvent::DriverEdgeReady { edge_id });
|
||||
}
|
||||
|
||||
while self.event_cursor < self.lifecycle.events().len() {
|
||||
let event = self.lifecycle.events()[self.event_cursor].clone();
|
||||
self.event_cursor += 1;
|
||||
progressed = true;
|
||||
match event {
|
||||
lifecycle::EdgeLifecycleEvent::EdgeReady { edge_id, .. } => {
|
||||
let report =
|
||||
match self.bindings.get(&edge_id).map(|binding| binding.direction) {
|
||||
Some(EdgeEndpointDirection::Inbound) => {
|
||||
DataPlaneReportMsg::InboundEdgeReady { edge_id }
|
||||
}
|
||||
Some(EdgeEndpointDirection::Outbound) => {
|
||||
DataPlaneReportMsg::OutboundEdgeReady { edge_id }
|
||||
}
|
||||
None => continue,
|
||||
};
|
||||
let _ = ctx.send(run.report_sink, report);
|
||||
}
|
||||
lifecycle::EdgeLifecycleEvent::EdgeFaulted { edge_id, reason } => {
|
||||
let _ = ctx.send(
|
||||
run.report_sink,
|
||||
DataPlaneReportMsg::EdgeFaulted { edge_id, reason },
|
||||
);
|
||||
}
|
||||
lifecycle::EdgeLifecycleEvent::EdgeStopped { edge_id } => {
|
||||
self.stopped_edges.insert(edge_id);
|
||||
let _ =
|
||||
ctx.send(run.report_sink, DataPlaneReportMsg::EdgeStopped { edge_id });
|
||||
self.maybe_report_local_edges_stopped();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for report in self.pending_reports.drain(..) {
|
||||
progressed = true;
|
||||
let _ = ctx.send(run.report_sink, report);
|
||||
}
|
||||
|
||||
if !progressed {
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl ActorInterface for DataPlaneNodeActor {
|
||||
type Incoming = DataPlaneNodeMsg;
|
||||
type Response = ();
|
||||
|
||||
fn handle(&mut self, ctx: &Ctx, msg: Self::Incoming) {
|
||||
self.observe(msg);
|
||||
self.drain(ctx);
|
||||
}
|
||||
}
|
||||
|
|
@ -1,7 +1,10 @@
|
|||
//! Reusable arena-backed ring allocation contracts.
|
||||
|
||||
use std::collections::{BTreeMap, VecDeque};
|
||||
use std::time::{SystemTime, UNIX_EPOCH};
|
||||
use std::time::{Duration, SystemTime, UNIX_EPOCH};
|
||||
|
||||
use datastream::Record;
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
pub struct ArenaSnapshot {
|
||||
|
|
@ -17,6 +20,44 @@ pub struct ArenaSnapshot {
|
|||
pub release_failures_total: u64,
|
||||
}
|
||||
|
||||
pub const ARENA_SAMPLE_CHANNEL: &str = "mvp.arena";
|
||||
pub const ARENA_SAMPLE_INTERVAL: Duration = Duration::from_secs(1);
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct ArenaSample {
|
||||
pub seq: u64,
|
||||
pub sample_unix_ms: u64,
|
||||
pub capacity_bytes: u64,
|
||||
pub live_bytes: u64,
|
||||
pub free_bytes: u64,
|
||||
pub active_leases: u64,
|
||||
pub pending_leases: u64,
|
||||
pub largest_free_range_bytes: u64,
|
||||
pub allocation_failures_total: u64,
|
||||
pub release_failures_total: u64,
|
||||
}
|
||||
|
||||
impl Record for ArenaSample {
|
||||
const CHANNEL: &'static str = ARENA_SAMPLE_CHANNEL;
|
||||
}
|
||||
|
||||
impl From<ArenaSnapshot> for ArenaSample {
|
||||
fn from(snapshot: ArenaSnapshot) -> Self {
|
||||
Self {
|
||||
seq: snapshot.seq,
|
||||
sample_unix_ms: snapshot.sample_unix_ms,
|
||||
capacity_bytes: snapshot.capacity_bytes,
|
||||
live_bytes: snapshot.live_bytes,
|
||||
free_bytes: snapshot.free_bytes,
|
||||
active_leases: snapshot.active_leases,
|
||||
pending_leases: snapshot.pending_leases,
|
||||
largest_free_range_bytes: snapshot.largest_free_range_bytes,
|
||||
allocation_failures_total: snapshot.allocation_failures_total,
|
||||
release_failures_total: snapshot.release_failures_total,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
|
||||
pub struct NodeId(pub u64);
|
||||
|
||||
|
|
|
|||
|
|
@ -7,7 +7,7 @@ pub struct EdgeId(pub u64);
|
|||
#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
|
||||
pub struct PortId(pub String);
|
||||
|
||||
pub use data_plane::object_record::{
|
||||
pub use crate::object_record::{
|
||||
FLAG_BEGIN_SEQUENCE, FLAG_END_OF_SEQUENCE, HEADER_LEN, KNOWN_FLAGS_MASK, OBJECT_MAGIC,
|
||||
OBJECT_MAGIC_BYTES, OBJECT_VERSION, ObjectFlags, ObjectHeader, ObjectId, ObjectLayout,
|
||||
ObjectSpec,
|
||||
|
|
@ -127,7 +127,6 @@ pub enum WakeHint {
|
|||
RingWritable { ring_id: RingId },
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
struct PendingStep {
|
||||
step_id: StepId,
|
||||
|
|
@ -135,7 +134,6 @@ struct PendingStep {
|
|||
role_state_updated: bool,
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
pub struct EgressProducerHarness {
|
||||
generation: WorkerGeneration,
|
||||
rings: std::collections::BTreeMap<RingId, InstallRing>,
|
||||
|
|
@ -149,7 +147,6 @@ pub struct EgressProducerHarness {
|
|||
events: Vec<WorkerEgressOut>,
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
impl EgressProducerHarness {
|
||||
pub fn new(generation: WorkerGeneration) -> Self {
|
||||
Self {
|
||||
|
|
@ -369,7 +366,6 @@ impl EgressProducerHarness {
|
|||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
fn encode_header(output: OutputBinding) -> Vec<u8> {
|
||||
ObjectHeader {
|
||||
object_id: output.object_id,
|
||||
|
|
@ -7,7 +7,7 @@ pub struct EdgeId(pub u64);
|
|||
#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
|
||||
pub struct PortId(pub String);
|
||||
|
||||
pub use data_plane::object_record::{
|
||||
pub use crate::object_record::{
|
||||
FLAG_BEGIN_SEQUENCE, FLAG_END_OF_SEQUENCE, HEADER_LEN, KNOWN_FLAGS_MASK, OBJECT_MAGIC,
|
||||
OBJECT_MAGIC_BYTES, OBJECT_VERSION, ObjectFailureReason, ObjectFlags, ObjectHeader, ObjectId,
|
||||
ObjectLayout, ObjectRecord, ObjectRecordBuilder, ObjectRecordRead, ObjectSpec,
|
||||
|
|
@ -106,7 +106,6 @@ pub struct DeviceCopyLog {
|
|||
pub byte_count: u64,
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||
struct PendingObject {
|
||||
record: ObjectRecord,
|
||||
|
|
@ -114,7 +113,6 @@ struct PendingObject {
|
|||
handle: Option<DeviceHandle>,
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
pub struct IngressParserHarness {
|
||||
generation: WorkerGeneration,
|
||||
install: Option<InstallRing>,
|
||||
|
|
@ -128,7 +126,6 @@ pub struct IngressParserHarness {
|
|||
events: Vec<WorkerIngressOut>,
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
impl IngressParserHarness {
|
||||
pub fn new(generation: WorkerGeneration) -> Self {
|
||||
Self {
|
||||
|
|
@ -300,7 +297,6 @@ impl IngressParserHarness {
|
|||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
fn object_failure_metadata(
|
||||
bytes: &[u8],
|
||||
reason: ObjectFailureReason,
|
||||
|
|
@ -1,5 +1,11 @@
|
|||
//! Reusable local data-plane contracts for arena-backed byte movement.
|
||||
//! Reusable actor-oriented data-plane contracts for wire edges, local IPC rings,
|
||||
//! arena-backed byte movement, and GPU worker object movement.
|
||||
|
||||
pub mod actor;
|
||||
pub mod arena;
|
||||
pub mod edge_actor;
|
||||
pub mod edge_lifecycle;
|
||||
pub mod egress;
|
||||
pub mod ingress;
|
||||
pub mod object_record;
|
||||
pub mod ring;
|
||||
|
|
|
|||
231
crates/data-plane/tests/data_plane_actor_guarantees.rs
Normal file
231
crates/data-plane/tests/data_plane_actor_guarantees.rs
Normal file
|
|
@ -0,0 +1,231 @@
|
|||
use data_plane::actor as dp;
|
||||
use data_plane::object_record;
|
||||
use swactor::config::RuntimeConfig;
|
||||
use swactor::runtime::Runtime;
|
||||
|
||||
fn object_spec() -> dp::ObjectSpec {
|
||||
dp::ObjectSpec {
|
||||
kind: dp::ObjectKind::Activation,
|
||||
dtype: dp::DType::F16,
|
||||
max_extent_bytes: 4096,
|
||||
}
|
||||
}
|
||||
|
||||
fn record_spec() -> object_record::ObjectSpec {
|
||||
object_record::ObjectSpec {
|
||||
max_extent: 4096,
|
||||
alignment: 4,
|
||||
layout: object_record::ObjectLayout::Token,
|
||||
}
|
||||
}
|
||||
|
||||
fn ring_spec() -> dp::RingSpec {
|
||||
dp::RingSpec {
|
||||
header_bytes: 128,
|
||||
data_bytes: 4096,
|
||||
alignment: 64,
|
||||
}
|
||||
}
|
||||
|
||||
fn layout() -> dp::RingLayout {
|
||||
dp::RingLayout {
|
||||
start_offset: 0,
|
||||
header_offset: 0,
|
||||
data_offset: 128,
|
||||
end_offset: 4224,
|
||||
data_bytes: 4096,
|
||||
alignment: 64,
|
||||
}
|
||||
}
|
||||
|
||||
fn inbound_endpoint() -> dp::WireEdgeEndpoint {
|
||||
dp::WireEdgeEndpoint {
|
||||
run_id: dp::RunId(55),
|
||||
edge_id: dp::EdgeId(7001),
|
||||
direction: dp::EdgeEndpointDirection::Inbound,
|
||||
kind: dp::EdgeKind::Activation,
|
||||
local_node_id: dp::NodeId(10),
|
||||
peer_node_id: Some(dp::NodeId(9)),
|
||||
peer_endpoint: Some(dp::PeerEndpoint("node-9".to_owned())),
|
||||
local_role_port: dp::PortId("input".to_owned()),
|
||||
object_spec: object_spec(),
|
||||
ring_spec: ring_spec(),
|
||||
object_record_spec: record_spec(),
|
||||
transport: dp::TransportBinding::Remote {
|
||||
endpoint: Some(dp::PeerEndpoint("node-9".to_owned())),
|
||||
},
|
||||
worker_ring: dp::WorkerRingBinding::Required,
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn inbound_wire_edge_establishes_through_arena_worker_transport_then_reports_ready() {
|
||||
let runtime = Runtime::new(RuntimeConfig::default());
|
||||
let arena = runtime
|
||||
.new_inbox::<dp::DataPlaneArenaMsg>()
|
||||
.expect("arena inbox");
|
||||
let worker = runtime
|
||||
.new_inbox::<dp::DataPlaneWorkerMsg>()
|
||||
.expect("worker inbox");
|
||||
let transport = runtime
|
||||
.new_inbox::<dp::DataPlaneTransportMsg>()
|
||||
.expect("transport inbox");
|
||||
let reports = runtime
|
||||
.new_inbox::<dp::DataPlaneReportMsg>()
|
||||
.expect("report inbox");
|
||||
let actor = runtime
|
||||
.spawn(dp::DataPlaneNodeActor::new(dp::NodeId(10)))
|
||||
.expect("spawn data-plane actor");
|
||||
|
||||
runtime
|
||||
.send_to(
|
||||
actor,
|
||||
dp::DataPlaneNodeMsg::ConfigureRun(dp::DataPlaneRunConfig {
|
||||
run_id: dp::RunId(55),
|
||||
local_node_id: dp::NodeId(10),
|
||||
arena_actor: *arena.addr(),
|
||||
worker_actor: *worker.addr(),
|
||||
transport_actor: *transport.addr(),
|
||||
report_sink: *reports.addr(),
|
||||
}),
|
||||
)
|
||||
.expect("configure run");
|
||||
runtime
|
||||
.send_to(
|
||||
actor,
|
||||
dp::DataPlaneNodeMsg::ProvisionWireEdgeEndpoint(inbound_endpoint()),
|
||||
)
|
||||
.expect("provision inbound");
|
||||
runtime.tick();
|
||||
|
||||
assert_eq!(
|
||||
arena.try_recv(),
|
||||
Some(dp::DataPlaneArenaMsg::LeaseRing {
|
||||
request_id: dp::LeaseRequestId(1),
|
||||
edge_id: dp::EdgeId(7001),
|
||||
direction: dp::RingDirection::Ingress,
|
||||
ring_spec: ring_spec(),
|
||||
})
|
||||
);
|
||||
|
||||
runtime
|
||||
.send_to(
|
||||
actor,
|
||||
dp::DataPlaneNodeMsg::Arena(dp::ArenaObservation::RingLeased {
|
||||
request_id: dp::LeaseRequestId(1),
|
||||
ring_id: dp::RingId(8001),
|
||||
layout: layout(),
|
||||
}),
|
||||
)
|
||||
.expect("ring leased");
|
||||
runtime.tick();
|
||||
|
||||
assert_eq!(
|
||||
worker.try_recv(),
|
||||
Some(dp::DataPlaneWorkerMsg::InstallRing {
|
||||
edge_id: dp::EdgeId(7001),
|
||||
ring_id: dp::RingId(8001),
|
||||
direction: dp::RingDirection::Ingress,
|
||||
layout: layout(),
|
||||
object_spec: object_spec(),
|
||||
ring_spec: ring_spec(),
|
||||
role_port: dp::PortId("input".to_owned()),
|
||||
})
|
||||
);
|
||||
|
||||
runtime
|
||||
.send_to(
|
||||
actor,
|
||||
dp::DataPlaneNodeMsg::Worker(dp::WorkerObservation::RingInstalled {
|
||||
edge_id: dp::EdgeId(7001),
|
||||
ring_id: dp::RingId(8001),
|
||||
}),
|
||||
)
|
||||
.expect("worker installed");
|
||||
runtime.tick();
|
||||
|
||||
assert_eq!(
|
||||
transport.try_recv(),
|
||||
Some(dp::DataPlaneTransportMsg::EstablishRecv {
|
||||
edge_id: dp::EdgeId(7001),
|
||||
ring_id: dp::RingId(8001),
|
||||
layout: layout(),
|
||||
})
|
||||
);
|
||||
|
||||
runtime
|
||||
.send_to(
|
||||
actor,
|
||||
dp::DataPlaneNodeMsg::Transport(dp::TransportObservation::EdgeReady {
|
||||
edge_id: dp::EdgeId(7001),
|
||||
}),
|
||||
)
|
||||
.expect("transport ready");
|
||||
runtime.tick();
|
||||
|
||||
assert_eq!(
|
||||
reports.try_recv(),
|
||||
Some(dp::DataPlaneReportMsg::InboundEdgeReady {
|
||||
edge_id: dp::EdgeId(7001)
|
||||
})
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn object_loaded_observation_is_reported_as_coarse_data_plane_outcome() {
|
||||
let runtime = Runtime::new(RuntimeConfig::default());
|
||||
let arena = runtime
|
||||
.new_inbox::<dp::DataPlaneArenaMsg>()
|
||||
.expect("arena inbox");
|
||||
let worker = runtime
|
||||
.new_inbox::<dp::DataPlaneWorkerMsg>()
|
||||
.expect("worker inbox");
|
||||
let transport = runtime
|
||||
.new_inbox::<dp::DataPlaneTransportMsg>()
|
||||
.expect("transport inbox");
|
||||
let reports = runtime
|
||||
.new_inbox::<dp::DataPlaneReportMsg>()
|
||||
.expect("report inbox");
|
||||
let actor = runtime
|
||||
.spawn(dp::DataPlaneNodeActor::new(dp::NodeId(10)))
|
||||
.expect("spawn data-plane actor");
|
||||
|
||||
runtime
|
||||
.send_to(
|
||||
actor,
|
||||
dp::DataPlaneNodeMsg::ConfigureRun(dp::DataPlaneRunConfig {
|
||||
run_id: dp::RunId(55),
|
||||
local_node_id: dp::NodeId(10),
|
||||
arena_actor: *arena.addr(),
|
||||
worker_actor: *worker.addr(),
|
||||
transport_actor: *transport.addr(),
|
||||
report_sink: *reports.addr(),
|
||||
}),
|
||||
)
|
||||
.expect("configure run");
|
||||
runtime
|
||||
.send_to(
|
||||
actor,
|
||||
dp::DataPlaneNodeMsg::Worker(dp::WorkerObservation::ObjectLoaded {
|
||||
edge_id: dp::EdgeId(7001),
|
||||
ring_id: dp::RingId(8001),
|
||||
object_id: object_record::ObjectId(9000),
|
||||
sequence: 7,
|
||||
extent: 16,
|
||||
handle: dp::DeviceHandle::new(dp::WorkerGeneration(3), 42),
|
||||
}),
|
||||
)
|
||||
.expect("object loaded");
|
||||
runtime.tick();
|
||||
|
||||
assert_eq!(
|
||||
reports.try_recv(),
|
||||
Some(dp::DataPlaneReportMsg::ObjectLoaded {
|
||||
edge_id: dp::EdgeId(7001),
|
||||
object_id: object_record::ObjectId(9000),
|
||||
sequence: 7,
|
||||
extent: 16,
|
||||
handle: dp::DeviceHandle::new(dp::WorkerGeneration(3), 42),
|
||||
})
|
||||
);
|
||||
}
|
||||
|
|
@ -1,4 +1,4 @@
|
|||
//! Black-box contract tests for MVP edge establishment.
|
||||
//! Black-box contract tests for data-plane edge establishment.
|
||||
//!
|
||||
//! These tests intentionally know only the public EdgeEstablisher surface:
|
||||
//!
|
||||
|
|
@ -7,9 +7,9 @@
|
|||
//! - arena, worker/token, driver, ready, fault, and release commands out
|
||||
//!
|
||||
//! They assert the guarantees in
|
||||
//! `specs/mvp_system/edge_establisher_contract.md`.
|
||||
//! the reusable data-plane edge lifecycle contract.
|
||||
|
||||
use mvp_system::node::edge_lifecycle as edge;
|
||||
use data_plane::edge_lifecycle as edge;
|
||||
|
||||
// A send provision carries the consumer node id because the driver must know
|
||||
// where to send. It deliberately carries no remote actor address.
|
||||
|
|
@ -1,16 +1,15 @@
|
|||
//! Black-box contract tests for MVP GPU worker egress production.
|
||||
//! Black-box contract tests for data-plane GPU worker egress production.
|
||||
//!
|
||||
//! These tests intentionally know only the public worker egress surface:
|
||||
//! These tests intentionally know only the public data-plane egress surface:
|
||||
//!
|
||||
//! - `InstallRing`, `ExecuteStep` output bindings, device-copy outcomes,
|
||||
//! backpressure, and shutdown events in
|
||||
//! - committed ring bytes, cursor publication, `ObjectProduced`,
|
||||
//! `StepCompleted`, and step failures out
|
||||
//!
|
||||
//! They assert the guarantees in
|
||||
//! `specs/mvp_system/gpu_worker_egress_producer_contract.md`.
|
||||
//! They assert the reusable data-plane egress producer contract.
|
||||
|
||||
use mvp_system::node_data::egress;
|
||||
use data_plane::egress;
|
||||
|
||||
// A valid egress ring config supplies the edge object spec and current worker
|
||||
// generation. The producer remains free to choose copy scheduling internally.
|
||||
|
|
@ -5,7 +5,6 @@ extern crate self as mvp_system;
|
|||
|
||||
pub mod chat;
|
||||
pub mod node;
|
||||
pub mod node_data;
|
||||
pub mod observability;
|
||||
pub mod orchestration;
|
||||
pub mod prompt;
|
||||
|
|
|
|||
69
crates/mvp-system/src/node/data_plane_bridge.rs
Normal file
69
crates/mvp-system/src/node/data_plane_bridge.rs
Normal file
|
|
@ -0,0 +1,69 @@
|
|||
//! MVP adapter for coarse reusable data-plane actor reports.
|
||||
|
||||
use data_plane::actor as dp;
|
||||
use swactor::actor::{ActorAddress, ActorInterface};
|
||||
use swactor::runtime::Ctx;
|
||||
|
||||
use crate::node::actor::NodeAgentMsg;
|
||||
|
||||
pub struct MvpDataPlaneReportSinkActor {
|
||||
node_agent: ActorAddress,
|
||||
}
|
||||
|
||||
impl MvpDataPlaneReportSinkActor {
|
||||
pub fn new(node_agent: ActorAddress) -> Self {
|
||||
Self { node_agent }
|
||||
}
|
||||
}
|
||||
|
||||
impl ActorInterface for MvpDataPlaneReportSinkActor {
|
||||
type Incoming = dp::DataPlaneReportMsg;
|
||||
type Response = ();
|
||||
|
||||
fn handle(&mut self, ctx: &Ctx, msg: Self::Incoming) {
|
||||
match msg {
|
||||
dp::DataPlaneReportMsg::InboundEdgeReady { edge_id } => {
|
||||
let _ = ctx.send(
|
||||
self.node_agent,
|
||||
NodeAgentMsg::MarkInboundEdgeReady { edge_id: edge_id.0 },
|
||||
);
|
||||
}
|
||||
dp::DataPlaneReportMsg::OutboundEdgeReady { edge_id } => {
|
||||
let _ = ctx.send(
|
||||
self.node_agent,
|
||||
NodeAgentMsg::MarkOutboundEdgeReady { edge_id: edge_id.0 },
|
||||
);
|
||||
}
|
||||
dp::DataPlaneReportMsg::ObjectLoaded {
|
||||
edge_id,
|
||||
object_id,
|
||||
sequence,
|
||||
handle,
|
||||
..
|
||||
} => {
|
||||
let _ = ctx.send(
|
||||
self.node_agent,
|
||||
NodeAgentMsg::ObjectLoaded {
|
||||
edge_id: edge_id.0,
|
||||
object_id: object_id.0,
|
||||
sequence,
|
||||
handle_generation: handle.generation.0,
|
||||
handle_id: handle.id,
|
||||
},
|
||||
);
|
||||
}
|
||||
dp::DataPlaneReportMsg::ObjectProduced { .. } => {}
|
||||
dp::DataPlaneReportMsg::EdgeFaulted { .. }
|
||||
| dp::DataPlaneReportMsg::WorkerDataPlaneFaulted { .. } => {
|
||||
let _ = ctx.send(self.node_agent, NodeAgentMsg::WorkerCrashed);
|
||||
}
|
||||
dp::DataPlaneReportMsg::EdgeStopped { .. } => {}
|
||||
dp::DataPlaneReportMsg::LocalEdgesStopped { run_id } => {
|
||||
let _ = ctx.send(
|
||||
self.node_agent,
|
||||
NodeAgentMsg::LocalEdgesStopped { run_id: run_id.0 },
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -5,5 +5,5 @@
|
|||
|
||||
pub mod actor;
|
||||
pub mod boot_lifecycle;
|
||||
pub mod edge_lifecycle;
|
||||
pub mod data_plane_bridge;
|
||||
pub mod worker_node_runtime;
|
||||
|
|
|
|||
|
|
@ -26,9 +26,6 @@ use crate::node::actor::{
|
|||
NodeAgentActor, NodeAgentMsg, NodeAgentReport, StageCommandWire, StageInboundEdgeWire,
|
||||
StageObjectSpecWire, StageOutboundEdgeWire, StageRingSpecWire,
|
||||
};
|
||||
use crate::node::edge_lifecycle as edge;
|
||||
use crate::node_data::arena;
|
||||
use crate::node_data::ingress;
|
||||
use crate::observability::benchmark;
|
||||
use crate::orchestration::distribution_stack::DistributionRuntimeStack;
|
||||
use crate::orchestration::provider_adapters::relay::relay_runtime_config_from_env;
|
||||
|
|
@ -41,6 +38,9 @@ use crate::transport::driver_pumps as driver_model;
|
|||
use crate::transport::endpoint_advertisement::{
|
||||
EndpointAddrMask, MVP_IROH_ENDPOINT_ADDR_MASK_ENV, advertised_endpoint,
|
||||
};
|
||||
use data_plane::arena;
|
||||
use data_plane::edge_lifecycle as edge;
|
||||
use data_plane::ingress;
|
||||
use distribution::node::DistributedNodeConfig;
|
||||
use distribution::swim::telemetry::ObservedProbeEvent;
|
||||
use distribution::telemetry::{MembershipTransition, SwimProbeEvent};
|
||||
|
|
@ -737,7 +737,7 @@ fn spawn_arena_sampler(
|
|||
loop {
|
||||
interval.tick().await;
|
||||
|
||||
let sample = arena_manager.lock().sample(seq);
|
||||
let sample: arena::ArenaSample = arena_manager.lock().sample(seq).into();
|
||||
seq = seq.saturating_add(1);
|
||||
producer.submit_record(channel, &sample);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,131 +0,0 @@
|
|||
//! MVP arena-manager adapter over reusable data-plane arena contracts.
|
||||
|
||||
use std::time::Duration;
|
||||
|
||||
use datastream::Record;
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
pub use data_plane::arena::{
|
||||
ArenaCommand, ArenaConfig, ArenaEvent, ArenaFault, ArenaRequest, LayoutPointer, LeaseRequestId,
|
||||
LeaseRing, NodeId, QuiescenceProof, RingId, RingLayout, RingLease, RingLeaseRejection,
|
||||
RingReleaseRejection, RingSpec,
|
||||
};
|
||||
|
||||
pub const ARENA_SAMPLE_CHANNEL: &str = "mvp.arena";
|
||||
pub const ARENA_SAMPLE_INTERVAL: Duration = Duration::from_secs(1);
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct ArenaSample {
|
||||
pub seq: u64,
|
||||
pub sample_unix_ms: u64,
|
||||
pub capacity_bytes: u64,
|
||||
pub live_bytes: u64,
|
||||
pub free_bytes: u64,
|
||||
pub active_leases: u64,
|
||||
pub pending_leases: u64,
|
||||
pub largest_free_range_bytes: u64,
|
||||
pub allocation_failures_total: u64,
|
||||
pub release_failures_total: u64,
|
||||
}
|
||||
|
||||
impl Record for ArenaSample {
|
||||
const CHANNEL: &'static str = ARENA_SAMPLE_CHANNEL;
|
||||
}
|
||||
|
||||
impl From<data_plane::arena::ArenaSnapshot> for ArenaSample {
|
||||
fn from(snapshot: data_plane::arena::ArenaSnapshot) -> Self {
|
||||
Self {
|
||||
seq: snapshot.seq,
|
||||
sample_unix_ms: snapshot.sample_unix_ms,
|
||||
capacity_bytes: snapshot.capacity_bytes,
|
||||
live_bytes: snapshot.live_bytes,
|
||||
free_bytes: snapshot.free_bytes,
|
||||
active_leases: snapshot.active_leases,
|
||||
pending_leases: snapshot.pending_leases,
|
||||
largest_free_range_bytes: snapshot.largest_free_range_bytes,
|
||||
allocation_failures_total: snapshot.allocation_failures_total,
|
||||
release_failures_total: snapshot.release_failures_total,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub struct ArenaManager {
|
||||
inner: data_plane::arena::ArenaManager,
|
||||
}
|
||||
|
||||
impl ArenaManager {
|
||||
pub fn boot(config: ArenaConfig) -> Result<Self, ArenaFault> {
|
||||
data_plane::arena::ArenaManager::boot(config).map(|inner| Self { inner })
|
||||
}
|
||||
|
||||
pub fn request(&mut self, request: ArenaRequest) -> Vec<ArenaEvent> {
|
||||
self.inner.request(request)
|
||||
}
|
||||
|
||||
pub fn live_leases(&self) -> &[RingLease] {
|
||||
self.inner.live_leases()
|
||||
}
|
||||
|
||||
pub fn lookup_lease(&self, ring_id: RingId) -> Option<&RingLease> {
|
||||
self.inner.lookup_lease(ring_id)
|
||||
}
|
||||
|
||||
pub fn sample(&self, seq: u64) -> ArenaSample {
|
||||
self.inner.sample(seq).into()
|
||||
}
|
||||
|
||||
#[cfg(target_os = "linux")]
|
||||
pub fn arena_fd(&self) -> std::os::fd::RawFd {
|
||||
self.inner.arena_fd()
|
||||
}
|
||||
|
||||
#[cfg(target_os = "linux")]
|
||||
pub fn write_arena(&self, offset: u64, bytes: &[u8]) -> Result<(), std::io::Error> {
|
||||
self.inner.write_arena(offset, bytes)
|
||||
}
|
||||
|
||||
#[cfg(target_os = "linux")]
|
||||
pub fn read_arena(&self, offset: u64, len: usize) -> Result<Vec<u8>, std::io::Error> {
|
||||
self.inner.read_arena(offset, len)
|
||||
}
|
||||
}
|
||||
|
||||
pub struct ArenaManagerHarness {
|
||||
manager: ArenaManager,
|
||||
events: Vec<ArenaEvent>,
|
||||
commands: Vec<ArenaCommand>,
|
||||
}
|
||||
|
||||
impl ArenaManagerHarness {
|
||||
pub fn boot(config: ArenaConfig) -> Result<Self, ArenaFault> {
|
||||
Ok(Self {
|
||||
manager: ArenaManager::boot(config)?,
|
||||
events: Vec::new(),
|
||||
commands: Vec::new(),
|
||||
})
|
||||
}
|
||||
|
||||
pub fn request(&mut self, request: ArenaRequest) {
|
||||
self.events.extend(self.manager.request(request));
|
||||
}
|
||||
|
||||
pub fn events(&self) -> &[ArenaEvent] {
|
||||
&self.events
|
||||
}
|
||||
|
||||
pub fn commands(&self) -> &[ArenaCommand] {
|
||||
&self.commands
|
||||
}
|
||||
|
||||
pub fn live_leases(&self) -> &[RingLease] {
|
||||
self.manager.live_leases()
|
||||
}
|
||||
|
||||
pub fn lookup_lease(&self, ring_id: RingId) -> Option<&RingLease> {
|
||||
self.manager.lookup_lease(ring_id)
|
||||
}
|
||||
|
||||
pub fn sample(&self, seq: u64) -> ArenaSample {
|
||||
self.manager.sample(seq)
|
||||
}
|
||||
}
|
||||
|
|
@ -1,25 +0,0 @@
|
|||
//! MVP node-local data-plane adapter public surface.
|
||||
//!
|
||||
//! Reusable arena, ring, and object-record contracts live in `data-plane`.
|
||||
//! This module binds those contracts to MVP worker ingress, worker egress,
|
||||
//! and edge actor behavior.
|
||||
|
||||
pub mod arena;
|
||||
pub mod edge_actor;
|
||||
pub mod ingress;
|
||||
|
||||
pub mod ring {
|
||||
pub use data_plane::ring::*;
|
||||
}
|
||||
|
||||
pub mod object {
|
||||
pub use data_plane::object_record::*;
|
||||
}
|
||||
|
||||
pub mod egress {
|
||||
pub use crate::worker::egress::*;
|
||||
}
|
||||
|
||||
pub mod reusable {
|
||||
pub use data_plane::{arena, object_record, ring};
|
||||
}
|
||||
|
|
@ -4,9 +4,9 @@ use datastream::hardware::net::HostNetSample;
|
|||
use datastream::{ChannelRegistry, Record};
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
use crate::node_data::arena::ArenaSample;
|
||||
use crate::observability::lifecycle as obs;
|
||||
use crate::orchestration::provisioning::{self, ProvisionLogStream};
|
||||
use data_plane::arena::ArenaSample;
|
||||
|
||||
/// Structured MVP lifecycle facts: run, node, stage, edge, ring, object, step, and worker events.
|
||||
pub const MVP_LIFECYCLE: &str = "mvp.lifecycle";
|
||||
|
|
|
|||
|
|
@ -22,7 +22,6 @@ use crate::orchestration::config::{DEFAULT_CONFIG_PATH, TomlConfigOverlay};
|
|||
use crate::transport::codec_registry::register_mvp_actor_codecs;
|
||||
const PROVIDER_START_MAX_ATTEMPTS: usize = 4;
|
||||
|
||||
use crate::node_data::object as ingress;
|
||||
use crate::observability::telemetry::{
|
||||
MVP_PROVISIONING_EVENTS, MvpProvisionEventRecord, MvpProvisionLogRecord,
|
||||
mvp_provision_log_channel,
|
||||
|
|
@ -53,6 +52,7 @@ use crate::staging::gguf_shard::{StageShardPlan, plan_stage_shard};
|
|||
use crate::transport::endpoint_advertisement::{
|
||||
EndpointAddrMask, MVP_IROH_ENDPOINT_ADDR_MASK_ENV, advertised_endpoint,
|
||||
};
|
||||
use data_plane::object_record as ingress;
|
||||
use datastream::{
|
||||
ChannelContent, ChannelId, ChannelRef, DatastreamEndpoint, DatastreamEvent, DatastreamProducer,
|
||||
DatastreamPublisherMsg, DatastreamSubscribe, Frame, Lifetime, NodeId, Record, StreamDescriptor,
|
||||
|
|
|
|||
|
|
@ -1,416 +0,0 @@
|
|||
//! Black-box contract tests for MVP ArenaManager behavior.
|
||||
//!
|
||||
//! These tests intentionally know only the public arena-manager surface:
|
||||
//!
|
||||
//! - arena boot configuration
|
||||
//! - lease, cancel, release, quiescence, and shutdown requests in
|
||||
//! - lease events, rejected requests, released ranges, and faults out
|
||||
//!
|
||||
//! They assert the guarantees in
|
||||
//! `specs/mvp_system/arena_manager_contract.md`.
|
||||
|
||||
use mvp_system::node_data::arena;
|
||||
|
||||
// A small deterministic arena makes overlap, alignment, queueing, and reuse
|
||||
// proofs easy to inspect. The concrete mmap strategy remains outside the test.
|
||||
fn arena_config() -> arena::ArenaConfig {
|
||||
arena::ArenaConfig {
|
||||
node_id: arena::NodeId(10),
|
||||
reservation_ceiling: 4096,
|
||||
base_alignment: 64,
|
||||
}
|
||||
}
|
||||
|
||||
// Lease requests name size and alignment only. They do not request pointers or
|
||||
// private allocator slots, which keeps layout authority inside ArenaManager.
|
||||
fn lease_request(request_id: u64, bytes: u64, alignment: u64) -> arena::LeaseRing {
|
||||
arena::LeaseRing {
|
||||
request_id: arena::LeaseRequestId(request_id),
|
||||
ring_spec: arena::RingSpec {
|
||||
header_bytes: 128,
|
||||
data_bytes: bytes,
|
||||
alignment,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// The harness exposes public lease events and lease snapshots. Tests use those
|
||||
// snapshots only after a RingLeased event, so private allocator state remains
|
||||
// unobservable.
|
||||
fn new_arena() -> arena::ArenaManagerHarness {
|
||||
arena::ArenaManagerHarness::boot(arena_config()).expect("test arena must boot")
|
||||
}
|
||||
|
||||
// This helper proves two public layouts are disjoint using half-open ranges.
|
||||
// It is more useful than comparing offsets directly because allocator choice is
|
||||
// intentionally implementation-defined.
|
||||
fn assert_non_overlapping(left: &arena::RingLayout, right: &arena::RingLayout) {
|
||||
let left_range = left.start_offset..left.end_offset;
|
||||
let right_range = right.start_offset..right.end_offset;
|
||||
assert!(
|
||||
left_range.end <= right_range.start || right_range.end <= left_range.start,
|
||||
"live leases overlap: {left:?} and {right:?}"
|
||||
);
|
||||
}
|
||||
|
||||
// This proves sampling an unused arena reports only reserved capacity and no
|
||||
// allocator activity.
|
||||
#[test]
|
||||
fn sample_reports_empty_arena_capacity_and_zero_activity() {
|
||||
let harness = new_arena();
|
||||
|
||||
let sample = harness.sample(41);
|
||||
|
||||
assert_eq!(sample.seq, 41);
|
||||
assert!(
|
||||
sample.sample_unix_ms > 0,
|
||||
"sample timestamp must be a populated Unix epoch millisecond"
|
||||
);
|
||||
assert_eq!(sample.capacity_bytes, arena_config().reservation_ceiling);
|
||||
assert_eq!(sample.live_bytes, 0);
|
||||
assert_eq!(sample.free_bytes, arena_config().reservation_ceiling);
|
||||
assert_eq!(sample.active_leases, 0);
|
||||
assert_eq!(sample.pending_leases, 0);
|
||||
assert_eq!(
|
||||
sample.largest_free_range_bytes,
|
||||
arena_config().reservation_ceiling
|
||||
);
|
||||
assert_eq!(sample.allocation_failures_total, 0);
|
||||
assert_eq!(sample.release_failures_total, 0);
|
||||
}
|
||||
|
||||
// This proves a granted lease is reflected in live capacity accounting and the
|
||||
// active lease count without relying on private allocator slots.
|
||||
#[test]
|
||||
fn sample_counts_live_bytes_and_active_leases_after_grant() {
|
||||
let mut harness = new_arena();
|
||||
harness.request(arena::ArenaRequest::LeaseRing(lease_request(1, 512, 64)));
|
||||
let lease = harness.live_leases()[0].clone();
|
||||
let lease_bytes = lease.layout.end_offset - lease.layout.start_offset;
|
||||
|
||||
let sample = harness.sample(42);
|
||||
|
||||
assert_eq!(sample.seq, 42);
|
||||
assert_eq!(sample.capacity_bytes, arena_config().reservation_ceiling);
|
||||
assert_eq!(sample.live_bytes, lease_bytes);
|
||||
assert_eq!(
|
||||
sample.free_bytes,
|
||||
arena_config().reservation_ceiling - lease_bytes
|
||||
);
|
||||
assert_eq!(sample.active_leases, 1);
|
||||
assert_eq!(sample.pending_leases, 0);
|
||||
assert_eq!(sample.allocation_failures_total, 0);
|
||||
assert_eq!(sample.release_failures_total, 0);
|
||||
}
|
||||
|
||||
// This proves a proof-backed release returns the full range to the free pool and
|
||||
// restores the largest allocatable span.
|
||||
#[test]
|
||||
fn sample_reports_full_free_space_after_release() {
|
||||
let mut harness = new_arena();
|
||||
harness.request(arena::ArenaRequest::LeaseRing(lease_request(1, 512, 64)));
|
||||
let ring_id = harness.live_leases()[0].ring_id;
|
||||
|
||||
harness.request(arena::ArenaRequest::ReleaseRing {
|
||||
ring_id,
|
||||
proof: arena::QuiescenceProof::verified(),
|
||||
});
|
||||
let sample = harness.sample(43);
|
||||
|
||||
assert_eq!(sample.live_bytes, 0);
|
||||
assert_eq!(sample.free_bytes, arena_config().reservation_ceiling);
|
||||
assert_eq!(sample.active_leases, 0);
|
||||
assert_eq!(sample.pending_leases, 0);
|
||||
assert_eq!(
|
||||
sample.largest_free_range_bytes,
|
||||
arena_config().reservation_ceiling
|
||||
);
|
||||
}
|
||||
|
||||
// This proves queued leases are visible as pending work while existing live
|
||||
// leases continue to own their bytes.
|
||||
#[test]
|
||||
fn sample_counts_queued_requests_as_pending_leases() {
|
||||
let mut harness = new_arena();
|
||||
harness.request(arena::ArenaRequest::LeaseRing(lease_request(1, 3072, 64)));
|
||||
let live = harness.live_leases()[0].clone();
|
||||
let live_bytes = live.layout.end_offset - live.layout.start_offset;
|
||||
harness.request(arena::ArenaRequest::LeaseRing(lease_request(2, 1024, 64)));
|
||||
|
||||
let sample = harness.sample(44);
|
||||
|
||||
assert_eq!(sample.live_bytes, live_bytes);
|
||||
assert_eq!(
|
||||
sample.free_bytes,
|
||||
arena_config().reservation_ceiling - live_bytes
|
||||
);
|
||||
assert_eq!(sample.active_leases, 1);
|
||||
assert_eq!(sample.pending_leases, 1);
|
||||
assert_eq!(sample.allocation_failures_total, 0);
|
||||
assert_eq!(sample.release_failures_total, 0);
|
||||
}
|
||||
|
||||
// This proves lease rejection increments the allocation failure counter without
|
||||
// changing the arena's free capacity.
|
||||
#[test]
|
||||
fn sample_counts_rejected_leases_as_allocation_failures() {
|
||||
let mut harness = new_arena();
|
||||
|
||||
harness.request(arena::ArenaRequest::LeaseRing(lease_request(1, 8192, 64)));
|
||||
let sample = harness.sample(45);
|
||||
|
||||
assert_eq!(sample.live_bytes, 0);
|
||||
assert_eq!(sample.free_bytes, arena_config().reservation_ceiling);
|
||||
assert_eq!(sample.active_leases, 0);
|
||||
assert_eq!(sample.pending_leases, 0);
|
||||
assert_eq!(
|
||||
sample.largest_free_range_bytes,
|
||||
arena_config().reservation_ceiling
|
||||
);
|
||||
assert_eq!(sample.allocation_failures_total, 1);
|
||||
assert_eq!(sample.release_failures_total, 0);
|
||||
}
|
||||
|
||||
// This proves release rejection increments the release failure counter and keeps
|
||||
// the live lease accounted as active.
|
||||
#[test]
|
||||
fn sample_counts_rejected_releases_as_release_failures() {
|
||||
let mut harness = new_arena();
|
||||
harness.request(arena::ArenaRequest::LeaseRing(lease_request(1, 512, 64)));
|
||||
let lease = harness.live_leases()[0].clone();
|
||||
let lease_bytes = lease.layout.end_offset - lease.layout.start_offset;
|
||||
|
||||
harness.request(arena::ArenaRequest::ReleaseRing {
|
||||
ring_id: lease.ring_id,
|
||||
proof: arena::QuiescenceProof::missing(),
|
||||
});
|
||||
let sample = harness.sample(46);
|
||||
|
||||
assert_eq!(sample.live_bytes, lease_bytes);
|
||||
assert_eq!(
|
||||
sample.free_bytes,
|
||||
arena_config().reservation_ceiling - lease_bytes
|
||||
);
|
||||
assert_eq!(sample.active_leases, 1);
|
||||
assert_eq!(sample.pending_leases, 0);
|
||||
assert_eq!(sample.allocation_failures_total, 0);
|
||||
assert_eq!(sample.release_failures_total, 1);
|
||||
}
|
||||
|
||||
// This proves arena boot creates one stable sparse arena with one reservation
|
||||
// ceiling, offset-only layouts, and typed boot failure.
|
||||
#[test]
|
||||
fn arena_boot_creates_stable_offset_only_layout_domain() {
|
||||
// Boot a valid arena.
|
||||
let mut harness = new_arena();
|
||||
|
||||
// Lease one ring so the public layout can be inspected.
|
||||
harness.request(arena::ArenaRequest::LeaseRing(lease_request(1, 512, 64)));
|
||||
let lease = harness
|
||||
.events()
|
||||
.iter()
|
||||
.find_map(|event| match event {
|
||||
arena::ArenaEvent::RingLeased { lease } => Some(lease),
|
||||
_ => None,
|
||||
})
|
||||
.expect("valid lease must be granted");
|
||||
|
||||
// Layout facts are arena offsets and stay under the reservation ceiling.
|
||||
assert!(lease.layout.start_offset < arena_config().reservation_ceiling);
|
||||
assert!(lease.layout.end_offset <= arena_config().reservation_ceiling);
|
||||
assert!(matches!(
|
||||
lease.layout.pointer,
|
||||
arena::LayoutPointer::NoProcessPointer
|
||||
));
|
||||
|
||||
// Boot failure is typed and emits no usable arena.
|
||||
let failed = arena::ArenaManagerHarness::boot(arena::ArenaConfig {
|
||||
reservation_ceiling: 0,
|
||||
..arena_config()
|
||||
});
|
||||
assert!(matches!(
|
||||
failed,
|
||||
Err(arena::ArenaFault::InvalidReservationCeiling)
|
||||
));
|
||||
}
|
||||
|
||||
// This proves LeaseRing either leases, queues, or rejects, and that live RingId
|
||||
// values are unique for the node lifetime.
|
||||
#[test]
|
||||
fn lease_requests_grant_queue_or_reject_with_unique_ring_ids() {
|
||||
// Fill most of the arena with one live lease.
|
||||
let mut harness = new_arena();
|
||||
harness.request(arena::ArenaRequest::LeaseRing(lease_request(1, 3072, 64)));
|
||||
harness.request(arena::ArenaRequest::LeaseRing(lease_request(2, 1024, 64)));
|
||||
|
||||
// A satisfiable request under pressure may queue instead of rejecting.
|
||||
assert!(harness.events().iter().any(|event| {
|
||||
matches!(
|
||||
event,
|
||||
arena::ArenaEvent::RingLeaseQueued {
|
||||
request_id: arena::LeaseRequestId(2)
|
||||
}
|
||||
)
|
||||
}));
|
||||
|
||||
// A request larger than the reservation ceiling must reject.
|
||||
harness.request(arena::ArenaRequest::LeaseRing(lease_request(3, 8192, 64)));
|
||||
assert!(harness.events().iter().any(|event| {
|
||||
matches!(
|
||||
event,
|
||||
arena::ArenaEvent::RingLeaseRejected {
|
||||
request_id: arena::LeaseRequestId(3),
|
||||
reason: arena::RingLeaseRejection::CannotFitWithinCeiling,
|
||||
}
|
||||
)
|
||||
}));
|
||||
|
||||
// Granted RingId values must be unique among all live leases.
|
||||
let ids = harness
|
||||
.live_leases()
|
||||
.iter()
|
||||
.map(|lease| lease.ring_id)
|
||||
.collect::<std::collections::BTreeSet<_>>();
|
||||
assert_eq!(ids.len(), harness.live_leases().len());
|
||||
}
|
||||
|
||||
// This proves live leases are non-overlapping, in-bounds, aligned, and stable
|
||||
// for the lease lifetime.
|
||||
#[test]
|
||||
fn live_layouts_are_non_overlapping_aligned_in_bounds_and_stable() {
|
||||
// Lease two rings with explicit alignment requirements.
|
||||
let mut harness = new_arena();
|
||||
harness.request(arena::ArenaRequest::LeaseRing(lease_request(1, 512, 64)));
|
||||
harness.request(arena::ArenaRequest::LeaseRing(lease_request(2, 512, 128)));
|
||||
|
||||
// Read the public lease snapshots.
|
||||
let leases = harness.live_leases().to_vec();
|
||||
assert_eq!(leases.len(), 2);
|
||||
|
||||
// Prove non-overlap and in-bounds without constraining allocator placement.
|
||||
assert_non_overlapping(&leases[0].layout, &leases[1].layout);
|
||||
for lease in &leases {
|
||||
assert!(lease.layout.end_offset <= arena_config().reservation_ceiling);
|
||||
assert_eq!(lease.layout.start_offset % lease.requested_alignment, 0);
|
||||
}
|
||||
|
||||
// Re-observing the same live lease must not change offsets.
|
||||
let before = leases[0].layout.clone();
|
||||
let after = harness
|
||||
.lookup_lease(leases[0].ring_id)
|
||||
.expect("live lease must be lookupable")
|
||||
.layout
|
||||
.clone();
|
||||
assert_eq!(after, before);
|
||||
}
|
||||
|
||||
// This proves CancelLease removes queued work and suppresses later hot-path
|
||||
// installation, including a fresh lease that races with cancellation.
|
||||
#[test]
|
||||
fn cancelled_queued_lease_never_installs_hot_path_state() {
|
||||
// Fill the arena and queue a second satisfiable lease.
|
||||
let mut harness = new_arena();
|
||||
harness.request(arena::ArenaRequest::LeaseRing(lease_request(1, 3072, 64)));
|
||||
harness.request(arena::ArenaRequest::LeaseRing(lease_request(2, 1024, 64)));
|
||||
|
||||
// Cancel the queued request before it is leased.
|
||||
harness.request(arena::ArenaRequest::CancelLease {
|
||||
request_id: arena::LeaseRequestId(2),
|
||||
});
|
||||
|
||||
// Releasing pressure must not install worker or pump state for the canceled
|
||||
// request.
|
||||
let live = harness.live_leases()[0].ring_id;
|
||||
harness.request(arena::ArenaRequest::ReleaseRing {
|
||||
ring_id: live,
|
||||
proof: arena::QuiescenceProof::verified(),
|
||||
});
|
||||
assert!(!harness.commands().iter().any(|command| {
|
||||
matches!(
|
||||
command,
|
||||
arena::ArenaCommand::InstallWorkerOrPumpState {
|
||||
request_id: arena::LeaseRequestId(2),
|
||||
..
|
||||
}
|
||||
)
|
||||
}));
|
||||
|
||||
// If a fresh lease was produced during the race, it must be released instead
|
||||
// of becoming hot-path state.
|
||||
assert!(harness.events().iter().any(|event| {
|
||||
matches!(
|
||||
event,
|
||||
arena::ArenaEvent::CancelledFreshLeaseReleased {
|
||||
request_id: arena::LeaseRequestId(2)
|
||||
}
|
||||
)
|
||||
}));
|
||||
}
|
||||
|
||||
// This proves ranges are released only with quiescence proof, are not reused
|
||||
// while live work owns them, and may be reused after release.
|
||||
#[test]
|
||||
fn release_requires_quiescence_and_reuse_happens_only_after_release() {
|
||||
// Lease one ring and record its range.
|
||||
let mut harness = new_arena();
|
||||
harness.request(arena::ArenaRequest::LeaseRing(lease_request(1, 1024, 64)));
|
||||
let first = harness.live_leases()[0].clone();
|
||||
|
||||
// Release without proof must reject and keep the range live.
|
||||
harness.request(arena::ArenaRequest::ReleaseRing {
|
||||
ring_id: first.ring_id,
|
||||
proof: arena::QuiescenceProof::missing(),
|
||||
});
|
||||
assert!(harness.lookup_lease(first.ring_id).is_some());
|
||||
|
||||
// A second lease while the first is live must not overlap the first range.
|
||||
harness.request(arena::ArenaRequest::LeaseRing(lease_request(2, 1024, 64)));
|
||||
let second = harness
|
||||
.live_leases()
|
||||
.iter()
|
||||
.find(|lease| lease.ring_id != first.ring_id)
|
||||
.expect("second live lease must exist")
|
||||
.clone();
|
||||
assert_non_overlapping(&first.layout, &second.layout);
|
||||
|
||||
// Verified quiescence allows release, after which reuse is legal.
|
||||
harness.request(arena::ArenaRequest::ReleaseRing {
|
||||
ring_id: first.ring_id,
|
||||
proof: arena::QuiescenceProof::verified(),
|
||||
});
|
||||
harness.request(arena::ArenaRequest::LeaseRing(lease_request(3, 1024, 64)));
|
||||
assert!(
|
||||
harness
|
||||
.events()
|
||||
.iter()
|
||||
.any(|event| { matches!(event, arena::ArenaEvent::RingLeased { .. }) })
|
||||
);
|
||||
}
|
||||
|
||||
// This proves shutdown rejects new leases, preserves live lease records, and
|
||||
// does not release live ranges without quiescence proof.
|
||||
#[test]
|
||||
fn shutdown_rejects_new_leases_without_corrupting_live_records() {
|
||||
// Create a live lease before shutdown.
|
||||
let mut harness = new_arena();
|
||||
harness.request(arena::ArenaRequest::LeaseRing(lease_request(1, 512, 64)));
|
||||
let live_before = harness.live_leases().to_vec();
|
||||
|
||||
// Shut the arena down.
|
||||
harness.request(arena::ArenaRequest::Shutdown);
|
||||
|
||||
// New leases are rejected after shutdown.
|
||||
harness.request(arena::ArenaRequest::LeaseRing(lease_request(2, 512, 64)));
|
||||
assert!(harness.events().iter().any(|event| {
|
||||
matches!(
|
||||
event,
|
||||
arena::ArenaEvent::RingLeaseRejected {
|
||||
request_id: arena::LeaseRequestId(2),
|
||||
reason: arena::RingLeaseRejection::ArenaShuttingDown,
|
||||
}
|
||||
)
|
||||
}));
|
||||
|
||||
// Existing live lease records remain intact until proof-backed release.
|
||||
assert_eq!(harness.live_leases(), live_before.as_slice());
|
||||
}
|
||||
96
crates/mvp-system/src/tests/data_plane_bridge_guarantees.rs
Normal file
96
crates/mvp-system/src/tests/data_plane_bridge_guarantees.rs
Normal file
|
|
@ -0,0 +1,96 @@
|
|||
use data_plane::actor as dp;
|
||||
use mvp_system::node::actor::NodeAgentMsg;
|
||||
use mvp_system::node::data_plane_bridge::MvpDataPlaneReportSinkActor;
|
||||
use swactor::config::RuntimeConfig;
|
||||
use swactor::runtime::Runtime;
|
||||
|
||||
fn spawn_bridge(
|
||||
runtime: &Runtime,
|
||||
) -> (
|
||||
swactor::actor::ActorAddress,
|
||||
swactor::runtime::Inbox<NodeAgentMsg>,
|
||||
) {
|
||||
let node_messages = runtime
|
||||
.new_inbox::<NodeAgentMsg>()
|
||||
.expect("node-agent inbox");
|
||||
let bridge = runtime
|
||||
.spawn(MvpDataPlaneReportSinkActor::new(*node_messages.addr()))
|
||||
.expect("spawn bridge");
|
||||
(bridge, node_messages)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn data_plane_reports_drive_mvp_stage_readiness_and_object_visibility() {
|
||||
let runtime = Runtime::new(RuntimeConfig::default());
|
||||
let (bridge, node_messages) = spawn_bridge(&runtime);
|
||||
|
||||
runtime
|
||||
.send_to(
|
||||
bridge,
|
||||
dp::DataPlaneReportMsg::InboundEdgeReady {
|
||||
edge_id: dp::EdgeId(7001),
|
||||
},
|
||||
)
|
||||
.expect("send inbound ready");
|
||||
runtime
|
||||
.send_to(
|
||||
bridge,
|
||||
dp::DataPlaneReportMsg::ObjectLoaded {
|
||||
edge_id: dp::EdgeId(7001),
|
||||
object_id: data_plane::object_record::ObjectId(9000),
|
||||
sequence: 3,
|
||||
extent: 8,
|
||||
handle: dp::DeviceHandle::new(dp::WorkerGeneration(2), 44),
|
||||
},
|
||||
)
|
||||
.expect("send object loaded");
|
||||
runtime.tick();
|
||||
|
||||
assert_eq!(
|
||||
node_messages.try_recv(),
|
||||
Some(NodeAgentMsg::MarkInboundEdgeReady { edge_id: 7001 })
|
||||
);
|
||||
assert_eq!(
|
||||
node_messages.try_recv(),
|
||||
Some(NodeAgentMsg::ObjectLoaded {
|
||||
edge_id: 7001,
|
||||
object_id: 9000,
|
||||
sequence: 3,
|
||||
handle_generation: 2,
|
||||
handle_id: 44,
|
||||
})
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn data_plane_fault_and_stop_reports_map_to_mvp_lifecycle_messages() {
|
||||
let runtime = Runtime::new(RuntimeConfig::default());
|
||||
let (bridge, node_messages) = spawn_bridge(&runtime);
|
||||
|
||||
runtime
|
||||
.send_to(
|
||||
bridge,
|
||||
dp::DataPlaneReportMsg::EdgeFaulted {
|
||||
edge_id: dp::EdgeId(7001),
|
||||
reason: dp::EdgeFaultReason::StreamFault(
|
||||
data_plane::edge_lifecycle::StreamFaultReason::ProtocolError,
|
||||
),
|
||||
},
|
||||
)
|
||||
.expect("send edge fault");
|
||||
runtime
|
||||
.send_to(
|
||||
bridge,
|
||||
dp::DataPlaneReportMsg::LocalEdgesStopped {
|
||||
run_id: dp::RunId(55),
|
||||
},
|
||||
)
|
||||
.expect("send stopped");
|
||||
runtime.tick();
|
||||
|
||||
assert_eq!(node_messages.try_recv(), Some(NodeAgentMsg::WorkerCrashed));
|
||||
assert_eq!(
|
||||
node_messages.try_recv(),
|
||||
Some(NodeAgentMsg::LocalEdgesStopped { run_id: 55 })
|
||||
);
|
||||
}
|
||||
|
|
@ -9,7 +9,7 @@
|
|||
//! They assert the guarantees in
|
||||
//! `specs/mvp_system/gpu_worker_ingress_parser_contract.md`.
|
||||
|
||||
use mvp_system::node_data::ingress;
|
||||
use data_plane::ingress;
|
||||
|
||||
// A valid ingress ring config supplies the edge object spec and current worker
|
||||
// generation. The parser remains a black box behind ring helper operations.
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
use std::collections::{BTreeMap, BTreeSet};
|
||||
|
||||
use mvp_system::node_data::edge_actor;
|
||||
use data_plane::edge_actor;
|
||||
use mvp_system::observability::lifecycle as obs;
|
||||
use mvp_system::orchestration::engine_builder as engine;
|
||||
use mvp_system::orchestration::run_fsm as fsm;
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
use mvp_system::node_data::edge_actor;
|
||||
use data_plane::edge_actor;
|
||||
use mvp_system::orchestration::run_plan as plan;
|
||||
use mvp_system::staging as stage;
|
||||
|
||||
|
|
|
|||
|
|
@ -1,11 +1,9 @@
|
|||
mod arena_manager_guarantees;
|
||||
mod bootstrap_datastream_guarantees;
|
||||
mod data_plane_bridge_guarantees;
|
||||
mod device_bridge_guarantees;
|
||||
mod docker_cluster_provisioning_guarantees;
|
||||
mod edge_establisher_guarantees;
|
||||
mod engine_builder_guarantees;
|
||||
mod gpu_worker_ctl_guarantees;
|
||||
mod gpu_worker_egress_producer_guarantees;
|
||||
mod gpu_worker_ingress_parser_guarantees;
|
||||
mod gpu_worker_process_adapter_guarantees;
|
||||
mod local_mock;
|
||||
|
|
|
|||
|
|
@ -3,9 +3,8 @@
|
|||
//! These tests assert only the observable public-path contract: target modules
|
||||
//! expose existing behavior without introducing copied type definitions.
|
||||
|
||||
use mvp_system::{
|
||||
chat, node, node_data, observability, orchestration, prompt, staging, transport, worker,
|
||||
};
|
||||
use data_plane::{arena as data_arena, edge_lifecycle as data_edge};
|
||||
use mvp_system::{chat, node, observability, orchestration, prompt, staging, transport, worker};
|
||||
|
||||
#[test]
|
||||
fn target_modules_offer_new_paths_to_existing_public_contracts() {
|
||||
|
|
@ -15,14 +14,14 @@ fn target_modules_offer_new_paths_to_existing_public_contracts() {
|
|||
let boot_node_id: node::boot_lifecycle::NodeId = node::boot_lifecycle::NodeId(11);
|
||||
assert_eq!(boot_node_id, node::boot_lifecycle::NodeId(11));
|
||||
|
||||
let arena_ring_id: node_data::arena::RingId = node_data::arena::RingId(3);
|
||||
assert_eq!(arena_ring_id, node_data::arena::RingId(3));
|
||||
let arena_ring_id: data_arena::RingId = data_arena::RingId(3);
|
||||
assert_eq!(arena_ring_id, data_arena::RingId(3));
|
||||
|
||||
let stage_edge_id: staging::EdgeId = staging::EdgeId(7001);
|
||||
assert_eq!(stage_edge_id, staging::EdgeId(7001));
|
||||
|
||||
let transport_edge_id: node::edge_lifecycle::EdgeId = node::edge_lifecycle::EdgeId(7002);
|
||||
assert_eq!(transport_edge_id, node::edge_lifecycle::EdgeId(7002));
|
||||
let transport_edge_id: data_edge::EdgeId = data_edge::EdgeId(7002);
|
||||
assert_eq!(transport_edge_id, data_edge::EdgeId(7002));
|
||||
|
||||
let worker_generation: worker::WorkerGeneration = worker::WorkerGeneration(2);
|
||||
assert_eq!(worker_generation, worker::WorkerGeneration(2));
|
||||
|
|
|
|||
|
|
@ -9,7 +9,7 @@
|
|||
//! They assert the guarantees in
|
||||
//! `specs/mvp_system/shared_ring_helper_abi_contract.md`.
|
||||
|
||||
use mvp_system::node_data::ring;
|
||||
use data_plane::ring;
|
||||
|
||||
// A small ring forces wraparound and full/empty transitions quickly. The helper
|
||||
// still owns the actual shared-memory atomics and process-local address math.
|
||||
|
|
|
|||
|
|
@ -8,7 +8,7 @@
|
|||
//! They assert the guarantees in
|
||||
//! `specs/mvp_system/tx_rx_edge_actor_contract.md`.
|
||||
|
||||
use mvp_system::node_data::edge_actor;
|
||||
use data_plane::edge_actor;
|
||||
|
||||
// The edge id fixture gives both actors a shared identity while keeping Tx and
|
||||
// Rx lifecycle tests independent from driver and ring internals.
|
||||
|
|
|
|||
|
|
@ -1,9 +1,9 @@
|
|||
use data_plane::object_record as ingress;
|
||||
use mvp_system::node::actor::{
|
||||
NodeAgentActor, NodeAgentMsg, NodeAgentReport, StageCommandWire, StageEdgeKindWire,
|
||||
StageInboundEdgeWire, StageObjectSpecWire, StageOutboundEdgeWire, StageProvisionWire,
|
||||
StageRingSpecWire,
|
||||
};
|
||||
use mvp_system::node_data::object as ingress;
|
||||
use mvp_system::orchestration::run_plan::{self, GgufSource, TokenizerSource};
|
||||
use mvp_system::staging as stage;
|
||||
use swactor::actor::ActorAddress;
|
||||
|
|
|
|||
|
|
@ -2,7 +2,6 @@
|
|||
|
||||
pub mod control;
|
||||
pub mod device_bridge;
|
||||
pub mod egress;
|
||||
pub mod process_adapter;
|
||||
|
||||
pub mod process {
|
||||
|
|
|
|||
Loading…
Reference in a new issue