diff --git a/MP2/peer.py b/MP2/peer.py index 571fa73..cf67227 100644 --- a/MP2/peer.py +++ b/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: