diff --git a/MP2/peer.py b/MP2/peer.py index 763d732..7e7dfdb 100644 --- a/MP2/peer.py +++ b/MP2/peer.py @@ -1,285 +1,563 @@ -from datetime import datetime, timezone import socket +from datetime import datetime, timezone import threading -import time +import json +import os import random -import json +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", "a") +log_lock = threading.Lock() +def log(message): + timestamp = datetime.now().strftime("%H:%M:%S.%f")[:-3] + with log_lock: + print(f"[{timestamp}] {message}") + log_file.write(f"[{timestamp}] {message}\n") + log_file.flush() def utc_time(): return datetime.now(timezone.utc).timestamp() -INTRODUCER_IP = "172.22.156.3" #hardcoded ip -IP = socket.gethostbyname(socket.gethostname()) #the vm IP address -SERVER_PORT = 8989 -INTRODUCER_FLAG = (INTRODUCER_IP == IP) -HEART_BEAT_INTERVAL = 0.25 +#fixed +INTRODUCER_IP = "172.22.156.3" +GLOBAL_PORT = 8989 +MY_IP = socket.gethostbyname(socket.gethostname()) + +HEARTBEAT_INTERVAL = 0.25 GOSSIP_INTERVAL = 0.25 -FAILURE_TIMEOUT = 2.5 -FAILURE_T_CLEANUP = 10.0 -RETRY_INTERVAL = 5.0 -FANOUT = 1 +JOIN_RETRY_INTERVAL = 1.0 +GOSSIP_FANOUT = 1 + +GOSSIP_FAILURE_TIMEOUT = 1.8 +GOSSIP_S_SUSPECT_TIMEOUT = 0.8 +GOSSIP_S_TIMEOUT_SUSPECT = 1.0 +GOSSIP_FAILURE_TIMEOUT_CLEANUP = 10.0 -#message json example: -# { -# type: "GOSSIP", or "JOIN", ... -# sender_id: (ip, port, timestamp), -# members: [ -# { -# ip: " -# port: 8989, -# timestamp: 1234567890.123456, -# heart_beat: 1, -# state: "alive" -# }, -# ... members_list = {} -dead_members = set() #[BASICALLY VERSION TRACKING]this tracks members that have been marked as failed and removed from the membership list. +dead_members = set() members_list_lock = threading.Lock() -#adding its self to the membership list -SELF_ID = (IP, SERVER_PORT, utc_time()) -members_list[SELF_ID] = { - "heart_beat": 1, - "local_last_seen": utc_time(), +suspect_history = [] +suspect_history_lock = threading.Lock() + +MY_ID = (MY_IP, GLOBAL_PORT, utc_time()) +members_list[MY_ID] = { + "heart_beat": 0, "state": "alive", - "local_failed_since" : None, - "local_suspected_since" : None + "incarnation_counter": 0, + "local_last_seen": utc_time(), + "local_suspected_since": None, + "local_failed_since": None, } -stop_event = threading.Event() #trigger to stop all threads when the program is exiting +stop_threads = threading.Event() #to trigger a full shut down +suspect_mode = threading.Event() #to trigger a switch to suspect mode -#one socket for all threads to use -server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) -server_socket.bind((IP, SERVER_PORT)) -server_socket.settimeout(1.0) -#responsible for handling incoming gossip messages, join requests on the introducer, and updating the membership list based on the received messages +#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(): - has_joined = INTRODUCER_FLAG #if this is the introducer, it has already joined - last_join_time = 0 #flag to track the last time a join request was sent to retry - - while not stop_event.is_set(): - # If the node has not joined and enough time has passed since the last join attempt, send a JOIN request to the introducer - 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() + 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: + server_socket.sendto(joining_message_json.encode('utf-8'), (INTRODUCER_IP, GLOBAL_PORT)) + except Exception as e: + log(f"JOIN send failed: {e}") + last_join_attempt = utc_time() try: - data, addr = server_socket.recvfrom(65535) - - msg = json.loads(data.decode('utf-8')) - - msg_type = msg.get("type") - - #handle join type - if msg_type == "JOIN": - join_ip, join_port, join_timestamp = msg["sender_id"] - join_id = (join_ip, join_port, join_timestamp) - - with members_list_lock: - #if in dead members, ignore the join request - if join_id in dead_members: - continue - - if join_id not in members_list: - members_list[join_id] = { - "heart_beat": 1, - "local_last_seen": utc_time(), - "state": "alive", - "local_failed_since" : None, - "local_suspected_since" : None - } - - members_payload = [] - for m_id, m_data in members_list.items(): - members_payload.append({ - "ip": m_id[0], - "port": m_id[1], - "timestamp": m_id[2], - "heart_beat": m_data["heart_beat"], - "state": m_data["state"] - }) - - #send the current membership list 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)) - - #handle gossip type - elif msg_type == "GOSSIP": - #first time receiving a gossip message, mark the node as joined - if not has_joined: - has_joined = True - - members_json = msg.get('members', []) - - with members_list_lock: - for member in members_json: - member_id = (member['ip'], member['port'], member['timestamp']) - - if member_id == SELF_ID or member_id in dead_members: - continue - - incoming_state = member['state'] - incoming_hb = member['heart_beat'] - - local_member = members_list.get(member_id) - - if local_member is None: - if incoming_state == "failed": - dead_members.add(member_id) - continue - else: - members_list[member_id] = { - "heart_beat": incoming_hb, - "local_last_seen": utc_time(), - "state": incoming_state, - "local_failed_since" : None, - "local_suspected_since" : None - } - continue - - if local_member["state"] == "failed": - continue - - 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() - + data, addr = server_socket.recvfrom(65535) except socket.timeout: continue -#handles incrementing the local node's heartbeat counter at regular intervals -def worker_2(): - while not stop_event.is_set(): - with members_list_lock: - members_list[SELF_ID]["heart_beat"] += 1 - time.sleep(HEART_BEAT_INTERVAL) + try: + msg = json.loads(data.decode('utf-8')) + except Exception: + continue -#handles sending gossip messages to a subset of alive peers at regular intervals -def worker_3(): - peer_queue = [] - - while not stop_event.is_set(): + 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): + 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: + server_socket.sendto(json.dumps(ack_payload).encode('utf-8'), (sender_id[0], sender_id[1])) + 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: - 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(): - members_payload.append({ - "ip": m_id[0], - "port": m_id[1], - "timestamp": m_id[2], - "heart_beat": m_data["heart_beat"], - "state": m_data["state"] - }) - - if current_peers: - #remove any peers from the queue that are no longer alive - peer_queue = [p for p in peer_queue if p in current_peers] - + 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 _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 = [] - num_to_select = min(FANOUT, len(current_peers)) - - for _ in range(num_to_select): - 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 - - payload = { - "type": "GOSSIP", - "sender_id": SELF_ID, - "members": members_payload - } - - json_string = json.dumps(payload) - server_socket.sendto(json_string.encode('utf-8'), (peer_ip, peer_port)) - + 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: + server_socket.sendto(json.dumps(gossip_payload).encode('utf-8'), (peer[0], peer[1])) + except Exception as e: + log(f"Failed to send GOSSIP to {peer[0]}:{peer[1]}: {e}") + time.sleep(GOSSIP_INTERVAL) -# handle failure detection and cleanup of failed members from the membership list +# THREAD 4 FAILURE DETECTION def worker_4(): - while not stop_event.is_set(): + while not stop_threads.is_set(): current_time = utc_time() - to_remove = [] - + members_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: + 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 - - 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: - dead_members.add(member_id) + 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] - - time.sleep(0.5) + dead_members.add(member_id) + log(f"Removed member {member_id[0]}:{member_id[1]}") -#for monitoring purposes -def worker_monitoring(): - while not stop_event.is_set(): + 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": + print(str(MY_ID)) + elif cmd_0 == "join": + print("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: + print("Usage: switch suspect|nosuspect") + elif cmd_0 == "display_protocol": + with cluster_state_lock: + print(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: + print("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 + print(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: + print("No suspects recorded.") + else: + print("Suspect history:") + for member_id, timestamp in suspect_history: + print(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: - local_ml = { - member_id: member_data.copy() - 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") + snapshot = dict(members_list) + try: + with open("membership_list.txt", "w") as f: + for m_id, m_data in snapshot.items(): + f.write(f"{m_id}: {m_data}\n") + except Exception: + pass time.sleep(0.5) -t1 = threading.Thread(target=worker_1) -t2 = threading.Thread(target=worker_2) -t3 = threading.Thread(target=worker_3) -t4 = threading.Thread(target=worker_4) -t5 = threading.Thread(target=worker_monitoring) -t1.start() -t2.start() -t3.start() -t4.start() -t5.start() +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) -try: - while not stop_event.is_set(): - time.sleep(0.5) -except KeyboardInterrupt: - print("\nExiting...") - stop_event.set() + t1.start() + t2.start() + t3.start() + t4.start() + t5.start() + t6.start() -t1.join() -t2.join() -t3.join() -t4.join() -t5.join() -server_socket.close() + 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() \ No newline at end of file