dscale 0.4.1

A fast & deterministic simulation framework for benchmarking and testing distributed systems
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

use std::collections::BinaryHeap;
use std::sync::Arc;

use log::debug;

use crate::jiffy::Jiffies;
use crate::random::Randomizer;
use crate::step::{Step, StepQueue, TimedStep};
use crate::topology::Topology;

pub(crate) struct LatencyQueue {
    topology: Arc<Topology>,
    randomizer: Randomizer,
    queue: StepQueue,
}

impl LatencyQueue {
    pub(crate) fn new(randomizer: Randomizer, topology: Arc<Topology>) -> Self {
        Self {
            randomizer,
            topology,
            queue: BinaryHeap::new(),
        }
    }

    pub(crate) fn push(&mut self, mut message: TimedStep) {
        debug!("Before latency: {}", message.invocation_time);
        let Step::NetworkStep { source, target, .. } = &message.step else {
            unreachable!("LatencyQueue only accepts NetworkSteps");
        };
        let distribution = self.topology.get_distribution(*source, *target);
        message.invocation_time += self.randomizer.random_usize(distribution);
        debug!("After latency: {}", message.invocation_time);
        self.queue.push(std::cmp::Reverse(message));
    }

    pub(crate) fn pop(&mut self) -> Option<TimedStep> {
        Some(self.queue.pop()?.0)
    }

    pub(crate) fn peek(&self) -> Option<Jiffies> {
        Some(self.queue.peek()?.0.invocation_time)
    }
}