better plots

Author: pseusys

evaluation/src/typhoon_eval/flow_plot.py

@@ -52,24 +52,35 @@
 }
 
 
-def _parse_lines(lines: list[str]) -> list[dict]:
-    """Extract capture JSONL records from env_logger output lines."""
-    records = []
+def _parse_lines(lines: list[str]) -> tuple[list[dict], list[dict]]:
+    """
+    Extract capture JSONL records from env_logger output lines.
+
+    Returns (packet_records, config_records) separated by ``kind``.
+    Config records (kind="Config") carry flow configuration metadata;
+    all other records are per-packet measurements.
+    """
+    packets: list[dict] = []
+    configs: list[dict] = []
     for line in lines:
         if "typhoon::capture" not in line:
             continue
         brace = line.find("{")
         if brace == -1:
             continue
         try:
-            records.append(json.loads(line[brace:]))
+            rec = json.loads(line[brace:])
         except json.JSONDecodeError:
-            pass
-    return records
+            continue
+        if rec.get("kind") == "Config":
+            configs.append(rec)
+        else:
+            packets.append(rec)
+    return packets, configs
 
 
-def _run_example(example: str, typhoon_dir: Path, timeout: int) -> list[dict]:
-    """Build and run a TYPHOON example with capture logging; return parsed records."""
+def _run_example(example: str, typhoon_dir: Path, timeout: int) -> tuple[list[dict], list[dict]]:
+    """Build and run a TYPHOON example with capture logging; return (packets, configs)."""
     env = {**os.environ, "RUST_LOG": "typhoon::capture=trace"}
     try:
         result = subprocess.run(
@@ -229,15 +240,57 @@ def _compute_xpos(timestamps: list[int]) -> np.ndarray:
     return xpos
 
 
-def _subplot_title(c2s_addr: str | None, s2c_addr: str | None) -> str:
+def _config_annotation(
+    c2s_addr: str | None,
+    s2c_addr: str | None,
+    configs: list[dict],
+) -> str:
+    """
+    Build a compact config annotation string for a subplot.
+
+    Looks up Config records matching each flow address and direction,
+    then formats: ``body=… header=…B decoy=…`` per direction.
+    Returns an empty string if no config records are found.
+    """
+    by_key: dict[tuple[str, str], dict] = {
+        (r.get("flow", ""), r.get("dir", "")): r for r in configs
+    }
+
+    def _fmt(addr: str | None, direction: str) -> str:
+        if addr is None:
+            return ""
+        rec = by_key.get((addr, direction))
+        if rec is None:
+            return ""
+        body = rec.get("body_mode", "?")
+        hdr = rec.get("header_len", "?")
+        decoy = rec.get("decoy", "?")
+        return f"{direction}: body={body}  header={hdr}B  decoy={decoy}"
+
+    parts = [p for p in (_fmt(c2s_addr, "c2s"), _fmt(s2c_addr, "s2c")) if p]
+    return "\n".join(parts)
+
+
+def _subplot_title(
+    c2s_addr: str | None,
+    s2c_addr: str | None,
+    configs: list[dict] | None = None,
+) -> str:
     if c2s_addr == s2c_addr:
-        return c2s_addr or "unknown"
-    parts = []
-    if c2s_addr:
-        parts.append(f"→ {c2s_addr} (c2s)")
-    if s2c_addr:
-        parts.append(f"← {s2c_addr} (s2c)")
-    return "   ".join(parts)
+        base = c2s_addr or "unknown"
+    else:
+        parts = []
+        if c2s_addr:
+            parts.append(f"→ {c2s_addr} (c2s)")
+        if s2c_addr:
+            parts.append(f"← {s2c_addr} (s2c)")
+        base = "   ".join(parts)
+
+    if configs:
+        annotation = _config_annotation(c2s_addr, s2c_addr, configs)
+        if annotation:
+            return f"{base}\n{annotation}"
+    return base
 
 
 def _draw_bars(ax, times_or_xpos, values_by_direction: dict[str, dict[int | float, dict]]) -> None:
@@ -279,6 +332,7 @@ def _plot_all(
     out_dir: Path,
     bucket_ms: int,
     name: str,
+    configs: list[dict] | None = None,
 ) -> None:
     """Render all paired flows as bucketed stacked-bar subplots in a single PNG."""
     pairs = [(c, s, b) for c, s, b in pairs if b]
@@ -309,7 +363,7 @@ def _empty_dirs() -> dict:
         ax.axhline(0, color="black", linewidth=0.8)
         ax.set_xlabel(f"Time ({bucket_ms} ms buckets)")
         ax.set_ylabel("Bytes")
-        ax.set_title(_subplot_title(c2s_addr, s2c_addr))
+        ax.set_title(_subplot_title(c2s_addr, s2c_addr, configs), fontsize=8)
         ax.legend(handles=component_patches + direction_patches, loc="upper right", fontsize=8)
 
     fig.suptitle(name, fontsize=13, fontweight="bold")
@@ -326,6 +380,7 @@ def _plot_all_per_packet(
     pairs_with_records: list[tuple[str | None, str | None, list[dict]]],
     out_dir: Path,
     name: str,
+    configs: list[dict] | None = None,
 ) -> None:
     """
     Render all paired flows as per-packet stacked-bar subplots in a single PNG.
@@ -359,7 +414,7 @@ def _plot_all_per_packet(
 
         ax.axhline(0, color="black", linewidth=0.8)
         ax.set_ylabel("Bytes")
-        ax.set_title(_subplot_title(c2s_addr, s2c_addr))
+        ax.set_title(_subplot_title(c2s_addr, s2c_addr, configs), fontsize=8)
         ax.legend(handles=component_patches + direction_patches, loc="upper right", fontsize=8)
 
         # Place ~10 time-reference ticks labelled with ms-since-start.
@@ -399,13 +454,13 @@ def main(example: str, log_file: str, out_dir: str, typhoon_dir: str, timeout: i
         raise click.UsageError("--example and --log are mutually exclusive.")
 
     if example:
-        records = _run_example(example, Path(typhoon_dir), timeout)
+        records, configs = _run_example(example, Path(typhoon_dir), timeout)
         name = example
     elif log_file == "-":
-        records = _parse_lines(sys.stdin.readlines())
+        records, configs = _parse_lines(sys.stdin.readlines())
         name = "capture"
     else:
-        records = _parse_lines(Path(log_file).read_text().splitlines())
+        records, configs = _parse_lines(Path(log_file).read_text().splitlines())
         name = Path(log_file).stem
 
     if not records:
@@ -420,12 +475,12 @@ def main(example: str, log_file: str, out_dir: str, typhoon_dir: str, timeout: i
     out = Path(out_dir)
     if per_packet:
         pairs = _pair_records(records)
-        _plot_all_per_packet(pairs, out, name)
+        _plot_all_per_packet(pairs, out, name, configs)
     else:
         bms = bucket_ms if bucket_ms > 0 else _auto_bucket_ms(records)
         flows = _bucket(records, bms)
         pairs = _pair_flows(flows)
-        _plot_all(pairs, out, bms, name)
+        _plot_all(pairs, out, bms, name, configs)
 
 
 if __name__ == "__main__":

typhoon/src/capture.rs

@@ -9,7 +9,11 @@
 //!
 //! Each line is a self-contained JSON object with fields:
 //! `t` (unix ms), `dir` (`c2s`/`s2c`), `flow` (addr), `kind`
-//! (`Data`/`Service`/`Decoy`), `tailor`, `crypto`, `header`, `payload`, `body`.
+//! (`Data`/`Service`/`Decoy`/`Config`), `tailor`, `crypto`, `header`, `payload`, `body`.
+//!
+//! All capture functions accept a lazy closure so that argument computation —
+//! string formatting, allocations, etc. — is entirely skipped at zero cost
+//! when the `capture` feature is disabled.
 
 use std::net::SocketAddr;
 
@@ -49,8 +53,16 @@ impl CaptureContext {
     }
 
     /// Emit a c2s (client-to-server) packet record.
+    ///
+    /// `f` is called only when the `capture` feature is enabled; its body
+    /// (including any string construction or arithmetic) is never executed
+    /// otherwise, giving true zero overhead.
     #[cfg(feature = "capture")]
-    pub(crate) fn record_send(&self, kind: &str, tailor: usize, crypto: usize, header: usize, payload: usize, body: usize) {
+    pub(crate) fn record_send<F>(&self, f: F)
+    where
+        F: FnOnce() -> (&'static str, usize, usize, usize, usize, usize),
+    {
+        let (kind, tailor, crypto, header, payload, body) = f();
         trace!(
             target: "typhoon::capture",
             "{{\"t\":{},\"dir\":\"c2s\",\"flow\":\"{}\",\"kind\":\"{kind}\",\"tailor\":{tailor},\"crypto\":{crypto},\"header\":{header},\"payload\":{payload},\"body\":{body}}}",
@@ -61,12 +73,47 @@ impl CaptureContext {
 
     #[cfg(not(feature = "capture"))]
     #[inline(always)]
-    pub(crate) fn record_send(&self, _: &str, _: usize, _: usize, _: usize, _: usize, _: usize) {}
+    pub(crate) fn record_send<F>(&self, _: F)
+    where
+        F: FnOnce() -> (&'static str, usize, usize, usize, usize, usize),
+    {
+    }
+}
+
+/// Emit a configuration record when a flow is established.
+///
+/// `f` is called only when the `capture` feature is enabled.
+/// It should return `(body_mode_description, header_len_bytes, decoy_name)`.
+#[cfg(feature = "capture")]
+pub(crate) fn record_flow_config<F>(flow_addr: SocketAddr, dir: &str, f: F)
+where
+    F: FnOnce() -> (String, usize, &'static str),
+{
+    let (body_mode, header_len, decoy) = f();
+    trace!(
+        target: "typhoon::capture",
+        "{{\"t\":{},\"kind\":\"Config\",\"dir\":\"{dir}\",\"flow\":\"{flow_addr}\",\"body_mode\":\"{body_mode}\",\"header_len\":{header_len},\"decoy\":\"{decoy}\"}}",
+        unix_timestamp_ms(),
+    );
+}
+
+#[cfg(not(feature = "capture"))]
+#[inline(always)]
+pub(crate) fn record_flow_config<F>(_: SocketAddr, _: &str, _: F)
+where
+    F: FnOnce() -> (String, usize, &'static str),
+{
 }
 
 /// Emit an s2c (server-to-client) packet record from the server send path.
+///
+/// `f` is called only when the `capture` feature is enabled.
 #[cfg(feature = "capture")]
-pub(crate) fn record_server_send(addr: SocketAddr, kind: &str, tailor: usize, crypto: usize, header: usize, payload: usize, body: usize) {
+pub(crate) fn record_server_send<F>(addr: SocketAddr, f: F)
+where
+    F: FnOnce() -> (&'static str, usize, usize, usize, usize, usize),
+{
+    let (kind, tailor, crypto, header, payload, body) = f();
     trace!(
         target: "typhoon::capture",
         "{{\"t\":{},\"dir\":\"s2c\",\"flow\":\"{addr}\",\"kind\":\"{kind}\",\"tailor\":{tailor},\"crypto\":{crypto},\"header\":{header},\"payload\":{payload},\"body\":{body}}}",
@@ -76,4 +123,8 @@ pub(crate) fn record_server_send(addr: SocketAddr, kind: &str, tailor: usize, cr
 
 #[cfg(not(feature = "capture"))]
 #[inline(always)]
-pub(crate) fn record_server_send(_: SocketAddr, _: &str, _: usize, _: usize, _: usize, _: usize, _: usize) {}
+pub(crate) fn record_server_send<F>(_: SocketAddr, _: F)
+where
+    F: FnOnce() -> (&'static str, usize, usize, usize, usize, usize),
+{
+}

typhoon/src/flow/client.rs

@@ -4,7 +4,7 @@ use std::sync::{Arc, Weak};
 
 use crate::bytes::DynamicByteBuffer;
 use crate::cache::CachedValue;
-use crate::capture::CaptureContext;
+use crate::capture::{CaptureContext, record_flow_config};
 use crate::crypto::ClientCryptoTool;
 use crate::flow::common::{FlowManager, FlowReceiveInternal, FlowSendInternal};
 use crate::flow::config::FlowConfig;
@@ -35,6 +35,9 @@ impl<T: IdentityType + Clone + 'static, AE: AsyncExecutor + 'static> ClientFlowM
         let manager_ref = Arc::new_cyclic(|m: &Weak<ClientFlowManager<T, AE>>| {
             let mgr: Weak<dyn DecoyFlowSender> = m.clone();
             let decoy = factory(mgr, settings.clone(), identity);
+            record_flow_config(addr, "c2s", || {
+                (config.fake_body_mode.description(), config.fake_header_mode.len(), decoy.name())
+            });
             ClientFlowManager {
                 decoy_provider: Mutex::new(decoy),
                 send_internal: Mutex::new(FlowSendInternal {

typhoon/src/flow/common.rs

@@ -111,8 +111,10 @@ impl<CP: FlowCryptoProvider> FlowSendInternal<CP> {
         self.config.fake_header_mode.fill(full_packet.rebuffer_end(fake_header_len));
         get_rng().fill(&mut full_packet.rebuffer_both(fake_header_len, full_packet_len));
 
-        let kind = if packet_flags.is_discardable() { "Decoy" } else if packet_flags.is_service() { "Service" } else { "Data" };
-        self.capture.record_send(kind, full_tailor_len, CP::tailor_overhead(), fake_header_len, data_len, full_packet_len - fake_header_len);
+        self.capture.record_send(|| {
+            let kind = if packet_flags.is_discardable() { "Decoy" } else if packet_flags.is_service() { "Service" } else { "Data" };
+            (kind, full_tailor_len, CP::tailor_overhead(), fake_header_len, data_len, full_packet_len - fake_header_len)
+        });
 
         Ok(full_packet)
     }

typhoon/src/flow/config.rs

@@ -37,6 +37,16 @@ pub enum FakeBodyMode {
 }
 
 impl FakeBodyMode {
+    /// Human-readable description of this mode for capture log records.
+    #[inline]
+    pub(crate) fn description(&self) -> String {
+        match self {
+            FakeBodyMode::Empty => "Empty".to_string(),
+            FakeBodyMode::Random { min_length, max_length, service } => format!("Random({min_length}..{max_length},svc={service})"),
+            FakeBodyMode::Constant { packet_length } => format!("Constant({packet_length})"),
+        }
+    }
+
     /// Maximum fake body length this mode can produce — used to bound MTU calculations.
     pub fn max_len(&self) -> usize {
         match self {

typhoon/src/flow/decoy/common.rs

@@ -227,6 +227,9 @@ impl<T: FlowManager + Send + Sync> DecoyFlowSender for T {
 /// flow managers. All async methods are boxed automatically by `async_trait`.
 #[async_trait]
 pub trait DecoyProvider: Send + Sync {
+    /// Short display name of this provider (e.g. "SparseDecoyProvider").
+    fn name(&self) -> &'static str;
+
     /// Start the background decoy generation timer.
     async fn start(&mut self);
 

typhoon/src/flow/decoy/heavy.rs

@@ -99,6 +99,9 @@ impl<T: IdentityType + Clone, AE: AsyncExecutor> HeavyDecoyProvider<T, AE> {
 
 #[async_trait]
 impl<T: IdentityType + Clone + 'static, AE: AsyncExecutor + 'static> DecoyProvider for HeavyDecoyProvider<T, AE> {
+    #[inline]
+    fn name(&self) -> &'static str { "HeavyDecoyProvider" }
+
     async fn start(&mut self) {
         let executor = {
             let lock = self.state.read().await;

typhoon/src/flow/decoy/mod.rs

@@ -42,7 +42,7 @@ where
     DP: DecoyCommunicationMode<T, AE> + 'static,
 {
     Arc::new(|manager, settings, identity| {
-        info!("decoy provider: {}", DP::name());
+        info!("decoy provider: {}", <DP as DecoyCommunicationMode<T, AE>>::name());
         Box::new(DP::new(manager, settings, identity))
     })
 }

typhoon/src/flow/decoy/noisy.rs

@@ -97,6 +97,9 @@ impl<T: IdentityType + Clone, AE: AsyncExecutor> NoisyDecoyProvider<T, AE> {
 
 #[async_trait]
 impl<T: IdentityType + Clone + 'static, AE: AsyncExecutor + 'static> DecoyProvider for NoisyDecoyProvider<T, AE> {
+    #[inline]
+    fn name(&self) -> &'static str { "NoisyDecoyProvider" }
+
     async fn start(&mut self) {
         let executor = {
             let lock = self.state.read().await;

typhoon/src/flow/decoy/simple.rs

@@ -14,6 +14,9 @@ pub struct SimpleDecoyProvider;
 
 #[async_trait]
 impl DecoyProvider for SimpleDecoyProvider {
+    #[inline]
+    fn name(&self) -> &'static str { "SimpleDecoyProvider" }
+
     async fn start(&mut self) {}
 
     async fn feed_input(&mut self, packet: DynamicByteBuffer) -> Option<DynamicByteBuffer> {