crab-net 0.0.4

<TCP|UDP> + TLS packets generator
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

use coarsetime::{Duration, Instant};
use kanal::AsyncSender;
use log::debug;
use tokio::{
    io::{AsyncWrite, AsyncWriteExt},
    net::UdpSocket,
    time::sleep,
};

use crate::{DtlsSession, statistics::StatPacket};

pub async fn sender_task_udp(
    id: usize,
    socket: UdpSocket,
    payload: Vec<u8>,
    rate: usize,
    stats_tx: AsyncSender<StatPacket>,
) {
    debug!("client {id} spawned");
    let one_sec = Duration::new(1, 0);

    loop {
        let start_time = Instant::now();
        let mut packets_error = 0;

        for _ in 0..rate {
            if socket.send(&payload).await.is_err() {
                packets_error += 1;
            }
        }

        send_stats(rate, payload.len(), packets_error, &stats_tx).await;
        maybe_sleep(start_time, one_sec).await;
    }
}

pub async fn sender_task_dtls(
    id: usize,
    mut session: DtlsSession,
    payload: Vec<u8>,
    rate: usize,
    stats_tx: AsyncSender<StatPacket>,
) {
    debug!("client {id} spawned");
    let one_sec = Duration::new(1, 0);

    loop {
        let start_time = Instant::now();
        let mut packets_error = 0;

        for _ in 0..rate {
            if session.write(&payload).await.is_err() {
                packets_error += 1;
            }
        }

        send_stats(rate, payload.len(), packets_error, &stats_tx).await;
        maybe_sleep(start_time, one_sec).await;
    }
}

pub async fn sender_task_tcp(
    id: usize,
    mut stream: Box<dyn AsyncWrite + Unpin + Send>,
    payload: Vec<u8>,
    rate: usize,
    stats_tx: AsyncSender<StatPacket>,
) {
    debug!("client {id} spawned");
    let one_sec = Duration::new(1, 0);

    loop {
        let start_time = Instant::now();
        let mut packets_error = 0;

        for _ in 0..rate {
            if stream.write_all(&payload).await.is_err() {
                packets_error += 1;
            }
        }

        send_stats(rate, payload.len(), packets_error, &stats_tx).await;
        maybe_sleep(start_time, one_sec).await;
    }
}

async fn send_stats(
    rate: usize,
    payload_len: usize,
    packets_error: usize,
    stats_tx: &AsyncSender<StatPacket>,
) {
    let packets_sent = rate - packets_error;
    let _ = stats_tx
        .send((packets_sent * payload_len, packets_sent))
        .await;
}

async fn maybe_sleep(start_time: Instant, duration: Duration) {
    let time_elapsed = Instant::now() - start_time;

    if time_elapsed < duration {
        let time_to_sleep = duration - time_elapsed;
        sleep(time_to_sleep.into()).await;
    }
}