concurrency/
concurrency.rs

1//! Concurrent RakNet echo validation; run only inside an isolated network.
2use rust_raknet::{RaknetSocket, Reliability};
3use std::{
4    error::Error,
5    sync::Arc,
6    time::{Duration, Instant},
7};
8use tokio::{sync::Semaphore, task::JoinSet};
9
10#[tokio::main(worker_threads = 4)]
11async fn main() -> Result<(), Box<dyn Error>> {
12    let args: Vec<_> = std::env::args().collect();
13    let address = args
14        .get(1)
15        .ok_or("usage: concurrency ADDRESS [CONNECTIONS] [MESSAGES]")?
16        .parse()?;
17    let connections: usize = args.get(2).map_or(Ok(512), |n| n.parse())?;
18    let messages: usize = args.get(3).map_or(Ok(100), |n| n.parse())?;
19    if connections == 0 || messages == 0 {
20        return Err("counts must be positive".into());
21    }
22    tokio::time::timeout(Duration::from_secs(120), async {
23        let started = Instant::now();
24        let dialing = Arc::new(Semaphore::new(8));
25        let mut tasks = JoinSet::new();
26        for index in 0..connections {
27            let dialing = dialing.clone();
28            tasks.spawn(async move {
29                let _permit = dialing.acquire().await.unwrap();
30                let socket = RaknetSocket::connect_with_version(&address, 11).await?;
31                Ok::<_, rust_raknet::error::RaknetError>((index, socket))
32            });
33        }
34        let mut sockets = Vec::with_capacity(connections);
35        while let Some(result) = tasks.join_next().await { sockets.push(result??); }
36        let handshake_seconds = started.elapsed().as_secs_f64();
37        let transfer = Instant::now();
38        let mut exchanges = JoinSet::new();
39        for (index, socket) in sockets {
40            exchanges.spawn(async move {
41                fn payload(connection: usize, message: usize) -> Vec<u8> {
42                    let size = [64, 800, 4096][message % 3];
43                    let mut bytes = vec![0xfe; size];
44                    bytes[1..9].copy_from_slice(&(connection as u64).to_le_bytes());
45                    bytes[9..17].copy_from_slice(&(message as u64).to_le_bytes());
46                    bytes
47                }
48                let sender = async {
49                    for message in 0..messages {
50                        socket.send(&payload(index, message), Reliability::ReliableOrdered).await?;
51                    }
52                    socket.flush().await
53                };
54                let receiver = async {
55                    // Exercise a slow application without blocking protocol ACKs.
56                    tokio::time::sleep(Duration::from_millis(30)).await;
57                    for message in 0..messages {
58                        let actual = socket.recv().await?;
59                        if actual != payload(index, message) {
60                            return Err(rust_raknet::error::RaknetError::PacketParseError);
61                        }
62                    }
63                    Ok(())
64                };
65                tokio::try_join!(sender, receiver)?;
66                socket.close().await
67            });
68        }
69        while let Some(result) = exchanges.join_next().await { result??; }
70        println!("PASS: {connections} simultaneous connections, {} unique ordered echoes; handshake={handshake_seconds:.3}s, transfer={:.3}s", connections * messages, transfer.elapsed().as_secs_f64());
71        Ok::<_, Box<dyn Error>>(())
72    }).await?
73}
concurrency/concurrency.rs

concurrency/
concurrency.rs
