From db037f4f42fa165680ce26dabb0ff94b2b426c4a Mon Sep 17 00:00:00 2001 From: Aryan Singh Date: Fri, 25 Sep 2026 22:57:39 -0500 Subject: [PATCH 01/11] adding monitoring --- MP2/peer.py | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/MP2/peer.py b/MP2/peer.py index 060d635..859237e 100644 --- a/MP2/peer.py +++ b/MP2/peer.py @@ -149,14 +149,31 @@ def worker_3(): time.sleep(GOSSIP_INTERVAL) +#only for debuging +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() + } + + 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) t1.start() t2.start() t3.start() +t4.start() try: while not stop_event.is_set(): @@ -168,4 +185,5 @@ except KeyboardInterrupt: t1.join() t2.join() t3.join() +t4.join() server_socket.close() From ce0cec697c192e28d7ed9413282e4b8f643b8395 Mon Sep 17 00:00:00 2001 From: Aryan Singh Date: Fri, 25 Sep 2026 23:03:11 -0500 Subject: [PATCH 02/11] Update peer.py --- MP2/peer.py | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/MP2/peer.py b/MP2/peer.py index 859237e..00af6d9 100644 --- a/MP2/peer.py +++ b/MP2/peer.py @@ -38,16 +38,14 @@ 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(): try: data, addr = server_socket.recvfrom(65535) - try: - msg = json.loads(data.decode('utf-8')) - except json.JSONDecodeError: - continue + msg = json.loads(data.decode('utf-8')) msg_type = msg.get("type") print(f"Processing message type={msg_type} from {addr}") @@ -75,7 +73,7 @@ def worker_1(): print(f"New member joined: {join_id}") - # Send the initial state directly to the joining node + # Send the initial state directly to the joining node - decreases the overhead of having that init state propogate through the cluster reply_payload = { "type": "GOSSIP", "sender_id": SELF_ID, From 87d4bb967c31bb7230acc491f17e57f970a84641 Mon Sep 17 00:00:00 2001 From: Aryan Singh Date: Fri, 25 Sep 2026 23:08:57 -0500 Subject: [PATCH 03/11] minor fix --- MP2/.gitignore | 1 + MP2/peer.py | 2 +- 2 files changed, 2 insertions(+), 1 deletion(-) create mode 100644 MP2/.gitignore diff --git a/MP2/.gitignore b/MP2/.gitignore new file mode 100644 index 0000000..528dc4e --- /dev/null +++ b/MP2/.gitignore @@ -0,0 +1 @@ +scripts \ No newline at end of file diff --git a/MP2/peer.py b/MP2/peer.py index 00af6d9..c149e2a 100644 --- a/MP2/peer.py +++ b/MP2/peer.py @@ -152,7 +152,7 @@ def worker_monitoring(): while not stop_event.is_set(): with members_list_lock: local_ml = { - member_id: member_data.copy(), + member_id: member_data.copy() for member_id , member_data in members_list.items() } From 13a2033b5f715c1bbab5c629d72670c352017cf5 Mon Sep 17 00:00:00 2001 From: Aryan Singh Date: Fri, 25 Sep 2026 23:12:29 -0500 Subject: [PATCH 04/11] changing port so it does not collide --- MP2/peer.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/MP2/peer.py b/MP2/peer.py index c149e2a..f01260e 100644 --- a/MP2/peer.py +++ b/MP2/peer.py @@ -10,7 +10,7 @@ def utc_time(): INTRODUCER_IP = "172.22.156.3" IP = socket.gethostbyname(socket.gethostname()) -SERVER_PORT = 6767 +SERVER_PORT = 8989 INTRODUCER_FLAG = (INTRODUCER_IP == IP) HEART_BEAT_INTERVAL = 5 GOSSIP_INTERVAL = 10.0 From a63bff32f914b0b29e980c9c87e7957b855e42c6 Mon Sep 17 00:00:00 2001 From: Aryan Singh Date: Sat, 26 Sep 2026 00:42:10 -0500 Subject: [PATCH 05/11] added gossip style failure detection test1 --- MP2/peer.py | 135 ++++++++++++++++++++++++++++++++++++---------------- 1 file changed, 95 insertions(+), 40 deletions(-) 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() From 6f5cfe02b81ead773c6da978facece247ca61f7f Mon Sep 17 00:00:00 2001 From: Aryan Singh Date: Sat, 26 Sep 2026 01:18:42 -0500 Subject: [PATCH 06/11] implementing retyring logic for joining --- MP2/peer.py | 56 ++++++++++++++++++++++++++++++++--------------------- 1 file changed, 34 insertions(+), 22 deletions(-) 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: From 9f085ea1b6193640f4f46e337af473601534950c Mon Sep 17 00:00:00 2001 From: Aryan Singh Date: Sat, 26 Sep 2026 01:23:35 -0500 Subject: [PATCH 07/11] minor fix --- MP2/peer.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/MP2/peer.py b/MP2/peer.py index cf67227..3b836fc 100644 --- a/MP2/peer.py +++ b/MP2/peer.py @@ -107,7 +107,7 @@ def worker_1(): incoming_state = member['state'] incoming_hb = member['heart_beat'] - if member_id not in members_list: + if member_id not in members_list and incoming_state != "failed": members_list[member_id] = { "heart_beat": incoming_hb, "local_last_seen": utc_time(), From 00fec9b46c6709a8e20752a0104569f8557af902 Mon Sep 17 00:00:00 2001 From: Aryan Singh Date: Sat, 26 Sep 2026 01:50:00 -0500 Subject: [PATCH 08/11] Update peer.py --- MP2/peer.py | 20 ++++++++++++++++---- 1 file changed, 16 insertions(+), 4 deletions(-) diff --git a/MP2/peer.py b/MP2/peer.py index 3b836fc..d392038 100644 --- a/MP2/peer.py +++ b/MP2/peer.py @@ -20,6 +20,7 @@ RETRY_INTERVAL = 5.0 FANOUT = 1 members_list = {} +dead_members = set() members_list_lock = threading.Lock() SELF_ID = (IP, SERVER_PORT, utc_time()) @@ -63,6 +64,9 @@ def worker_1(): join_id = (join_ip, join_port, join_timestamp) with members_list_lock: + if join_id in dead_ids: + continue + if join_id not in members_list: members_list[join_id] = { "heart_beat": 1, @@ -107,13 +111,21 @@ def worker_1(): incoming_state = member['state'] incoming_hb = member['heart_beat'] - if member_id not in members_list and incoming_state != "failed": + if member_id in dead_ids: + continue + + if member_id not in members_list: + + if incoming_state == "failed": + dead_ids.add(member_id) + continue + members_list[member_id] = { "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 + "local_failed_since": None, + "local_suspected_since": None } continue @@ -209,8 +221,8 @@ def worker_4(): to_remove.append(member_id) for member_id in to_remove: + dead_ids.add(member_id) del members_list[member_id] - print(f"Purged member {member_id} from list.") time.sleep(0.5) From 20f4cc124dba79ddcd3065f123b1f4d930742d14 Mon Sep 17 00:00:00 2001 From: Aryan Singh Date: Sat, 26 Sep 2026 01:51:18 -0500 Subject: [PATCH 09/11] minor fix --- MP2/peer.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/MP2/peer.py b/MP2/peer.py index d392038..5a9b024 100644 --- a/MP2/peer.py +++ b/MP2/peer.py @@ -64,7 +64,7 @@ def worker_1(): join_id = (join_ip, join_port, join_timestamp) with members_list_lock: - if join_id in dead_ids: + if join_id in dead_members: continue if join_id not in members_list: @@ -111,13 +111,13 @@ def worker_1(): incoming_state = member['state'] incoming_hb = member['heart_beat'] - if member_id in dead_ids: + if member_id in dead_members: continue if member_id not in members_list: if incoming_state == "failed": - dead_ids.add(member_id) + dead_members.add(member_id) continue members_list[member_id] = { @@ -221,7 +221,7 @@ def worker_4(): to_remove.append(member_id) for member_id in to_remove: - dead_ids.add(member_id) + dead_members.add(member_id) del members_list[member_id] time.sleep(0.5) From 611a86830e600ca3a04fd1c13098475b7dd5a665 Mon Sep 17 00:00:00 2001 From: Aryan Singh Date: Sat, 26 Sep 2026 02:24:15 -0500 Subject: [PATCH 10/11] V1 DONE --- MP2/peer.py | 80 ++++++++++++++++++++++++++++++++--------------------- 1 file changed, 49 insertions(+), 31 deletions(-) diff --git a/MP2/peer.py b/MP2/peer.py index 5a9b024..aad448d 100644 --- a/MP2/peer.py +++ b/MP2/peer.py @@ -8,8 +8,8 @@ import json def utc_time(): return datetime.now(timezone.utc).timestamp() -INTRODUCER_IP = "172.22.156.3" -IP = socket.gethostbyname(socket.gethostname()) +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 @@ -19,10 +19,25 @@ FAILURE_T_CLEANUP = 10.0 RETRY_INTERVAL = 5.0 FANOUT = 1 +#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() +dead_members = set() #[BASICALLY VERSION TRACKING]this tracks members that have been marked as failed and removed from the membership list. 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, @@ -31,19 +46,21 @@ members_list[SELF_ID] = { "local_failed_since" : None, "local_suspected_since" : None } -print(f"Starting peer: IP={IP}, port={SERVER_PORT}, introducer={INTRODUCER_FLAG}, self_id={SELF_ID}") -stop_event = threading.Event() +stop_event = threading.Event() #trigger to stop all threads when the program is exiting +#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 def worker_1(): - has_joined = INTRODUCER_FLAG - last_join_time = 0 + 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", @@ -59,11 +76,13 @@ def worker_1(): 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 @@ -75,7 +94,6 @@ def worker_1(): "local_failed_since" : None, "local_suspected_since" : None } - print(f"Added member: {join_id}") members_payload = [] for m_id, m_data in members_list.items(): @@ -86,16 +104,18 @@ def worker_1(): "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)) - print(f"Sent initial membership state to {(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 @@ -105,32 +125,27 @@ def worker_1(): for member in members_json: member_id = (member['ip'], member['port'], member['timestamp']) - if member_id == SELF_ID: + if member_id == SELF_ID or member_id in dead_members: continue incoming_state = member['state'] incoming_hb = member['heart_beat'] - - if member_id in dead_members: - continue - - if member_id not in members_list: - - if incoming_state == "failed": - dead_members.add(member_id) - continue - - 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 local_member = members_list[member_id] + if local_member is None: + if incoming_state == "failed": + dead_members.add(member_id) + 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 @@ -145,12 +160,14 @@ def worker_1(): 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) +#handles sending gossip messages to a subset of alive peers at regular intervals def worker_3(): peer_queue = [] @@ -169,14 +186,13 @@ def worker_3(): }) if current_peers: - # Clean up peers from queue that might have failed or left + #remove any peers from the queue that are no longer alive 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) @@ -198,6 +214,7 @@ def worker_3(): time.sleep(GOSSIP_INTERVAL) +# handle failure detection and cleanup of failed members from the membership list def worker_4(): while not stop_event.is_set(): current_time = utc_time() @@ -226,6 +243,7 @@ def worker_4(): time.sleep(0.5) +#for monitoring purposes def worker_monitoring(): while not stop_event.is_set(): with members_list_lock: From c23085954957efb3bc31ff5459bef57ad84a5cf8 Mon Sep 17 00:00:00 2001 From: Aryan Singh Date: Sat, 26 Sep 2026 02:27:42 -0500 Subject: [PATCH 11/11] V1 Done --- MP2/peer.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/MP2/peer.py b/MP2/peer.py index aad448d..763d732 100644 --- a/MP2/peer.py +++ b/MP2/peer.py @@ -131,11 +131,12 @@ def worker_1(): incoming_state = member['state'] incoming_hb = member['heart_beat'] - local_member = members_list[member_id] + 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,