aggligator 0.12.1

Aggregates multiple links (TCP, WebSocket, USB or similar) into one connection having their combined bandwidth and provides resiliency against failure of individual links, allowing failover and roaming between networks in user space.
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

//! Dump data for performance analysis.
//!
//! The purpose of the data is to debug connection performance issues
//! and to help with the development of Aggligator.
//!
//! The data can be visualized using the `PlotDump.ipynb` script
//! from the repository.
//!

use serde::{Deserialize, Serialize};
use std::{io::Error, path::Path};
use tokio::{
    fs::File,
    io::{AsyncWriteExt, BufWriter},
    sync::mpsc,
};

/// Link dump data for analysis.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct LinkDump {
    /// Whether the link is present.
    pub present: bool,
    /// The link id.
    pub link_id: u128,
    /// Whether the link is currently unconfirmed.
    pub unconfirmed: bool,
    /// Whether the sender is idle.
    pub tx_idle: bool,
    /// Whether the sender is being polled for readiness.
    pub tx_pending: bool,
    /// Whether the sender is being flushed.
    pub tx_flushing: bool,
    /// Whether the sender has been flushed.
    pub tx_flushed: bool,
    /// Then length of the acknowledgement queue.
    pub tx_ack_queue: usize,
    /// Amount of sent, unacknowledged data.
    pub txed_unacked_data: usize,
    /// Current limit of sent, unacknowledged data.
    pub txed_unacked_data_limit: usize,
    /// How many times `txed_unacked_data_limit` has been increased
    /// without send buffer overrun.
    pub txed_unacked_data_limit_increased_consecutively: usize,
    /// Total bytes sent.
    pub total_sent: u64,
    /// Total bytes received.
    pub total_recved: u64,
    /// Roundtrip time in milliseconds.
    pub roundtrip: f32,
}

/// Connection dump data for analysis.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ConnDump {
    /// Connection id.
    pub conn_id: u128,
    /// Running time in seconds.
    pub runtime: f32,
    /// Amount of sent, unacknowledged data.
    pub txed_unacked: usize,
    /// Amount of sent, unconsumed data.
    pub txed_unconsumed: usize,
    /// Amount of sent, unconsumable data.
    pub txed_unconsumable: usize,
    /// Send buffer size.
    pub send_buffer: u32,
    /// Receive buffer size of remote endpoint.
    pub remote_receive_buffer: u32,
    /// Resend queue length.
    pub resend_queue: usize,
    /// Amount of received, unconsumable data.
    pub rxed_reliable_size: usize,
    /// Amount of received data consumed since sending last
    /// Consumed message.
    pub rxed_reliable_consumed_since_last_ack: usize,
    /// Link 0.
    pub link0: LinkDump,
    /// Link 1.
    pub link1: LinkDump,
    /// Link 2.
    pub link2: LinkDump,
    /// Link 3.
    pub link3: LinkDump,
    /// Link 4.
    pub link4: LinkDump,
    /// Link 5.
    pub link5: LinkDump,
    /// Link 6.
    pub link6: LinkDump,
    /// Link 7.
    pub link7: LinkDump,
    /// Link 8.
    pub link8: LinkDump,
    /// Link 9.
    pub link9: LinkDump,
}

/// Dumps analysis data from the channel to a JSON line file.
///
/// The file has one JSON object per line.
pub async fn dump_to_json_line_file(
    path: impl AsRef<Path>, mut rx: mpsc::Receiver<ConnDump>,
) -> Result<(), Error> {
    let file = File::create(path).await?;
    let mut writer = BufWriter::new(file);

    while let Some(dump) = rx.recv().await {
        let line = serde_json::to_vec(&dump).unwrap();
        writer.write_all(&line).await?;
        writer.write_u8(b'\n').await?;
    }

    writer.flush().await?;

    Ok(())
}