use core::convert::Infallible;
use core::fmt::{Display, Formatter, Result as FormatResult};
use core::ops::Range;
use pipecircuit::{
Circuit, CircuitFlowOutcome, CircuitReactionOutcome, CircuitSendOutcome, DuplexTransport,
Inline, InstalledPipe, Pipe, Pipecircuit, Reactor, ReceiveFromOutcome, StateMachine,
StepInputOf, Switchboard, TransportFrom, TransportTo, flow,
};
const EXTENT_COUNT: usize = 2;
const EXTENT_BYTE_COUNT: usize = 8;
const OBSERVATION_CAPACITY: usize = 3;
const PHYSICAL_PAYLOADS: [&[u8]; 3] = [b"echo", b"more", b"again"];
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
struct ArenaLease(usize);
struct DuplexEndpoint;
impl Pipe for DuplexEndpoint {
type In<'message> = SendPayload<'message>;
type Out<'message> = ReceivedPayload<'message>;
}
struct ReceivedPayload<'message> {
lease: ArenaLease,
payload: &'message mut [u8],
}
struct SendPayload<'message> {
lease: ArenaLease,
payload: &'message mut [u8],
}
struct EchoPayload<'message>(ReceivedPayload<'message>);
struct EchoedPayload<'message>(SendPayload<'message>);
struct EchoMachine {
step_count: usize,
observed_addresses: [usize; OBSERVATION_CAPACITY],
}
impl StateMachine for EchoMachine {}
impl Pipe for EchoMachine {
type In<'message> = EchoPayload<'message>;
type Out<'message> = EchoedPayload<'message>;
}
impl<'message> StepInputOf<EchoMachine> for EchoPayload<'message> {
type Outcome = EchoedPayload<'message>;
fn step(self, machine: &mut EchoMachine) -> Self::Outcome {
let ReceivedPayload { lease, payload } = self.0;
if let Some(address) = machine.observed_addresses.get_mut(machine.step_count) {
*address = payload.as_ptr() as usize;
}
machine.step_count = machine.step_count.saturating_add(1);
if let Some(first) = payload.first_mut() {
first.make_ascii_uppercase();
}
EchoedPayload(SendPayload { lease, payload })
}
}
enum FromProgram<'message> {
Endpoint(ReceivedPayload<'message>),
Echo(EchoedPayload<'message>),
}
enum ToProgram<'message> {
Echo(EchoPayload<'message>),
Endpoint(SendPayload<'message>),
}
struct EchoSwitchboard;
impl Switchboard for EchoSwitchboard {
type From<'message> = FromProgram<'message>;
type To<'message> = ToProgram<'message>;
fn route<'message>(&mut self, from: Self::From<'message>) -> Self::To<'message> {
match from {
FromProgram::Endpoint(payload) => ToProgram::Echo(EchoPayload(payload)),
FromProgram::Echo(EchoedPayload(payload)) => ToProgram::Endpoint(payload),
}
}
}
#[derive(Clone, Copy)]
enum PhysicalReaction {
Readable,
Writable,
}
struct SequenceReactor {
reactions: [PhysicalReaction; 2],
next: usize,
wait_count: usize,
}
impl Reactor for SequenceReactor {
type Reaction = PhysicalReaction;
type Failure = Infallible;
fn wait_for_reaction(&mut self) -> Result<Self::Reaction, Self::Failure> {
self.wait_count = self.wait_count.saturating_add(1);
let reaction = self
.reactions
.get(self.next)
.copied()
.unwrap_or(PhysicalReaction::Writable);
self.next = self.next.saturating_add(1);
Ok(reaction)
}
}
struct IncomingState {
payload_count: usize,
next_payload: usize,
active_leases: [bool; EXTENT_COUNT],
receive_count: usize,
recycle_count: usize,
}
struct OutgoingState {
pending: [Option<PendingWrite>; EXTENT_COUNT],
pending_start: usize,
pending_count: usize,
immediate_write_byte_limit: usize,
submission_count: usize,
resume_count: usize,
submitted_addresses: [usize; OBSERVATION_CAPACITY],
resumed_addresses: [usize; OBSERVATION_CAPACITY],
resumed_leases: [Option<ArenaLease>; OBSERVATION_CAPACITY],
observed_byte_sum: u64,
}
#[derive(Clone)]
struct PendingWrite {
lease: ArenaLease,
remaining: Range<usize>,
}
enum PendingWriteRetentionOutcome {
Retained,
Full,
}
impl OutgoingState {
fn retain(&mut self, pending: PendingWrite) -> PendingWriteRetentionOutcome {
if self.pending_count == self.pending.len() {
return PendingWriteRetentionOutcome::Full;
}
let Some(position) = self
.pending_start
.checked_add(self.pending_count)
.and_then(|position| position.checked_rem(self.pending.len()))
else {
return PendingWriteRetentionOutcome::Full;
};
let Some(destination) = self.pending.get_mut(position) else {
return PendingWriteRetentionOutcome::Full;
};
*destination = Some(pending);
self.pending_count = self.pending_count.saturating_add(1);
PendingWriteRetentionOutcome::Retained
}
fn take_first(&mut self) -> Option<PendingWrite> {
if self.pending_count == 0 {
return None;
}
let pending = self.pending.get_mut(self.pending_start)?.take()?;
self.pending_start = self
.pending_start
.checked_add(1)
.and_then(|position| position.checked_rem(self.pending.len()))?;
self.pending_count = self.pending_count.saturating_sub(1);
Some(pending)
}
fn first(&self) -> Option<&PendingWrite> {
self.pending.get(self.pending_start)?.as_ref()
}
fn observe_submission(&mut self, address: usize) {
if let Some(observed) = self.submitted_addresses.get_mut(self.submission_count) {
*observed = address;
}
self.submission_count = self.submission_count.saturating_add(1);
}
fn observe_resume(&mut self, lease: ArenaLease, address: usize) {
if let Some(observed) = self.resumed_addresses.get_mut(self.resume_count) {
*observed = address;
}
if let Some(observed) = self.resumed_leases.get_mut(self.resume_count) {
*observed = Some(lease);
}
self.resume_count = self.resume_count.saturating_add(1);
}
}
struct ZeroCopyDuplexFitting {
arena: [[u8; EXTENT_BYTE_COUNT]; EXTENT_COUNT],
incoming: IncomingState,
outgoing: OutgoingState,
}
struct IncomingDirection<'fitting> {
state: &'fitting mut IncomingState,
available_extent: Option<(ArenaLease, &'fitting mut [u8; EXTENT_BYTE_COUNT])>,
}
struct OutgoingDirection<'fitting> {
state: &'fitting mut OutgoingState,
retained_before_available: &'fitting [[u8; EXTENT_BYTE_COUNT]],
retained_after_available: &'fitting [[u8; EXTENT_BYTE_COUNT]],
available_extent_index: Option<usize>,
extent_addresses: [usize; EXTENT_COUNT],
}
impl DuplexTransport<DuplexEndpoint> for ZeroCopyDuplexFitting {
type Incoming<'fitting> = IncomingDirection<'fitting>;
type Outgoing<'fitting> = OutgoingDirection<'fitting>;
fn split<'fitting>(&'fitting mut self) -> (Self::Incoming<'fitting>, Self::Outgoing<'fitting>) {
let Self {
arena,
incoming,
outgoing,
} = self;
let extent_addresses = core::array::from_fn(|index| {
arena
.get(index)
.map_or(0, |extent| extent.as_ptr() as usize)
});
let available_extent_index = incoming.active_leases.iter().position(|active| !active);
match available_extent_index {
Some(index) => {
let (before, selected_and_after) = arena.split_at_mut(index);
let Some((available, after)) = selected_and_after.split_first_mut() else {
return (
IncomingDirection {
state: incoming,
available_extent: None,
},
OutgoingDirection {
state: outgoing,
retained_before_available: before,
retained_after_available: &[],
available_extent_index: None,
extent_addresses,
},
);
};
(
IncomingDirection {
state: incoming,
available_extent: Some((ArenaLease(index), available)),
},
OutgoingDirection {
state: outgoing,
retained_before_available: before,
retained_after_available: after,
available_extent_index: Some(index),
extent_addresses,
},
)
}
None => (
IncomingDirection {
state: incoming,
available_extent: None,
},
OutgoingDirection {
state: outgoing,
retained_before_available: arena,
retained_after_available: &[],
available_extent_index: None,
extent_addresses,
},
),
}
}
}
impl TransportFrom<DuplexEndpoint> for IncomingDirection<'_> {
type IncomingMessage<'message>
= PhysicalReaction
where
Self: 'message;
type Failure = Infallible;
fn receive_from<'message>(
&'message mut self,
reaction: Self::IncomingMessage<'message>,
) -> ReceiveFromOutcome<ReceivedPayload<'message>, Self::IncomingMessage<'message>, Self::Failure>
{
if !matches!(reaction, PhysicalReaction::Readable)
|| self.state.next_payload == self.state.payload_count
{
return ReceiveFromOutcome::Quiescent;
}
let Some((lease, extent)) = self.available_extent.as_mut() else {
return ReceiveFromOutcome::Backpressured { rejected: reaction };
};
let Some(active) = self.state.active_leases.get_mut(lease.0) else {
return ReceiveFromOutcome::Backpressured { rejected: reaction };
};
if *active {
return ReceiveFromOutcome::Backpressured { rejected: reaction };
}
let Some(source) = PHYSICAL_PAYLOADS.get(self.state.next_payload).copied() else {
return ReceiveFromOutcome::Quiescent;
};
let Some(payload) = extent.get_mut(..source.len()) else {
return ReceiveFromOutcome::Quiescent;
};
payload.copy_from_slice(source);
*active = true;
self.state.next_payload = self.state.next_payload.saturating_add(1);
self.state.receive_count = self.state.receive_count.saturating_add(1);
ReceiveFromOutcome::Received {
output: ReceivedPayload {
lease: *lease,
payload,
},
}
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum RecycleLeaseOutcome {
Recycled,
Absent,
}
impl IncomingDirection<'_> {
fn recycle(&mut self, lease: ArenaLease) -> RecycleLeaseOutcome {
let Some(active) = self.state.active_leases.get_mut(lease.0) else {
return RecycleLeaseOutcome::Absent;
};
if !*active {
return RecycleLeaseOutcome::Absent;
}
*active = false;
self.state.recycle_count = self.state.recycle_count.saturating_add(1);
RecycleLeaseOutcome::Recycled
}
}
enum SubmissionOutcome<'message> {
Written { lease: ArenaLease },
Retained,
Backpressured { rejected: SendPayload<'message> },
ForeignPayload { rejected: SendPayload<'message> },
}
impl OutgoingDirection<'_> {
fn payload_range(&self, input: &SendPayload<'_>) -> Option<Range<usize>> {
let extent_start = *self.extent_addresses.get(input.lease.0)?;
let payload_start = input.payload.as_ptr() as usize;
let payload_offset = payload_start.checked_sub(extent_start)?;
let payload_end = payload_offset.checked_add(input.payload.len())?;
if payload_end > EXTENT_BYTE_COUNT {
return None;
}
Some(payload_offset..payload_end)
}
fn retained_extent(&self, lease: ArenaLease) -> Option<&[u8; EXTENT_BYTE_COUNT]> {
match self.available_extent_index {
Some(available) if lease.0 == available => None,
Some(available) if lease.0 < available => self.retained_before_available.get(lease.0),
Some(available) => self
.retained_after_available
.get(lease.0.saturating_sub(available).saturating_sub(1)),
None => self.retained_before_available.get(lease.0),
}
}
fn pending_count(&self) -> usize {
self.state.pending_count
}
}
impl TransportTo<DuplexEndpoint> for OutgoingDirection<'_> {
type Outcome<'transport, 'message>
= SubmissionOutcome<'message>
where
Self: 'transport;
fn send<'transport, 'message>(
&'transport mut self,
input: <DuplexEndpoint as Pipe>::In<'message>,
) -> Self::Outcome<'transport, 'message> {
let Some(payload_range) = self.payload_range(&input) else {
return SubmissionOutcome::ForeignPayload { rejected: input };
};
if self.state.pending_count == self.state.pending.len() {
return SubmissionOutcome::Backpressured { rejected: input };
}
self.state
.observe_submission(input.payload.as_ptr() as usize);
if self.state.pending_count != 0 {
return match self.state.retain(PendingWrite {
lease: input.lease,
remaining: payload_range,
}) {
PendingWriteRetentionOutcome::Retained => SubmissionOutcome::Retained,
PendingWriteRetentionOutcome::Full => {
SubmissionOutcome::Backpressured { rejected: input }
}
};
}
let immediately_written = input
.payload
.len()
.min(self.state.immediate_write_byte_limit);
let Some(immediate_payload) = input.payload.get(..immediately_written) else {
return SubmissionOutcome::ForeignPayload { rejected: input };
};
self.state.observed_byte_sum = self
.state
.observed_byte_sum
.saturating_add(byte_sum(immediate_payload));
if immediately_written == input.payload.len() {
return SubmissionOutcome::Written { lease: input.lease };
}
let remaining = payload_range.start.saturating_add(immediately_written)..payload_range.end;
match self.state.retain(PendingWrite {
lease: input.lease,
remaining,
}) {
PendingWriteRetentionOutcome::Retained => SubmissionOutcome::Retained,
PendingWriteRetentionOutcome::Full => {
SubmissionOutcome::Backpressured { rejected: input }
}
}
}
}
fn byte_sum(bytes: &[u8]) -> u64 {
bytes
.iter()
.fold(0_u64, |sum, byte| sum.saturating_add(u64::from(*byte)))
}
enum ResumeWriteOutcome {
Completed {
lease: ArenaLease,
},
Idle,
ArenaUnavailable,
RangeUnavailable {
remaining: Range<usize>,
},
QueueUnavailable {
lease: ArenaLease,
remaining: Range<usize>,
},
}
impl OutgoingDirection<'_> {
fn resume(&mut self) -> ResumeWriteOutcome {
let Some(pending) = self.state.first().cloned() else {
return ResumeWriteOutcome::Idle;
};
let Some(extent) = self.retained_extent(pending.lease) else {
return ResumeWriteOutcome::ArenaUnavailable;
};
let Some(remaining) = extent.get(pending.remaining.clone()) else {
return ResumeWriteOutcome::RangeUnavailable {
remaining: pending.remaining,
};
};
let resumed_address = remaining.as_ptr() as usize;
let resumed_byte_sum = byte_sum(remaining);
let Some(completed) = self.state.take_first() else {
return ResumeWriteOutcome::QueueUnavailable {
lease: pending.lease,
remaining: pending.remaining,
};
};
self.state.observe_resume(completed.lease, resumed_address);
self.state.observed_byte_sum = self
.state
.observed_byte_sum
.saturating_add(resumed_byte_sum);
ResumeWriteOutcome::Completed {
lease: completed.lease,
}
}
}
#[derive(Debug)]
enum ZeroCopyCircuitFailure {
Backpressured {
rejected_lease: ArenaLease,
rejected_byte_count: usize,
},
ForeignPayload {
rejected_lease: ArenaLease,
rejected_byte_count: usize,
},
LeaseNotRecycled,
PendingWriteMissing,
PendingArenaUnavailable,
PendingRangeUnavailable {
remaining: Range<usize>,
},
PendingQueueUnavailable {
lease: ArenaLease,
remaining: Range<usize>,
},
}
impl Display for ZeroCopyCircuitFailure {
fn fmt(&self, formatter: &mut Formatter<'_>) -> FormatResult {
match self {
Self::Backpressured {
rejected_lease,
rejected_byte_count,
} => write!(
formatter,
"the outgoing fitting rejected lease {} containing {rejected_byte_count} bytes",
rejected_lease.0
),
Self::ForeignPayload {
rejected_lease,
rejected_byte_count,
} => write!(
formatter,
"lease {} containing {rejected_byte_count} bytes was foreign to the fitting",
rejected_lease.0
),
Self::LeaseNotRecycled => formatter.write_str("the receive lease was not recycled"),
Self::PendingWriteMissing => formatter.write_str("the pending write was absent"),
Self::PendingArenaUnavailable => {
formatter.write_str("the pending write arena was unavailable")
}
Self::PendingRangeUnavailable { remaining } => write!(
formatter,
"the pending write range {}..{} was unavailable",
remaining.start, remaining.end
),
Self::PendingQueueUnavailable { lease, remaining } => write!(
formatter,
"lease {} with pending range {}..{} disappeared from the outgoing queue",
lease.0, remaining.start, remaining.end
),
}
}
}
struct ZeroCopyCircuit {
reactor: SequenceReactor,
endpoint_pipe: InstalledPipe<DuplexEndpoint, ZeroCopyDuplexFitting>,
echo_pipe: InstalledPipe<EchoMachine, Inline<EchoMachine>>,
}
fn zero_copy_fitting(
immediate_write_byte_limit: usize,
payload_count: usize,
) -> ZeroCopyDuplexFitting {
ZeroCopyDuplexFitting {
arena: [[0; EXTENT_BYTE_COUNT]; EXTENT_COUNT],
incoming: IncomingState {
payload_count,
next_payload: 0,
active_leases: [false; EXTENT_COUNT],
receive_count: 0,
recycle_count: 0,
},
outgoing: OutgoingState {
pending: [None, None],
pending_start: 0,
pending_count: 0,
immediate_write_byte_limit,
submission_count: 0,
resume_count: 0,
submitted_addresses: [0; OBSERVATION_CAPACITY],
resumed_addresses: [0; OBSERVATION_CAPACITY],
resumed_leases: [None; OBSERVATION_CAPACITY],
observed_byte_sum: 0,
},
}
}
fn echo_machine() -> EchoMachine {
EchoMachine {
step_count: 0,
observed_addresses: [0; OBSERVATION_CAPACITY],
}
}
impl ZeroCopyCircuit {
fn flow<'message>(
switchboard: &mut EchoSwitchboard,
from: FromProgram<'message>,
outgoing: &mut OutgoingDirection<'_>,
echo_pipe: &mut InstalledPipe<EchoMachine, Inline<EchoMachine>>,
released_lease: &mut Option<ArenaLease>,
) -> Result<CircuitFlowOutcome<()>, ZeroCopyCircuitFailure> {
flow(switchboard, from, |to| match to {
ToProgram::Echo(input) => Ok(CircuitSendOutcome::Immediate {
from: FromProgram::Echo(echo_pipe.send(input)),
}),
ToProgram::Endpoint(input) => match outgoing.send(input) {
SubmissionOutcome::Written { lease } => {
*released_lease = Some(lease);
Ok(CircuitSendOutcome::Submitted)
}
SubmissionOutcome::Retained => Ok(CircuitSendOutcome::Submitted),
SubmissionOutcome::Backpressured { rejected } => {
let failure = ZeroCopyCircuitFailure::Backpressured {
rejected_lease: rejected.lease,
rejected_byte_count: rejected.payload.len(),
};
*released_lease = Some(rejected.lease);
Err(failure)
}
SubmissionOutcome::ForeignPayload { rejected } => {
let failure = ZeroCopyCircuitFailure::ForeignPayload {
rejected_lease: rejected.lease,
rejected_byte_count: rejected.payload.len(),
};
*released_lease = Some(rejected.lease);
Err(failure)
}
},
})
}
}
impl Circuit<EchoSwitchboard> for ZeroCopyCircuit {
type Reactor = SequenceReactor;
type Completion = ();
type Failure = ZeroCopyCircuitFailure;
fn reactor(&mut self) -> &mut Self::Reactor {
&mut self.reactor
}
fn react(
&mut self,
switchboard: &mut EchoSwitchboard,
reaction: &PhysicalReaction,
) -> Result<CircuitReactionOutcome<Self::Completion>, Self::Failure> {
let Self {
reactor: _,
endpoint_pipe,
echo_pipe,
} = self;
let (mut incoming, mut outgoing) = endpoint_pipe.split();
match reaction {
PhysicalReaction::Readable => match incoming.receive_from(*reaction) {
ReceiveFromOutcome::Received { output } => {
let mut released_lease = None;
let outcome = Self::flow(
switchboard,
FromProgram::Endpoint(output),
&mut outgoing,
echo_pipe,
&mut released_lease,
);
if let Some(lease) = released_lease
&& incoming.recycle(lease) != RecycleLeaseOutcome::Recycled
{
return Err(ZeroCopyCircuitFailure::LeaseNotRecycled);
}
match outcome? {
CircuitFlowOutcome::Submitted => {
Ok(CircuitReactionOutcome::ContinueReaction)
}
CircuitFlowOutcome::Completed { completion } => {
Ok(CircuitReactionOutcome::Completed { completion })
}
}
}
ReceiveFromOutcome::Progressed => Ok(CircuitReactionOutcome::ContinueReaction),
ReceiveFromOutcome::Quiescent
| ReceiveFromOutcome::Backpressured { rejected: _ } => {
Ok(CircuitReactionOutcome::AwaitingReaction)
}
ReceiveFromOutcome::Failed { failure } => match failure {},
},
PhysicalReaction::Writable => match outgoing.resume() {
ResumeWriteOutcome::Completed { lease } => {
if incoming.recycle(lease) != RecycleLeaseOutcome::Recycled {
return Err(ZeroCopyCircuitFailure::LeaseNotRecycled);
}
if outgoing.pending_count() == 0 {
Ok(CircuitReactionOutcome::Completed { completion: () })
} else {
Ok(CircuitReactionOutcome::ContinueReaction)
}
}
ResumeWriteOutcome::Idle => Err(ZeroCopyCircuitFailure::PendingWriteMissing),
ResumeWriteOutcome::ArenaUnavailable => {
Err(ZeroCopyCircuitFailure::PendingArenaUnavailable)
}
ResumeWriteOutcome::RangeUnavailable { remaining } => {
Err(ZeroCopyCircuitFailure::PendingRangeUnavailable { remaining })
}
ResumeWriteOutcome::QueueUnavailable { lease, remaining } => {
Err(ZeroCopyCircuitFailure::PendingQueueUnavailable { lease, remaining })
}
},
}
}
}
#[test]
fn retained_output_extents_do_not_block_reads_into_free_extents() {
let circuit = ZeroCopyCircuit {
reactor: SequenceReactor {
reactions: [PhysicalReaction::Readable, PhysicalReaction::Writable],
next: 0,
wait_count: 0,
},
endpoint_pipe: InstalledPipe::new(zero_copy_fitting(2, 2)),
echo_pipe: InstalledPipe::new(Inline::new(echo_machine())),
};
let mut program = Pipecircuit::new(EchoSwitchboard, circuit);
assert!(program.conduct().is_ok());
let circuit = program.circuit();
let fitting = circuit.endpoint_pipe.transport();
let first_extent_address = fitting.arena[0].as_ptr() as usize;
let second_extent_address = fitting.arena[1].as_ptr() as usize;
let echo = circuit.echo_pipe.transport().destination();
assert_eq!(circuit.reactor.wait_count, 2);
assert_eq!(fitting.incoming.receive_count, 2);
assert_eq!(fitting.incoming.recycle_count, 2);
assert_eq!(fitting.incoming.active_leases, [false, false]);
assert_eq!(fitting.outgoing.submission_count, 2);
assert_eq!(fitting.outgoing.resume_count, 2);
assert_eq!(fitting.outgoing.pending_count, 0);
assert_eq!(echo.step_count, 2);
assert_eq!(
echo.observed_addresses.get(..2),
Some([first_extent_address, second_extent_address].as_slice())
);
assert_eq!(
fitting.outgoing.submitted_addresses.get(..2),
Some([first_extent_address, second_extent_address].as_slice())
);
assert_eq!(
fitting.outgoing.resumed_addresses.get(..2),
Some([first_extent_address + 2, second_extent_address].as_slice())
);
assert_eq!(
fitting.outgoing.resumed_leases.get(..2),
Some([Some(ArenaLease(0)), Some(ArenaLease(1))].as_slice())
);
assert_eq!(
fitting.outgoing.observed_byte_sum,
byte_sum(b"Echo").saturating_add(byte_sum(b"More"))
);
assert_eq!(fitting.arena[0].get(..4), Some(b"Echo".as_slice()));
assert_eq!(fitting.arena[1].get(..4), Some(b"More".as_slice()));
}
#[test]
fn completed_writes_recycle_the_mutable_arena_lease_in_the_same_turn() -> Result<(), &'static str> {
let mut endpoint_pipe =
InstalledPipe::<DuplexEndpoint, _>::new(zero_copy_fitting(usize::MAX, 1));
let mut echo_pipe = InstalledPipe::<EchoMachine, _>::new(Inline::new(echo_machine()));
let mut switchboard = EchoSwitchboard;
{
let (mut incoming, mut outgoing) = endpoint_pipe.split();
let received = match incoming.receive_from(PhysicalReaction::Readable) {
ReceiveFromOutcome::Received { output } => output,
ReceiveFromOutcome::Progressed => return Err("the readable fitting only progressed"),
ReceiveFromOutcome::Quiescent => return Err("the readable fitting was quiescent"),
ReceiveFromOutcome::Backpressured { rejected: _ } => {
return Err("the readable fitting was backpressured");
}
ReceiveFromOutcome::Failed { failure } => match failure {},
};
let ToProgram::Echo(input) = switchboard.route(FromProgram::Endpoint(received)) else {
return Err("the received payload did not route to the echo machine");
};
let echoed = echo_pipe.send(input);
let ToProgram::Endpoint(payload) = switchboard.route(FromProgram::Echo(echoed)) else {
return Err("the echoed payload did not route to the duplex endpoint");
};
let lease = match outgoing.send(payload) {
SubmissionOutcome::Written { lease } => lease,
SubmissionOutcome::Retained => return Err("the complete write was retained"),
SubmissionOutcome::Backpressured { .. } => {
return Err("the complete write was backpressured");
}
SubmissionOutcome::ForeignPayload { .. } => {
return Err("the arena payload was reported as foreign");
}
};
if incoming.recycle(lease) != RecycleLeaseOutcome::Recycled {
return Err("the completed write lease was not recycled");
}
}
let fitting = endpoint_pipe.transport();
let arena_address = fitting.arena[0].as_ptr() as usize;
assert_eq!(fitting.incoming.recycle_count, 1);
assert_eq!(fitting.incoming.active_leases, [false, false]);
assert_eq!(fitting.outgoing.pending_count, 0);
assert_eq!(fitting.outgoing.submitted_addresses[0], arena_address);
assert_eq!(fitting.outgoing.observed_byte_sum, byte_sum(b"Echo"));
assert_eq!(fitting.arena[0].get(..4), Some(b"Echo".as_slice()));
Ok(())
}
#[test]
fn exhausted_extent_capacity_is_explicit_and_recovered_by_output_progress()
-> Result<(), &'static str> {
let mut endpoint_pipe = InstalledPipe::<DuplexEndpoint, _>::new(zero_copy_fitting(0, 3));
let mut echo_pipe = InstalledPipe::<EchoMachine, _>::new(Inline::new(echo_machine()));
let mut switchboard = EchoSwitchboard;
for _ in 0..EXTENT_COUNT {
let (mut incoming, mut outgoing) = endpoint_pipe.split();
let received = match incoming.receive_from(PhysicalReaction::Readable) {
ReceiveFromOutcome::Received { output } => output,
ReceiveFromOutcome::Progressed => return Err("the readable fitting only progressed"),
ReceiveFromOutcome::Quiescent => return Err("the readable fitting was quiescent"),
ReceiveFromOutcome::Backpressured { rejected: _ } => {
return Err("a free extent was unexpectedly unavailable");
}
ReceiveFromOutcome::Failed { failure } => match failure {},
};
let ToProgram::Echo(input) = switchboard.route(FromProgram::Endpoint(received)) else {
return Err("the received payload did not route to the echo machine");
};
let echoed = echo_pipe.send(input);
let ToProgram::Endpoint(payload) = switchboard.route(FromProgram::Echo(echoed)) else {
return Err("the echoed payload did not route to the duplex endpoint");
};
if !matches!(outgoing.send(payload), SubmissionOutcome::Retained) {
return Err("the output extent was not retained");
}
}
{
let (mut incoming, mut outgoing) = endpoint_pipe.split();
if !matches!(
incoming.receive_from(PhysicalReaction::Readable),
ReceiveFromOutcome::Backpressured { .. }
) {
return Err("exhausted extent capacity was not reported as backpressure");
}
let ResumeWriteOutcome::Completed { lease } = outgoing.resume() else {
return Err("the oldest retained extent did not complete");
};
if incoming.recycle(lease) != RecycleLeaseOutcome::Recycled {
return Err("the completed retained extent was not recycled");
}
}
{
let (mut incoming, _outgoing) = endpoint_pipe.split();
let received = match incoming.receive_from(PhysicalReaction::Readable) {
ReceiveFromOutcome::Received { output } => output,
ReceiveFromOutcome::Progressed => {
return Err("the recovered receive fitting only progressed");
}
ReceiveFromOutcome::Quiescent => {
return Err("the recovered receive extent was quiescent");
}
ReceiveFromOutcome::Backpressured { rejected: _ } => {
return Err("output progress did not restore receive capacity");
}
ReceiveFromOutcome::Failed { failure } => match failure {},
};
if received.payload != b"again" {
return Err("the recovered receive extent contained the wrong payload");
}
let lease = received.lease;
let _ = received;
if incoming.recycle(lease) != RecycleLeaseOutcome::Recycled {
return Err("the recovered receive extent was not recycled");
}
}
let fitting = endpoint_pipe.transport();
assert_eq!(fitting.incoming.receive_count, 3);
assert_eq!(fitting.incoming.recycle_count, 2);
assert_eq!(fitting.incoming.next_payload, 3);
assert_eq!(fitting.outgoing.pending_count, 1);
Ok(())
}