Fixed port

Removing custom proxy headers from tests

MP1/README.md

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

MP1/scripts/deploy_mp.sh

@@ -1,9 +1,9 @@
 #/usr/bin/env bash
 
-/Users/aryan/Documents/Github/fa26_cs425_.group-mp_.team-01/MP1/scripts/kill_mp.sh all
+~/fa26_cs425_.group-mp_.team-01/MP1/scripts/kill_mp.sh all
 
 for i in {1..10}; do
-    ssh -n -f -o BatchMode=yes "server$i" "cd fa26_cs425_.group-mp_.team-01/MP1 && git pull origin aryan && ~/fa26_cs425_.group-mp_.team-01/MP1/scripts/install_cmd.sh"
+    ssh -n -f -o BatchMode=yes "vm$i" "cd fa26_cs425_.group-mp_.team-01/MP1 && git switch justin && git pull origin justin && ~/fa26_cs425_.group-mp_.team-01/MP1/scripts/install_cmd.sh"
 done
 
-/Users/aryan/Documents/Github/fa26_cs425_.group-mp_.team-01/MP1/scripts/run_mp.sh all
+~/fa26_cs425_.group-mp_.team-01/MP1/scripts/run_mp.sh all

MP1/scripts/kill_mp.sh

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

MP1/scripts/run_mp.sh

@@ -1,6 +1,6 @@
 #!/usr/bin/env bash
 
-CMD='cd /home/asing271/fa26_cs425_.group-mp_.team-01/MP1 && nohup python3 server.py </dev/null >/dev/null 2>&1 &'
+CMD='cd 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."
@@ -16,18 +16,18 @@ fi
 echo "Starting servers"
 
 for i in $TARGETS; do
-    ssh -n -f -o BatchMode=yes "server$i" "$CMD"
+    ssh -n -f -o BatchMode=yes "vm$i" "$CMD"
 done
 
 sleep 2
 
 for i in $TARGETS; do
-    pid=$(ssh -n -o BatchMode=yes "server$i" "pgrep -o -u \$USER -f '[p]ython3 server.py'")
+    pid=$(ssh -n -o BatchMode=yes "vm$i" "pgrep -o -u \$USER -f '[p]ython3 server.py'")
     
     if [[ -n "$pid" ]]; then
         pid_clean=$(echo "$pid" | tr '\n' ' ')
-        echo "Server$i is running with PID(s): $pid_clean"
+        echo "vm$i is running with PID(s): $pid_clean"
     else
-        echo "Server$i is not running."
+        echo "vm$i is not running."
     fi
 done

MP1/server.py

@@ -5,7 +5,7 @@ import subprocess
 from pathlib import Path
 
 HOST = "0.0.0.0"
-PORT = 5056
+PORT = 5054
 
 server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 
@@ -44,23 +44,25 @@ while True:
     conn, addr = server_socket.accept()
     print(f"Connected by {addr}")
 
-    try:
+    command = receive_message(conn)
-        command = receive_message(conn)
+    if command is None:
-        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

@@ -290,19 +290,19 @@ Total matches: 100
 #generate log files on all the machines before running this test
 print("Generating log files on all machines...")
 for i, ip in enumerate(ALL_IPS):
-    subprocess.run(["ssh", f"asing271@{ip}", f"cd ~/fa26_cs425_.group-mp_.team-01/MP1 &&  python3 generate_log.py {i + 1} 100"])
+    subprocess.run(["ssh", f"vm{i + 1}", f"cd ~/fa26_cs425_.group-mp_.team-01/MP1 &&  python3 generate_log.py {i + 1} 100"])
 
 #start the server on all the machines before running this test
 print("Starting server on all machines...")
 for i, ip in enumerate(ALL_IPS):
     print(f"Starting server on machine {i + 1} ({ip})...")
-    subprocess.run(["ssh", f"asing271@{ip}", "cd ~/fa26_cs425_.group-mp_.team-01/MP1 && setsid -f python3 server.py > server.log 2>&1 < /dev/null"])
+    subprocess.run(["ssh", f"vm{i + 1}", "cd ~/fa26_cs425_.group-mp_.team-01/MP1 && setsid -f python3 server.py > server.log 2>&1 < /dev/null"])
 
 #ssh into one machine at randome
 machine_num = random.randint(1, len(ALL_IPS))
 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)
+    result = subprocess.run(["ssh", f"vm{machine_num}", 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(f"PASS {i} test")
     else:
@@ -311,4 +311,4 @@ for i in tests:
 #delete log files on all the machines after running this test
 print("Deleting log files on all machines...")
 for i, ip in enumerate(ALL_IPS):
-    subprocess.run(["ssh", f"asing271@{ip}", "cd ~/fa26_cs425_.group-mp_.team-01/MP1 &&  rm -f ./machine.*.test.log"])
+    subprocess.run(["ssh", f"vm{i + 1}", "cd ~/fa26_cs425_.group-mp_.team-01/MP1 &&  rm -f ./machine.*.test.log"])

MP2/README.md

@@ -1,63 +0,0 @@
-# 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

@@ -1,620 +0,0 @@
-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

@@ -1,405 +0,0 @@
-#!/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

@@ -1,264 +0,0 @@
-#!/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

@@ -1,147 +0,0 @@
-#!/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

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

MP3/eventual.go

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

MP3/go.mod

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

MP3/main.go

@@ -1,35 +0,0 @@
-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

@@ -1,634 +0,0 @@
-
-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

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

MP3/replair.go

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

MP3/replication.go

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

MP3/storage.go

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

MP3/strong.go

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

MP3/utils.go

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