Merge pull request #10 from illinois-cs-coursework/aryan

DONE
This commit is contained in:
Aryan Singh authored and GitHub committed 2026-09-27 22:41:47 -05:00 Unverified
commit 891bb79770
5 files changed
+878 -27

No files matched your search

-1
View File
@@ -1 +0,0 @@
scripts
+62 -26
View File
@@ -14,32 +14,35 @@ def _machine_index():
return int(idx) return int(idx)
return None return None
log_file = open(f"machine.{_machine_index()}.log", "a") log_file = open(f"machine.{_machine_index()}.log", "w")
log_lock = threading.Lock() log_lock = threading.Lock()
def log(message): def log(message):
timestamp = datetime.now().strftime("%H:%M:%S.%f")[:-3] timestamp = datetime.now().strftime("%H:%M:%S.%f")[:-3]
with log_lock: with log_lock:
print(f"[{timestamp}] {message}")
log_file.write(f"[{timestamp}] {message}\n") log_file.write(f"[{timestamp}] {message}\n")
log_file.flush() log_file.flush()
def utc_time(): def utc_time():
return datetime.now(timezone.utc).timestamp() return datetime.now(timezone.utc).timestamp()
TOTAL_SENT_BYTES = 0
TOTAL_RECEIVED_BYTES = 0
traffic_lock = threading.Lock()
#fixed #fixed
INTRODUCER_IP = "172.22.156.3" INTRODUCER_IP = "172.22.156.3"
GLOBAL_PORT = 8989 GLOBAL_PORT = 8989
MY_IP = socket.gethostbyname(socket.gethostname()) MY_IP = socket.gethostbyname(socket.gethostname())
HEARTBEAT_INTERVAL = 0.25 HEARTBEAT_INTERVAL = 0.25
GOSSIP_INTERVAL = 0.15 GOSSIP_INTERVAL = 0.25
JOIN_RETRY_INTERVAL = 1.0 JOIN_RETRY_INTERVAL = 1.0
GOSSIP_FANOUT = 1 GOSSIP_FANOUT = 2
GOSSIP_FAILURE_TIMEOUT = 2.8 GOSSIP_FAILURE_TIMEOUT = 3
GOSSIP_S_SUSPECT_TIMEOUT = 1 GOSSIP_S_SUSPECT_TIMEOUT = 1.0
GOSSIP_S_TIMEOUT_SUSPECT = 1.8 GOSSIP_S_TIMEOUT_SUSPECT = 2.0
GOSSIP_FAILURE_TIMEOUT_CLEANUP = 10.0 GOSSIP_FAILURE_TIMEOUT_CLEANUP = 6.0
members_list = {} members_list = {}
@@ -117,6 +120,7 @@ def global_switcher(cmd, value):
#THREAD 1 #THREAD 1
def worker_1(): def worker_1():
global TOTAL_SENT_BYTES, TOTAL_RECEIVED_BYTES
has_joined = (INTRODUCER_IP == MY_IP) has_joined = (INTRODUCER_IP == MY_IP)
last_join_attempt = 0.0 last_join_attempt = 0.0
@@ -128,7 +132,10 @@ def worker_1():
joining_message_json = json.dumps(joining_message_payload) joining_message_json = json.dumps(joining_message_payload)
try: try:
server_socket.sendto(joining_message_json.encode('utf-8'), (INTRODUCER_IP, GLOBAL_PORT)) 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: except Exception as e:
log(f"JOIN send failed: {e}") log(f"JOIN send failed: {e}")
last_join_attempt = utc_time() last_join_attempt = utc_time()
@@ -138,6 +145,9 @@ def worker_1():
except socket.timeout: except socket.timeout:
continue continue
with traffic_lock:
TOTAL_RECEIVED_BYTES += len(data)
try: try:
msg = json.loads(data.decode('utf-8')) msg = json.loads(data.decode('utf-8'))
except Exception: except Exception:
@@ -196,6 +206,7 @@ def current_cluster_state():
} }
def handle_join(msg): def handle_join(msg):
global TOTAL_SENT_BYTES
sender_id = tuple(msg["sender_id"]) sender_id = tuple(msg["sender_id"])
with members_list_lock: with members_list_lock:
#check if the sender is in the dead members list #check if the sender is in the dead members list
@@ -228,7 +239,10 @@ def handle_join(msg):
} }
try: try:
server_socket.sendto(json.dumps(ack_payload).encode('utf-8'), (sender_id[0], sender_id[1])) 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: except Exception as e:
log(f"Failed to send GOSSIP response to {sender_id[0]}:{sender_id[1]}: {e}") log(f"Failed to send GOSSIP response to {sender_id[0]}:{sender_id[1]}: {e}")
@@ -378,6 +392,7 @@ _peer_queue = []
_peer_queue_lock = threading.Lock() _peer_queue_lock = threading.Lock()
def worker_3(): def worker_3():
global TOTAL_SENT_BYTES
global _peer_queue global _peer_queue
while not stop_threads.is_set(): while not stop_threads.is_set():
with members_list_lock: with members_list_lock:
@@ -408,7 +423,10 @@ def worker_3():
} }
try: try:
server_socket.sendto(json.dumps(gossip_payload).encode('utf-8'), (peer[0], peer[1])) 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: except Exception as e:
log(f"Failed to send GOSSIP to {peer[0]}:{peer[1]}: {e}") log(f"Failed to send GOSSIP to {peer[0]}:{peer[1]}: {e}")
@@ -475,9 +493,9 @@ def worker_5():
if cmd_0 == "list_mem": if cmd_0 == "list_mem":
cmd_list_mem() cmd_list_mem()
elif cmd_0 == "list_self": elif cmd_0 == "list_self":
print(str(MY_ID)) log(str(MY_ID))
elif cmd_0 == "join": elif cmd_0 == "join":
print("auto joins at start up") log("auto joins at start up")
elif cmd_0 == "leave": elif cmd_0 == "leave":
log("Leaving") log("Leaving")
stop_threads.set() stop_threads.set()
@@ -487,34 +505,46 @@ def worker_5():
if len(cmd) >= 2: if len(cmd) >= 2:
global_switcher("switch", cmd[1]) global_switcher("switch", cmd[1])
else: else:
print("Usage: switch suspect|nosuspect") log("Usage: switch suspect|nosuspect")
elif cmd_0 == "display_protocol": elif cmd_0 == "display_protocol":
with cluster_state_lock: with cluster_state_lock:
print(f"Mode: {cluster_state['mode']}, drop rate: {cluster_state['drop_rate']}") log(f"Mode: {cluster_state['mode']}, drop rate: {cluster_state['drop_rate']}")
elif cmd_0 == "set_drop_rate": elif cmd_0 == "set_drop_rate":
if len(cmd) >= 2: if len(cmd) >= 2:
global_switcher("drop_rate", cmd[1]) global_switcher("drop_rate", cmd[1])
else: else:
print("Usage: set_drop_rate <number>") log("Usage: set_drop_rate <number>")
else: else:
log(f"Unknown command: {cmd_0}") log(f"Unknown command: {cmd_0}")
#THREAD 5 HELPERS #THREAD 5 HELPERS
def cmd_list_mem(): def cmd_list_mem():
with members_list_lock: with members_list_lock:
for member_id, member_data in members_list.items(): cluster_state = [
if member_data["state"] == "failed": (
continue m_id,
print(f"{member_id} state = {member_data['state']}, heartbeat = {member_data['heart_beat']}, incarnation = {member_data['incarnation_counter']}") 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(): def cmd_display_suspects():
with suspect_history_lock: with suspect_history_lock:
if not suspect_history: history_snapshot = list(suspect_history)
print("No suspects recorded.")
if not history_snapshot:
log("No suspects recorded.")
else: else:
print("Suspect history:") log("Suspect history:")
for member_id, timestamp in suspect_history: for member_id, timestamp in history_snapshot:
print(f"{member_id} suspected at {timestamp}") log(f"{member_id} suspected at {timestamp}")
# THREAD 6 - Display membership list in a pretty format # THREAD 6 - Display membership list in a pretty format
@@ -531,6 +561,10 @@ def worker_monitor():
mode = cluster_state["mode"] mode = cluster_state["mode"]
drop_rate = cluster_state["drop_rate"] drop_rate = cluster_state["drop_rate"]
with traffic_lock:
sent_bytes = TOTAL_SENT_BYTES
received_bytes = TOTAL_RECEIVED_BYTES
try: try:
with open("membership_list.txt", "w") as f: with open("membership_list.txt", "w") as f:
for member_id, member_data in snapshot.items(): for member_id, member_data in snapshot.items():
@@ -538,7 +572,9 @@ def worker_monitor():
f.write( f.write(
f"Mode: {mode}, " f"Mode: {mode}, "
f"Drop Rate: {drop_rate}\n" f"Drop Rate: {drop_rate}, "
f"Sent Bytes: {sent_bytes}, "
f"Received Bytes: {received_bytes}\n"
) )
except Exception as e: except Exception as e:
+405
View File
@@ -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
+264
View File
@@ -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
)
+147
View File
@@ -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"