refactor: p2p measuring

mixnet/v2/sim/p2p.py

@@ -39,11 +39,23 @@ class P2P(ABC):
         # Without any yield, SimPy complains that the broadcast func is not a generator.
         yield self.env.timeout(0)
 
-    @abstractmethod
-    def receive(self, msg: SphinxPacket | bytes, sender: "Node", receiver: "Node"):
+    def send(self, msg: SphinxPacket | bytes, sender: "Node", receiver: "Node", is_first_of_msg: bool):
+        if is_first_of_msg:
+            self.adversary.inspect_message_size(msg)
+            self.adversary.observe_sending_node(sender)
+        self.measurement.measure_egress(sender, msg)
+
         # simulate network latency
         yield self.env.timeout(random.uniform(0, self.config.p2p.max_network_latency))
 
+        self.measurement.measure_ingress(receiver, msg)
+        self.adversary.observe_receiving_node(receiver)
+        self.receive(msg, sender, receiver)
+
+    @abstractmethod
+    def receive(self, msg: SphinxPacket | bytes, sender: "Node", receiver: "Node"):
+        pass
+
     def log(self, msg):
         print(f"t={self.env.now:.3f}: P2P: {msg}")
 
@@ -57,27 +69,12 @@ class NaiveBroadcastP2P(P2P):
     # but we accept SphinxPacket as well because we don't implement Sphinx deserialization.
     def broadcast(self, sender: "Node", msg: SphinxPacket | bytes):
         yield from super().broadcast(sender, msg)
+
         self.log(f"Node:{sender.id}: Broadcasting a msg: {len(msg)} bytes")
-
-        cnt = 0
-        for receiver in self.nodes:
-            self.measurement.measure_egress(sender, msg)
-            self.env.process(self.receive(msg, sender, receiver))
-            cnt += 1
-
-        if cnt > 0:
-            # Adversary
-            self.adversary.inspect_message_size(msg)
-            self.adversary.observe_sending_node(sender)
+        for i, receiver in enumerate(self.nodes):
+            self.env.process(self.send(msg, sender, receiver, i == 0))
 
     def receive(self, msg: SphinxPacket | bytes, sender: "Node", receiver: "Node"):
-        yield from super().receive(msg, sender, receiver)
-
-        # Measurement
-        self.measurement.measure_ingress(receiver, msg)
-        # Adversary
-        self.adversary.observe_receiving_node(receiver)
-
         self.env.process(receiver.receive_message(msg))
 
 
@@ -122,24 +119,10 @@ class GossipP2P(P2P):
             # Don't gossip the message if it was received from the node who is going to be the receiver,
             # which means that the node already knows the message.
             if receiver != self.message_cache[sender][msg_hash]:
-                self.measurement.measure_egress(sender, msg)
-                self.env.process(self.receive(msg, sender, receiver))
+                self.env.process(self.send(msg, sender, receiver, cnt == 0))
                 cnt += 1
 
-        if cnt > 0:
-            # Adversary
-            self.adversary.inspect_message_size(msg)
-            self.adversary.observe_sending_node(sender)
-
     def receive(self, msg: SphinxPacket | bytes, sender: "Node", receiver: "Node"):
-        yield from super().receive(msg, sender, receiver)
-
-        # Measure ingress regardless of whether the message has been received before,
-        # because the node doesn't know if the message is a duplicate before receiving it from the network.
-        self.measurement.measure_ingress(receiver, msg)
-        # Adversary
-        self.adversary.observe_receiving_node(receiver)
-
         # Receive/gossip the msg only if it hasn't been received before. If not, just ignore the msg.
         # i.e. each message is received/gossiped at most once by each node.
         msg_hash = hashlib.sha256(bytes(msg)).digest()