use crate::{FlowObservation, FlowObserver};
const _: () = assert!(core::mem::size_of::<usize>() <= core::mem::size_of::<u64>());
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub struct TransitionCounts {
started: u64,
completed: u64,
input_representation_bytes: u64,
output_representation_bytes: u64,
saturated: bool,
}
impl TransitionCounts {
pub const fn new() -> Self {
Self {
started: 0,
completed: 0,
input_representation_bytes: 0,
output_representation_bytes: 0,
saturated: false,
}
}
pub const fn started(self) -> u64 {
self.started
}
pub const fn completed(self) -> u64 {
self.completed
}
pub const fn input_representation_bytes(self) -> u64 {
self.input_representation_bytes
}
pub const fn output_representation_bytes(self) -> u64 {
self.output_representation_bytes
}
pub const fn saturated(self) -> bool {
self.saturated
}
fn record_started(&mut self, input_representation_bytes: usize) {
accumulate(&mut self.started, 1, &mut self.saturated);
accumulate_usize(
&mut self.input_representation_bytes,
input_representation_bytes,
&mut self.saturated,
);
}
fn record_completed(&mut self, output_representation_bytes: usize) {
accumulate(&mut self.completed, 1, &mut self.saturated);
accumulate_usize(
&mut self.output_representation_bytes,
output_representation_bytes,
&mut self.saturated,
);
}
}
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub struct ExecutionCounts {
circuit_turns: TransitionCounts,
pipe_sends: TransitionCounts,
participant_steps: TransitionCounts,
switchboard_routes: TransitionCounts,
}
impl ExecutionCounts {
pub const fn new() -> Self {
Self {
circuit_turns: TransitionCounts::new(),
pipe_sends: TransitionCounts::new(),
participant_steps: TransitionCounts::new(),
switchboard_routes: TransitionCounts::new(),
}
}
pub const fn circuit_turns(self) -> TransitionCounts {
self.circuit_turns
}
pub const fn pipe_sends(self) -> TransitionCounts {
self.pipe_sends
}
pub const fn participant_steps(self) -> TransitionCounts {
self.participant_steps
}
pub const fn switchboard_routes(self) -> TransitionCounts {
self.switchboard_routes
}
pub const fn saturated(self) -> bool {
self.circuit_turns.saturated()
|| self.pipe_sends.saturated()
|| self.participant_steps.saturated()
|| self.switchboard_routes.saturated()
}
}
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub struct CountingFlowObserver {
counts: ExecutionCounts,
}
impl CountingFlowObserver {
pub const fn new() -> Self {
Self {
counts: ExecutionCounts::new(),
}
}
pub const fn counts(&self) -> ExecutionCounts {
self.counts
}
pub fn take_counts(&mut self) -> ExecutionCounts {
core::mem::take(&mut self.counts)
}
pub const fn into_counts(self) -> ExecutionCounts {
self.counts
}
}
impl FlowObserver for CountingFlowObserver {
fn observe(&mut self, observation: FlowObservation) {
match observation {
FlowObservation::CircuitTurnBegan {
input_representation_bytes,
..
} => self
.counts
.circuit_turns
.record_started(input_representation_bytes),
FlowObservation::CircuitTurnCompleted {
output_representation_bytes,
..
} => self
.counts
.circuit_turns
.record_completed(output_representation_bytes),
FlowObservation::PipeSendBegan {
input_representation_bytes,
..
} => self
.counts
.pipe_sends
.record_started(input_representation_bytes),
FlowObservation::PipeSendCompleted {
output_representation_bytes,
..
} => self
.counts
.pipe_sends
.record_completed(output_representation_bytes),
FlowObservation::ParticipantStepBegan {
input_representation_bytes,
..
} => self
.counts
.participant_steps
.record_started(input_representation_bytes),
FlowObservation::ParticipantStepCompleted {
output_representation_bytes,
..
} => self
.counts
.participant_steps
.record_completed(output_representation_bytes),
FlowObservation::SwitchboardRouteBegan {
input_representation_bytes,
..
} => self
.counts
.switchboard_routes
.record_started(input_representation_bytes),
FlowObservation::SwitchboardRouteCompleted {
output_representation_bytes,
..
} => self
.counts
.switchboard_routes
.record_completed(output_representation_bytes),
}
}
}
fn accumulate(target: &mut u64, increment: u64, saturated: &mut bool) {
match target.checked_add(increment) {
Some(total) => *target = total,
None => {
*target = u64::MAX;
*saturated = true;
}
}
}
fn accumulate_usize(target: &mut u64, increment: usize, saturated: &mut bool) {
accumulate(target, increment as u64, saturated);
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn taking_counts_starts_one_fresh_measurement_window() {
let mut observer = CountingFlowObserver::new();
observer.observe(FlowObservation::PipeSendBegan {
participant: "participant",
input: "input",
transport: "transport",
input_representation_bytes: 24,
});
observer.observe(FlowObservation::PipeSendCompleted {
participant: "participant",
transport: "transport",
outcome: "outcome",
output_representation_bytes: 16,
});
let measured = observer.take_counts().pipe_sends();
assert_eq!(measured.started(), 1);
assert_eq!(measured.completed(), 1);
assert_eq!(measured.input_representation_bytes(), 24);
assert_eq!(measured.output_representation_bytes(), 16);
assert_eq!(observer.counts(), ExecutionCounts::default());
}
#[test]
fn overflowing_measurements_saturate_explicitly() {
let mut counts = TransitionCounts {
input_representation_bytes: u64::MAX,
..TransitionCounts::default()
};
counts.record_started(1);
assert_eq!(counts.input_representation_bytes(), u64::MAX);
assert!(counts.saturated());
}
#[test]
fn every_observation_contributes_to_its_transition_family() {
let mut observer = CountingFlowObserver::new();
for observation in [
FlowObservation::CircuitTurnBegan {
circuit: "circuit",
input: "input",
input_representation_bytes: 1,
},
FlowObservation::CircuitTurnCompleted {
circuit: "circuit",
output: "output",
output_representation_bytes: 2,
},
FlowObservation::PipeSendBegan {
participant: "participant",
input: "input",
transport: "transport",
input_representation_bytes: 3,
},
FlowObservation::PipeSendCompleted {
participant: "participant",
transport: "transport",
outcome: "outcome",
output_representation_bytes: 4,
},
FlowObservation::ParticipantStepBegan {
participant: "participant",
input: "input",
input_representation_bytes: 5,
},
FlowObservation::ParticipantStepCompleted {
participant: "participant",
outcome: "outcome",
output_representation_bytes: 6,
},
FlowObservation::SwitchboardRouteBegan {
switchboard: "switchboard",
input: "input",
input_representation_bytes: 7,
},
FlowObservation::SwitchboardRouteCompleted {
switchboard: "switchboard",
output: "output",
output_representation_bytes: 8,
},
] {
observer.observe(observation);
}
let counts = observer.into_counts();
assert_transition(counts.circuit_turns(), 1, 2);
assert_transition(counts.pipe_sends(), 3, 4);
assert_transition(counts.participant_steps(), 5, 6);
assert_transition(counts.switchboard_routes(), 7, 8);
assert!(!counts.saturated());
}
#[test]
fn aggregate_saturation_includes_every_transition_family() {
for counts in [
ExecutionCounts {
circuit_turns: saturated_transition(),
..ExecutionCounts::new()
},
ExecutionCounts {
pipe_sends: saturated_transition(),
..ExecutionCounts::new()
},
ExecutionCounts {
participant_steps: saturated_transition(),
..ExecutionCounts::new()
},
ExecutionCounts {
switchboard_routes: saturated_transition(),
..ExecutionCounts::new()
},
] {
assert!(counts.saturated());
}
}
fn assert_transition(counts: TransitionCounts, input_bytes: u64, output_bytes: u64) {
assert_eq!(counts.started(), 1);
assert_eq!(counts.completed(), 1);
assert_eq!(counts.input_representation_bytes(), input_bytes);
assert_eq!(counts.output_representation_bytes(), output_bytes);
assert!(!counts.saturated());
}
fn saturated_transition() -> TransitionCounts {
TransitionCounts {
saturated: true,
..TransitionCounts::new()
}
}
}