563 lines
21 KiB
Python
563 lines
21 KiB
Python
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", "a")
|
|
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()
|
|
|
|
def utc_time():
|
|
return datetime.now(timezone.utc).timestamp()
|
|
|
|
#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 = 1
|
|
|
|
GOSSIP_FAILURE_TIMEOUT = 1.8
|
|
GOSSIP_S_SUSPECT_TIMEOUT = 0.8
|
|
GOSSIP_S_TIMEOUT_SUSPECT = 1.0
|
|
GOSSIP_FAILURE_TIMEOUT_CLEANUP = 10.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():
|
|
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:
|
|
server_socket.sendto(joining_message_json.encode('utf-8'), (INTRODUCER_IP, GLOBAL_PORT))
|
|
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
|
|
|
|
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):
|
|
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:
|
|
server_socket.sendto(json.dumps(ack_payload).encode('utf-8'), (sender_id[0], sender_id[1]))
|
|
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 _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:
|
|
server_socket.sendto(json.dumps(gossip_payload).encode('utf-8'), (peer[0], peer[1]))
|
|
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":
|
|
print(str(MY_ID))
|
|
elif cmd_0 == "join":
|
|
print("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:
|
|
print("Usage: switch suspect|nosuspect")
|
|
elif cmd_0 == "display_protocol":
|
|
with cluster_state_lock:
|
|
print(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:
|
|
print("Usage: set_drop_rate <number>")
|
|
else:
|
|
log(f"Unknown command: {cmd_0}")
|
|
|
|
#THREAD 5 HELPERS
|
|
def cmd_list_mem():
|
|
with members_list_lock:
|
|
for member_id, member_data in members_list.items():
|
|
if member_data["state"] == "failed":
|
|
continue
|
|
print(f"{member_id} state = {member_data['state']}, heartbeat = {member_data['heart_beat']}, incarnation = {member_data['incarnation_counter']}")
|
|
|
|
def cmd_display_suspects():
|
|
with suspect_history_lock:
|
|
if not suspect_history:
|
|
print("No suspects recorded.")
|
|
else:
|
|
print("Suspect history:")
|
|
for member_id, timestamp in suspect_history:
|
|
print(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 = dict(members_list)
|
|
try:
|
|
with open("membership_list.txt", "w") as f:
|
|
for m_id, m_data in snapshot.items():
|
|
f.write(f"{m_id}: {m_data}\n")
|
|
except Exception:
|
|
pass
|
|
time.sleep(0.5)
|
|
|
|
|
|
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() |