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

V1 DONE

Browse Source
This commit is contained in:
Aryan Singh committed 2026-09-26 02:24:15 -05:00
1 parent 20f4cc124d
commit 611a86830e
1 file changed
+49 -31
Download Patch File Download Diff File Expand all files Collapse all files
MP2/peer.py
+49 -31
View File
@@ -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: