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()