aryansingh/fa26_cs425_.group-mp_.team-01
0
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #9 from illinois-cs-coursework/aryan

Browse Source 
Aryan
This commit is contained in:
Aryan Singh authored and GitHub committed 2026-09-27 18:47:14 -05:00 Unverified
2 parents dc20f8f72f 4dbe907e1f
commit d027c8b59f
1 file changed
+534 -241
Download Patch File Download Diff File Expand all files Collapse all files
MP2/peer.py
+534 -241
View File
@@ -1,285 +1,578 @@
from datetime import datetime, timezone
import socket import socket
from datetime import datetime, timezone
import threading import threading
import time import json
import os
import random 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(): def utc_time():
return datetime.now(timezone.utc).timestamp() return datetime.now(timezone.utc).timestamp()
INTRODUCER_IP = "172.22.156.3" #hardcoded ip #fixed
IP = socket.gethostbyname(socket.gethostname()) #the vm IP address INTRODUCER_IP = "172.22.156.3"
SERVER_PORT = 8989 GLOBAL_PORT = 8989
INTRODUCER_FLAG = (INTRODUCER_IP == IP) MY_IP = socket.gethostbyname(socket.gethostname())
HEART_BEAT_INTERVAL = 0.25
GOSSIP_INTERVAL = 0.25 HEARTBEAT_INTERVAL = 0.25
FAILURE_TIMEOUT = 2.5 GOSSIP_INTERVAL = 0.15
FAILURE_T_CLEANUP = 10.0 JOIN_RETRY_INTERVAL = 1.0
RETRY_INTERVAL = 5.0 GOSSIP_FANOUT = 1
FANOUT = 1
GOSSIP_FAILURE_TIMEOUT = 2.8
GOSSIP_S_SUSPECT_TIMEOUT = 1
GOSSIP_S_TIMEOUT_SUSPECT = 1.8
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 = {} 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() members_list_lock = threading.Lock()
#adding its self to the membership list suspect_history = []
SELF_ID = (IP, SERVER_PORT, utc_time()) suspect_history_lock = threading.Lock()
members_list[SELF_ID] = {
"heart_beat": 1, MY_ID = (MY_IP, GLOBAL_PORT, utc_time())
"local_last_seen": utc_time(), members_list[MY_ID] = {
"heart_beat": 0,
"state": "alive", "state": "alive",
"local_failed_since" : None, "incarnation_counter": 0,
"local_suspected_since" : None "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(): def worker_1():
has_joined = INTRODUCER_FLAG #if this is the introducer, it has already joined has_joined = (INTRODUCER_IP == MY_IP)
last_join_time = 0 #flag to track the last time a join request was sent to retry last_join_attempt = 0.0
while not stop_event.is_set(): while not stop_threads.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): #we keep retrying until we have joined the cluster
join_payload = { if not has_joined and (utc_time() - last_join_attempt) >= JOIN_RETRY_INTERVAL:
"type": "JOIN", joining_message_payload = {"type": "JOIN", "sender_id": MY_ID}
"sender_id": SELF_ID joining_message_json = json.dumps(joining_message_payload)
}
server_socket.sendto(json.dumps(join_payload).encode('utf-8'), (INTRODUCER_IP, SERVER_PORT)) try:
last_join_time = utc_time() 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: try:
data, addr = server_socket.recvfrom(65535) 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()
except socket.timeout: except socket.timeout:
continue continue
#handles incrementing the local node's heartbeat counter at regular intervals try:
def worker_2(): msg = json.loads(data.decode('utf-8'))
while not stop_event.is_set(): except Exception:
with members_list_lock: continue
members_list[SELF_ID]["heart_beat"] += 1
time.sleep(HEART_BEAT_INTERVAL)
#handles sending gossip messages to a subset of alive peers at regular intervals msg_type = msg.get("type")
def worker_3():
peer_queue = [] #handle dropping messages only on GOSSIP
if msg_type == "GOSSIP":
while not stop_event.is_set(): 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: 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_list[MY_ID]["heart_beat"] += 1
time.sleep(HEARTBEAT_INTERVAL)
members_payload = []
for m_id, m_data in members_list.items(): #THREAD 3 - send gossip messages to random members
members_payload.append({ #threshold, pure random can starve a peer for a few rounds and cause false positives.
"ip": m_id[0], _peer_queue = []
"port": m_id[1], _peer_queue_lock = threading.Lock()
"timestamp": m_id[2],
"heart_beat": m_data["heart_beat"], def worker_3():
"state": m_data["state"] global _peer_queue
}) while not stop_threads.is_set():
with members_list_lock:
if current_peers: current_peers = [member_id for member_id in members_list if member_id != MY_ID and members_list[member_id]["state"] in ["alive", "suspect"]]
#remove any peers from the queue that are no longer alive
peer_queue = [p for p in peer_queue if p in current_peers] 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 = [] selected_peers = []
num_to_select = min(FANOUT, len(current_peers)) for _ in range(min(GOSSIP_FANOUT, len(current_peers))):
if not _peer_queue:
for _ in range(num_to_select): _peer_queue = list(current_peers)
if not peer_queue: random.shuffle(_peer_queue)
peer_queue = list(current_peers) if _peer_queue:
random.shuffle(peer_queue) selected_peers.append(_peer_queue.pop(0))
if peer_queue: for peer in selected_peers:
selected_peers.append(peer_queue.pop(0)) gossip_payload = {
"type": "GOSSIP",
for peer in selected_peers: "sender_id": MY_ID,
peer_ip, peer_port, _ = peer "members": gossip_members_payload,
"mode": gossip_control_payload["mode"],
payload = { "cluster_version": gossip_control_payload["version"],
"type": "GOSSIP", "drop_rate": gossip_control_payload["drop_rate"],
"sender_id": SELF_ID, }
"members": members_payload
} try:
server_socket.sendto(json.dumps(gossip_payload).encode('utf-8'), (peer[0], peer[1]))
json_string = json.dumps(payload) except Exception as e:
server_socket.sendto(json_string.encode('utf-8'), (peer_ip, peer_port)) log(f"Failed to send GOSSIP to {peer[0]}:{peer[1]}: {e}")
time.sleep(GOSSIP_INTERVAL) time.sleep(GOSSIP_INTERVAL)
# handle failure detection and cleanup of failed members from the membership list # THREAD 4 FAILURE DETECTION
def worker_4(): def worker_4():
while not stop_event.is_set(): while not stop_threads.is_set():
current_time = utc_time() current_time = utc_time()
to_remove = [] members_to_remove = []
with members_list_lock: with members_list_lock:
for member_id, member_data in members_list.items(): for member_id, member_data in members_list.items():
if member_id == SELF_ID: if member_id == MY_ID:
continue continue # Skip self
if member_data["state"] == "alive": #gossip
last_seen = member_data["local_last_seen"] if not suspect_mode.is_set():
time_since_last_seen = current_time - last_seen if member_data["state"] == "alive" and (current_time - member_data["local_last_seen"]) > GOSSIP_FAILURE_TIMEOUT:
if time_since_last_seen > FAILURE_TIMEOUT:
member_data["state"] = "failed" member_data["state"] = "failed"
member_data["local_failed_since"] = current_time member_data["local_failed_since"] = current_time
log(f"Member {member_id[0]}:{member_id[1]} is now failed (gossip timeout)")
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: #gossip + Suspect
to_remove.append(member_id) else:
if member_data["state"] == "alive" and (current_time - member_data["local_last_seen"]) > GOSSIP_S_SUSPECT_TIMEOUT:
for member_id in to_remove: 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) dead_members.add(member_id)
del members_list[member_id] log(f"Removed member {member_id[0]}:{member_id[1]}")
time.sleep(0.5) 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 <number>")
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():
#for monitoring purposes
def worker_monitoring():
while not stop_event.is_set():
with members_list_lock: with members_list_lock:
local_ml = { snapshot = {
member_id: member_data.copy() 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) with cluster_state_lock:
t2 = threading.Thread(target=worker_2) mode = cluster_state["mode"]
t3 = threading.Thread(target=worker_3) drop_rate = cluster_state["drop_rate"]
t4 = threading.Thread(target=worker_4)
t5 = threading.Thread(target=worker_monitoring)
t1.start() try:
t2.start() with open("membership_list.txt", "w") as f:
t3.start() for member_id, member_data in snapshot.items():
t4.start() f.write(f"{member_id}: {member_data}\n")
t5.start()
try: f.write(
while not stop_event.is_set(): f"Mode: {mode}, "
time.sleep(0.5) f"Drop Rate: {drop_rate}\n"
except KeyboardInterrupt: )
print("\nExiting...")
stop_event.set()
t1.join() except Exception as e:
t2.join() pass
t3.join() time.sleep(0.25)
t4.join()
t5.join()
server_socket.close() 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()