#![cfg(feature = "storage")]
extern crate alloc;
use alloc::string::String;
use core::num::{NonZeroU32, NonZeroU64};
use pipecircuit::storage::warp_table::Capacity;
use pipecircuit::{
ContinuationAdmissionOutcome, ContinuationArmOutcome, ContinuationCancellationOutcome,
ContinuationFailure, ContinuationLedger, ContinuationResolutionOutcome, ContinuationResult,
Continued, InstalledPipe, Owed, Pipe, StateMachine, StepInputOf, Switchboard, TimeoutAfter,
TimerOut, TransportTo,
};
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
struct PlayerId(u64);
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum ProfileLookupOutcome {
Found { player_id: PlayerId },
Absent { player_id: PlayerId },
}
struct ProfileDatabase;
impl StateMachine for ProfileDatabase {}
impl Pipe for ProfileDatabase {
type In<'message> = PlayerId;
type Out<'message> = ProfileLookupOutcome;
}
impl StepInputOf<ProfileDatabase> for PlayerId {
type Outcome = ProfileLookupOutcome;
fn step(self, _database: &mut ProfileDatabase) -> Self::Outcome {
if self.0 == 0 {
ProfileLookupOutcome::Absent { player_id: self }
} else {
ProfileLookupOutcome::Found { player_id: self }
}
}
}
struct Player {
last_lookup: Option<ProfileLookupOutcome>,
timed_out: bool,
}
impl StateMachine for Player {}
struct LoadProfile {
player_id: PlayerId,
}
enum PlayerOutcome {
Owed(Owed<PlayerId>),
Recorded,
Unavailable,
}
impl StepInputOf<Player> for LoadProfile {
type Outcome = PlayerOutcome;
fn step(self, _player: &mut Player) -> Self::Outcome {
PlayerOutcome::Owed(Owed::new(self.player_id, profile_timeout()))
}
}
struct ProfileLookupFinished<'message> {
result: ContinuationResult<'message, ProfileDatabase>,
}
impl<'message> core::convert::From<ContinuationResult<'message, ProfileDatabase>>
for ProfileLookupFinished<'message>
{
fn from(result: ContinuationResult<'message, ProfileDatabase>) -> Self {
Self { result }
}
}
impl<'message> StepInputOf<Player> for ProfileLookupFinished<'message> {
type Outcome = PlayerOutcome;
fn step(self, player: &mut Player) -> Self::Outcome {
match self.result {
Ok(reply) => {
player.last_lookup = Some(reply);
PlayerOutcome::Recorded
}
Err(ContinuationFailure::Expired) => {
player.timed_out = true;
PlayerOutcome::Unavailable
}
Err(ContinuationFailure::Unavailable { .. }) => PlayerOutcome::Unavailable,
}
}
}
enum From<'message> {
Player(PlayerOutcome),
ProfileDatabase(ContinuationResult<'message, ProfileDatabase>),
}
enum To<'message> {
ProfileDatabase(Owed<PlayerId>),
Player(ProfileLookupFinished<'message>),
Finished,
}
struct Board;
impl Switchboard for Board {
type From<'message> = From<'message>;
type To<'message> = To<'message>;
fn route<'message>(
&mut self,
from: Self::From<'message>,
) -> pipecircuit::Route<Self::To<'message>> {
pipecircuit::Route::one({
match from {
From::Player(PlayerOutcome::Owed(owed)) => To::ProfileDatabase(owed),
From::ProfileDatabase(result) => To::Player(result.into()),
From::Player(PlayerOutcome::Recorded | PlayerOutcome::Unavailable) => To::Finished,
}
})
}
}
struct ImmediateContinuedProfileDatabase {
continuations: ContinuationLedger<()>,
database: ProfileDatabase,
}
impl ImmediateContinuedProfileDatabase {
fn try_new() -> Result<Self, &'static str> {
let capacity = NonZeroU32::new(8).ok_or("continuation capacity was zero")?;
let continuations = ContinuationLedger::try_new(Capacity::new(capacity))
.map_err(|_| "continuation ledger unavailable")?;
Ok(Self {
continuations,
database: ProfileDatabase,
})
}
}
impl TransportTo<Continued<ProfileDatabase>> for ImmediateContinuedProfileDatabase {
type Outcome<'transport, 'message>
= Result<ContinuationResult<'message, ProfileDatabase>, &'static str>
where
Self: 'transport;
fn send<'transport, 'message>(
&'transport mut self,
owed: Owed<PlayerId>,
) -> Self::Outcome<'transport, 'message> {
let ContinuationAdmissionOutcome::Admitted { admission } =
self.continuations.admit::<ProfileDatabase>((), owed)
else {
return Err("continued fitting exhausted its configured capacity");
};
let token = admission.timer_token();
let arm_outcome = TimerOut::Armed { token }
.into_arm_outcome()
.map_err(|_| "continued fitting produced the wrong timer arm output")?;
let ContinuationArmOutcome::Armed { work } = admission
.settle_timer_arm(arm_outcome)
.map_err(|_| "continued fitting received the wrong timer token")?
else {
return Err("continued fitting refused its own timer");
};
let completion = work.step(&mut self.database);
let cancellation_outcome = TimerOut::Cancelled { token }
.into_cancellation_outcome()
.map_err(|_| "continued fitting produced the wrong cancellation output")?;
let ContinuationCancellationOutcome::Cancelled { completion } = completion
.settle_timer_cancellation(cancellation_outcome)
.map_err(|_| "continued fitting cancelled the wrong timer")?
else {
return Err("continued fitting completed stale work");
};
let ContinuationResolutionOutcome::Resolved { result, .. } =
self.continuations.resolve(completion)
else {
return Err("continued fitting lost its admitted work");
};
Ok(result)
}
}
const fn profile_timeout() -> TimeoutAfter {
TimeoutAfter::milliseconds(NonZeroU64::MIN)
}
#[test]
fn switchboard_routes_only_owed_work_and_its_result() -> Result<(), &'static str> {
let mut board = Board;
let mut player = Player {
last_lookup: None,
timed_out: false,
};
let mut database = InstalledPipe::<Continued<ProfileDatabase>, _>::new(
ImmediateContinuedProfileDatabase::try_new()?,
);
let player_outcome = LoadProfile {
player_id: PlayerId(7),
}
.step(&mut player);
let pipecircuit::Route::One(To::ProfileDatabase(owed)) =
board.route(From::Player(player_outcome))
else {
return Err("profile work was not routed to its continued pipe");
};
let database_result = database.send(owed)?;
let pipecircuit::Route::One(To::Player(finished)) =
board.route(From::ProfileDatabase(database_result))
else {
return Err("profile result was not routed to the player");
};
assert!(matches!(
finished.step(&mut player),
PlayerOutcome::Recorded
));
assert_eq!(
player.last_lookup,
Some(ProfileLookupOutcome::Found {
player_id: PlayerId(7),
})
);
Ok(())
}
#[test]
fn expiration_uses_the_same_typed_pipe_result() -> Result<(), &'static str> {
let mut board = Board;
let mut player = Player {
last_lookup: None,
timed_out: false,
};
let expired: ContinuationResult<'_, ProfileDatabase> = Err(ContinuationFailure::Expired);
let pipecircuit::Route::One(To::Player(finished)) = board.route(From::ProfileDatabase(expired))
else {
return Err("expiration was not routed to the player");
};
assert!(matches!(
finished.step(&mut player),
PlayerOutcome::Unavailable
));
assert!(player.timed_out);
Ok(())
}
struct OwnedEcho;
impl Pipe for OwnedEcho {
type In<'message> = String;
type Out<'message> = String;
}
#[test]
fn private_envelope_moves_across_a_thread_without_copying_its_payload() -> Result<(), &'static str>
{
let capacity = NonZeroU32::new(1).ok_or("continuation capacity was zero")?;
let mut ledger = ContinuationLedger::try_new(Capacity::new(capacity))
.map_err(|_| "continuation ledger unavailable")?;
let payload = String::from("owned payload");
let payload_address = payload.as_ptr();
let ContinuationAdmissionOutcome::Admitted { admission } =
ledger.admit::<OwnedEcho>((), Owed::new(payload, profile_timeout()))
else {
return Err("owned work was refused");
};
let token = admission.timer_token();
let arm_outcome = TimerOut::Armed { token }
.into_arm_outcome()
.map_err(|_| "armed output was rejected")?;
let ContinuationArmOutcome::Armed { work } = admission
.settle_timer_arm(arm_outcome)
.map_err(|_| "timer returned the wrong token")?
else {
return Err("timer refused owned work");
};
let worker = std::thread::spawn(move || work.fulfill_with(|input| input));
let completion = worker.join().map_err(|_| "worker thread failed")?;
assert_eq!(completion.output().as_ptr(), payload_address);
let cancellation_outcome = TimerOut::Cancelled { token }
.into_cancellation_outcome()
.map_err(|_| "cancelled output was rejected")?;
let ContinuationCancellationOutcome::Cancelled { completion } = completion
.settle_timer_cancellation(cancellation_outcome)
.map_err(|_| "timer returned the wrong token")?
else {
return Err("owned completion was stale");
};
let ContinuationResolutionOutcome::Resolved { result, .. } = ledger.resolve(completion) else {
return Err("owned completion was absent from the ledger");
};
let Ok(reply) = result else {
return Err("owned continuation did not complete");
};
assert_eq!(reply.as_ptr(), payload_address);
Ok(())
}