strest 0.1.10

Blazing-fast async HTTP load tester in Rust - lock-free design, real-time stats, distributed runs, and optional chart exports for high-load API testing.
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

use std::sync::{
    Arc,
    atomic::{AtomicU64, Ordering},
};

use tokio::{sync::Semaphore, time::Instant};

use crate::{
    metrics::Metrics,
    shutdown::{ShutdownReceiver, ShutdownSender},
};

use super::data::{RequestLimiter, WorkerContext};

pub(super) struct InflightGuard<'counter> {
    counter: &'counter AtomicU64,
}

impl<'counter> InflightGuard<'counter> {
    pub(super) fn acquire(counter: &'counter Arc<AtomicU64>) -> Self {
        counter.fetch_add(1, Ordering::Relaxed);
        Self {
            counter: counter.as_ref(),
        }
    }
}

impl Drop for InflightGuard<'_> {
    fn drop(&mut self) {
        loop {
            let current = self.counter.load(Ordering::Relaxed);
            let Some(next) = current.checked_sub(1) else {
                break;
            };
            if self
                .counter
                .compare_exchange(current, next, Ordering::Relaxed, Ordering::Relaxed)
                .is_ok()
            {
                break;
            }
        }
    }
}

pub(super) async fn run_and_record(
    shutdown_rx: &mut ShutdownReceiver,
    worker: &WorkerContext<'_>,
    latency_start: Option<Instant>,
    run_request: impl std::future::Future<Output = (u16, bool, bool, u64)>,
) -> bool {
    let start = latency_start.unwrap_or_else(Instant::now);
    let in_flight_guard = InflightGuard::acquire(worker.in_flight_counter);
    let (status, timed_out, transport_error, response_bytes) = if worker.wait_ongoing {
        run_request.await
    } else {
        tokio::select! {
            _ = shutdown_rx.recv() => return true,
            result = run_request => result,
        }
    };
    drop(in_flight_guard);

    let in_flight_ops = worker.in_flight_counter.load(Ordering::Relaxed);
    let metric = Metrics::new(
        start,
        status,
        timed_out,
        transport_error,
        response_bytes,
        in_flight_ops,
    );
    if let Some(log_sink) = worker.log_sink
        && !log_sink.send(metric)
    {
        return true;
    }
    if worker.metrics_tx.try_send(metric).is_err() {
        // Ignore UI backpressure; summary and charts use log pipeline.
    }
    false
}

pub(super) async fn prepare_iteration(
    shutdown_rx: &mut ShutdownReceiver,
    shutdown_tx: &ShutdownSender,
    request_limiter: Option<&Arc<RequestLimiter>>,
    rate_limiter: Option<&Arc<Semaphore>>,
    wait_ongoing: bool,
    latency_correction: bool,
) -> Option<Option<Instant>> {
    if wait_ongoing && shutdown_rx.try_recv().is_ok() {
        return None;
    }
    if let Some(request_limiter) = request_limiter
        && !request_limiter.try_reserve(shutdown_tx)
    {
        return None;
    }
    let mut latency_start = if latency_correction {
        Some(Instant::now())
    } else {
        None
    };
    if let Some(rate_limiter) = rate_limiter {
        let denied = tokio::select! {
            _ = shutdown_rx.recv() => true,
            permit = rate_limiter.acquire() => permit.is_err(),
        };
        if denied {
            return None;
        }
        if !latency_correction {
            latency_start = None;
        }
    }

    Some(latency_start)
}