import socket from datetime import datetime, timezone import threading import json import os import random import time import sys #setup logging def _machine_index(): idx = os.environ.get("MACHINE_INDEX") if idx is not None: return int(idx) return None log_file = open(f"machine.{_machine_index()}.log", "w") log_lock = threading.Lock() def log(message): timestamp = datetime.now().strftime("%H:%M:%S.%f")[:-3] with log_lock: log_file.write(f"[{timestamp}] {message}\n") log_file.flush() def utc_time(): return datetime.now(timezone.utc).timestamp() TOTAL_SENT_BYTES = 0 TOTAL_RECEIVED_BYTES = 0 traffic_lock = threading.Lock() #fixed INTRODUCER_IP = "172.22.156.3" GLOBAL_PORT = 8989 MY_IP = socket.gethostbyname(socket.gethostname()) HEARTBEAT_INTERVAL = 0.25 GOSSIP_INTERVAL = 0.25 JOIN_RETRY_INTERVAL = 1.0 GOSSIP_FANOUT = 2 GOSSIP_FAILURE_TIMEOUT = 3 GOSSIP_S_SUSPECT_TIMEOUT = 1.0 GOSSIP_S_TIMEOUT_SUSPECT = 2.0 GOSSIP_FAILURE_TIMEOUT_CLEANUP = 10.0 members_list = {} dead_members = set() members_list_lock = threading.Lock() suspect_history = [] suspect_history_lock = threading.Lock() MY_ID = (MY_IP, GLOBAL_PORT, utc_time()) members_list[MY_ID] = { "heart_beat": 0, "state": "alive", "incarnation_counter": 0, "local_last_seen": utc_time(), "local_suspected_since": None, "local_failed_since": None, } stop_threads = threading.Event() #to trigger a full shut down suspect_mode = threading.Event() #to trigger a switch to suspect mode #variable state #the ip is a tie breaker for the version number, so we can deterministically resolve conflicts #where two nodes have the same version number but different modes or drop rates #extremely unlikely, but possible if two nodes are started at the same time and both switch modes simultaneously cluster_state = { "mode": "nosuspect", "drop_rate": 0.0, "version": (0, "0.0.0.0") } cluster_state_lock = threading.Lock() #start the main shared socket server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) #UDP server_socket.bind((MY_IP, GLOBAL_PORT)) server_socket.settimeout(0.05) #global state switcher def global_switcher(cmd, value): if cmd == "switch": cmd = "mode" if cmd == "mode": with cluster_state_lock: if value in ["suspect", "nosuspect"]: cluster_state["mode"] = value cluster_state["version"] = (cluster_state["version"][0] + 1, MY_IP) if value == "suspect": suspect_mode.set() else: suspect_mode.clear() with members_list_lock: for m_id, m_data in members_list.items(): if m_id != MY_ID and m_data["state"] == "suspect": m_data["state"] = "alive" m_data["local_suspected_since"] = None log(f"Resolved {m_id[0]}:{m_id[1]} suspect -> alive on switch to nosuspect") log(f"Switched mode to {value}") else: log(f"Invalid mode value: {value}") elif cmd == "drop_rate": try: drop_rate_value = float(value) if 0.0 <= drop_rate_value <= 100.0: with cluster_state_lock: cluster_state["drop_rate"] = drop_rate_value cluster_state["version"] = (cluster_state["version"][0] + 1, MY_IP) log(f"Set drop rate to {drop_rate_value}%") else: log(f"Invalid drop rate value: {value}. Must be between 0 and 100.") except ValueError: log(f"Invalid drop rate value: {value}. Must be a number.") #THREAD 1 def worker_1(): global TOTAL_SENT_BYTES, TOTAL_RECEIVED_BYTES has_joined = (INTRODUCER_IP == MY_IP) last_join_attempt = 0.0 while not stop_threads.is_set(): #we keep retrying until we have joined the cluster if not has_joined and (utc_time() - last_join_attempt) >= JOIN_RETRY_INTERVAL: joining_message_payload = {"type": "JOIN", "sender_id": MY_ID} joining_message_json = json.dumps(joining_message_payload) try: joining_message_bytes = joining_message_json.encode('utf-8') server_socket.sendto(joining_message_bytes, (INTRODUCER_IP, GLOBAL_PORT)) with traffic_lock: TOTAL_SENT_BYTES += len(joining_message_bytes) except Exception as e: log(f"JOIN send failed: {e}") last_join_attempt = utc_time() try: data, addr = server_socket.recvfrom(65535) except socket.timeout: continue with traffic_lock: TOTAL_RECEIVED_BYTES += len(data) try: msg = json.loads(data.decode('utf-8')) except Exception: continue msg_type = msg.get("type") #handle dropping messages only on GOSSIP if msg_type == "GOSSIP": drop_rate = 0 with cluster_state_lock: drop_rate = cluster_state["drop_rate"] if drop_rate > 0.0 and (random.random() * 100) < drop_rate: log(f"DROPPED GOSSIP message from {addr[0]}:{addr[1]}") continue # Skip processing this message #handle different message types if msg_type == "JOIN": if INTRODUCER_IP == MY_IP: handle_join(msg) elif msg_type == "GOSSIP": handle_gossip(msg) if not has_joined: has_joined = True log(f"Successfully joined the cluster via {addr[0]}:{addr[1]}") else: log(f"Unknown message type received from {addr[0]}:{addr[1]}: {msg_type}") # THREAD 1 HELPERS #this fixes the deadlock where handle_join held members_list_lock then called current_membership() def _current_membership_locked(): result = [] for member_id, member_data in members_list.items(): result.append({ "ip": member_id[0], "port": member_id[1], "timestamp": member_id[2], "heartbeat": member_data["heart_beat"], "state": member_data["state"], "incarnation": member_data["incarnation_counter"], }) return result def current_membership(): with members_list_lock: return _current_membership_locked() def current_cluster_state(): with cluster_state_lock: return { "mode": cluster_state["mode"], "drop_rate": cluster_state["drop_rate"], "version": list(cluster_state["version"]), } def handle_join(msg): global TOTAL_SENT_BYTES sender_id = tuple(msg["sender_id"]) with members_list_lock: #check if the sender is in the dead members list if sender_id in dead_members: return if sender_id not in members_list: members_list[sender_id] = { "heart_beat": 0, "state": "alive", "incarnation_counter": 0, "local_last_seen": utc_time(), "local_suspected_since": None, "local_failed_since": None, } log(f"Added new member: {sender_id[0]}:{sender_id[1]}") members_snapshot = _current_membership_locked() control_state = current_cluster_state() #we have to send back a GOSSIP message with our current membership list and cluster state ack_payload = { "type": "GOSSIP", "sender_id": MY_ID, "members": members_snapshot, "mode": control_state["mode"], "cluster_version": control_state["version"], "drop_rate": control_state["drop_rate"], } try: ack_payload_bytes = json.dumps(ack_payload).encode('utf-8') server_socket.sendto(ack_payload_bytes, (sender_id[0], sender_id[1])) with traffic_lock: TOTAL_SENT_BYTES += len(ack_payload_bytes) except Exception as e: log(f"Failed to send GOSSIP response to {sender_id[0]}:{sender_id[1]}: {e}") def handle_member_update(member): # 4 cases: # 1. if i am being suspected, clear our name # 2. new member, add to membership list # 3. [hard] incarnation update # 4. increase heartbeat, update last seen member_id = (member["ip"], member["port"], member["timestamp"]) if member_id in dead_members: return recieved_incarnation = member["incarnation"] recieved_heartbeat = member["heartbeat"] recieved_state = member["state"] #1 if member_id == MY_ID and recieved_state == "suspect" and recieved_incarnation >= members_list[MY_ID]["incarnation_counter"]: log(f"Clearing my own name from suspect state due to gossip from {member_id[0]}:{member_id[1]}") members_list[MY_ID]["state"] = "alive" members_list[MY_ID]["incarnation_counter"] = recieved_incarnation + 1 members_list[MY_ID]["local_last_seen"] = utc_time() return #2 local_member_data = members_list.get(member_id) if local_member_data is None: if recieved_state == "failed": dead_members.add(member_id) log(f"Received failed state for unknown member {member_id[0]}:{member_id[1]}, ignoring") return members_list[member_id] = { "heart_beat": recieved_heartbeat, "state": recieved_state, "incarnation_counter": recieved_incarnation, "local_last_seen": utc_time(), "local_suspected_since": utc_time() if recieved_state == "suspect" else None, "local_failed_since": None, } if recieved_state == "suspect": with suspect_history_lock: suspect_history.append((member_id, utc_time())) log(f"Added new member: {member_id[0]}:{member_id[1]} with state suspect") else: log(f"Added new member: {member_id[0]}:{member_id[1]}") return if local_member_data["state"] == "failed": return #waiting for cleanup, ignore any updates #3 if recieved_incarnation > local_member_data["incarnation_counter"]: local_member_data["incarnation_counter"] = recieved_incarnation if recieved_state == "alive": if local_member_data["state"] != "alive": log(f"Member {member_id[0]}:{member_id[1]} is now alive (incarnation update)") local_member_data["state"] = "alive" local_member_data["local_last_seen"] = utc_time() local_member_data["local_suspected_since"] = None local_member_data["local_failed_since"] = None elif recieved_state == "suspect": if local_member_data["state"] != "suspect": with suspect_history_lock: suspect_history.append((member_id, utc_time())) log(f"Member {member_id[0]}:{member_id[1]} is now suspect (incarnation update)") local_member_data["state"] = "suspect" local_member_data["local_suspected_since"] = utc_time() elif recieved_state == "failed": if local_member_data["state"] != "failed": log(f"Member {member_id[0]}:{member_id[1]} is now failed (incarnation update)") local_member_data["state"] = "failed" local_member_data["local_failed_since"] = utc_time() #This is needed to handle the case where two nodes have the same incarnation number but different states. We need to resolve this deterministically. # We will use the following rules: # - If the received state is "suspect" and the local state is "alive", we will update the local state to "suspect". # - If the received state is "failed" and the local state is "alive" or "suspect", we will update the local state to "failed". elif recieved_incarnation == local_member_data["incarnation_counter"]: if recieved_state == "suspect" and local_member_data["state"] == "alive": local_member_data["state"] = "suspect" local_member_data["local_suspected_since"] = utc_time() with suspect_history_lock: suspect_history.append((member_id, utc_time())) log(f"Member {member_id[0]}:{member_id[1]} is now suspect (equal-incarnation update)") elif recieved_state == "failed" and local_member_data["state"] in ["alive", "suspect"]: local_member_data["state"] = "failed" local_member_data["local_failed_since"] = utc_time() log(f"Member {member_id[0]}:{member_id[1]} is now failed (equal-incarnation update)") local_member_data["local_suspected_since"] = None #4 if recieved_heartbeat > local_member_data["heart_beat"]: local_member_data["heart_beat"] = recieved_heartbeat local_member_data["local_last_seen"] = utc_time() #we have to check the currect control state version and update our local state if the received version is newer def handle_gossip(msg): incoming_version = None if "cluster_version" in msg: try: incoming_version = tuple(msg["cluster_version"]) except Exception: incoming_version = None if incoming_version is not None: with cluster_state_lock: if incoming_version > cluster_state["version"]: cluster_state["mode"] = msg.get("mode", cluster_state["mode"]) cluster_state["drop_rate"] = msg.get("drop_rate", cluster_state["drop_rate"]) cluster_state["version"] = incoming_version #also have to send the suspect thread event if cluster_state["mode"] == "suspect": suspect_mode.set() else: suspect_mode.clear() #when switching to nosuspect, we have to clear any suspect states in our membership list with members_list_lock: for m_id, m_data in members_list.items(): if m_id != MY_ID and m_data["state"] == "suspect": m_data["state"] = "alive" m_data["local_suspected_since"] = None log(f"Resolved {m_id[0]}:{m_id[1]} suspect -> alive on switch to nosuspect") log(f"Updated cluster state to version {incoming_version}: mode={cluster_state['mode']}, drop_rate={cluster_state['drop_rate']}") members_json = msg.get("members", []) with members_list_lock: for member in members_json: handle_member_update(member) #THREAD 2 - update my heartbeat def worker_2(): while not stop_threads.is_set(): with members_list_lock: members_list[MY_ID]["heart_beat"] += 1 time.sleep(HEARTBEAT_INTERVAL) #THREAD 3 - send gossip messages to random members #threshold, pure random can starve a peer for a few rounds and cause false positives. _peer_queue = [] _peer_queue_lock = threading.Lock() def worker_3(): global TOTAL_SENT_BYTES global _peer_queue while not stop_threads.is_set(): with members_list_lock: current_peers = [member_id for member_id in members_list if member_id != MY_ID and members_list[member_id]["state"] in ["alive", "suspect"]] gossip_members_payload = current_membership() gossip_control_payload = current_cluster_state() #shuffled round-robin target selection with _peer_queue_lock: _peer_queue = [p for p in _peer_queue if p in current_peers] selected_peers = [] for _ in range(min(GOSSIP_FANOUT, len(current_peers))): 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: gossip_payload = { "type": "GOSSIP", "sender_id": MY_ID, "members": gossip_members_payload, "mode": gossip_control_payload["mode"], "cluster_version": gossip_control_payload["version"], "drop_rate": gossip_control_payload["drop_rate"], } try: gossip_payload_bytes = json.dumps(gossip_payload).encode('utf-8') server_socket.sendto(gossip_payload_bytes, (peer[0], peer[1])) with traffic_lock: TOTAL_SENT_BYTES += len(gossip_payload_bytes) except Exception as e: log(f"Failed to send GOSSIP to {peer[0]}:{peer[1]}: {e}") time.sleep(GOSSIP_INTERVAL) # THREAD 4 FAILURE DETECTION def worker_4(): while not stop_threads.is_set(): current_time = utc_time() members_to_remove = [] with members_list_lock: for member_id, member_data in members_list.items(): if member_id == MY_ID: continue # Skip self #gossip if not suspect_mode.is_set(): if member_data["state"] == "alive" and (current_time - member_data["local_last_seen"]) > GOSSIP_FAILURE_TIMEOUT: member_data["state"] = "failed" member_data["local_failed_since"] = current_time log(f"Member {member_id[0]}:{member_id[1]} is now failed (gossip timeout)") #gossip + Suspect else: if member_data["state"] == "alive" and (current_time - member_data["local_last_seen"]) > GOSSIP_S_SUSPECT_TIMEOUT: member_data["state"] = "suspect" member_data["local_suspected_since"] = current_time with suspect_history_lock: suspect_history.append((member_id, current_time)) log(f"Member {member_id[0]}:{member_id[1]} is now suspect (gossip + suspect timeout)") elif member_data["state"] == "suspect" and (current_time - member_data["local_suspected_since"]) > GOSSIP_S_TIMEOUT_SUSPECT: member_data["state"] = "failed" member_data["local_failed_since"] = current_time member_data["local_suspected_since"] = None log(f"Member {member_id[0]}:{member_id[1]} is now failed (suspect timeout)") #clean up for both modes if member_data["state"] == "failed" and (current_time - member_data["local_failed_since"]) > GOSSIP_FAILURE_TIMEOUT_CLEANUP: members_to_remove.append(member_id) for member_id in members_to_remove: del members_list[member_id] dead_members.add(member_id) log(f"Removed member {member_id[0]}:{member_id[1]}") time.sleep(0.1) # THREAD 5 Handle stdin input for control commands def worker_5(): while not stop_threads.is_set(): line = sys.stdin.readline() if not line: time.sleep(0.1) continue cmd = line.strip().split() if not cmd: continue cmd_0 = cmd[0].lower() if cmd_0 == "list_mem": cmd_list_mem() elif cmd_0 == "list_self": log(str(MY_ID)) elif cmd_0 == "join": log("auto joins at start up") elif cmd_0 == "leave": log("Leaving") stop_threads.set() elif cmd_0 == "display_suspects": cmd_display_suspects() elif cmd_0 == "switch": if len(cmd) >= 2: global_switcher("switch", cmd[1]) else: log("Usage: switch suspect|nosuspect") elif cmd_0 == "display_protocol": with cluster_state_lock: log(f"Mode: {cluster_state['mode']}, drop rate: {cluster_state['drop_rate']}") elif cmd_0 == "set_drop_rate": if len(cmd) >= 2: global_switcher("drop_rate", cmd[1]) else: log("Usage: set_drop_rate ") else: log(f"Unknown command: {cmd_0}") #THREAD 5 HELPERS def cmd_list_mem(): with members_list_lock: for member_id, member_data in members_list.items(): if member_data["state"] == "failed": continue log(f"{member_id} state = {member_data['state']}, heartbeat = {member_data['heart_beat']}, incarnation = {member_data['incarnation_counter']}") def cmd_display_suspects(): with suspect_history_lock: if not suspect_history: log("No suspects recorded.") else: log("Suspect history:") for member_id, timestamp in suspect_history: log(f"{member_id} suspected at {timestamp}") # THREAD 6 - Display membership list in a pretty format def worker_monitor(): while not stop_threads.is_set(): with members_list_lock: snapshot = { member_id: member_data.copy() for member_id, member_data in members_list.items() } with cluster_state_lock: mode = cluster_state["mode"] drop_rate = cluster_state["drop_rate"] with traffic_lock: sent_bytes = TOTAL_SENT_BYTES received_bytes = TOTAL_RECEIVED_BYTES try: with open("membership_list.txt", "w") as f: for member_id, member_data in snapshot.items(): f.write(f"{member_id}: {member_data}\n") f.write( f"Mode: {mode}, " f"Drop Rate: {drop_rate}, " f"Sent Bytes: {sent_bytes}, " f"Received Bytes: {received_bytes}\n" ) except Exception as e: pass time.sleep(0.25) if __name__ == "__main__": t1 = threading.Thread(target=worker_1, daemon=True) t2 = threading.Thread(target=worker_2, daemon=True) t3 = threading.Thread(target=worker_3, daemon=True) t4 = threading.Thread(target=worker_4, daemon=True) t5 = threading.Thread(target=worker_5, daemon=True) t6 = threading.Thread(target=worker_monitor, daemon=True) t1.start() t2.start() t3.start() t4.start() t5.start() t6.start() try: while not stop_threads.is_set(): time.sleep(0.1) except KeyboardInterrupt: log("Keyboard interrupt, shutting down...") log("Shutting down...") stop_threads.set() t1.join(timeout=1.0) t2.join(timeout=1.0) t3.join(timeout=1.0) t4.join(timeout=1.0) t6.join(timeout=1.0) log("Shutdown complete.") server_socket.close()