script/research/event_graph: some modifications & add probability of losing messages

script/research/event_graph/main.py

@@ -1,6 +1,6 @@
 from hashlib import sha256
 from datetime import datetime
-from random import randint
+from random import randint, random
 import math
 import asyncio
 
@@ -11,6 +11,32 @@ import networkx as nx
 EventId = str
 EventIds = list[EventId]
 
+# Number of nodes
+NODES_N = 10
+# Broadcast attempt for each node
+BROADCAST_ATTEMPT = 3
+
+MAX_BROADCAST_DELAY = round(math.log(NODES_N))
+MIN_BROADCAST_DELAY = 0
+
+PROBABILITY_OF_DROPPING_MSG = 0.50  # 1/2
+
+# Timeout for sending tasks to finish
+BROADCAST_TIMEOUT = NODES_N * BROADCAST_ATTEMPT * MAX_BROADCAST_DELAY
+
+
+class NetworkPool:
+    def __init__(self, nodes):
+        self.nodes = nodes
+
+    def request(self, event_id: EventId):
+        for n in self.nodes:
+            event = n.get_event(event_id)
+            if event != None:
+                return event
+
+        return None
+
 
 class Event:
     def __init__(self, parents: EventIds):
@@ -82,10 +108,11 @@ class Graph:
 
 
 class Node:
-    def __init__(self, name: str):
+    def __init__(self, name: str, queue):
         self.name = name
         self.orphan_pool = Graph()
         self.active_pool = Graph()
+        self.queue = queue
 
         # The active pool should always start with one event
         genesis_event = Event([])
@@ -109,7 +136,7 @@ class Node:
         self.heads.append(event_hash)
 
     # On receive new event
-    def receive_new_event(self, event: Event):
+    def receive_new_event(self, event: Event, np):
         event_hash = event.hash()
 
         # Reject event with no parents
@@ -156,6 +183,16 @@ class Node:
             # Send all the missing parents in request_list
             # to the node who send this event
 
+            # For simulation purpose the node fetch the missed parents from the
+            # network pool which contains all the nodes and its messages
+            for event in request_list:
+                requested_event = np.request(event)
+                if requested_event != None:
+                    self.receive_new_event(requested_event, np)
+                else:
+                    # It must always find the missed event from the network
+                    print(f"Error: {self.name} requested {event} not found")
+
     # This will check if passed parents are in the orphan pool, and fill
     # request_list with missing parents
     def check_parents(self, request_list, parents: EventIds, visited=[]):
@@ -205,6 +242,15 @@ class Node:
                 # Recursive call
                 self.relink(orphan, remove_list)
 
+    def get_event(self, event_id: EventId):
+        # Check the active_pool
+        event = self.active_pool.events.get(event_id)
+        # Check the orphan_pool
+        if event == None:
+            event = self.orphan_pool.events.get(event)
+
+        return event
+
     def __str__(self):
         return f"""------
 	        \n Name: {self.name}
@@ -212,32 +258,25 @@ class Node:
 	        \n Orphan Pool: {self.orphan_pool}"""
 
 
-# Number of nodes
-NODES_N = 15
-# Broadcast attempt for each node
-BROADCAST_ATTEMPT = 3
-
-MAX_BROADCAST_DELAY = round(math.log(NODES_N))
-MIN_BROADCAST_DELAY = 0
-
-# Timeout for sending tasks to finish
-BROADCAST_TIMEOUT = NODES_N * BROADCAST_ATTEMPT * MAX_BROADCAST_DELAY
-
-
 # Each node has NODES_N of this function running in the background
 # for receiving events from each node separately
-async def recv_loop(node, peer, queue):
+async def recv_loop(node, peer, queue, np):
     while True:
         # Wait new event
         event = await queue.get()
-        node.receive_new_event(event)
         queue.task_done()
+
+        if random() <= PROBABILITY_OF_DROPPING_MSG:
+            print(f"{node.name} dropped: \n  {event}")
+            continue
+
+        node.receive_new_event(event, np)
         print(f"{node.name} receive event from {peer}: \n {event}")
 
 
 # Send new event at random intervals
 # Each node has this function running in the background
-async def send_loop(node, queue):
+async def send_loop(node):
     for _ in range(BROADCAST_ATTEMPT):
 
         await asyncio.sleep(randint(MIN_BROADCAST_DELAY, MAX_BROADCAST_DELAY))
@@ -247,45 +286,52 @@ async def send_loop(node, queue):
 
         print(f"{node.name} broadcast event: \n {event}")
         for _ in range(NODES_N):
-            await queue.put(event)
-            await queue.join()
+            await node.queue.put(event)
+            await node.queue.join()
 
 
 async def main():
     nodes = []
-    queues = dict()
 
     print(f"Run {NODES_N} Nodes")
 
     try:
-        # Initialize new nodes and create coroutine task for each node
-        # contains send_loop function
+
+        # Initialize NODES_N nodes
         for i in range(NODES_N):
-            node = Node(f"Node{i}")
+            queue = asyncio.Queue()
+            node = Node(f"Node{i}", queue)
             nodes.append(node)
 
-            queue = asyncio.Queue()
-            queues[node.name] = queue
+        # Initialize NetworkPool contains all nodes
+        np = NetworkPool(nodes)
 
         # Initialize NODES_N * NODES_N coroutine tasks for receiving events
+        # Each node listen to all queues from the running nodes
         for node in nodes:
-            for (peer, queue) in queues.items():
-                asyncio.create_task(recv_loop(node, peer, queue))
+            for n in nodes:
+                asyncio.create_task(recv_loop(node, n.name, n.queue, np))
 
+        # Create coroutine task contains send_loop function for each node
         # Run and wait for send tasks
-        s_g = asyncio.gather(*[send_loop(n, queues[n.name]) for n in nodes])
+        s_g = asyncio.gather(*[send_loop(n) for n in nodes])
         await asyncio.wait_for(s_g, BROADCAST_TIMEOUT)
 
         # Assert if all nodes share the same active pool graph
-        assert (all(n.active_pool.events.keys() ==
-                nodes[0].active_pool.events.keys() for n in nodes))
+        # assert (all(n.active_pool.events.keys() ==
+        #        nodes[0].active_pool.events.keys() for n in nodes))
 
         # Assert if all nodes share the same orphan pool graph
-        assert (all(n.orphan_pool.events.keys() ==
-                nodes[0].orphan_pool.events.keys() for n in nodes))
+        # assert (all(n.orphan_pool.events.keys() ==
+        #        nodes[0].orphan_pool.events.keys() for n in nodes))
 
-        print_graph([nodes[0]])
+        # Assert if all nodes heads are equal
+        #assert (all(n.heads == nodes[0].heads for n in nodes))
 
+        # print_graph([nodes[0]])
+
+    except KeyboardInterrupt:
+        print("Done")
     except asyncio.exceptions.TimeoutError:
         print("Broadcast TimeoutError")
 
@@ -316,28 +362,5 @@ def print_graph(nodes):
     plt.show()
 
 
-def test_node():
-    node_a = Node("NodeA")
-
-    event0 = node_a.genesis_event
-    event1 = Event([event0.hash()])
-    event2 = Event([event1.hash()])
-    event3 = Event([event2.hash(), event0.hash()])
-    event4 = Event([event1.hash(), event3.hash()])
-    event5 = Event([event4.hash(), "FAKEHASH"])
-    event6 = Event([event5.hash(), event3.hash()])
-
-    node_a.receive_new_event(event3)
-    node_a.receive_new_event(event2)
-    node_a.receive_new_event(event1)
-    node_a.receive_new_event(event5)
-    node_a.receive_new_event(event6)
-    node_a.receive_new_event(event4)
-
-    print(node_a)
-
-
 if __name__ == "__main__":
-
-    # test_node()
     asyncio.run(main())