tropel-engine 0.6.0

The tropel load-test engine: scenario execution, VU scheduling, the protocol registry and the metrics pipeline.
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
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123

//! Think-time and duration-parsing helpers shared by the VU loop: pacing
//! between iterations (`apply_think_time`), execution-config extraction
//! (`extract_think_time`) and k6-style duration parsing (`parse_duration_str`).
//! Split out of the former `vu_loop.rs` god-file.

use rand::RngExt;
use std::sync::Arc;
use std::time::Duration;
use tropel_core::config::{ExecutionConfig, ThinkTimeConfig};
use tropel_sdk::Result;

pub(crate) fn extract_think_time(exec_cfg: &ExecutionConfig) -> ThinkTimeConfig {
    match exec_cfg {
        ExecutionConfig::ConstantVus { think_time, .. } => think_time.clone(),
        ExecutionConfig::RampingVus { think_time, .. } => think_time.clone(),
        ExecutionConfig::ConstantArrivalRate { think_time, .. } => think_time.clone(),
        ExecutionConfig::SharedIterations { think_time, .. } => think_time.clone(),
        ExecutionConfig::PerVUIterations { think_time, .. } => think_time.clone(),
        ExecutionConfig::RampingArrivalRate { think_time, .. } => think_time.clone(),
        ExecutionConfig::ExternallyControlled { think_time, .. } => think_time.clone(),
    }
}

// ── Duration parsing (from old engine.rs) ──

pub(crate) fn parse_duration_str(s: &str) -> Result<Duration> {
    let s = s.trim();
    if s.is_empty() || s == "0" || s == "0s" {
        return Ok(Duration::ZERO);
    }
    tropel_sdk::parse_duration(s)
}

// ── Think time ──

pub(crate) async fn apply_think_time(
    config: &ThinkTimeConfig,
    iter_duration: Option<Duration>,
    stop: Option<&Arc<tokio::sync::Notify>>,
) {
    // Helper: sleep that can be interrupted by a stop signal.
    async fn interruptible_sleep(dur: Duration, stop: Option<&Arc<tokio::sync::Notify>>) {
        match stop {
            Some(s) => {
                let notified = s.notified();
                tokio::pin!(notified);
                tokio::select! {
                    _ = tokio::time::sleep(dur) => {}
                    _ = notified => {}
                }
            }
            None => {
                tokio::time::sleep(dur).await;
            }
        }
    }

    if let Some(pacing_str) = &config.iteration_pacing {
        match parse_duration_str(pacing_str) {
            Ok(pacing) => {
                if let Some(actual_dur) = iter_duration {
                    if actual_dur < pacing {
                        let remaining = pacing - actual_dur;
                        if remaining > Duration::from_millis(1) {
                            interruptible_sleep(remaining, stop).await;
                        }
                    }
                }
                return;
            }
            Err(e) => {
                tracing::warn!(
                    "Malformed iterationPacing '{}' ({}): ignoring",
                    pacing_str,
                    e
                );
            }
        }
    }

    if let Some(delay_str) = &config.delay {
        match parse_duration_str(delay_str) {
            Ok(delay) => {
                if delay > Duration::from_millis(1) {
                    interruptible_sleep(delay, stop).await;
                    return;
                }
            }
            Err(e) => {
                tracing::warn!(
                    "Malformed thinkTime.delay '{}' ({}): using zero think time",
                    delay_str,
                    e
                );
            }
        }
    }

    if let (Some(min_str), Some(max_str)) = (&config.min_delay, &config.max_delay) {
        match (parse_duration_str(min_str), parse_duration_str(max_str)) {
            (Ok(min), Ok(max)) if max > Duration::ZERO && max > min => {
                let range_ms = (max - min).as_millis() as u64;
                let rand_ms = rand::rng().random_range(0..=range_ms);
                interruptible_sleep(min + Duration::from_millis(rand_ms), stop).await;
            }
            (Err(e), _) => {
                tracing::warn!(
                    "Malformed thinkTime.minDelay '{}' ({}): ignoring",
                    min_str,
                    e
                );
            }
            (_, Err(e)) => {
                tracing::warn!(
                    "Malformed thinkTime.maxDelay '{}' ({}): ignoring",
                    max_str,
                    e
                );
            }
            _ => {}
        }
    }
}