Add utilities for plotting latency distribution for libp2p

Author: AlbertoSoutullo

analysis/src/mesh_analysis/analyzers/nimlibp2p_analyzer.py

@@ -1,7 +1,6 @@
 # Python Imports
 import logging
 from pathlib import Path
-
 import pandas as pd
 from typing import List, Optional, Tuple
 from result import Result, Err, Ok
@@ -10,7 +9,7 @@
 from src.mesh_analysis.readers.tracers.nimlibp2p_tracer import Nimlibp2pTracer
 from src.mesh_analysis.stacks.vaclab_stack_analysis import VaclabStackAnalysis
 # Project Imports
-from src.utils import path_utils, file_utils
+
 
 logger = logging.getLogger(__name__)
 
@@ -42,6 +41,21 @@ def _set_up_paths(self, dump_analysis_dir: str, local_folder_to_analyze: str):
             logger.error(result.err_value)
             exit(1)
 
+    def grab_latencies(self):
+        tracer = Nimlibp2pTracer(extra_fields=self._kwargs['extra_fields']) \
+            .with_latency_pattern()
+
+        queries = ['milliseconds']
+        reader_builder = VictoriaReaderBuilder(tracer, queries, **self._kwargs)
+        stack_analysis = VaclabStackAnalysis(reader_builder, **self._kwargs)
+
+        dfs = stack_analysis.get_all_node_dataframes(4)
+        df = self._merge_latency_dfs(dfs)
+        result = self._dump_latency_df(df)
+        if result.is_err():
+            logger.error(f'Issue dumping message summary. {result.err_value}')
+
+
     def analyze_reliability(self, n_jobs: int):
         """
         This function automatically assumes two scenarios, analysis from online data, and analysis from local data.
@@ -151,6 +165,15 @@ def _merge_mix_dfs(self, dfs: List[List[pd.DataFrame]]) -> List[pd.DataFrame]:
 
         return [received_df, sent_df, mix_df]
 
+    def _merge_latency_dfs(self, dfs: List[List[pd.DataFrame]]) -> pd.DataFrame:
+        logger.info("Merging and sorting information")
+
+        received_df = pd.concat([pd.concat(group[0], ignore_index=True) for group in dfs], ignore_index=True)
+        received_df.set_index(['latency'], inplace=True)
+        received_df.sort_index(inplace=True)
+
+        return received_df
+
     def _dump_dfs(self, dfs: List[pd.DataFrame]) -> Result:
         received = dfs[0].reset_index()
         received = received.astype(str)
@@ -197,6 +220,17 @@ def _dump_mix_dfs(self, dfs: List[pd.DataFrame]) -> Result:
 
         return Ok(None)
 
+    def _dump_latency_df(self, df: pd.DataFrame) -> Result:
+        received = df.reset_index()
+        received = received.astype(str)
+        logger.info("Dumping received information")
+        result = file_utils.dump_df_as_csv(received, self._dump_analysis_path / 'summary' / 'latencies.csv', False)
+        if result.is_err():
+            logger.warning(result.err_value)
+            return Err(result.err_value)
+
+        return Ok(None)
+
     def _has_message_reliability_issues(self, msg_identifier: str, peer_identifier: str,
                                         received_df: pd.DataFrame, sent_df: pd.DataFrame,
                                         issue_dump_location: Path) -> Optional[List[str]]:

analysis/src/mesh_analysis/analyzers/waku/waku_plots.py

@@ -137,6 +137,57 @@ def get_mixnet_and_outside_time(group):
 
     return Ok(None)
 
+def plot_message_latency_distributions(
+    received_summary_paths: list[Path],
+    plot_title: str,
+    dump_path: Path
+) -> Result[None, str]:
+
+    # Validate input files
+    for p in received_summary_paths:
+        if not p.exists():
+            error = f"Received summary file {p} does not exist"
+            logger.error(error)
+            return Err(error)
+
+    sns.set_theme()
+
+    all_ranges = []
+
+    # Process each file separately
+    for path in received_summary_paths:
+        df = pd.read_csv(path)
+        #df.set_index(['shard', 'msg_hash', 'timestamp'], inplace=True)
+
+        df["source"] = path.stem     # label dataset
+
+        all_ranges.append(df)
+
+    # Merge all datasets into one
+    merged = pd.concat(all_ranges, ignore_index=True)
+
+    plt.figure(figsize=(14, 6))
+    ax = sns.boxplot(
+        x='source',
+        y='latency',
+        data=merged,
+        showfliers=False
+    )
+
+    add_boxplot_stat_labels(ax, value_type="min")
+    add_boxplot_stat_labels(ax, value_type="max")
+    add_boxplot_stat_labels(ax, value_type="median")
+
+    plt.ylabel("Latency distribution (ms)")
+    plt.xlabel("Dataset")
+    plt.title(plot_title)
+
+    plt.savefig(dump_path)
+    plt.show()
+
+    return Ok(None)
+
+
 def plot_message_distribution(received_summary_path: Path, plot_title: str, dump_path: Path) -> Result[
     None, str]:
     """

analysis/src/mesh_analysis/readers/tracers/nimlibp2p_tracer.py

@@ -30,6 +30,20 @@ def get_num_patterns_group(self) -> int:
     def get_patterns(self) -> List[List[str]]:
         return self._patterns
 
+    def with_latency_pattern(self):
+        patterns = [
+            r'(\d+)$'
+        ]
+
+        tracers = [
+            self._trace_latencies_in_logs,
+        ]
+
+        self._patterns.append(patterns)
+        self._tracings.append(tracers)
+
+        return self
+
     def with_received_pattern_group(self) -> Self:
         patterns = [
             r'Received message.*?msgId=([\w*]+).*?sentAt=([\w*]+).*?current=([\w*]+).*?delayMs=([\w*]+)'
@@ -101,6 +115,16 @@ def _trace_received_in_logs(self, parsed_logs: List) -> pd.DataFrame:
 
         return df
 
+    def _trace_latencies_in_logs(self, parsed_logs: List) -> pd.DataFrame:
+        columns = ['latency']
+        if self._extra_fields is not None:
+            columns.extend(self._extra_fields)
+
+        df = pd.DataFrame(parsed_logs, columns=columns)
+        df['latency'] = pd.to_numeric(df['latency'], errors='coerce').fillna(-1).astype(int)
+
+        return df
+
     def _trace_sent_in_logs(self, parsed_logs: List) -> pd.DataFrame:
         columns = ['msgId', 'timestamp']
         if self._extra_fields is not None: