membership

test1

MP1/README.md

@@ -0,0 +1,40 @@
+# MP1 Usage
+
+## Running MP1
+
+Use `run_mp.sh <1..10|all>` to start the socket servers on specific VMs:
+
+```bash
+./run_mp.sh 3
+./run_mp.sh all
+```
+
+Use `kill_mp.sh <1..10|all>` to stop the socket servers on specific VMs:
+
+```bash
+./kill_mp.sh 3
+./kill_mp.sh all
+```
+
+Use `deploy_mp.sh` to kill the servers on all VMs, pull the latest version of `main`, and restart every server using `run_mp.sh`:
+
+```bash
+./deploy_mp.sh
+```
+
+After deployment, the `grepp` command should be available. Use it like regular `grep`, but it searches the log files across all ten machines:
+
+```bash
+grepp ERROR
+grepp -E 'GET|POST'
+```
+
+## Testing
+
+Run `test.py` on your local machine:
+
+```bash
+python3 test.py
+```
+
+The script generates test log files, runs `grepp` with the `-t` flag, checks the output for correctness, and deletes the generated test files afterward.

MP1/client.py

@@ -1,8 +1,10 @@
 #!/usr/bin/env python3
 import socket
 import sys
+import time
+import shlex
 
-PORT = 5055
+PORT = 5056
 ALL_IPS = [
     "172.22.156.3",
     "172.22.158.2",
@@ -48,10 +50,9 @@ def grep_all_ips(ips, cmd, if_unit_test):
             client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
             client_socket.connect((ip, PORT))
 
-            if if_unit_test:
+            filename = f"machine.{i + 1}.test.log" if if_unit_test else f"machine.{i + 1}.log"
-                full_cmd = cmd + f" ./machine.{i + 1}.test.log"
+
-            else:
+            full_cmd = shlex.join(cmd + [filename])
-                full_cmd = cmd + f" ./machine.{i + 1}.log"
 
             send_message(client_socket, full_cmd)
             data = receive_message(client_socket)
@@ -73,8 +74,12 @@ def grep_all_ips(ips, cmd, if_unit_test):
 
 #main
 #if unit test flag exists, then use the test log files instead of the actual log files
-cmd = "grep " + " ".join([arg for arg in sys.argv[1:] if arg != "-t"])
+cmd = ["grep", *[arg for arg in sys.argv[1:] if arg != "-t"]]
-match_count = grep_all_ips(ALL_IPS, cmd, if_unit_test=[x for x in sys.argv].count("-t") > 0)
+#track time
+start_time = time.perf_counter()
+is_unit_test = "-t" in sys.argv
+match_count = grep_all_ips(ALL_IPS, cmd, if_unit_test=is_unit_test)
+end_time = time.perf_counter()
 
 print("=========================== SUMMARY ===========================")
 for i, count in enumerate(match_count):
@@ -88,5 +93,7 @@ for i, count in enumerate(match_count):
             print(f"machine.{i + 1}.test.log: {count} matches")
         else:
             print(f"machine.{i + 1}.log: {count} matches")
-print(f"Total matches: {max(0, sum(match_count))}")
+total_matches = sum(count for count in match_count if count > 0)
+print(f"Total matches: {total_matches}")
 print("===============================================================")
+print(f"Time taken: {end_time - start_time:.2f} seconds")

MP1/scripts/kill_mp.sh

@@ -3,13 +3,13 @@
 if [[ "$1" == "all" ]]; then
     for i in {1..10}; do
         echo "Killing server$i..."
-        ssh -o BatchMode=yes "server$i" "fuser -k -9 5055/tcp"
+        ssh -o BatchMode=yes "server$i" "fuser -k -9 5056/tcp"
         echo ""
     done
 else
     for i in "$@"; do
         echo "Killing server$i..."
-        ssh -o BatchMode=yes "server$i" "fuser -k -9 5055/tcp"
+        ssh -o BatchMode=yes "server$i" "fuser -k -9 5056/tcp"
         echo ""
     done
 fi

MP1/scripts/run_mp.sh

@@ -1,6 +1,6 @@
 #!/usr/bin/env bash
 
-CMD='cd fa26_cs425_.group-mp_.team-01/MP1 && nohup python3 server.py </dev/null >/dev/null 2>&1 &'
+CMD='cd /home/asing271/fa26_cs425_.group-mp_.team-01/MP1 && nohup python3 server.py </dev/null >/dev/null 2>&1 &'
 
 if [[ "$#" -eq 0 ]]; then
     echo "Need at least one server number as an argument."

MP1/server.py

@@ -5,7 +5,7 @@ import subprocess
 from pathlib import Path
 
 HOST = "0.0.0.0"
-PORT = 5055
+PORT = 5056
 
 server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 
@@ -44,25 +44,23 @@ while True:
     conn, addr = server_socket.accept()
     print(f"Connected by {addr}")
 
-    command = receive_message(conn)
+    try:
-    if command is None:
+        command = receive_message(conn)
+        if command is None:
+            conn.close()
+            continue
+
+        command_args = shlex.split(command)
+
+        command_args.insert(1, "-n")
+
+        result = subprocess.run(command_args, cwd=SCRIPT_DIR, capture_output=True, text=True)
+    
+        send_message(conn, result.stdout + result.stderr)
+    
+    except Exception as e:
+        None
+    finally:
         conn.close()
-        continue
-
-    command_args = shlex.split(command)
-
-    if not command_args or command_args[0] != "grep":
-        response = "Invalid command. Please use the 'grep' command."
-        send_message(conn, response)
-        conn.close()
-        continue
-
-    command_args.insert(1, "-n")
-
-    result = subprocess.run(command_args, cwd=SCRIPT_DIR, capture_output=True, text=True)
-   
-    send_message(conn, result.stdout + result.stderr)
-
-    conn.close()
 
 server_socket.close()

MP1/test.py

@@ -304,9 +304,9 @@ print(f"Running grep command on a {machine_num} random machine...")
 for i in tests:
     result = subprocess.run(["ssh", f"asing271@{ALL_IPS[machine_num - 1]}", f"cd ~/fa26_cs425_.group-mp_.team-01/MP1 &&  grepp {i} -t"], capture_output=True, text=True)
     if result.stdout.strip() == results[tests.index(i)].strip():
-        print("PASS {i} test")
+        print(f"PASS {i} test")
     else:
-        print("FAIL {i} test")
+        print(f"FAIL {i} test")
 
 #delete log files on all the machines after running this test
 print("Deleting log files on all machines...")

MP2/README.md

@@ -0,0 +1,63 @@
+# MP2 Distributed Group Membership
+
+## Files
+- `peer.py` - membership daemon
+- `scripts/cmd.sh` - start/kill/control VMs
+- `scripts/view.sh` - tmux membership dashboard
+- `scripts/pretty_membership.py` - dashboard formatter
+
+## Running a node
+Set `MACHINE_INDEX` to the VM's index, then run:
+
+```bash
+MACHINE_INDEX=1 python3 peer.py
+```
+
+## Starting the demo
+Start the dashboard. `tmux` is required:
+
+```bash
+./scripts/view.sh
+```
+
+From another terminal, start the VMs:
+
+```bash
+./scripts/cmd.sh start 1
+./scripts/cmd.sh start 2 3 4 5
+./scripts/cmd.sh start 6 7 8 9 10
+```
+
+## Commands
+The following commands are available in the peer prompt:
+
+```text
+list_mem
+list_self
+leave
+display_suspects
+switch suspect
+switch nosuspect
+display_protocol
+set_drop_rate <percent>
+```
+
+## Demo/testing helpers
+Run these from the `MP2` directory:
+
+```bash
+./scripts/cmd.sh kill 4 8
+./scripts/cmd.sh start 4 8
+./scripts/cmd.sh cmd all "switch suspect"
+./scripts/cmd.sh cmd all "set_drop_rate 10"
+./scripts/cmd.sh status
+./scripts/cmd.sh log
+```
+
+## Introducer
+`VM1` / `172.22.156.3`
+
+## Notes
+- Gossip is default mode.
+- Artificial drops are applied at receivers.
+- Rejoining processes receive a new timestamp in their node ID.

MP2/peer.py

@@ -0,0 +1,620 @@
+import socket
+from datetime import datetime, timezone
+import threading
+import json
+import os
+import random
+import time
+import sys
+
+#setup logging
+def _machine_index():
+    idx = os.environ.get("MACHINE_INDEX")
+    if idx is not None:
+        return int(idx)
+    return None
+
+log_file = open(f"machine.{_machine_index()}.log", "w")
+log_file_mp1 = open(f"/home/asing271/fa26_cs425_.group-mp_.team-01/MP1/machine.{_machine_index()}.log","w")
+
+log_lock = threading.Lock()
+def log(message):
+    timestamp = datetime.now().strftime("%H:%M:%S.%f")[:-3]
+    with log_lock:
+        print(f"[{timestamp}] {message}")
+        log_file.write(f"[{timestamp}] {message}\n")
+        log_file.flush()
+
+        log_file_mp1.write(f"[{timestamp}] {message}\n")
+        log_file_mp1.flush()
+
+def utc_time():
+    return datetime.now(timezone.utc).timestamp()
+
+TOTAL_SENT_BYTES = 0
+TOTAL_RECEIVED_BYTES = 0
+traffic_lock = threading.Lock()
+
+#fixed
+INTRODUCER_IP = "172.22.156.3"
+GLOBAL_PORT = 8989
+MY_IP = socket.gethostbyname(socket.gethostname())
+
+HEARTBEAT_INTERVAL = 0.25
+GOSSIP_INTERVAL = 0.25
+JOIN_RETRY_INTERVAL = 1.0
+GOSSIP_FANOUT = 2
+
+GOSSIP_FAILURE_TIMEOUT = 3
+GOSSIP_S_SUSPECT_TIMEOUT = 1.0
+GOSSIP_S_TIMEOUT_SUSPECT = 2.0
+GOSSIP_FAILURE_TIMEOUT_CLEANUP = 6.0
+
+
+members_list = {}
+dead_members = set()
+members_list_lock = threading.Lock()
+
+suspect_history = []
+suspect_history_lock = threading.Lock()
+
+MY_ID = (MY_IP, GLOBAL_PORT, utc_time())
+members_list[MY_ID] = {
+    "heart_beat": 0,
+    "state": "alive",
+    "incarnation_counter": 0,
+    "local_last_seen": utc_time(),
+    "local_suspected_since": None,
+    "local_failed_since": None,
+}
+
+stop_threads = threading.Event() #to trigger a full shut down
+suspect_mode = threading.Event() #to trigger a switch to suspect mode
+
+
+#variable state
+#the ip is a tie breaker for the version number, so we can deterministically resolve conflicts
+#where two nodes have the same version number but different modes or drop rates
+#extremely unlikely, but possible if two nodes are started at the same time and both switch modes simultaneously
+cluster_state = {
+    "mode": "nosuspect",
+    "drop_rate": 0.0,
+    "version": (0, "0.0.0.0")
+}
+cluster_state_lock = threading.Lock()
+
+#start the main shared socket
+server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) #UDP
+server_socket.bind((MY_IP, GLOBAL_PORT))
+server_socket.settimeout(0.05)
+
+#global state switcher
+def global_switcher(cmd, value):
+    if cmd == "switch":
+        cmd = "mode"
+    if cmd == "mode":
+        with cluster_state_lock:
+            if value in ["suspect", "nosuspect"]:
+                cluster_state["mode"] = value
+                cluster_state["version"] = (cluster_state["version"][0] + 1, MY_IP)
+                if value == "suspect":
+                    suspect_mode.set()
+                else:
+                    suspect_mode.clear()
+                    with members_list_lock:
+                        for m_id, m_data in members_list.items():
+                            if m_id != MY_ID and m_data["state"] == "suspect":
+                                m_data["state"] = "alive"
+                                m_data["local_suspected_since"] = None
+                                log(f"Resolved {m_id[0]}:{m_id[1]} suspect -> alive on switch to nosuspect")
+                log(f"Switched mode to {value}")
+            else:
+                log(f"Invalid mode value: {value}")
+    elif cmd == "drop_rate":
+        try:
+            drop_rate_value = float(value)
+            if 0.0 <= drop_rate_value <= 100.0:
+                with cluster_state_lock:
+                    cluster_state["drop_rate"] = drop_rate_value
+                    cluster_state["version"] = (cluster_state["version"][0] + 1, MY_IP)
+                log(f"Set drop rate to {drop_rate_value}%")
+            else:
+                log(f"Invalid drop rate value: {value}. Must be between 0 and 100.")
+        except ValueError:
+            log(f"Invalid drop rate value: {value}. Must be a number.")
+
+
+#THREAD 1
+def worker_1():
+    global TOTAL_SENT_BYTES, TOTAL_RECEIVED_BYTES
+    has_joined = (INTRODUCER_IP == MY_IP)
+    last_join_attempt = 0.0
+
+    while not stop_threads.is_set():
+
+        #we keep retrying until we have joined the cluster
+        if not has_joined and (utc_time() - last_join_attempt) >= JOIN_RETRY_INTERVAL:
+            joining_message_payload = {"type": "JOIN", "sender_id": MY_ID}
+            joining_message_json = json.dumps(joining_message_payload)
+
+            try:
+                joining_message_bytes = joining_message_json.encode('utf-8')
+                server_socket.sendto(joining_message_bytes, (INTRODUCER_IP, GLOBAL_PORT))
+                with traffic_lock:
+                    TOTAL_SENT_BYTES += len(joining_message_bytes)
+            except Exception as e:
+                log(f"JOIN send failed: {e}")
+            last_join_attempt = utc_time()
+
+        try:
+            data, addr = server_socket.recvfrom(65535)
+        except socket.timeout:
+            continue
+
+        with traffic_lock:
+            TOTAL_RECEIVED_BYTES += len(data)
+
+        try:
+            msg = json.loads(data.decode('utf-8'))
+        except Exception:
+            continue
+
+        msg_type = msg.get("type")
+
+        #handle dropping messages only on GOSSIP
+        if msg_type == "GOSSIP":
+            drop_rate = 0
+            with cluster_state_lock:
+                drop_rate = cluster_state["drop_rate"]
+            if drop_rate > 0.0 and (random.random() * 100) < drop_rate:
+                log(f"DROPPED GOSSIP message from {addr[0]}:{addr[1]}")
+                continue  # Skip processing this message
+
+        #handle different message types
+        if msg_type == "JOIN":
+            if INTRODUCER_IP == MY_IP:
+                handle_join(msg)
+
+        elif msg_type == "GOSSIP":
+            handle_gossip(msg)
+            if not has_joined:
+                has_joined = True
+                log(f"Successfully joined the cluster via {addr[0]}:{addr[1]}")
+
+        else:
+            log(f"Unknown message type received from {addr[0]}:{addr[1]}: {msg_type}")
+
+# THREAD 1 HELPERS
+#this fixes the deadlock where handle_join held members_list_lock then called current_membership()
+def _current_membership_locked():
+    result = []
+    for member_id, member_data in members_list.items():
+        result.append({
+            "ip": member_id[0],
+            "port": member_id[1],
+            "timestamp": member_id[2],
+            "heartbeat": member_data["heart_beat"],
+            "state": member_data["state"],
+            "incarnation": member_data["incarnation_counter"],
+        })
+    return result
+
+def current_membership():
+    with members_list_lock:
+        return _current_membership_locked()
+
+def current_cluster_state():
+    with cluster_state_lock:
+        return {
+            "mode": cluster_state["mode"],
+            "drop_rate": cluster_state["drop_rate"],
+            "version": list(cluster_state["version"]),
+        }
+
+def handle_join(msg):
+    global TOTAL_SENT_BYTES
+    sender_id = tuple(msg["sender_id"])
+    with members_list_lock:
+        #check if the sender is in the dead members list
+        if sender_id in dead_members:
+            return
+
+        if sender_id not in members_list:
+            members_list[sender_id] = {
+                "heart_beat": 0,
+                "state": "alive",
+                "incarnation_counter": 0,
+                "local_last_seen": utc_time(),
+                "local_suspected_since": None,
+                "local_failed_since": None,
+            }
+            log(f"Added new member: {sender_id[0]}:{sender_id[1]}")
+
+        members_snapshot = _current_membership_locked()
+
+    control_state = current_cluster_state()
+
+    #we have to send back a GOSSIP message with our current membership list and cluster state
+    ack_payload = {
+        "type": "GOSSIP",
+        "sender_id": MY_ID,
+        "members": members_snapshot,
+        "mode": control_state["mode"],
+        "cluster_version": control_state["version"],
+        "drop_rate": control_state["drop_rate"],
+    }
+
+    try:
+        ack_payload_bytes = json.dumps(ack_payload).encode('utf-8')
+        server_socket.sendto(ack_payload_bytes, (sender_id[0], sender_id[1]))
+        with traffic_lock:
+            TOTAL_SENT_BYTES += len(ack_payload_bytes)
+    except Exception as e:
+        log(f"Failed to send GOSSIP response to {sender_id[0]}:{sender_id[1]}: {e}")
+
+def handle_member_update(member):
+    # 4 cases:
+    #     1. if i am being suspected, clear our name
+    #     2. new member, add to membership list
+    #     3. [hard] incarnation update
+    #     4. increase heartbeat, update last seen
+
+    member_id = (member["ip"], member["port"], member["timestamp"])
+
+    if member_id in dead_members:
+        return
+
+    recieved_incarnation = member["incarnation"]
+    recieved_heartbeat = member["heartbeat"]
+    recieved_state = member["state"]
+
+    #1
+    if member_id == MY_ID and recieved_state == "suspect" and recieved_incarnation >= members_list[MY_ID]["incarnation_counter"]:
+        log(f"Clearing my own name from suspect state due to gossip from {member_id[0]}:{member_id[1]}")
+        members_list[MY_ID]["state"] = "alive"
+        members_list[MY_ID]["incarnation_counter"] = recieved_incarnation + 1
+        members_list[MY_ID]["local_last_seen"] = utc_time()
+        return
+
+    #2
+    local_member_data = members_list.get(member_id)
+    if local_member_data is None:
+        if recieved_state == "failed":
+            dead_members.add(member_id)
+            log(f"Received failed state for unknown member {member_id[0]}:{member_id[1]}, ignoring")
+            return
+        members_list[member_id] = {
+            "heart_beat": recieved_heartbeat,
+            "state": recieved_state,
+            "incarnation_counter": recieved_incarnation,
+            "local_last_seen": utc_time(),
+            "local_suspected_since": utc_time() if recieved_state == "suspect" else None,
+            "local_failed_since": None,
+        }
+
+        if recieved_state == "suspect":
+            with suspect_history_lock:
+                suspect_history.append((member_id, utc_time()))
+            log(f"Added new member: {member_id[0]}:{member_id[1]} with state suspect")
+        else:
+            log(f"Added new member: {member_id[0]}:{member_id[1]}")
+        return
+
+    if local_member_data["state"] == "failed":
+        return #waiting for cleanup, ignore any updates
+
+    #3
+    if recieved_incarnation > local_member_data["incarnation_counter"]:
+        local_member_data["incarnation_counter"] = recieved_incarnation
+
+        if recieved_state == "alive":
+            if local_member_data["state"] != "alive":
+                log(f"Member {member_id[0]}:{member_id[1]} is now alive (incarnation update)")
+            local_member_data["state"] = "alive"
+            local_member_data["local_last_seen"] = utc_time()
+            local_member_data["local_suspected_since"] = None
+            local_member_data["local_failed_since"] = None
+
+        elif recieved_state == "suspect":
+            if local_member_data["state"] != "suspect":
+                with suspect_history_lock:
+                    suspect_history.append((member_id, utc_time()))
+                log(f"Member {member_id[0]}:{member_id[1]} is now suspect (incarnation update)")
+            local_member_data["state"] = "suspect"
+            local_member_data["local_suspected_since"] = utc_time()
+
+        elif recieved_state == "failed":
+            if local_member_data["state"] != "failed":
+                log(f"Member {member_id[0]}:{member_id[1]} is now failed (incarnation update)")
+            local_member_data["state"] = "failed"
+            local_member_data["local_failed_since"] = utc_time()
+
+    #This is needed to handle the case where two nodes have the same incarnation number but different states. We need to resolve this deterministically.
+    # We will use the following rules:
+    # - If the received state is "suspect" and the local state is "alive", we will update the local state to "suspect".
+    # - If the received state is "failed" and the local state is "alive" or "suspect", we will update the local state to "failed".
+    elif recieved_incarnation == local_member_data["incarnation_counter"]:
+        if recieved_state == "suspect" and local_member_data["state"] == "alive":
+            local_member_data["state"] = "suspect"
+            local_member_data["local_suspected_since"] = utc_time()
+            with suspect_history_lock:
+                suspect_history.append((member_id, utc_time()))
+            log(f"Member {member_id[0]}:{member_id[1]} is now suspect (equal-incarnation update)")
+        elif recieved_state == "failed" and local_member_data["state"] in ["alive", "suspect"]:
+            local_member_data["state"] = "failed"
+            local_member_data["local_failed_since"] = utc_time()
+            log(f"Member {member_id[0]}:{member_id[1]} is now failed (equal-incarnation update)")
+            local_member_data["local_suspected_since"] = None
+
+    #4
+    if recieved_heartbeat > local_member_data["heart_beat"]:
+        local_member_data["heart_beat"] = recieved_heartbeat
+        local_member_data["local_last_seen"] = utc_time()
+
+#we have to check the currect control state version and update our local state if the received version is newer
+def handle_gossip(msg):
+    incoming_version = None
+    if "cluster_version" in msg:
+        try:
+            incoming_version = tuple(msg["cluster_version"])
+        except Exception:
+            incoming_version = None
+
+    if incoming_version is not None:
+        with cluster_state_lock:
+            if incoming_version > cluster_state["version"]:
+                cluster_state["mode"] = msg.get("mode", cluster_state["mode"])
+                cluster_state["drop_rate"] = msg.get("drop_rate", cluster_state["drop_rate"])
+                cluster_state["version"] = incoming_version
+                #also have to send the suspect thread event
+                if cluster_state["mode"] == "suspect":
+                    suspect_mode.set()
+                else:
+                    suspect_mode.clear()
+                    #when switching to nosuspect, we have to clear any suspect states in our membership list
+                    with members_list_lock:
+                        for m_id, m_data in members_list.items():
+                            if m_id != MY_ID and m_data["state"] == "suspect":
+                                m_data["state"] = "alive"
+                                m_data["local_suspected_since"] = None
+                                log(f"Resolved {m_id[0]}:{m_id[1]} suspect -> alive on switch to nosuspect")
+                log(f"Updated cluster state to version {incoming_version}: mode={cluster_state['mode']}, drop_rate={cluster_state['drop_rate']}")
+
+    members_json = msg.get("members", [])
+    with members_list_lock:
+        for member in members_json:
+            handle_member_update(member)
+
+#THREAD 2 - update my heartbeat
+def worker_2():
+    while not stop_threads.is_set():
+        with members_list_lock:
+            members_list[MY_ID]["heart_beat"] += 1
+        time.sleep(HEARTBEAT_INTERVAL)
+
+#THREAD 3 - send gossip messages to random members
+#threshold, pure random can starve a peer for a few rounds and cause false positives.
+_peer_queue = []
+_peer_queue_lock = threading.Lock()
+
+def worker_3():
+    global TOTAL_SENT_BYTES
+    global _peer_queue
+    while not stop_threads.is_set():
+        with members_list_lock:
+            current_peers = [member_id for member_id in members_list if member_id != MY_ID and members_list[member_id]["state"] in ["alive", "suspect"]]
+
+        gossip_members_payload = current_membership()
+        gossip_control_payload = current_cluster_state()
+
+        #shuffled round-robin target selection
+        with _peer_queue_lock:
+            _peer_queue = [p for p in _peer_queue if p in current_peers]
+            selected_peers = []
+            for _ in range(min(GOSSIP_FANOUT, len(current_peers))):
+                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:
+            gossip_payload = {
+                "type": "GOSSIP",
+                "sender_id": MY_ID,
+                "members": gossip_members_payload,
+                "mode": gossip_control_payload["mode"],
+                "cluster_version": gossip_control_payload["version"],
+                "drop_rate": gossip_control_payload["drop_rate"],
+            }
+
+            try:
+                gossip_payload_bytes = json.dumps(gossip_payload).encode('utf-8')
+                server_socket.sendto(gossip_payload_bytes, (peer[0], peer[1]))
+                with traffic_lock:
+                    TOTAL_SENT_BYTES += len(gossip_payload_bytes)
+            except Exception as e:
+                log(f"Failed to send GOSSIP to {peer[0]}:{peer[1]}: {e}")
+
+        time.sleep(GOSSIP_INTERVAL)
+
+# THREAD 4 FAILURE DETECTION
+def worker_4():
+    while not stop_threads.is_set():
+        current_time = utc_time()
+        members_to_remove = []
+
+        with members_list_lock:
+            for member_id, member_data in members_list.items():
+                if member_id == MY_ID:
+                    continue  # Skip self
+
+                #gossip
+                if not suspect_mode.is_set():
+                    if member_data["state"] == "alive" and (current_time - member_data["local_last_seen"]) > GOSSIP_FAILURE_TIMEOUT:
+                        member_data["state"] = "failed"
+                        member_data["local_failed_since"] = current_time
+                        log(f"Member {member_id[0]}:{member_id[1]} is now failed (gossip timeout)")
+
+                #gossip + Suspect
+                else:
+                    if member_data["state"] == "alive" and (current_time - member_data["local_last_seen"]) > GOSSIP_S_SUSPECT_TIMEOUT:
+                        member_data["state"] = "suspect"
+                        member_data["local_suspected_since"] = current_time
+                        with suspect_history_lock:
+                            suspect_history.append((member_id, current_time))
+                        log(f"Member {member_id[0]}:{member_id[1]} is now suspect (gossip + suspect timeout)")
+
+                    elif member_data["state"] == "suspect" and (current_time - member_data["local_suspected_since"]) > GOSSIP_S_TIMEOUT_SUSPECT:
+                        member_data["state"] = "failed"
+                        member_data["local_failed_since"] = current_time
+                        member_data["local_suspected_since"] = None
+                        log(f"Member {member_id[0]}:{member_id[1]} is now failed (suspect timeout)")
+
+                #clean up for both modes
+                if member_data["state"] == "failed" and (current_time - member_data["local_failed_since"]) > GOSSIP_FAILURE_TIMEOUT_CLEANUP:
+                    members_to_remove.append(member_id)
+
+            for member_id in members_to_remove:
+                del members_list[member_id]
+                dead_members.add(member_id)
+                log(f"Removed member {member_id[0]}:{member_id[1]}")
+
+        time.sleep(0.1)
+
+# THREAD 5 Handle stdin input for control commands
+def worker_5():
+    while not stop_threads.is_set():
+        line = sys.stdin.readline()
+
+        if not line:
+            time.sleep(0.1)
+            continue
+
+        cmd = line.strip().split()
+        if not cmd:
+            continue
+        cmd_0 = cmd[0].lower()
+
+        if cmd_0 == "list_mem":
+            cmd_list_mem()
+        elif cmd_0 == "list_self":
+            log(str(MY_ID))
+        elif cmd_0 == "join":
+            log("auto joins at start up")
+        elif cmd_0 == "leave":
+            log("Leaving")
+            stop_threads.set()
+        elif cmd_0 == "display_suspects":
+            cmd_display_suspects()
+        elif cmd_0 == "switch":
+            if len(cmd) >= 2:
+                global_switcher("switch", cmd[1])
+            else:
+                log("Usage: switch suspect|nosuspect")
+        elif cmd_0 == "display_protocol":
+            with cluster_state_lock:
+                log(f"Mode: {cluster_state['mode']}, drop rate: {cluster_state['drop_rate']}")
+        elif cmd_0 == "set_drop_rate":
+            if len(cmd) >= 2:
+                global_switcher("drop_rate", cmd[1])
+            else:
+                log("Usage: set_drop_rate <number>")
+        else:
+            log(f"Unknown command: {cmd_0}")
+
+#THREAD 5 HELPERS
+def cmd_list_mem():
+    with members_list_lock:
+        cluster_state = [
+            (
+                m_id,
+                m_data["state"],
+                m_data["heart_beat"],
+                m_data["incarnation_counter"],
+            )
+            for m_id, m_data in members_list.items()
+            if m_data["state"] != "failed"
+        ]
+
+    for m_id, state, hb, inc in cluster_state:
+        log(f"{m_id} state = {state}, heartbeat = {hb}, incarnation = {inc}")
+
+def cmd_display_suspects():
+    with suspect_history_lock:
+        history_snapshot = list(suspect_history)
+
+    if not history_snapshot:
+        log("No suspects recorded.")
+    else:
+        log("Suspect history:")
+        for member_id, timestamp in history_snapshot:
+            log(f"{member_id} suspected at {timestamp}")
+
+
+
+# THREAD 6 - Display membership list in a pretty format
+def worker_monitor():
+    while not stop_threads.is_set():
+
+        with members_list_lock:
+            snapshot = {
+                member_id: member_data.copy()
+                for member_id, member_data in members_list.items()
+            }
+
+        with cluster_state_lock:
+            mode = cluster_state["mode"]
+            drop_rate = cluster_state["drop_rate"]
+
+        with traffic_lock:
+            sent_bytes = TOTAL_SENT_BYTES
+            received_bytes = TOTAL_RECEIVED_BYTES
+
+        try:
+            with open("membership_list.txt", "w") as f:
+                for member_id, member_data in snapshot.items():
+                    f.write(f"{member_id}: {member_data}\n")
+
+                f.write(
+                    f"Mode: {mode}, "
+                        f"Drop Rate: {drop_rate}, "
+                        f"Sent Bytes: {sent_bytes}, "
+                        f"Received Bytes: {received_bytes}\n"
+                )
+
+        except Exception as e:
+            pass
+        time.sleep(0.25)
+
+
+if __name__ == "__main__":
+    t1 = threading.Thread(target=worker_1, daemon=True)
+    t2 = threading.Thread(target=worker_2, daemon=True)
+    t3 = threading.Thread(target=worker_3, daemon=True)
+    t4 = threading.Thread(target=worker_4, daemon=True)
+    t5 = threading.Thread(target=worker_5, daemon=True)
+    t6 = threading.Thread(target=worker_monitor, daemon=True)
+
+    t1.start()
+    t2.start()
+    t3.start()
+    t4.start()
+    t5.start()
+    t6.start()
+
+    try:
+        while not stop_threads.is_set():
+            time.sleep(0.1)
+    except KeyboardInterrupt:
+        log("Keyboard interrupt, shutting down...")
+
+    log("Shutting down...")
+    stop_threads.set()
+    t1.join(timeout=1.0)
+    t2.join(timeout=1.0)
+    t3.join(timeout=1.0)
+    t4.join(timeout=1.0)
+    t6.join(timeout=1.0)
+    log("Shutdown complete.")
+    server_socket.close()

MP2/scripts/cmd.sh

@@ -0,0 +1,405 @@
+#!/usr/bin/env bash
+
+SESSION="cs425-debug"
+CMDS_WINDOW="cmds"
+
+NUM_SERVERS=10
+
+PORT=8989
+
+WORKDIR="fa26_cs425_.group-mp_.team-01/MP2"
+SCRIPT="peer.py"
+
+SSH_OPTS="-o BatchMode=yes \
+-o ConnectTimeout=2 \
+-o ServerAliveInterval=1 \
+-o ServerAliveCountMax=1"
+
+
+# HELPERS
+
+
+expand_targets() {
+
+    if [ "$1" = "all" ]; then
+        seq 1 "$NUM_SERVERS"
+    else
+        echo "$@"
+    fi
+}
+
+
+get_pane() {
+
+    local vm="$1"
+
+    # VM1  -> pane 0
+    # VM2  -> pane 1
+
+    echo "$SESSION:$CMDS_WINDOW.$((vm - 1))"
+}
+
+
+is_running() {
+
+    local vm="$1"
+
+    ssh $SSH_OPTS "server$vm" \
+        "pgrep -f '[p]ython3.*$SCRIPT' >/dev/null" \
+        2>/dev/null
+}
+
+
+
+# START
+
+
+start_vm() {
+
+    local vm="$1"
+    local pane
+
+    pane=$(get_pane "$vm")
+
+    if ! tmux has-session \
+        -t "$SESSION" \
+        2>/dev/null
+    then
+        echo "VM$vm: tmux session not running"
+        return 1
+    fi
+
+    echo "starting VM$vm..."
+
+    # Kill old process on remote VM
+
+    ssh $SSH_OPTS "server$vm" \
+        "fuser -k -9 $PORT/udp 2>/dev/null || true; \
+         fuser -k -9 $PORT/tcp 2>/dev/null || true; \
+         pkill -9 -f '[p]ython3.*$SCRIPT' 2>/dev/null || true" \
+        >/dev/null 2>&1
+
+    sleep 0.3
+
+    # Reset pane
+
+    tmux send-keys \
+        -t "$pane" \
+        C-c
+
+    sleep 0.05
+
+    tmux send-keys \
+        -t "$pane" \
+        "clear" \
+        C-m
+
+    # Start peer.py
+
+    tmux send-keys \
+        -t "$pane" \
+        "ssh -t $SSH_OPTS server$vm 'cd \$HOME/$WORKDIR && MACHINE_INDEX=$vm python3 $SCRIPT'" \
+        C-m
+}
+
+
+
+# KILL
+
+
+kill_vm() {
+
+    local vm="$1"
+
+    echo "killing VM$vm..."
+
+    ssh $SSH_OPTS "server$vm" \
+        "fuser -k -9 $PORT/udp 2>/dev/null || true; \
+         fuser -k -9 $PORT/tcp 2>/dev/null || true; \
+         pkill -9 -f '[p]ython3.*$SCRIPT' 2>/dev/null || true" \
+        >/dev/null 2>&1
+}
+
+
+
+# COMMAND
+
+
+cmd_to_vm() {
+
+    local vm="$1"
+    shift
+
+    local command="$*"
+    local pane
+    local pane_command
+
+    pane=$(get_pane "$vm")
+
+    if ! is_running "$vm"; then
+        echo "VM$vm skipped (STOPPED)"
+        return
+    fi
+
+    pane_command=$(
+        tmux display-message \
+            -p \
+            -t "$pane" \
+            '#{pane_current_command}' \
+            2>/dev/null
+    )
+
+    if [ "$pane_command" != "ssh" ]; then
+        echo "VM$vm skipped (pane not attached)"
+        return
+    fi
+
+    echo "VM$vm <- $command"
+
+    tmux send-keys \
+        -t "$pane" \
+        "$command" \
+        C-m
+}
+
+
+
+# LIST
+
+
+list_vm() {
+
+    local vm="$1"
+
+    echo
+    echo "========== VM$vm =========="
+
+    ssh $SSH_OPTS "server$vm" \
+        "cat \$HOME/$WORKDIR/membership_list.txt" \
+        2>/dev/null
+}
+
+
+
+# LOG
+
+
+log_vm() {
+
+    local vm="$1"
+
+    echo
+    echo "========== VM$vm =========="
+
+    ssh $SSH_OPTS "server$vm" \
+        "tail -n 30 \$HOME/$WORKDIR/machine.$vm.log" \
+        2>/dev/null
+}
+
+
+
+# STATUS
+
+
+status_vm() {
+
+    local vm="$1"
+    local pid
+
+    pid=$(
+        ssh $SSH_OPTS "server$vm" \
+            "pgrep -f '[p]ython3.*$SCRIPT' | tr '\n' ' '" \
+            2>/dev/null
+    )
+
+    if [ -n "$pid" ]; then
+        echo "VM$vm RUNNING, PID(s): $pid"
+    else
+        echo "VM$vm STOPPED"
+    fi
+}
+
+
+
+# MAIN
+
+
+action="$1"
+shift || true
+
+case "$action" in
+
+
+# START
+
+
+start)
+
+    if [ "$#" -eq 0 ]; then
+        echo "Usage: ./cmd.sh start <all|N...>"
+        exit 1
+    fi
+
+    targets=$(expand_targets "$@")
+
+    for vm in $targets; do
+        start_vm "$vm" &
+    done
+
+    wait
+
+    sleep 0.5
+
+    for vm in $targets; do
+        status_vm "$vm" &
+    done
+
+    wait
+    ;;
+
+
+
+# KILL
+
+
+kill)
+
+    if [ "$#" -eq 0 ]; then
+        echo "Usage: ./cmd.sh kill <all|N...>"
+        exit 1
+    fi
+
+    targets=$(expand_targets "$@")
+
+    for vm in $targets; do
+        kill_vm "$vm" &
+    done
+
+    wait
+    ;;
+
+
+
+# CMD
+
+
+cmd)
+
+    target="$1"
+    shift || true
+
+    if [ -z "$target" ] || [ "$#" -eq 0 ]; then
+        echo "Usage:"
+        echo "./cmd.sh cmd <all|N> \"command\""
+        exit 1
+    fi
+
+    if [ "$target" = "all" ]; then
+
+        for vm in $(seq 1 "$NUM_SERVERS"); do
+            cmd_to_vm "$vm" "$@" &
+        done
+
+        wait
+
+    else
+
+        cmd_to_vm "$target" "$@"
+
+    fi
+    ;;
+
+
+
+# LIST
+
+
+list)
+
+    if [ "$#" -eq 0 ]; then
+        targets=$(seq 1 "$NUM_SERVERS")
+    else
+        targets=$(expand_targets "$@")
+    fi
+
+    for vm in $targets; do
+        list_vm "$vm" &
+    done
+
+    wait
+    ;;
+
+
+
+# LOG
+
+
+log)
+
+    if [ "$#" -eq 0 ]; then
+        targets=$(seq 1 "$NUM_SERVERS")
+    else
+        targets=$(expand_targets "$@")
+    fi
+
+    for vm in $targets; do
+        log_vm "$vm" &
+    done
+
+    wait
+    ;;
+
+
+
+# STATUS
+
+
+status)
+
+    if [ "$#" -eq 0 ]; then
+        targets=$(seq 1 "$NUM_SERVERS")
+    else
+        targets=$(expand_targets "$@")
+    fi
+
+    for vm in $targets; do
+        status_vm "$vm" &
+    done
+
+    wait
+    ;;
+
+
+
+# HELP
+
+
+*)
+
+    echo
+    echo "Usage:"
+    echo
+    echo "  ./cmd.sh start  <all|N...>"
+    echo "  ./cmd.sh kill   <all|N...>"
+    echo "  ./cmd.sh cmd    <all|N> \"command\""
+    echo "  ./cmd.sh list   [all|N...]"
+    echo "  ./cmd.sh log    [all|N...]"
+    echo "  ./cmd.sh status [all|N...]"
+    echo
+    echo "Examples:"
+    echo
+    echo "  ./cmd.sh start 1"
+    echo "  ./cmd.sh start 2 3 4 5"
+    echo "  ./cmd.sh start 6 7 8 9 10"
+    echo
+    echo "  ./cmd.sh cmd all \"switch suspect\""
+    echo "  ./cmd.sh cmd all \"switch nosuspect\""
+    echo "  ./cmd.sh cmd all \"set_drop_rate 10\""
+    echo "  ./cmd.sh cmd all \"display_protocol\""
+    echo
+    echo "  ./cmd.sh kill 8 9"
+    echo "  ./cmd.sh log 8 9"
+    echo
+    exit 1
+    ;;
+esac

MP2/scripts/pretty_membership.py

@@ -0,0 +1,264 @@
+#!/usr/bin/env python3
+
+import sys
+import ast
+import time
+
+
+IP_TO_VM = {
+    "172.22.156.3": "1",
+    "172.22.158.2": "2",
+    "172.22.94.2":  "3",
+    "172.22.156.4": "4",
+    "172.22.158.3": "5",
+    "172.22.94.3":  "6",
+    "172.22.156.5": "7",
+    "172.22.158.4": "8",
+    "172.22.94.4":  "9",
+    "172.22.156.6": "10",
+}
+
+
+
+# COLORS
+
+
+GREEN  = "\033[92m"
+YELLOW = "\033[93m"
+RED    = "\033[91m"
+CYAN   = "\033[96m"
+WHITE  = "\033[97m"
+GRAY   = "\033[90m"
+
+BOLD  = "\033[1m"
+RESET = "\033[0m"
+
+CLEAR_LINE = "\033[K"
+
+
+STATE_STYLE = {
+    "ALIVE":   (GREEN, "A"),
+    "SUSPECT": (YELLOW, "S"),
+    "FAILED":  (RED, "F"),
+}
+
+
+def render(lines):
+
+   
+    # STOPPED / SSH ERROR
+   
+
+    for line in lines:
+        if line.startswith("**STOPPED**"):
+
+            reason = line.replace(
+                "**STOPPED**",
+                ""
+            ).strip()
+
+            return (
+                f"{CLEAR_LINE}"
+                f"{BOLD}{RED}>>> STOPPED <<<{RESET}"
+                f"{CLEAR_LINE}\n"
+                f"{GRAY}{reason}{RESET}"
+                f"{CLEAR_LINE}\n"
+            )
+
+   
+    # PARSE
+   
+
+    members = []
+
+    mode = "?"
+    drop_rate = "?"
+    sent_bytes = "?"
+    received_bytes = "?"
+
+    now = time.time()
+
+    for raw_line in lines:
+
+        line = raw_line.strip()
+
+        if not line:
+            continue
+
+        # ----------------------------------------------------
+        # MODE / DROP RATE LINE
+        # ----------------------------------------------------
+
+        if line.startswith("Mode:"):
+
+            try:
+                pieces = line.split(",")
+
+                mode = pieces[0].split(":", 1)[1].strip()
+                drop_rate = pieces[1].split(":", 1)[1].strip()
+                if len(pieces) > 2:
+                    sent_bytes = pieces[2].split(":", 1)[1].strip()
+                if len(pieces) > 3:
+                    received_bytes = pieces[3].split(":", 1)[1].strip()
+
+            except Exception:
+                pass
+
+            continue
+
+        # ----------------------------------------------------
+        # MEMBERSHIP ENTRY
+        # ----------------------------------------------------
+
+        try:
+
+            id_text, data_text = line.split(": ", 1)
+
+            member_id = ast.literal_eval(id_text)
+            member_data = ast.literal_eval(data_text)
+
+            ip, port, timestamp = member_id
+
+            local_last_seen = member_data.get(
+                "local_last_seen",
+                now
+            )
+
+            members.append({
+                "vm": IP_TO_VM.get(ip, "?"),
+
+                "hb": member_data.get(
+                    "heart_beat",
+                    0
+                ),
+
+                "state": member_data.get(
+                    "state",
+                    "?"
+                ).upper(),
+
+                "age": max(
+                    0.0,
+                    now - local_last_seen
+                ),
+
+                "inc": member_data.get(
+                    "incarnation_counter",
+                    0
+                ),
+            })
+
+        except Exception:
+            continue
+
+   
+    # HEADER
+   
+
+    out = []
+
+    mode_upper = str(mode).upper()
+
+    if mode_upper == "SUSPECT":
+        mode_color = YELLOW
+    elif mode_upper == "NOSUSPECT":
+        mode_color = GREEN
+    else:
+        mode_color = GRAY
+
+    out.append(
+        f"{BOLD}{CYAN}MODE:{RESET} "
+        f"{mode_color}{BOLD}{mode_upper}{RESET} "
+        f"{GRAY}|{RESET} "
+        f"{BOLD}{CYAN}DROP:{RESET} "
+        f"{WHITE}{drop_rate}%{RESET}"
+        f" {GRAY}|{RESET} "
+        f"{BOLD}{CYAN}TOTAL TX:{RESET} {WHITE}{sent_bytes} B{RESET}"
+        f" {GRAY}|{RESET} "
+        f"{BOLD}{CYAN}TOTAL RX:{RESET} {WHITE}{received_bytes} B{RESET}"
+        f"{CLEAR_LINE}"
+    )
+
+   
+    # NO DATA
+   
+
+    if not members:
+
+        out.append(
+            f"{BOLD}{YELLOW}"
+            f"waiting for membership..."
+            f"{RESET}"
+            f"{CLEAR_LINE}"
+        )
+
+        return "\n".join(out) + "\n"
+
+   
+    # SORT
+   
+
+    def vm_number(member):
+
+        try:
+            return int(member["vm"])
+
+        except Exception:
+            return 999
+
+    members.sort(
+        key=vm_number
+    )
+
+   
+    # TABLE HEADER
+   
+
+    out.append(
+        f"{BOLD}{CYAN}"
+        f"{'VM':<3} "
+        f"{'S':<2} "
+        f"{'HB':<6} "
+        f"{'AGE':<5} "
+        f"{'INC'}"
+        f"{RESET}"
+        f"{CLEAR_LINE}"
+    )
+
+   
+    # MEMBERS
+   
+
+    for member in members:
+
+        color, code = STATE_STYLE.get(
+            member["state"],
+            (GRAY, "?")
+        )
+
+        out.append(
+            f"{WHITE}"
+            f"{member['vm']:<3}"
+            f"{RESET} "
+
+            f"{color}{BOLD}"
+            f"{code:<2}"
+            f"{RESET} "
+
+            f"{member['hb']:<6} "
+
+            f"{member['age']:>4.1f} "
+
+            f"{member['inc']}"
+
+            f"{CLEAR_LINE}"
+        )
+
+    return "\n".join(out) + "\n"
+
+
+print(
+    render(list(sys.stdin)),
+    end="",
+    flush=True
+)

MP2/scripts/view.sh

@@ -0,0 +1,147 @@
+#!/usr/bin/env bash
+
+SESSION="cs425-debug"
+NUM_SERVERS=10
+
+WORKDIR="fa26_cs425_.group-mp_.team-01/MP2"
+FORMATTER="$(cd "$(dirname "$0")" && pwd)/pretty_membership.py"
+
+SSH_OPTS="-o ConnectTimeout=2 \
+-o ServerAliveInterval=1 \
+-o ServerAliveCountMax=1 \
+-o BatchMode=yes"
+
+
+# RESET TMUX
+
+
+tmux kill-session -t "$SESSION" 2>/dev/null || true
+
+tmux new-session -d -s "$SESSION" -n "display"
+
+tmux set-option -t "$SESSION" pane-border-status top
+tmux set-option -t "$SESSION" pane-border-format " #{pane_title} "
+
+
+# DISPLAY COMMAND
+
+
+make_display_cmd() {
+    local vm="$1"
+
+    cat <<EOF
+clear; while true; do
+    printf '\033[H'
+
+    {
+        ssh $SSH_OPTS server$vm \
+        'FILE="\$HOME/$WORKDIR/membership_list.txt";
+         if [ ! -f "\$FILE" ]; then
+             echo "**STOPPED** no membership file";
+         else
+             NOW=\$(date +%s);
+             MTIME=\$(stat -c %Y "\$FILE" 2>/dev/null || echo 0);
+             AGE=\$((NOW - MTIME));
+
+             if [ "\$AGE" -gt 3 ]; then
+                 echo "**STOPPED** \${AGE}s since last update";
+             else
+                 cat "\$FILE";
+             fi;
+         fi' 2>&1 \
+        || echo "**STOPPED** SSH unavailable"
+    } | python3 "$FORMATTER"
+
+    printf '\033[J'
+    sleep 0.75
+done
+EOF
+}
+
+
+# DISPLAY WINDOW
+
+
+tmux select-pane \
+    -t "$SESSION:display.0" \
+    -T "VM1"
+
+tmux send-keys \
+    -t "$SESSION:display.0" \
+    "$(make_display_cmd 1)" \
+    C-m
+
+for i in $(seq 2 "$NUM_SERVERS"); do
+
+    PANE=$(
+        tmux split-window \
+            -t "$SESSION:display" \
+            -P \
+            -F '#{pane_id}'
+    )
+
+    tmux select-layout \
+        -t "$SESSION:display" \
+        tiled
+
+    tmux select-pane \
+        -t "$PANE" \
+        -T "VM$i"
+
+    tmux send-keys \
+        -t "$PANE" \
+        "$(make_display_cmd "$i")" \
+        C-m
+done
+
+tmux select-layout \
+    -t "$SESSION:display" \
+    tiled
+
+
+# COMMAND WINDOW
+
+
+tmux new-window \
+    -t "$SESSION" \
+    -n "cmds"
+
+tmux set-window-option \
+    -t "$SESSION:cmds" \
+    pane-base-index 0
+
+tmux select-pane \
+    -t "$SESSION:cmds.0" \
+    -T "VM1"
+
+for i in $(seq 2 "$NUM_SERVERS"); do
+
+    PANE=$(
+        tmux split-window \
+            -t "$SESSION:cmds" \
+            -P \
+            -F '#{pane_id}'
+    )
+
+    tmux select-layout \
+        -t "$SESSION:cmds" \
+        tiled
+
+    tmux select-pane \
+        -t "$PANE" \
+        -T "VM$i"
+done
+
+tmux select-layout \
+    -t "$SESSION:cmds" \
+    tiled
+
+
+# OPEN DISPLAY
+
+
+tmux select-window \
+    -t "$SESSION:display"
+
+tmux attach-session \
+    -t "$SESSION"

MP3/consistency.go

@@ -0,0 +1 @@
+package main

MP3/eventual.go

@@ -0,0 +1 @@
+package main

MP3/go.mod

@@ -0,0 +1,3 @@
+module mp3/m
+
+go 1.27.2

MP3/main.go

@@ -0,0 +1,35 @@
+package main
+
+import (
+	// "fmt"
+	"net"
+	"os"
+	"time"
+	// "net"
+	"context"
+	"log/slog"
+)
+
+type Config struct{
+	introducer_ip string
+	port int
+	my_ip *net.IP
+	gossip_heartbeat_interval time.Duration
+	join_retry_interval time.Duration
+	gossip_fanout int
+	gossip_failure_timeout time.Duration
+}
+
+func main(){
+	ctx := context.TODO()
+	
+	logger := slog.New(slog.NewTextHandler(os.Stdout, nil))
+
+	//dial up trick to get my_ip
+	conn, _ := net.Dial("udp", "8.8.8.8:80"); defer conn.Close(); my_ip := conn.LocalAddr().(*net.UDPAddr).IP
+
+	config := Config{"172.22.156.3", 8989, &my_ip, 250 * time.Millisecond, 1 * time.Second, 2, 6 * time.Second}
+
+	start_membership(ctx, config, logger)
+
+}

MP3/membership.go

@@ -0,0 +1,634 @@
+
+package main
+
+import (
+	"bytes"
+	"context"
+	"encoding/gob"
+	"log/slog"
+	"math/rand"
+	"net"
+	"sync"
+	"time"
+)
+
+// this file handles gossip protocol from MP2
+// thread 1 - handles receiving gossip
+// thread 2 - handles sending gossips and updating heart beat
+// thread 3 - handles failure detection
+
+type Member_ID struct {
+	IP   string
+	Port int
+	Time time.Time
+}
+
+type MemberInfo struct {
+	HeartBeat          int
+	State              string
+	local_failed_since *time.Time
+	local_last_seen    *time.Time
+}
+
+type MembershipService struct {
+	MembersList  map[Member_ID]MemberInfo
+	SelfID       Member_ID
+	deadSessions map[Member_ID]struct{}
+	mutex        sync.RWMutex
+}
+
+type GossipMessage struct {
+	Type       string
+	Senders_id Member_ID
+	Members    map[Member_ID]MemberInfo
+}
+
+// One helper for gob encoding and UDP sending
+func send_gossip_message(
+	socket *net.UDPConn,
+	addr *net.UDPAddr,
+	gossip_msg GossipMessage,
+) error {
+	var buffer bytes.Buffer
+
+	if err := gob.NewEncoder(&buffer).Encode(gossip_msg); err != nil {
+		return err
+	}
+
+	_, err := socket.WriteToUDP(buffer.Bytes(), addr)
+	return err
+}
+
+func (m *MembershipService) get_current_members() map[Member_ID]MemberInfo {
+	m.mutex.RLock()
+	defer m.mutex.RUnlock()
+
+	// Create a copy of the members_list to avoid race conditions
+	current_members := make(map[Member_ID]MemberInfo)
+
+	for id, info := range m.MembersList {
+		current_members[id] = info
+	}
+
+	return current_members
+}
+
+func (m *MembershipService) get_self_id() Member_ID {
+	m.mutex.RLock()
+	defer m.mutex.RUnlock()
+
+	return m.SelfID
+}
+
+func get_all_member_ids(
+	members map[Member_ID]MemberInfo,
+	self Member_ID,
+) []Member_ID {
+	member_ids := make([]Member_ID, 0, len(members))
+
+	for id, _ := range members {
+		// Exclude self and DOWN nodes
+		if id != self {
+			member_ids = append(member_ids, id)
+		}
+	}
+
+	return member_ids
+}
+
+func (m *MembershipService) member_init(ip string, port int) {
+	now := time.Now()
+
+	// Round(0) removes monotonic clock data for gob map keys
+	my_id := Member_ID{ip, port, now.UTC().Round(0)}
+	members_info := MemberInfo{0, "alive", nil, &now}
+
+	m.mutex.Lock()
+	defer m.mutex.Unlock()
+
+	m.SelfID = my_id
+	m.MembersList[my_id] = members_info
+}
+
+// Add a new process session.
+// Caller must hold m.mutex.Lock().
+func (m *MembershipService) add_new_member(
+	id Member_ID,
+	info MemberInfo,
+	now time.Time,
+) bool {
+	// Never accept a session already declared dead
+	if _, dead := m.deadSessions[id]; dead {
+		return false
+	}
+
+	// Check whether we already know a newer session
+	for old_id := range m.MembersList {
+		if old_id.IP == id.IP && old_id.Port == id.Port {
+			if !id.Time.After(old_id.Time) {
+				return false
+			}
+		}
+	}
+
+	// New process lifetime: retire older sessions
+	for old_id := range m.MembersList {
+		if old_id.IP == id.IP && old_id.Port == id.Port {
+			m.deadSessions[old_id] = struct{}{}
+			delete(m.MembersList, old_id)
+		}
+	}
+
+	info.local_last_seen = &now
+	info.local_failed_since = nil
+
+	if info.State == "down" {
+		info.local_failed_since = &now
+		m.deadSessions[id] = struct{}{}
+	}
+
+	m.MembersList[id] = info
+	return true
+}
+
+func (m *MembershipService) handle_join(
+	new_member Member_ID,
+	logger *slog.Logger,
+	socket *net.UDPConn,
+) {
+	if net.ParseIP(new_member.IP) == nil {
+		logger.Warn("Invalid JOIN member", "member", new_member)
+		return
+	}
+
+	m.mutex.Lock()
+
+	// Add only if this is a new process session.
+	// Repeated JOIN does not reset heartbeat tracking.
+	now := time.Now()
+	m.add_new_member(
+		new_member,
+		MemberInfo{0, "alive", nil, nil},
+		now,
+	)
+
+	m.mutex.Unlock()
+
+	// ACK with current membership list
+	gossip_msg := GossipMessage{
+		Type:       "join_ack",
+		Senders_id: m.get_self_id(),
+		Members:    m.get_current_members(),
+	}
+
+	err := send_gossip_message(
+		socket,
+		&net.UDPAddr{
+			IP:   net.ParseIP(new_member.IP),
+			Port: new_member.Port,
+		},
+		gossip_msg,
+	)
+
+	if err != nil {
+		logger.Warn("Error sending JOIN acknowledgment", "error", err)
+		return
+	}
+
+	logger.Info("JOIN acknowledgment sent", "member", new_member)
+}
+
+func (m *MembershipService) handle_gossip_receive(
+	ctx context.Context,
+	socket *net.UDPConn,
+	logger *slog.Logger,
+	config Config,
+	wg *sync.WaitGroup,
+) {
+	defer wg.Done()
+
+	buffer := make([]byte, 65535)
+
+	// Introducer doesn't need to JOIN
+	is_introducer := config.introducer_ip == config.my_ip.String()
+	joined := is_introducer
+	last_join_time := time.Time{}
+
+	for {
+		if ctx.Err() != nil {
+			return
+		}
+
+		if !joined && time.Since(last_join_time) >= config.join_retry_interval {
+			// Send JOIN to introducer
+			join_msg := GossipMessage{
+				Type:       "join",
+				Senders_id: m.get_self_id(),
+				Members:    map[Member_ID]MemberInfo{},
+			}
+
+			introducer_addr := &net.UDPAddr{
+				IP:   net.ParseIP(config.introducer_ip),
+				Port: config.port,
+			}
+
+			if err := send_gossip_message(socket, introducer_addr, join_msg); err != nil {
+				logger.Warn("Error sending JOIN", "error", err)
+			}
+
+			last_join_time = time.Now()
+		}
+
+		socket.SetReadDeadline(time.Now().Add(500 * time.Millisecond))
+
+		n, _, err := socket.ReadFromUDP(buffer)
+
+		if err != nil {
+			if ctx.Err() != nil {
+				return
+			}
+
+			if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
+				continue
+			}
+
+			logger.Warn("Error reading UDP socket", "error", err)
+			continue
+		}
+
+		var gossip_msg GossipMessage
+
+		// Decode gob message
+		err = gob.NewDecoder(
+			bytes.NewReader(buffer[:n]),
+		).Decode(&gossip_msg)
+
+		if err != nil {
+			logger.Warn("Invalid gossip packet", "error", err)
+			continue
+		}
+
+		switch gossip_msg.Type {
+
+		case "gossip":
+			// Merge incoming membership records
+			// and rejoin if our own session was falsely marked DOWN
+			if m.handle_gossip(gossip_msg.Members, logger) && !is_introducer {
+				joined = false
+				last_join_time = time.Time{}
+			}
+
+		case "join":
+			if is_introducer {
+				m.handle_join(gossip_msg.Senders_id, logger, socket)
+			}
+
+		case "join_ack":
+			// Ignore delayed ACKs for an older local session
+			self_id := m.get_self_id()
+			if _, exists := gossip_msg.Members[self_id]; !exists {
+				continue
+			}
+
+			restarted := m.handle_gossip(gossip_msg.Members, logger)
+
+			if restarted {
+				joined = false
+				last_join_time = time.Time{}
+				continue
+			}
+
+			if !joined {
+				logger.Info("Successfully joined membership group")
+			}
+
+			joined = true
+
+		default:
+			logger.Warn("Unknown gossip message", "type", gossip_msg.Type)
+		}
+	}
+}
+
+func select_peers(
+	current_members map[Member_ID]MemberInfo,
+	fanout int,
+	self Member_ID,
+) []Member_ID {
+	member_ids := get_all_member_ids(current_members, self)
+
+	if fanout <= 0 {
+		return nil
+	}
+
+	if len(member_ids) <= fanout {
+		return member_ids
+	}
+
+	selected := make([]Member_ID, 0, fanout)
+	perm := rand.Perm(len(member_ids))
+
+	for i := 0; i < fanout; i++ {
+		selected = append(selected, member_ids[perm[i]])
+	}
+
+	return selected
+}
+
+func (m *MembershipService) handle_gossip_send(
+	ctx context.Context,
+	socket *net.UDPConn,
+	logger *slog.Logger,
+	config Config,
+	wg *sync.WaitGroup,
+) {
+	defer wg.Done()
+
+	ticker := time.NewTicker(config.gossip_heartbeat_interval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+
+		case <-ticker.C:
+			// Increment heartbeat for self
+			m.mutex.Lock()
+
+			self_id := m.SelfID
+			info := m.MembersList[self_id]
+
+			info.HeartBeat++
+			m.MembersList[self_id] = info
+
+			m.mutex.Unlock()
+
+			current_members := m.get_current_members()
+
+			selected_peers := select_peers(
+				current_members,
+				config.gossip_fanout,
+				m.get_self_id(),
+			)
+
+			gossip_msg := GossipMessage{
+				Type:       "gossip",
+				Senders_id: m.get_self_id(),
+				Members:    current_members,
+			}
+
+			for _, peer := range selected_peers {
+				peer_addr := &net.UDPAddr{
+					IP:   net.ParseIP(peer.IP),
+					Port: peer.Port,
+				}
+
+				if err := send_gossip_message(socket, peer_addr, gossip_msg); err != nil {
+					logger.Warn(
+						"Error sending gossip",
+						"peer", peer,
+						"error", err,
+					)
+					continue
+				}
+			}
+		}
+	}
+}
+
+// Merge incoming gossip into local membership.
+// Returns true if our own process session had to be replaced.
+func (m *MembershipService) handle_gossip(
+	incoming_members map[Member_ID]MemberInfo,
+	logger *slog.Logger,
+) bool {
+	m.mutex.Lock()
+	defer m.mutex.Unlock()
+
+	now := time.Now()
+	restarted := false
+
+	for id, incoming := range incoming_members {
+		if incoming.State != "alive" && incoming.State != "down" {
+			continue
+		}
+
+		// Handle information about our own machine
+		if id.IP == m.SelfID.IP && id.Port == m.SelfID.Port {
+
+			// Ignore reports about older local sessions
+			if id != m.SelfID {
+				continue
+			}
+
+			// Someone marked our current session DOWN.
+			// Create a NEW session timestamp to refute it.
+			if incoming.State == "down" {
+				old_id := m.SelfID
+				m.deadSessions[old_id] = struct{}{}
+				delete(m.MembersList, old_id)
+
+				new_time := now.UTC().Round(0)
+				if !new_time.After(old_id.Time) {
+					new_time = old_id.Time.Add(time.Nanosecond)
+				}
+
+				new_id := Member_ID{
+					IP:   old_id.IP,
+					Port: old_id.Port,
+					Time: new_time,
+				}
+
+				m.SelfID = new_id
+				m.MembersList[new_id] = MemberInfo{
+					HeartBeat:       0,
+					State:           "alive",
+					local_last_seen: &now,
+				}
+
+				restarted = true
+				logger.Info("Refuted DOWN report with new session")
+			}
+
+			continue
+		}
+
+		// Never resurrect an already dead process session
+		if _, dead := m.deadSessions[id]; dead {
+			continue
+		}
+
+		current, exists := m.MembersList[id]
+
+		// New session discovered through gossip
+		if !exists {
+			m.add_new_member(id, incoming, now)
+			continue
+		}
+
+		// Same session: DOWN cannot be overridden by ALIVE
+		if current.State == "down" {
+			continue
+		}
+
+		// Incoming DOWN report
+		if incoming.State == "down" {
+			if incoming.HeartBeat >= current.HeartBeat {
+				current.State = "down"
+				current.local_failed_since = &now
+
+				m.MembersList[id] = current
+				m.deadSessions[id] = struct{}{}
+
+				logger.Info("Member marked DOWN through gossip", "member", id)
+			}
+			continue
+		}
+
+		// Same session + ALIVE: accept newer heartbeat
+		if incoming.HeartBeat > current.HeartBeat {
+			current.HeartBeat = incoming.HeartBeat
+			current.local_last_seen = &now
+
+			m.MembersList[id] = current
+		}
+	}
+
+	return restarted
+}
+
+func (m *MembershipService) handle_failure_detection(
+	ctx context.Context,
+	socket *net.UDPConn,
+	logger *slog.Logger,
+	config Config,
+	wg *sync.WaitGroup,
+) {
+	defer wg.Done()
+
+	ticker := time.NewTicker(500 * time.Millisecond)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+
+		case <-ticker.C:
+			m.mutex.Lock()
+			now := time.Now()
+
+			for id, info := range m.MembersList {
+				// Don't check self or already DOWN nodes
+				if id == m.SelfID || info.State == "down" {
+					continue
+				}
+
+				if info.local_last_seen == nil {
+					info.local_last_seen = &now
+					m.MembersList[id] = info
+					continue
+				}
+
+				// No heartbeat advancement beyond timeout
+				if now.Sub(*info.local_last_seen) > config.gossip_failure_timeout {
+					info.State = "down"
+					info.local_failed_since = &now
+
+					m.MembersList[id] = info
+
+					// Remember this specific process session
+					m.deadSessions[id] = struct{}{}
+
+					logger.Warn("Member marked DOWN", "member", id)
+				}
+			}
+
+			m.mutex.Unlock()
+		}
+	}
+}
+
+func (m *MembershipService) print_membership(logger *slog.Logger) {
+	ticker := time.NewTicker(150 * time.Millisecond)
+	defer ticker.Stop()
+
+	for range ticker.C {
+		current_members := m.get_current_members()
+		logger.Info("Current Membership List:")
+		for id, info := range current_members {
+			logger.Info("Member", "id", id, "state", info.State)
+		}
+	}
+}
+
+func start_membership(
+	ctx context.Context,
+	config Config,
+	logger *slog.Logger,
+) {
+	logger.Info("Starting Membership")
+
+	if config.my_ip == nil ||
+		config.gossip_heartbeat_interval <= 0 ||
+		config.gossip_failure_timeout <= 0 ||
+		config.join_retry_interval <= 0 {
+		logger.Error("Invalid membership configuration")
+		return
+	}
+
+	var membership_service MembershipService
+
+	membership_service.MembersList = make(map[Member_ID]MemberInfo)
+	membership_service.deadSessions = make(map[Member_ID]struct{})
+
+	membership_service.member_init(
+		config.my_ip.String(),
+		config.port,
+	)
+
+	socket, err := net.ListenUDP("udp", &net.UDPAddr{
+		IP:   *config.my_ip,
+		Port: config.port,
+	})
+
+	if err != nil {
+		logger.Error("Failed to create UDP listener", "error", err)
+		return
+	}
+
+	defer socket.Close()
+
+	logger.Info(
+		"Membership Service Listening on",
+		"IP", config.my_ip.String(),
+		"Port", config.port,
+	)
+
+	var wg sync.WaitGroup
+	wg.Add(3)
+
+	go membership_service.handle_gossip_receive(
+		ctx, socket, logger, config, &wg,
+	)
+
+	go membership_service.handle_gossip_send(
+		ctx, socket, logger, config, &wg,
+	)
+
+	go membership_service.handle_failure_detection(
+		ctx, socket, logger, config, &wg,
+	)
+
+	// Wait for cancellation
+	<-ctx.Done()
+
+	// Close socket to unblock receiver
+	socket.Close()
+
+	// Wait for all three goroutines to finish
+	wg.Wait()
+
+	logger.Info("Membership Service Stopped")
+}

MP3/membership2.go

@@ -0,0 +1,9 @@
+package main
+
+import (
+	"fmt"
+)
+
+func main() {
+	
+}

MP3/replair.go

@@ -0,0 +1 @@
+package main

MP3/replication.go

@@ -0,0 +1 @@
+package main

MP3/storage.go

@@ -0,0 +1 @@
+package main

MP3/strong.go

@@ -0,0 +1 @@
+package main

MP3/utils.go

@@ -0,0 +1,2 @@
+package main
+