MGen/mgentools/src/main.rs

use std::collections::HashMap;
use std::{error::Error, fs::File, io, io::Write, process};

use serde::Deserialize;

#[derive(Debug, Deserialize)]
struct Record {
    _hr_time: String,
    time: f64, // FIXME: use a better type, either a real datetime or our own two ints
    user: String,
    group: String,
    action_data: Vec<String>,
}

struct RunningValue {
    sent_count: u32,
    receipt_count: u32,
    running_rtt_mean: f64,
}

fn core(path: Option<String>) -> Result<(), Box<dyn Error>> {
    let mut rtt_all_file = File::create("rtt_all.mgen.json")?;
    writeln!(rtt_all_file, "[")?;

    let mut sent_times: HashMap<(String, String, u32), f64> = HashMap::new();
    let mut running_values: HashMap<(String, String), RunningValue> = HashMap::new();
    let file: Box<dyn io::Read> = if let Some(path) = path {
        Box::new(File::open(path)?)
    } else {
        Box::new(io::stdin())
    };
    let mut rdr = csv::ReaderBuilder::new()
        .has_headers(false)
        .flexible(true)
        .from_reader(file);
    for result in rdr.deserialize() {
        let record: Record = result?;
        match record.action_data[0].as_str() {
            "send" => {
                let id = record.action_data[4].parse()?;
                sent_times.insert((record.user.clone(), record.group.clone(), id), record.time);
                running_values
                    .entry((record.user, record.group))
                    .and_modify(|running_value| running_value.sent_count += 1)
                    .or_insert(RunningValue {
                        sent_count: 1,
                        receipt_count: 0,
                        running_rtt_mean: 0.0,
                    });
            }
            "receive" => {
                if record.action_data[4] == "receipt" {
                    let id = record.action_data[5].parse()?;
                    let key = (record.user, record.group, id);
                    let Some(sent_time) = sent_times.get(&key) else {
                        // this should never happen in the client-server case,
                        // but we filter out early conversation in the p2p case
                        //eprintln!("receipt for unknown message: {:?}", key);
                        //panic!();
                        continue;
                    };
                    let rtt = record.time - sent_time;
                    writeln!(rtt_all_file, "  {},", rtt)?;

                    let key = (key.0, key.1);
                    running_values.entry(key).and_modify(|running_value| {
                        running_value.receipt_count += 1;
                        running_value.running_rtt_mean = running_value.running_rtt_mean
                            + (rtt - running_value.running_rtt_mean)
                                / (running_value.receipt_count as f64);
                    });
                }
            }
            _ => (),
        }
    }
    writeln!(rtt_all_file, "]")?;
    drop(rtt_all_file);

    let mut rtt_count_file = File::create("counts.mgen.json")?;
    let mut rtt_mean_file = File::create("rtt_mean.mgen.json")?;
    writeln!(rtt_count_file, "[")?;
    writeln!(rtt_mean_file, "[")?;
    for value in running_values.into_values() {
        writeln!(rtt_count_file, "  {},", value.sent_count)?;
        writeln!(rtt_mean_file, "  {},", value.running_rtt_mean)?;
    }
    writeln!(rtt_count_file, "]")?;
    writeln!(rtt_mean_file, "]")?;

    Ok(())
}

fn main() {
    let mut args = std::env::args();
    let _ = args.next();
    let path = args.next();
    if let Err(err) = core(path) {
        println!("error running example: {}", err);
        process::exit(1);
    }
}