Skip to main content

poolsim_core/
monte_carlo.rs

1//! Monte Carlo queue simulation primitives.
2//!
3//! This module is used when `poolsim` needs a sampled queue-wait distribution
4//! rather than only analytical queueing metrics.
5//!
6//! Typical uses:
7//!
8//! - full simulation through crate-root [`crate::simulate`]
9//! - fixed-size evaluation through [`crate::evaluate`]
10//! - M/D/c probing during optimization and sensitivity analysis
11//!
12//! The public entrypoint is [`crate::monte_carlo::run`], which returns
13//! [`crate::monte_carlo::MonteCarloResult`].
14
15use rand::{rngs::StdRng, Rng, SeedableRng};
16use rand_distr::{Distribution, Exp};
17
18use crate::{
19    distribution::LatencyDistribution, error::PoolsimError, types::QueueModel,
20    types::SimulationOptions, WorkloadConfig,
21};
22
23/// Monte Carlo queue-wait summary and raw sampled waits.
24#[derive(Debug, Clone)]
25pub struct MonteCarloResult {
26    /// Sorted per-request queue waits in milliseconds.
27    pub wait_times_ms: Vec<f64>,
28    /// p50 queue wait (milliseconds).
29    pub p50: f64,
30    /// p95 queue wait (milliseconds).
31    pub p95: f64,
32    /// p99 queue wait (milliseconds).
33    pub p99: f64,
34    /// Mean queue wait (milliseconds).
35    pub mean: f64,
36}
37
38/// Runs Monte Carlo queue-wait simulation.
39///
40/// # Errors
41///
42/// Returns [`PoolsimError::InvalidInput`] for invalid pool size, and
43/// [`PoolsimError::Simulation`] when no wait samples are produced.
44pub fn run(
45    workload: &WorkloadConfig,
46    pool_size: u32,
47    dist: &LatencyDistribution,
48    opts: &SimulationOptions,
49) -> Result<MonteCarloResult, PoolsimError> {
50    run_with_overhead(workload, pool_size, 0.0, dist, opts)
51}
52
53pub(crate) fn run_with_overhead(
54    workload: &WorkloadConfig,
55    pool_size: u32,
56    connection_overhead_ms: f64,
57    dist: &LatencyDistribution,
58    opts: &SimulationOptions,
59) -> Result<MonteCarloResult, PoolsimError> {
60    if pool_size == 0 {
61        return Err(PoolsimError::invalid_input(
62            "INVALID_POOL_SIZE",
63            "pool_size must be > 0",
64            None,
65        ));
66    }
67
68    let iterations = opts.iterations as usize;
69    let lambda = workload.requests_per_second;
70    let base_seed = opts.seed.unwrap_or_else(rand::random::<u64>);
71    let deterministic_service_ms = dist.mean_ms() + connection_overhead_ms;
72
73    #[cfg(feature = "parallel")]
74    let waits = {
75        use rayon::prelude::*;
76
77        let workers = rayon::current_num_threads().max(1);
78        let chunk_count = workers.min(iterations.max(1));
79        let chunk_size = (iterations + chunk_count - 1) / chunk_count;
80        let chunk_config = SimChunkConfig {
81            lambda,
82            pool_size,
83            connection_overhead_ms,
84            deterministic_service_ms,
85            queue_model: opts.queue_model,
86            dist,
87        };
88
89        (0..chunk_count)
90            .into_par_iter()
91            .map(|chunk_id| {
92                let start = chunk_id * chunk_size;
93                let end = ((chunk_id + 1) * chunk_size).min(iterations);
94                if start >= end {
95                    return Vec::new();
96                }
97
98                let seed = base_seed ^ ((chunk_id as u64 + 1).wrapping_mul(0x9E37_79B9_7F4A_7C15));
99                let mut rng = StdRng::seed_from_u64(seed);
100                simulate_chunk(&mut rng, end - start, &chunk_config)
101            })
102            .reduce(Vec::new, |mut left, mut right| {
103                left.append(&mut right);
104                left
105            })
106    };
107
108    #[cfg(not(feature = "parallel"))]
109    let waits = {
110        let mut rng = StdRng::seed_from_u64(base_seed);
111        let chunk_config = SimChunkConfig {
112            lambda,
113            pool_size,
114            connection_overhead_ms,
115            deterministic_service_ms,
116            queue_model: opts.queue_model,
117            dist,
118        };
119        simulate_chunk(&mut rng, iterations, &chunk_config)
120    };
121
122    build_result(waits)
123}
124
125struct SimChunkConfig<'a> {
126    lambda: f64,
127    pool_size: u32,
128    connection_overhead_ms: f64,
129    deterministic_service_ms: f64,
130    queue_model: QueueModel,
131    dist: &'a LatencyDistribution,
132}
133
134fn simulate_chunk<R: Rng + ?Sized>(
135    rng: &mut R,
136    iterations: usize,
137    config: &SimChunkConfig<'_>,
138) -> Vec<f64> {
139    let mut waits = Vec::with_capacity(iterations);
140    let mut arrival_time_s = 0.0;
141    let mut server_free_at = vec![0.0f64; config.pool_size as usize];
142
143    let inter_arrival = Exp::new(config.lambda).expect("lambda > 0 for exponential inter-arrival");
144
145    for _ in 0..iterations {
146        arrival_time_s += inter_arrival.sample(rng);
147
148        let mut min_idx = 0usize;
149        let mut min_free = server_free_at[0];
150        for (idx, free_at) in server_free_at.iter().copied().enumerate().skip(1) {
151            if free_at < min_free {
152                min_idx = idx;
153                min_free = free_at;
154            }
155        }
156
157        let wait_s = (min_free - arrival_time_s).max(0.0);
158        let service_ms = match config.queue_model {
159            QueueModel::MMC => config.dist.sample_ms(rng) + config.connection_overhead_ms,
160            QueueModel::MDC => config.deterministic_service_ms,
161        };
162        let service_s = service_ms / 1_000.0;
163
164        server_free_at[min_idx] = arrival_time_s + wait_s + service_s;
165        waits.push(wait_s * 1_000.0);
166    }
167
168    waits
169}
170
171fn build_result(mut waits: Vec<f64>) -> Result<MonteCarloResult, PoolsimError> {
172    if waits.is_empty() {
173        return Err(PoolsimError::Simulation(
174            "no wait times were generated during simulation".to_string(),
175        ));
176    }
177
178    let mean = waits.iter().sum::<f64>() / waits.len() as f64;
179    waits.sort_by(|a, b| a.total_cmp(b));
180
181    Ok(MonteCarloResult {
182        p50: percentile(&waits, 0.50),
183        p95: percentile(&waits, 0.95),
184        p99: percentile(&waits, 0.99),
185        mean,
186        wait_times_ms: waits,
187    })
188}
189
190fn percentile(sorted: &[f64], p: f64) -> f64 {
191    if sorted.is_empty() {
192        return 0.0;
193    }
194    let p = p.clamp(0.0, 1.0);
195    let idx = ((sorted.len() - 1) as f64 * p).round() as usize;
196    sorted[idx]
197}
198
199#[cfg(test)]
200mod tests {
201    use super::*;
202
203    #[test]
204    fn percentile_returns_zero_for_empty_input() {
205        assert_eq!(percentile(&[], 0.5), 0.0);
206    }
207}