add bandwidth measurement

mixnet/v2/sim/README.md

@@ -47,6 +47,13 @@ For more details, please see the [Time and Scheduling](https://simpy.readthedocs
   - [x] Naive random delays
   - [ ] More sophisticated delays (e.g. Poisson) if necessary
 
+## Performance Measurements
+
+- [ ] Bandwidth Usage
+  - DRAFT with the naive 1-to-all broadcasting
+    ![](./docs/bandwidth.png)
+  - Should be measured with realistic parameters and P2P gossiping.
+
 ## [Adversary Models](https://www.notion.so/Mixnet-v2-Proof-of-Concept-102d0563e75345a3a6f1c11791fbd746?pvs=4#c5ffa49486ce47ed81d25028bc0d9d40)
 - [x] Inspecting message sizes to analyze how far each message has traveled since emitted by the original sender.
   - Currently, all messages have the same size (including messages broadcasted after being fully unwrapped). Thus, the adversary can learn nothing.

mixnet/v2/sim/analysis.py

@@ -11,11 +11,37 @@ class Analysis:
         self.sim = sim
 
     def run(self):
+        self.bandwidth()
         self.message_size_distribution()
         self.messages_emitted_around_interval()
         self.mixed_messages_per_node_over_time()
         self.node_states()
 
+    def bandwidth(self):
+        dataframes = []
+        for ingress_bandwidths, egress_bandwidths in zip(self.sim.p2p.measurement.ingress_bandwidth_per_time, self.sim.p2p.measurement.egress_bandwidth_per_time):
+            rows = []
+            for node in self.sim.p2p.nodes:
+                rows.append((node.id, ingress_bandwidths[node]/1024.0, egress_bandwidths[node]/1024.0))
+            df = pd.DataFrame(rows, columns=["node_id", "ingress", "egress"])
+            dataframes.append(df)
+        times = range(len(dataframes))
+        df = pd.concat([df.assign(Time=time) for df, time in zip(dataframes, times)], ignore_index=True)
+        df = df.pivot(index="Time", columns="node_id", values=["ingress", "egress"])
+        plt.figure(figsize=(12, 6))
+        for column in df.columns:
+            marker = "x" if column[0] == "ingress" else "o"
+            plt.plot(df.index, df[column], marker=marker, label=column[0])
+        plt.title("Ingress/egress bandwidth of each node over time")
+        plt.xlabel("Time")
+        plt.ylabel("Bandwidth (KiB/s)")
+        # Customize the legend to show only 'ingress' and 'egress' regardless of node_id
+        handles, labels = plt.gca().get_legend_handles_labels()
+        by_label = dict(zip(labels, handles))
+        plt.legend(by_label.values(), by_label.keys())
+        plt.grid(True)
+        plt.show()
+
     def message_size_distribution(self):
         df = pd.DataFrame(self.sim.p2p.adversary.message_sizes, columns=["message_size"])
         print(df.describe())

mixnet/v2/sim/config.py

@@ -11,6 +11,7 @@ class Config:
     simulation: SimulationConfig
     mixnet: MixnetConfig
     p2p: P2pConfig
+    measurement: MeasurementConfig
     adversary: AdversaryConfig
 
     @classmethod
@@ -23,6 +24,7 @@ class Config:
         config.simulation.validate()
         config.mixnet.validate()
         config.p2p.validate()
+        config.measurement.validate()
         config.adversary.validate()
         
         return config
@@ -78,10 +80,19 @@ class P2pConfig:
         assert self.max_network_latency >= 0
 
 
+@dataclass
+class MeasurementConfig:
+    # How many times in simulation represent 1 second in real time
+    sim_time_per_second: float
+
+    def validate(self):
+        assert self.sim_time_per_second > 0
+
+
 @dataclass
 class AdversaryConfig:
     # A discrete time window for the adversary to observe inputs and outputs of a certain node
     io_observation_window: int
 
     def validate(self):
-        assert self.io_observation_window >= 0
+        assert self.io_observation_window >= 1

mixnet/v2/sim/config.yaml

@@ -25,6 +25,10 @@ p2p:
   # A maximum network latency between nodes directly connected with each other
   max_network_latency: 0.5
 
+measurement:
+    # How many times in simulation represent 1 second in real time
+    sim_time_per_second: 1
+
 adversary:
   # A discrete time window for the adversary to observe inputs and outputs of a certain node
   io_observation_window: 1

mixnet/v2/sim/measurement.py

@@ -0,0 +1,32 @@
+from collections import defaultdict
+from typing import TYPE_CHECKING
+
+import simpy
+
+from config import Config
+from sphinx import SphinxPacket
+
+if TYPE_CHECKING:
+    from node import Node
+
+class Measurement:
+    def __init__(self, env: simpy.Environment, config: Config):
+        self.env = env
+        self.config = config
+        self.ingress_bandwidth_per_time = []
+        self.egress_bandwidth_per_time = []
+
+        self.env.process(self.update_bandwidth_window())
+
+    def measure_ingress(self, node: "Node", msg: SphinxPacket | bytes):
+        self.ingress_bandwidth_per_time[-1][node] += len(msg)
+
+    def measure_egress(self, node: "Node", msg: SphinxPacket | bytes):
+        self.egress_bandwidth_per_time[-1][node] += len(msg)
+
+    def update_bandwidth_window(self):
+        while True:
+            self.ingress_bandwidth_per_time.append(defaultdict(int))
+            self.egress_bandwidth_per_time.append(defaultdict(int))
+            yield self.env.timeout(self.config.measurement.sim_time_per_second)
+

mixnet/v2/sim/p2p.py

@@ -6,6 +6,7 @@ import simpy
 
 from adversary import Adversary
 from config import Config
+from measurement import Measurement
 from sphinx import SphinxPacket
 
 if TYPE_CHECKING:
@@ -17,6 +18,7 @@ class P2p:
         self.env = env
         self.config = config
         self.nodes = []
+        self.measurement = Measurement(env, config)
         self.adversary = Adversary(env, config)
 
     def add_node(self, nodes: list["Node"]):
@@ -27,10 +29,11 @@ class P2p:
 
     # This should accept only bytes in practice,
     # but we accept SphinxPacket as well because we don't implement Sphinx deserialization.
-    def broadcast(self, sender, msg: SphinxPacket | bytes):
+    def broadcast(self, sender: "Node", msg: SphinxPacket | bytes):
         self.log("Broadcasting a msg: %d bytes" % len(msg))
 
         # Adversary
+        self.measurement.measure_egress(sender, msg)
         self.adversary.inspect_message_size(msg)
         self.adversary.observe_outgoing_message(sender)
 
@@ -42,10 +45,11 @@ class P2p:
         for node in self.nodes:
             self.env.process(self.send(msg, node))
 
-    def send(self, msg: SphinxPacket | bytes, node):
+    def send(self, msg: SphinxPacket | bytes, node: "Node"):
         # simulate network latency
         yield self.env.timeout(random.uniform(0, self.config.p2p.max_network_latency))
 
+        self.measurement.measure_ingress(node, msg)
         self.adversary.observe_incoming_message(node)
         self.env.process(node.receive_message(msg))