pharos-plugins 0.1.0

CoreDNS-compatible Plugins for the Pharos DNS server
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

#![forbid(unsafe_code)]

// Run with: cargo test --release -p pharos-plugins --test proxy_throughput -- --nocapture
// The fake upstream delays every response asynchronously by 10 ms. The results
// measure concurrency under upstream latency, not maximum loopback throughput.

use std::{sync::Arc, time::Instant};

use pharos_plugins::proxy::{Options, Proxy, Transport};
use tokio::{net::UdpSocket, sync::Barrier, time::Duration};

const QUERY: &[u8] = &[0x12, 0x34, 1, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 2, 0, 1];
const REQUESTS: usize = 128;
const CONCURRENCY: usize = 16;
const DELAY: Duration = Duration::from_millis(10);

async fn measure(proxy: Arc<Proxy>, concurrency: usize) -> Duration {
    let ready = Arc::new(Barrier::new(concurrency + 1));
    let start = Arc::new(Barrier::new(concurrency + 1));
    let mut workers = Vec::with_capacity(concurrency);
    for _ in 0..concurrency {
        let proxy = Arc::clone(&proxy);
        let ready = Arc::clone(&ready);
        let start = Arc::clone(&start);
        workers.push(tokio::spawn(async move {
            let mut response = [0; 512];
            proxy
                .exchange(QUERY, &mut response, Transport::Udp, Options::default())
                .await
                .unwrap();
            ready.wait().await;
            start.wait().await;
            for _ in 0..REQUESTS / concurrency {
                proxy
                    .exchange(QUERY, &mut response, Transport::Udp, Options::default())
                    .await
                    .unwrap();
            }
        }));
    }
    ready.wait().await;
    let began = Instant::now();
    start.wait().await;
    for worker in workers {
        worker.await.unwrap();
    }
    began.elapsed()
}

#[test]
fn proxy_throughput() {
    let runtime = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(2)
        .enable_all()
        .build()
        .unwrap();
    runtime.block_on(async {
        let socket = Arc::new(UdpSocket::bind("127.0.0.1:0").await.unwrap());
        let proxy = Arc::new(Proxy::new(socket.local_addr().unwrap()));
        let upstream = tokio::spawn(async move {
            loop {
                let mut wire = [0; 512];
                let (len, peer) = socket.recv_from(&mut wire).await.unwrap();
                wire[2] |= 0x80;
                let socket = Arc::clone(&socket);
                tokio::spawn(async move {
                    tokio::time::sleep(DELAY).await;
                    socket.send_to(&wire[..len], peer).await.unwrap();
                });
            }
        });

        let single = tokio::time::timeout(Duration::from_secs(15), measure(Arc::clone(&proxy), 1))
            .await
            .unwrap();
        let parallel = tokio::time::timeout(Duration::from_secs(15), measure(proxy, CONCURRENCY))
            .await
            .unwrap();
        println!(
            "UDP fake upstream: {DELAY:?} asynchronous response delay; {REQUESTS} requests per run"
        );
        for (concurrency, elapsed) in [(1, single), (CONCURRENCY, parallel)] {
            println!(
                "concurrency {concurrency}: {elapsed:?}, {:.1} requests/s",
                REQUESTS as f64 / elapsed.as_secs_f64()
            );
        }
        println!(
            "throughput ratio: {:.2}x",
            single.as_secs_f64() / parallel.as_secs_f64()
        );
        upstream.abort();
    });
}