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: