use core::fmt;
pub const DEFAULT_INTERVAL_SECONDS: f64 = 0.1;
pub const BOUNDARY_TOLERANCE_SCALE: f64 = 1e-12;
pub const MAX_SAMPLE_INDEX: u64 = 1 << f64::MANTISSA_DIGITS;
#[derive(Clone, Copy, Debug, PartialEq)]
pub struct FixedRateScheduleConfig {
pub interval_seconds: f64,
}
impl Default for FixedRateScheduleConfig {
fn default() -> Self {
Self {
interval_seconds: DEFAULT_INTERVAL_SECONDS,
}
}
}
impl FixedRateScheduleConfig {
#[must_use]
pub fn is_valid(self) -> bool {
self.interval_seconds.is_finite() && self.interval_seconds > 0.0
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum FixedRateScheduleConfigError {
InvalidInterval,
InvalidPhaseOffset,
}
impl fmt::Display for FixedRateScheduleConfigError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(match self {
Self::InvalidInterval => "sample interval must be finite and positive",
Self::InvalidPhaseOffset => {
"phase offset must be finite, at least zero and less than the interval"
}
})
}
}
impl std::error::Error for FixedRateScheduleConfigError {}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum FixedRateScheduleAdvanceError {
InvalidElapsed,
}
impl fmt::Display for FixedRateScheduleAdvanceError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(match self {
Self::InvalidElapsed => {
"elapsed time must be finite, not negative, and reach a representable next sample"
}
})
}
}
impl std::error::Error for FixedRateScheduleAdvanceError {}
#[derive(Clone, Copy, Debug, PartialEq)]
pub struct FixedRateSample {
pub scheduled_simulation_time_seconds: f64,
pub offset_within_update_seconds: f64,
pub elapsed_since_previous_sample_seconds: f64,
}
#[derive(Clone, Copy, Debug, PartialEq)]
pub struct FixedRateSchedule {
config: FixedRateScheduleConfig,
phase_offset_seconds: f64,
simulation_time_seconds: f64,
last_sample_time_seconds: f64,
next_sample_index: u64,
}
impl Default for FixedRateSchedule {
fn default() -> Self {
Self::new(FixedRateScheduleConfig::default(), 0.0)
.expect("default fixed-rate schedule is valid")
}
}
impl FixedRateSchedule {
pub fn new(
config: FixedRateScheduleConfig,
phase_offset_seconds: f64,
) -> Result<Self, FixedRateScheduleConfigError> {
if !config.is_valid() {
return Err(FixedRateScheduleConfigError::InvalidInterval);
}
if !phase_offset_seconds.is_finite()
|| phase_offset_seconds < 0.0
|| phase_offset_seconds >= config.interval_seconds
{
return Err(FixedRateScheduleConfigError::InvalidPhaseOffset);
}
let next_sample_index = u64::from(phase_offset_seconds == 0.0);
Ok(Self {
config,
phase_offset_seconds,
simulation_time_seconds: 0.0,
last_sample_time_seconds: 0.0,
next_sample_index,
})
}
#[must_use]
pub fn phase_offset_seconds(&self) -> f64 {
self.phase_offset_seconds
}
#[must_use]
pub fn simulation_time_seconds(&self) -> f64 {
self.simulation_time_seconds
}
pub fn skip_elapsed(
&mut self,
elapsed_seconds: f64,
) -> Result<(), FixedRateScheduleAdvanceError> {
if !elapsed_seconds.is_finite() || elapsed_seconds < 0.0 {
return Err(FixedRateScheduleAdvanceError::InvalidElapsed);
}
if elapsed_seconds == 0.0 {
return Ok(());
}
let skipped_end_seconds = self.simulation_time_seconds + elapsed_seconds;
if !skipped_end_seconds.is_finite() {
return Err(FixedRateScheduleAdvanceError::InvalidElapsed);
}
let mut next_sample_index = self.next_sample_index;
if due_at_or_before(
self.sample_time_seconds(next_sample_index),
skipped_end_seconds,
) {
let estimate = ((skipped_end_seconds - self.phase_offset_seconds)
/ self.config.interval_seconds)
.floor()
+ 1.0;
#[allow(clippy::cast_precision_loss)]
let max_index = MAX_SAMPLE_INDEX as f64;
if !estimate.is_finite() || estimate > max_index {
return Err(FixedRateScheduleAdvanceError::InvalidElapsed);
}
#[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
let estimate = estimate as u64;
next_sample_index = estimate.max(next_sample_index + 1);
if next_sample_index - 1 > self.next_sample_index
&& !due_at_or_before(
self.sample_time_seconds(next_sample_index - 1),
skipped_end_seconds,
)
{
next_sample_index -= 1;
}
if due_at_or_before(
self.sample_time_seconds(next_sample_index),
skipped_end_seconds,
) {
next_sample_index += 1;
}
}
let next_sample_time_seconds = self.sample_time_seconds(next_sample_index);
if next_sample_index > MAX_SAMPLE_INDEX
|| !next_sample_time_seconds.is_finite()
|| due_at_or_before(next_sample_time_seconds, skipped_end_seconds)
{
return Err(FixedRateScheduleAdvanceError::InvalidElapsed);
}
self.simulation_time_seconds = skipped_end_seconds;
self.last_sample_time_seconds = skipped_end_seconds;
self.next_sample_index = next_sample_index;
Ok(())
}
pub fn advance(
&mut self,
elapsed_seconds: f64,
mut emit: impl FnMut(FixedRateSample),
) -> Result<(), FixedRateScheduleAdvanceError> {
if !elapsed_seconds.is_finite() || elapsed_seconds < 0.0 {
return Err(FixedRateScheduleAdvanceError::InvalidElapsed);
}
let update_start_seconds = self.simulation_time_seconds;
let update_end_seconds = update_start_seconds + elapsed_seconds;
if !update_end_seconds.is_finite() {
return Err(FixedRateScheduleAdvanceError::InvalidElapsed);
}
loop {
let scheduled_simulation_time_seconds = self.next_sample_time_seconds();
if !due_at_or_before(scheduled_simulation_time_seconds, update_end_seconds) {
break;
}
emit(FixedRateSample {
scheduled_simulation_time_seconds,
offset_within_update_seconds: scheduled_simulation_time_seconds
- update_start_seconds,
elapsed_since_previous_sample_seconds: scheduled_simulation_time_seconds
- self.last_sample_time_seconds,
});
self.last_sample_time_seconds = scheduled_simulation_time_seconds;
self.next_sample_index += 1;
}
self.simulation_time_seconds = update_end_seconds;
Ok(())
}
#[must_use]
pub fn next_sample_time_seconds(&self) -> f64 {
self.sample_time_seconds(self.next_sample_index)
}
fn sample_time_seconds(&self, index: u64) -> f64 {
#[allow(clippy::cast_precision_loss)]
let index = index as f64;
self.phase_offset_seconds + index * self.config.interval_seconds
}
}
fn due_at_or_before(scheduled_seconds: f64, update_end_seconds: f64) -> bool {
scheduled_seconds <= update_end_seconds
|| scheduled_seconds - update_end_seconds
<= BOUNDARY_TOLERANCE_SCALE
* scheduled_seconds
.abs()
.max(update_end_seconds.abs())
.max(1.0)
}
#[cfg(test)]
mod tests {
use super::*;
const EPSILON: f64 = 1e-12;
fn event_times(events: &[FixedRateSample]) -> Vec<f64> {
events
.iter()
.map(|event| event.scheduled_simulation_time_seconds)
.collect()
}
fn assert_times_match(actual: &[f64], expected: &[f64]) {
assert_eq!(actual.len(), expected.len());
for (actual, expected) in actual.iter().zip(expected) {
assert!((actual - expected).abs() < EPSILON);
}
}
fn collect(schedule: &mut FixedRateSchedule, steps: &[f64]) -> Vec<FixedRateSample> {
let mut events = Vec::new();
for &step in steps {
schedule.advance(step, |event| events.push(event)).unwrap();
}
events
}
#[test]
fn default_schedule_emits_at_ten_hertz_boundaries() {
let mut schedule = FixedRateSchedule::default();
let mut events = Vec::new();
schedule.advance(0.099, |event| events.push(event)).unwrap();
assert!(events.is_empty());
schedule.advance(0.001, |event| events.push(event)).unwrap();
assert_eq!(events.len(), 1);
assert!((events[0].scheduled_simulation_time_seconds - 0.1).abs() < EPSILON);
assert!((events[0].offset_within_update_seconds - 0.001).abs() < EPSILON);
assert!((events[0].elapsed_since_previous_sample_seconds - 0.1).abs() < EPSILON);
events.clear();
schedule.advance(0.2, |event| events.push(event)).unwrap();
assert_times_match(&event_times(&events), &[0.2, 0.3]);
}
#[test]
fn coarse_and_partitioned_updates_emit_the_same_schedule() {
let mut coarse = FixedRateSchedule::default();
let mut partitioned = FixedRateSchedule::default();
let mut coarse_events = Vec::new();
let mut partitioned_events = Vec::new();
coarse
.advance(0.35, |event| coarse_events.push(event))
.unwrap();
for _ in 0..7 {
partitioned
.advance(0.05, |event| partitioned_events.push(event))
.unwrap();
}
assert_times_match(
&event_times(&coarse_events),
&event_times(&partitioned_events),
);
assert!(
(coarse.simulation_time_seconds() - partitioned.simulation_time_seconds()).abs()
< EPSILON
);
}
#[test]
fn phase_offsets_stagger_the_first_and_later_samples() {
let mut schedule = FixedRateSchedule::new(FixedRateScheduleConfig::default(), 0.025)
.expect("phase within the interval");
let mut events = Vec::new();
schedule.advance(0.024, |event| events.push(event)).unwrap();
assert!(events.is_empty());
schedule.advance(0.001, |event| events.push(event)).unwrap();
assert_times_match(&event_times(&events), &[0.025]);
assert!((events[0].elapsed_since_previous_sample_seconds - 0.025).abs() < EPSILON);
events.clear();
schedule.advance(0.1, |event| events.push(event)).unwrap();
assert_times_match(&event_times(&events), &[0.125]);
}
#[test]
fn invalid_configuration_and_elapsed_time_preserve_schedule_state() {
let invalid_interval = FixedRateScheduleConfig {
interval_seconds: 0.0,
};
assert_eq!(
FixedRateSchedule::new(invalid_interval, 0.0),
Err(FixedRateScheduleConfigError::InvalidInterval)
);
assert_eq!(
FixedRateSchedule::new(FixedRateScheduleConfig::default(), 0.1),
Err(FixedRateScheduleConfigError::InvalidPhaseOffset)
);
let mut schedule = FixedRateSchedule::default();
schedule.advance(0.04, |_| {}).unwrap();
let before = schedule;
assert_eq!(
schedule.advance(f64::NAN, |_| {}),
Err(FixedRateScheduleAdvanceError::InvalidElapsed)
);
assert_eq!(schedule, before);
assert_eq!(
schedule.advance(-0.1, |_| {}),
Err(FixedRateScheduleAdvanceError::InvalidElapsed)
);
assert_eq!(schedule, before);
assert_eq!(
schedule.skip_elapsed(f64::INFINITY),
Err(FixedRateScheduleAdvanceError::InvalidElapsed)
);
assert_eq!(schedule, before);
}
#[test]
fn skipping_elapsed_time_keeps_phase_without_replaying_samples() {
let mut schedule = FixedRateSchedule::new(
FixedRateScheduleConfig {
interval_seconds: 0.25,
},
0.1,
)
.unwrap();
schedule.advance(0.05, |_| {}).unwrap();
schedule.skip_elapsed(0.96).unwrap();
assert!((schedule.simulation_time_seconds - 1.01).abs() < EPSILON);
assert!((schedule.last_sample_time_seconds - 1.01).abs() < EPSILON);
assert!((schedule.next_sample_time_seconds() - 1.1).abs() < EPSILON);
let mut events = Vec::new();
schedule.advance(0.09, |event| events.push(event)).unwrap();
assert_times_match(&event_times(&events), &[1.1]);
assert!((events[0].elapsed_since_previous_sample_seconds - 0.09).abs() < EPSILON);
}
#[test]
fn skipping_unrepresentable_or_tolerance_due_time_preserves_schedule() {
let mut tiny_interval = FixedRateSchedule::new(
FixedRateScheduleConfig {
interval_seconds: f64::MIN_POSITIVE,
},
0.0,
)
.unwrap();
let tiny_before = tiny_interval;
assert_eq!(
tiny_interval.skip_elapsed(1.0),
Err(FixedRateScheduleAdvanceError::InvalidElapsed)
);
assert_eq!(tiny_interval, tiny_before);
let mut tolerance_due = FixedRateSchedule::new(
FixedRateScheduleConfig {
interval_seconds: 1e-13,
},
0.0,
)
.unwrap();
let tolerance_before = tolerance_due;
assert_eq!(
tolerance_due.skip_elapsed(1.0),
Err(FixedRateScheduleAdvanceError::InvalidElapsed)
);
assert_eq!(tolerance_due, tolerance_before);
}
#[test]
fn skipping_a_large_elapsed_interval_uses_a_future_sample_boundary() {
let mut schedule = FixedRateSchedule::default();
schedule.skip_elapsed(1_000_000_000.0).unwrap();
assert!(!due_at_or_before(
schedule.next_sample_time_seconds(),
schedule.simulation_time_seconds
));
assert_eq!(
schedule.last_sample_time_seconds,
schedule.simulation_time_seconds
);
}
#[test]
fn every_sample_lands_exactly_on_the_index_formula_over_a_long_run() {
let phase = 0.03;
let config = FixedRateScheduleConfig::default();
let mut schedule = FixedRateSchedule::new(config, phase).unwrap();
let events = collect(&mut schedule, &[1.0 / 60.0; 60 * 600]);
assert!(events.len() > 5000, "the run emitted {}", events.len());
for (k, event) in events.iter().enumerate() {
let expected = phase + k as f64 * config.interval_seconds;
assert_eq!(
event.scheduled_simulation_time_seconds.to_bits(),
expected.to_bits(),
"sample {k}"
);
}
}
#[test]
fn a_sample_on_the_update_boundary_fires_in_that_update() {
let mut schedule = FixedRateSchedule::default();
let mut per_update = Vec::new();
for _ in 0..10 {
let mut count = 0;
schedule.advance(0.1, |_| count += 1).unwrap();
per_update.push(count);
}
assert!(schedule.simulation_time_seconds() < 1.0);
assert_eq!(per_update, [1; 10]);
}
#[test]
fn split_and_whole_updates_emit_bit_identical_samples() {
let config = FixedRateScheduleConfig {
interval_seconds: 0.07,
};
let mut whole = FixedRateSchedule::new(config, 0.011).unwrap();
let mut split = FixedRateSchedule::new(config, 0.011).unwrap();
let whole_events = collect(&mut whole, &[12.5]);
let steps: Vec<f64> = [0.013, 0.2, 0.0, 0.041, 0.5, 0.0005]
.iter()
.copied()
.cycle()
.take(1000)
.collect();
let total: f64 = steps.iter().sum();
let split_events = collect(&mut split, &steps);
let covered = total.min(12.5) - config.interval_seconds;
let whole_times: Vec<u64> = whole_events
.iter()
.filter(|event| event.scheduled_simulation_time_seconds < covered)
.map(|event| event.scheduled_simulation_time_seconds.to_bits())
.collect();
let split_times: Vec<u64> = split_events
.iter()
.filter(|event| event.scheduled_simulation_time_seconds < covered)
.map(|event| event.scheduled_simulation_time_seconds.to_bits())
.collect();
assert!(whole_times.len() > 100);
assert_eq!(whole_times, split_times);
}
#[test]
fn two_schedules_fed_the_same_inputs_agree_bit_for_bit() {
let config = FixedRateScheduleConfig {
interval_seconds: 1.0 / 30.0,
};
let steps = [0.016, 0.017, 0.2, 0.001, 0.033];
let mut a = FixedRateSchedule::new(config, 0.005).unwrap();
let mut b = FixedRateSchedule::new(config, 0.005).unwrap();
let a_events = collect(&mut a, &steps);
let b_events = collect(&mut b, &steps);
assert!(!a_events.is_empty());
let bits = |event: &FixedRateSample| {
[
event.scheduled_simulation_time_seconds.to_bits(),
event.offset_within_update_seconds.to_bits(),
event.elapsed_since_previous_sample_seconds.to_bits(),
]
};
assert_eq!(
a_events.iter().map(bits).collect::<Vec<_>>(),
b_events.iter().map(bits).collect::<Vec<_>>()
);
assert_eq!(a, b);
}
#[test]
fn samples_are_ordered_and_offsets_lie_within_their_update() {
let mut schedule = FixedRateSchedule::new(
FixedRateScheduleConfig {
interval_seconds: 0.02,
},
0.007,
)
.unwrap();
let mut previous = f64::NEG_INFINITY;
for step in [0.1, 0.013, 0.5, 0.019] {
schedule
.advance(step, |event| {
assert!(event.scheduled_simulation_time_seconds > previous);
previous = event.scheduled_simulation_time_seconds;
assert!(event.offset_within_update_seconds > 0.0);
assert!(event.offset_within_update_seconds <= step + EPSILON);
})
.unwrap();
}
}
#[test]
fn skipping_emits_nothing_and_resumes_where_advancing_would() {
let config = FixedRateScheduleConfig {
interval_seconds: 0.1,
};
for phase in [0.0, 0.013, 0.05, 0.099] {
for skip in [0.05, 0.3, 1.0, 7.77, 1234.5678] {
let mut skipped = FixedRateSchedule::new(config, phase).unwrap();
let mut advanced = skipped;
skipped.advance(0.02, |_| {}).unwrap();
advanced.advance(0.02, |_| {}).unwrap();
skipped.skip_elapsed(skip).unwrap();
advanced.advance(skip, |_| {}).unwrap();
let after = 0.25;
let skipped_events = collect(&mut skipped, &[after]);
let advanced_events = collect(&mut advanced, &[after]);
assert_eq!(
event_times(&skipped_events)
.iter()
.map(|time| time.to_bits())
.collect::<Vec<_>>(),
event_times(&advanced_events)
.iter()
.map(|time| time.to_bits())
.collect::<Vec<_>>(),
"phase {phase}, skip {skip}"
);
let first = skipped_events[0];
assert!(
(first.elapsed_since_previous_sample_seconds
- first.offset_within_update_seconds)
.abs()
< EPSILON
);
}
}
}
#[test]
fn skipping_emits_nothing_on_the_next_update_for_the_skipped_time() {
let mut schedule = FixedRateSchedule::default();
schedule.skip_elapsed(5.0).unwrap();
assert!(collect(&mut schedule, &[0.0]).is_empty());
assert!(collect(&mut schedule, &[0.09]).is_empty());
}
#[test]
fn a_rejected_skip_past_the_index_bound_leaves_the_schedule_alone() {
let mut schedule = FixedRateSchedule::new(
FixedRateScheduleConfig {
interval_seconds: 1e-3,
},
0.0,
)
.unwrap();
schedule.advance(0.0105, |_| {}).unwrap();
let before = schedule;
assert_eq!(
schedule.skip_elapsed(1e14),
Err(FixedRateScheduleAdvanceError::InvalidElapsed)
);
assert_eq!(schedule, before);
assert_eq!(
schedule.skip_elapsed(-1.0),
Err(FixedRateScheduleAdvanceError::InvalidElapsed)
);
assert_eq!(schedule, before);
assert_eq!(
schedule.advance(f64::INFINITY, |_| panic!("emitted on a rejected update")),
Err(FixedRateScheduleAdvanceError::InvalidElapsed)
);
assert_eq!(schedule, before);
}
#[test]
fn phase_stagger_spreads_agents_across_the_interval() {
let config = FixedRateScheduleConfig::default();
let agents = 8_u32;
let mut schedules: Vec<FixedRateSchedule> = (0..agents)
.map(|i| {
let phase = f64::from(i) * config.interval_seconds / f64::from(agents);
FixedRateSchedule::new(config, phase).unwrap()
})
.collect();
for schedule in &mut schedules {
schedule.advance(config.interval_seconds, |_| {}).unwrap();
}
let mut times = Vec::new();
for schedule in &mut schedules {
let events = collect(schedule, &[config.interval_seconds]);
assert_eq!(events.len(), 1, "phase {}", schedule.phase_offset_seconds());
times.push(events[0].scheduled_simulation_time_seconds);
}
times.sort_by(f64::total_cmp);
let spacing = config.interval_seconds / f64::from(agents);
for pair in times.windows(2) {
assert!((pair[1] - pair[0] - spacing).abs() < EPSILON);
}
}
#[test]
fn errors_display_what_was_wrong() {
assert!(
FixedRateScheduleConfigError::InvalidInterval
.to_string()
.contains("interval")
);
assert!(
FixedRateScheduleConfigError::InvalidPhaseOffset
.to_string()
.contains("phase")
);
assert!(
FixedRateScheduleAdvanceError::InvalidElapsed
.to_string()
.contains("elapsed")
);
}
fn first_emitted(schedule: &FixedRateSchedule, step: f64) -> Option<f64> {
let mut copy = *schedule;
let mut first = None;
copy.advance(step, |event| {
first.get_or_insert(event.scheduled_simulation_time_seconds);
})
.unwrap();
first
}
#[test]
fn the_next_sample_time_is_the_sample_emitted_next_and_reading_it_takes_nothing() {
let config = FixedRateScheduleConfig::default();
let reach = config.interval_seconds * 2.0;
for phase in [0.0, 0.025, 0.099] {
let mut schedule = FixedRateSchedule::new(config, phase).expect("phase within");
let first = schedule.next_sample_time_seconds();
if phase == 0.0 {
assert_eq!(first.to_bits(), config.interval_seconds.to_bits());
} else {
assert_eq!(first.to_bits(), phase.to_bits());
}
for (round, step) in [0.013, 0.05, 0.0, 0.31, 0.1, 0.007].into_iter().enumerate() {
let next = schedule.next_sample_time_seconds();
assert_eq!(
next.to_bits(),
schedule.next_sample_time_seconds().to_bits(),
"phase {phase}, round {round}: reading it twice moved it"
);
assert!(next > schedule.simulation_time_seconds());
assert_eq!(
Some(next.to_bits()),
first_emitted(&schedule, reach).map(f64::to_bits),
"phase {phase}, round {round}: not the sample emitted next"
);
if round % 2 == 0 {
schedule.advance(step, |_| {}).unwrap();
} else {
schedule.skip_elapsed(step).unwrap();
}
}
}
}
}