From a4f6d78ab93ba37711bda12c37461e2f4fe9c217 Mon Sep 17 00:00:00 2001 From: Aryan Singh Date: Fri, 25 Sep 2026 18:46:04 -0500 Subject: [PATCH 01/11] init test --- MP2/peer.py | 32 ++++++++++++++++++++++++++++++++ 1 file changed, 32 insertions(+) create mode 100644 MP2/peer.py diff --git a/MP2/peer.py b/MP2/peer.py new file mode 100644 index 0000000..31aa924 --- /dev/null +++ b/MP2/peer.py @@ -0,0 +1,32 @@ +from datetime import datetime, timezone +import socket + +INTRODUCER_IP = "172.22.156.3" +IP = socket.gethostbyname(socket.gethostname()) +SERVER_PORT = 6767 +INTRODUCER_FLAG = (INTRODUCER_IP == IP) + +#check if this VM is the introducer and add it self +members_list = {} +if INTRODUCER_FLAG: + members_list[(IP, PORT)] = { + "heart_beat": 1, + "last_seen": utc_time() + } + + +#start UDP socket +server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) +server.bind((IP, SERVER_PORT)) + +if not INTRODUCER_FLAG: + server.sendto(b"JOIN", (INTRODUCER_IP, SERVER_PORT)) + +while True: + data, addr = server.recvfrom(4096) + +#increment heart beat + +#utc time +def utc_time(): + return datetime.now(timezone.utc) From 492b93544c7cf776b9517ecd366392a60992fd54 Mon Sep 17 00:00:00 2001 From: Aryan Singh Date: Fri, 25 Sep 2026 18:57:35 -0500 Subject: [PATCH 02/11] Update peer.py --- MP2/peer.py | 38 ++++++++++++++++---------------------- 1 file changed, 16 insertions(+), 22 deletions(-) diff --git a/MP2/peer.py b/MP2/peer.py index 31aa924..f9bad3b 100644 --- a/MP2/peer.py +++ b/MP2/peer.py @@ -1,32 +1,26 @@ from datetime import datetime, timezone import socket -INTRODUCER_IP = "172.22.156.3" -IP = socket.gethostbyname(socket.gethostname()) -SERVER_PORT = 6767 -INTRODUCER_FLAG = (INTRODUCER_IP == IP) +def main(): + INTRODUCER_IP = "172.22.156.3" + IP = socket.gethostbyname(socket.gethostname()) + SERVER_PORT = 6767 + INTRODUCER_FLAG = (INTRODUCER_IP == IP) -#check if this VM is the introducer and add it self -members_list = {} -if INTRODUCER_FLAG: - members_list[(IP, PORT)] = { - "heart_beat": 1, - "last_seen": utc_time() + members_list = { + (IP, SERVER_PORT): {"heart_beat": 1, "last_seen": utc_time()} } - -#start UDP socket -server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) -server.bind((IP, SERVER_PORT)) + server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + server.bind(("", SERVER_PORT)) -if not INTRODUCER_FLAG: - server.sendto(b"JOIN", (INTRODUCER_IP, SERVER_PORT)) + if not INTRODUCER_FLAG: + server.sendto(b"JOIN", (INTRODUCER_IP, SERVER_PORT)) + data, addr = server.recvfrom(4096) + print(addr, data) -while True: - data, addr = server.recvfrom(4096) - -#increment heart beat - -#utc time def utc_time(): return datetime.now(timezone.utc) + +if __name__ == "__main__": + main() From 5ecc71d6f879160e02eb0471437b7dbf002ea616 Mon Sep 17 00:00:00 2001 From: Aryan Singh Date: Fri, 25 Sep 2026 18:59:22 -0500 Subject: [PATCH 03/11] Update peer.py --- MP2/peer.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/MP2/peer.py b/MP2/peer.py index f9bad3b..f26beee 100644 --- a/MP2/peer.py +++ b/MP2/peer.py @@ -12,7 +12,7 @@ def main(): } server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) - server.bind(("", SERVER_PORT)) + server.bind((IP, SERVER_PORT)) if not INTRODUCER_FLAG: server.sendto(b"JOIN", (INTRODUCER_IP, SERVER_PORT)) From 1312a154a6a22dbf0abededbc3a7d82d03fcdd78 Mon Sep 17 00:00:00 2001 From: Aryan Singh Date: Fri, 25 Sep 2026 19:00:50 -0500 Subject: [PATCH 04/11] Update peer.py --- MP2/peer.py | 46 ++++++++++++++++++++++++++-------------------- 1 file changed, 26 insertions(+), 20 deletions(-) diff --git a/MP2/peer.py b/MP2/peer.py index f26beee..8c102d2 100644 --- a/MP2/peer.py +++ b/MP2/peer.py @@ -1,26 +1,32 @@ from datetime import datetime, timezone import socket -def main(): - INTRODUCER_IP = "172.22.156.3" - IP = socket.gethostbyname(socket.gethostname()) - SERVER_PORT = 6767 - INTRODUCER_FLAG = (INTRODUCER_IP == IP) - - members_list = { - (IP, SERVER_PORT): {"heart_beat": 1, "last_seen": utc_time()} - } - - server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) - server.bind((IP, SERVER_PORT)) - - if not INTRODUCER_FLAG: - server.sendto(b"JOIN", (INTRODUCER_IP, SERVER_PORT)) - data, addr = server.recvfrom(4096) - print(addr, data) - def utc_time(): return datetime.now(timezone.utc) -if __name__ == "__main__": - main() + +INTRODUCER_IP = "172.22.156.3" +IP = socket.gethostbyname(socket.gethostname()) +SERVER_PORT = 6767 +INTRODUCER_FLAG = (INTRODUCER_IP == IP) + +#check if this VM is the introducer and add it self +members_list = {} +if INTRODUCER_FLAG: + members_list[(IP, PORT)] = { + "heart_beat": 1, + "last_seen": utc_time() + } + + +#start UDP socket +server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) +server.bind((IP, SERVER_PORT)) + +if not INTRODUCER_FLAG: + server.sendto(b"JOIN", (INTRODUCER_IP, SERVER_PORT)) + +while True: + data, addr = server.recvfrom(4096) + +#increment heart beat \ No newline at end of file From 80592318621a5fa39d81d27d88d92d622a449b00 Mon Sep 17 00:00:00 2001 From: Aryan Singh Date: Fri, 25 Sep 2026 19:01:15 -0500 Subject: [PATCH 05/11] Update peer.py --- MP2/peer.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/MP2/peer.py b/MP2/peer.py index 8c102d2..ec8b720 100644 --- a/MP2/peer.py +++ b/MP2/peer.py @@ -13,7 +13,7 @@ INTRODUCER_FLAG = (INTRODUCER_IP == IP) #check if this VM is the introducer and add it self members_list = {} if INTRODUCER_FLAG: - members_list[(IP, PORT)] = { + members_list[(IP, SERVER_PORT)] = { "heart_beat": 1, "last_seen": utc_time() } From b54210aa62cae015e35a75ba37d04fbdd96a57d1 Mon Sep 17 00:00:00 2001 From: Aryan Singh Date: Fri, 25 Sep 2026 19:01:48 -0500 Subject: [PATCH 06/11] Update peer.py --- MP2/peer.py | 1 + 1 file changed, 1 insertion(+) diff --git a/MP2/peer.py b/MP2/peer.py index ec8b720..bc7518d 100644 --- a/MP2/peer.py +++ b/MP2/peer.py @@ -28,5 +28,6 @@ if not INTRODUCER_FLAG: while True: data, addr = server.recvfrom(4096) + print(data, addr) #increment heart beat \ No newline at end of file From 654c4e42580190d5293bc8b6c1ec0c3afccc0309 Mon Sep 17 00:00:00 2001 From: Aryan Singh Date: Fri, 25 Sep 2026 19:03:39 -0500 Subject: [PATCH 07/11] Update peer.py --- MP2/peer.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/MP2/peer.py b/MP2/peer.py index bc7518d..d54669a 100644 --- a/MP2/peer.py +++ b/MP2/peer.py @@ -28,6 +28,9 @@ if not INTRODUCER_FLAG: while True: data, addr = server.recvfrom(4096) - print(data, addr) + if data == "JOIN": + print(True) + print(data) + #increment heart beat \ No newline at end of file From 4b6516fc992286a62ee2b683b9f85187a626a10a Mon Sep 17 00:00:00 2001 From: Aryan Singh Date: Fri, 25 Sep 2026 19:04:35 -0500 Subject: [PATCH 08/11] Update peer.py --- MP2/peer.py | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/MP2/peer.py b/MP2/peer.py index d54669a..efd7dfd 100644 --- a/MP2/peer.py +++ b/MP2/peer.py @@ -28,9 +28,12 @@ if not INTRODUCER_FLAG: while True: data, addr = server.recvfrom(4096) - if data == "JOIN": - print(True) - print(data) + if data == b"JOIN": + members_list[addr] = { + "heart_beat": 1, + "last_seen": utc_time() + } + print(members_list) #increment heart beat \ No newline at end of file From 4525387b04fafcd77f8b81edbe165df933e2e204 Mon Sep 17 00:00:00 2001 From: Aryan Singh Date: Fri, 25 Sep 2026 19:06:25 -0500 Subject: [PATCH 09/11] Update peer.py --- MP2/peer.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/MP2/peer.py b/MP2/peer.py index efd7dfd..128c16e 100644 --- a/MP2/peer.py +++ b/MP2/peer.py @@ -33,7 +33,7 @@ while True: "heart_beat": 1, "last_seen": utc_time() } - print(members_list) + print(members_list[addr]) #increment heart beat \ No newline at end of file From 8915eab2092d71c7b4d2180a5d9d650ed6133408 Mon Sep 17 00:00:00 2001 From: Aryan Singh Date: Fri, 25 Sep 2026 21:24:47 -0500 Subject: [PATCH 10/11] Update peer.py --- MP2/peer.py | 168 +++++++++++++++++++++++++++++++++++++++++++++------- 1 file changed, 146 insertions(+), 22 deletions(-) diff --git a/MP2/peer.py b/MP2/peer.py index 128c16e..49cf850 100644 --- a/MP2/peer.py +++ b/MP2/peer.py @@ -1,39 +1,163 @@ from datetime import datetime, timezone import socket +import threading +import time +import random +import json def utc_time(): - return datetime.now(timezone.utc) - + return datetime.now(timezone.utc).timestamp() INTRODUCER_IP = "172.22.156.3" IP = socket.gethostbyname(socket.gethostname()) SERVER_PORT = 6767 INTRODUCER_FLAG = (INTRODUCER_IP == IP) +HEART_BEAT_INTERVAL = 5 +GOSSIP_INTERVAL = 10.0 +FAILURE_TIMEOUT = 3.0 -#check if this VM is the introducer and add it self members_list = {} -if INTRODUCER_FLAG: - members_list[(IP, SERVER_PORT)] = { - "heart_beat": 1, - "last_seen": utc_time() - } - +members_list_lock = threading.Lock() -#start UDP socket -server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) -server.bind((IP, SERVER_PORT)) +SELF_ID = (IP, SERVER_PORT, utc_time()) +members_list[SELF_ID] = { + "heart_beat": 1, + "last_seen": utc_time() +} -if not INTRODUCER_FLAG: - server.sendto(b"JOIN", (INTRODUCER_IP, SERVER_PORT)) +stop_event = threading.Event() -while True: - data, addr = server.recvfrom(4096) - if data == b"JOIN": - members_list[addr] = { - "heart_beat": 1, - "last_seen": utc_time() +server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) +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 } - print(members_list[addr]) + 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_type = msg.get("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 join_id not in members_list: + members_list[join_id] = { + "heart_beat": 1, + "last_seen": utc_time() + } + + 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"] + }) + + print(f"New member joined: {join_id}") + + # 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)) + + elif msg_type == "GOSSIP": + members_json = msg.get('members', []) + + 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 + + if member_id not in members_list: + members_list[member_id] = { + "heart_beat": member_hb, + "last_seen": utc_time() + } + 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() + + except socket.timeout: + continue + +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) + +def worker_3(): + while not stop_event.is_set(): + with members_list_lock: + peers = list(members_list.keys()) + + 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"] + }) + + if SELF_ID in peers: + peers.remove(SELF_ID) + + if peers: + random_peer = random.choice(peers) + peer_ip, peer_port, _ = random_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)) + + time.sleep(GOSSIP_INTERVAL) -#increment heart beat \ No newline at end of file +t1 = threading.Thread(target=worker_1) +t2 = threading.Thread(target=worker_2) +t3 = threading.Thread(target=worker_3) + +t1.start() +t2.start() +t3.start() + +try: + while not stop_event.is_set(): + time.sleep(0.5) +except KeyboardInterrupt: + print("\nExiting...") + stop_event.set() + +t1.join() +t2.join() +t3.join() +server_socket.close() From d6938f088e3e705da0dd9004eab9ada76337d297 Mon Sep 17 00:00:00 2001 From: Aryan Singh Date: Fri, 25 Sep 2026 21:28:32 -0500 Subject: [PATCH 11/11] Update peer.py add print statements --- MP2/peer.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/MP2/peer.py b/MP2/peer.py index 49cf850..060d635 100644 --- a/MP2/peer.py +++ b/MP2/peer.py @@ -24,6 +24,7 @@ members_list[SELF_ID] = { "heart_beat": 1, "last_seen": utc_time() } +print(f"Starting peer: IP={IP}, port={SERVER_PORT}, introducer={INTRODUCER_FLAG}, self_id={SELF_ID}") stop_event = threading.Event() @@ -49,6 +50,7 @@ def worker_1(): continue 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"] @@ -60,6 +62,7 @@ def worker_1(): "heart_beat": 1, "last_seen": utc_time() } + print(f"Added member: {join_id}") members_payload = [] for m_id, m_data in members_list.items(): @@ -79,6 +82,7 @@ def worker_1(): "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}") elif msg_type == "GOSSIP": members_json = msg.get('members', []) @@ -96,9 +100,11 @@ def worker_1(): "heart_beat": member_hb, "last_seen": utc_time() } + 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}") except socket.timeout: continue @@ -107,6 +113,7 @@ 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(): @@ -137,6 +144,7 @@ def worker_3(): } 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)) time.sleep(GOSSIP_INTERVAL)