diff --git a/MP2/peer.py b/MP2/peer.py index f01260e..571fa73 100644 --- a/MP2/peer.py +++ b/MP2/peer.py @@ -12,9 +12,11 @@ INTRODUCER_IP = "172.22.156.3" IP = socket.gethostbyname(socket.gethostname()) SERVER_PORT = 8989 INTRODUCER_FLAG = (INTRODUCER_IP == IP) -HEART_BEAT_INTERVAL = 5 -GOSSIP_INTERVAL = 10.0 -FAILURE_TIMEOUT = 3.0 +HEART_BEAT_INTERVAL = 0.5 +GOSSIP_INTERVAL = 0.5 +FAILURE_TIMEOUT = 2.5 +FAILURE_T_CLEANUP = 10.0 +FANOUT = 1 members_list = {} members_list_lock = threading.Lock() @@ -22,7 +24,10 @@ members_list_lock = threading.Lock() SELF_ID = (IP, SERVER_PORT, utc_time()) members_list[SELF_ID] = { "heart_beat": 1, - "last_seen": utc_time() + "local_last_seen": utc_time(), + "state": "alive", + "local_failed_since" : None, + "local_suspected_since" : None } print(f"Starting peer: IP={IP}, port={SERVER_PORT}, introducer={INTRODUCER_FLAG}, self_id={SELF_ID}") @@ -38,7 +43,6 @@ def worker_1(): "type": "JOIN", "sender_id": SELF_ID } - time.sleep(5) #just so the introducer has time to start the socket server_socket.sendto(json.dumps(join_payload).encode('utf-8'), (INTRODUCER_IP, SERVER_PORT)) while not stop_event.is_set(): @@ -58,7 +62,10 @@ def worker_1(): if join_id not in members_list: members_list[join_id] = { "heart_beat": 1, - "last_seen": utc_time() + "local_last_seen": utc_time(), + "state": "alive", + "local_failed_since" : None, + "local_suspected_since" : None } print(f"Added member: {join_id}") @@ -68,19 +75,20 @@ def worker_1(): "ip": m_id[0], "port": m_id[1], "timestamp": m_id[2], - "heart_beat": m_data["heart_beat"] + "heart_beat": m_data["heart_beat"], + "state": m_data["state"] }) print(f"New member joined: {join_id}") - # Send the initial state directly to the joining node - decreases the overhead of having that init state propogate through the cluster + # Send the initial state directly to the joining node reply_payload = { "type": "GOSSIP", "sender_id": SELF_ID, "members": members_payload } server_socket.sendto(json.dumps(reply_payload).encode('utf-8'), (join_ip, join_port)) - print(f"Sent initial membership state to {(join_ip, join_port)}: {reply_payload}") + print(f"Sent initial membership state to {(join_ip, join_port)}") elif msg_type == "GOSSIP": members_json = msg.get('members', []) @@ -88,21 +96,39 @@ def worker_1(): with members_list_lock: for member in members_json: member_id = (member['ip'], member['port'], member['timestamp']) - member_hb = member['heart_beat'] - + if member_id == SELF_ID: continue + incoming_state = member['state'] + incoming_hb = member['heart_beat'] + if member_id not in members_list: members_list[member_id] = { - "heart_beat": member_hb, - "last_seen": utc_time() + "heart_beat": incoming_hb, + "local_last_seen": utc_time(), + "state": incoming_state, + "local_failed_since" : utc_time() if incoming_state == "failed" else None, + "local_suspected_since" : None } - print(f"Added gossip member: {member_id}, heartbeat={member_hb}") - elif member_hb > members_list[member_id]["heart_beat"]: - members_list[member_id]["heart_beat"] = member_hb - members_list[member_id]["last_seen"] = utc_time() - print(f"Updated member: {member_id}, heartbeat={member_hb}") + 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() + print(f"Member {member_id} marked as failed due to gossip.") + 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 @@ -117,7 +143,7 @@ def worker_2(): def worker_3(): while not stop_event.is_set(): with members_list_lock: - peers = list(members_list.keys()) + 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(): @@ -125,53 +151,81 @@ def worker_3(): "ip": m_id[0], "port": m_id[1], "timestamp": m_id[2], - "heart_beat": m_data["heart_beat"] + "heart_beat": m_data["heart_beat"], + "state": m_data["state"] }) - - if SELF_ID in peers: - peers.remove(SELF_ID) - + if peers: - random_peer = random.choice(peers) - peer_ip, peer_port, _ = random_peer + selected_peers = random.sample(peers, min(FANOUT, len(peers))) - payload = { - "type": "GOSSIP", - "sender_id": SELF_ID, - "members": members_payload - } - - json_string = json.dumps(payload) - print(f"Sending GOSSIP to {(peer_ip, peer_port)}: {payload}") - server_socket.sendto(json_string.encode('utf-8'), (peer_ip, peer_port)) + for peer in selected_peers: + peer_ip, peer_port, _ = peer + + payload = { + "type": "GOSSIP", + "sender_id": SELF_ID, + "members": members_payload + } + + 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) -#only for debuging +def worker_4(): + while not stop_event.is_set(): + current_time = utc_time() + to_remove = [] + + with members_list_lock: + for member_id, member_data in members_list.items(): + if member_id == SELF_ID: + continue + + if member_data["state"] == "alive": + last_seen = member_data["local_last_seen"] + time_since_last_seen = current_time - last_seen + + if time_since_last_seen > FAILURE_TIMEOUT: + member_data["state"] = "failed" + member_data["local_failed_since"] = current_time + + last_failed_since = member_data.get("local_failed_since") + if last_failed_since is not None and (current_time - last_failed_since) > FAILURE_T_CLEANUP: + to_remove.append(member_id) + + for member_id in to_remove: + del members_list[member_id] + print(f"Purged member {member_id} from list.") + + time.sleep(0.5) + +# Only for debugging def worker_monitoring(): while not stop_event.is_set(): with members_list_lock: local_ml = { member_id: member_data.copy() - for member_id , member_data in members_list.items() + for member_id, member_data in members_list.items() } with open("membership_list.txt", "w") as f: for member_id, member_data in local_ml.items(): f.write(f"{member_id}: {member_data}\n") time.sleep(0.5) - t1 = threading.Thread(target=worker_1) t2 = threading.Thread(target=worker_2) t3 = threading.Thread(target=worker_3) -#only for debugging -t4 = threading.Thread(target=worker_monitoring) +t4 = threading.Thread(target=worker_4) +t5 = threading.Thread(target=worker_monitoring) t1.start() t2.start() t3.start() t4.start() +t5.start() try: while not stop_event.is_set(): @@ -184,4 +238,5 @@ t1.join() t2.join() t3.join() t4.join() +t5.join() server_socket.close()