implementing retyring logic for joining

MP2/peer.py

@@ -12,10 +12,11 @@ INTRODUCER_IP = "172.22.156.3"
 IP = socket.gethostbyname(socket.gethostname())
 SERVER_PORT = 8989
 INTRODUCER_FLAG = (INTRODUCER_IP == IP)
-HEART_BEAT_INTERVAL = 0.5
-GOSSIP_INTERVAL = 0.5
+HEART_BEAT_INTERVAL = 0.25
+GOSSIP_INTERVAL = 0.25
 FAILURE_TIMEOUT = 2.5
 FAILURE_T_CLEANUP = 10.0
+RETRY_INTERVAL = 5.0
 FANOUT = 1
 
 members_list = {}
@@ -38,21 +39,24 @@ server_socket.bind((IP, SERVER_PORT))
 server_socket.settimeout(1.0)
 
 def worker_1():
-    if not INTRODUCER_FLAG:
-        join_payload = {
-            "type": "JOIN",
-            "sender_id": SELF_ID
-        }
-        server_socket.sendto(json.dumps(join_payload).encode('utf-8'), (INTRODUCER_IP, SERVER_PORT))
+    has_joined = INTRODUCER_FLAG
+    last_join_time = 0
     
     while not stop_event.is_set():
+        if not has_joined and (utc_time() - last_join_time > RETRY_INTERVAL):
+            join_payload = {
+                "type": "JOIN",
+                "sender_id": SELF_ID
+            }
+            server_socket.sendto(json.dumps(join_payload).encode('utf-8'), (INTRODUCER_IP, SERVER_PORT))
+            last_join_time = utc_time()
+
         try:
             data, addr = server_socket.recvfrom(65535) 
             
             msg = json.loads(data.decode('utf-8'))
                 
             msg_type = msg.get("type")
-            print(f"Processing message type={msg_type} from {addr}")
             
             if msg_type == "JOIN":
                 join_ip, join_port, join_timestamp = msg["sender_id"]
@@ -79,9 +83,6 @@ def worker_1():
                             "state": m_data["state"]
                         })
                         
-                print(f"New member joined: {join_id}")
-                
-                # Send the initial state directly to the joining node
                 reply_payload = {
                     "type": "GOSSIP",
                     "sender_id": SELF_ID,
@@ -91,6 +92,9 @@ def worker_1():
                 print(f"Sent initial membership state to {(join_ip, join_port)}")
 
             elif msg_type == "GOSSIP":
+                if not has_joined:
+                    has_joined = True
+
                 members_json = msg.get('members', [])
 
                 with members_list_lock:
@@ -111,16 +115,13 @@ def worker_1():
                                 "local_failed_since" : utc_time() if incoming_state == "failed" else None,
                                 "local_suspected_since" : None
                             }
-                            print(f"Added gossip member: {member_id}, state={incoming_state}")
                             continue
                             
                         local_member = members_list[member_id]
 
-                        # Prevent stale ALIVE messages from resurrecting a node we already marked FAILED
                         if local_member["state"] == "failed":
                             continue
 
-                        # Merge FAILED state independently of heartbeat comparison
                         if incoming_state == "failed":
                             local_member["state"] = "failed"
                             local_member["local_failed_since"] = utc_time()
@@ -128,7 +129,6 @@ def worker_1():
                         elif incoming_hb > local_member["heart_beat"]:
                             local_member["heart_beat"] = incoming_hb
                             local_member["local_last_seen"] = utc_time()
-                            print(f"Updated member: {member_id}, heartbeat={incoming_hb}")
                 
         except socket.timeout:
             continue
@@ -137,13 +137,14 @@ def worker_2():
     while not stop_event.is_set():
         with members_list_lock:
             members_list[SELF_ID]["heart_beat"] += 1
-            print(f"Updated my heartbeat: {members_list[SELF_ID]['heart_beat']}")
         time.sleep(HEART_BEAT_INTERVAL)
 
 def worker_3():
+    peer_queue = []
+    
     while not stop_event.is_set():
         with members_list_lock:
-            peers = [member_id for member_id in members_list.keys() if member_id != SELF_ID and members_list[member_id]["state"] == "alive"]
+            current_peers = [member_id for member_id in members_list.keys() if member_id != SELF_ID and members_list[member_id]["state"] == "alive"]
             
             members_payload = []
             for m_id, m_data in members_list.items():
@@ -155,8 +156,21 @@ def worker_3():
                     "state": m_data["state"]
                 })
         
-        if peers:
-            selected_peers = random.sample(peers, min(FANOUT, len(peers)))
+        if current_peers:
+            # Clean up peers from queue that might have failed or left
+            peer_queue = [p for p in peer_queue if p in current_peers]
+            
+            selected_peers = []
+            num_to_select = min(FANOUT, len(current_peers))
+            
+            for _ in range(num_to_select):
+                # Reshuffle if the queue is empty
+                if not peer_queue:
+                    peer_queue = list(current_peers)
+                    random.shuffle(peer_queue)
+                
+                if peer_queue:
+                    selected_peers.append(peer_queue.pop(0))
             
             for peer in selected_peers:
                 peer_ip, peer_port, _ = peer
@@ -168,7 +182,6 @@ def worker_3():
                 }
                 
                 json_string = json.dumps(payload)
-                print(f"Sending GOSSIP to {(peer_ip, peer_port)}")
                 server_socket.sendto(json_string.encode('utf-8'), (peer_ip, peer_port))
             
         time.sleep(GOSSIP_INTERVAL)
@@ -201,7 +214,6 @@ def worker_4():
                 
         time.sleep(0.5)
 
-# Only for debugging
 def worker_monitoring():
     while not stop_event.is_set():
         with members_list_lock: