267 lines
9.1 KiB
Python
267 lines
9.1 KiB
Python
from datetime import datetime, timezone
|
|
import socket
|
|
import threading
|
|
import time
|
|
import random
|
|
import json
|
|
|
|
def utc_time():
|
|
return datetime.now(timezone.utc).timestamp()
|
|
|
|
INTRODUCER_IP = "172.22.156.3"
|
|
IP = socket.gethostbyname(socket.gethostname())
|
|
SERVER_PORT = 8989
|
|
INTRODUCER_FLAG = (INTRODUCER_IP == IP)
|
|
HEART_BEAT_INTERVAL = 0.25
|
|
GOSSIP_INTERVAL = 0.25
|
|
FAILURE_TIMEOUT = 2.5
|
|
FAILURE_T_CLEANUP = 10.0
|
|
RETRY_INTERVAL = 5.0
|
|
FANOUT = 1
|
|
|
|
members_list = {}
|
|
dead_members = set()
|
|
members_list_lock = threading.Lock()
|
|
|
|
SELF_ID = (IP, SERVER_PORT, utc_time())
|
|
members_list[SELF_ID] = {
|
|
"heart_beat": 1,
|
|
"local_last_seen": utc_time(),
|
|
"state": "alive",
|
|
"local_failed_since" : None,
|
|
"local_suspected_since" : None
|
|
}
|
|
print(f"Starting peer: IP={IP}, port={SERVER_PORT}, introducer={INTRODUCER_FLAG}, self_id={SELF_ID}")
|
|
|
|
stop_event = threading.Event()
|
|
|
|
server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
server_socket.bind((IP, SERVER_PORT))
|
|
server_socket.settimeout(1.0)
|
|
|
|
def worker_1():
|
|
has_joined = INTRODUCER_FLAG
|
|
last_join_time = 0
|
|
|
|
while not stop_event.is_set():
|
|
if not has_joined and (utc_time() - last_join_time > RETRY_INTERVAL):
|
|
join_payload = {
|
|
"type": "JOIN",
|
|
"sender_id": SELF_ID
|
|
}
|
|
server_socket.sendto(json.dumps(join_payload).encode('utf-8'), (INTRODUCER_IP, SERVER_PORT))
|
|
last_join_time = utc_time()
|
|
|
|
try:
|
|
data, addr = server_socket.recvfrom(65535)
|
|
|
|
msg = json.loads(data.decode('utf-8'))
|
|
|
|
msg_type = msg.get("type")
|
|
|
|
if msg_type == "JOIN":
|
|
join_ip, join_port, join_timestamp = msg["sender_id"]
|
|
join_id = (join_ip, join_port, join_timestamp)
|
|
|
|
with members_list_lock:
|
|
if join_id in dead_ids:
|
|
continue
|
|
|
|
if join_id not in members_list:
|
|
members_list[join_id] = {
|
|
"heart_beat": 1,
|
|
"local_last_seen": utc_time(),
|
|
"state": "alive",
|
|
"local_failed_since" : None,
|
|
"local_suspected_since" : None
|
|
}
|
|
print(f"Added member: {join_id}")
|
|
|
|
members_payload = []
|
|
for m_id, m_data in members_list.items():
|
|
members_payload.append({
|
|
"ip": m_id[0],
|
|
"port": m_id[1],
|
|
"timestamp": m_id[2],
|
|
"heart_beat": m_data["heart_beat"],
|
|
"state": m_data["state"]
|
|
})
|
|
|
|
reply_payload = {
|
|
"type": "GOSSIP",
|
|
"sender_id": SELF_ID,
|
|
"members": members_payload
|
|
}
|
|
server_socket.sendto(json.dumps(reply_payload).encode('utf-8'), (join_ip, join_port))
|
|
print(f"Sent initial membership state to {(join_ip, join_port)}")
|
|
|
|
elif msg_type == "GOSSIP":
|
|
if not has_joined:
|
|
has_joined = True
|
|
|
|
members_json = msg.get('members', [])
|
|
|
|
with members_list_lock:
|
|
for member in members_json:
|
|
member_id = (member['ip'], member['port'], member['timestamp'])
|
|
|
|
if member_id == SELF_ID:
|
|
continue
|
|
|
|
incoming_state = member['state']
|
|
incoming_hb = member['heart_beat']
|
|
|
|
if member_id in dead_ids:
|
|
continue
|
|
|
|
if member_id not in members_list:
|
|
|
|
if incoming_state == "failed":
|
|
dead_ids.add(member_id)
|
|
continue
|
|
|
|
members_list[member_id] = {
|
|
"heart_beat": incoming_hb,
|
|
"local_last_seen": utc_time(),
|
|
"state": incoming_state,
|
|
"local_failed_since": None,
|
|
"local_suspected_since": None
|
|
}
|
|
continue
|
|
|
|
local_member = members_list[member_id]
|
|
|
|
if local_member["state"] == "failed":
|
|
continue
|
|
|
|
if incoming_state == "failed":
|
|
local_member["state"] = "failed"
|
|
local_member["local_failed_since"] = utc_time()
|
|
print(f"Member {member_id} marked as failed due to gossip.")
|
|
elif incoming_hb > local_member["heart_beat"]:
|
|
local_member["heart_beat"] = incoming_hb
|
|
local_member["local_last_seen"] = utc_time()
|
|
|
|
except socket.timeout:
|
|
continue
|
|
|
|
def worker_2():
|
|
while not stop_event.is_set():
|
|
with members_list_lock:
|
|
members_list[SELF_ID]["heart_beat"] += 1
|
|
time.sleep(HEART_BEAT_INTERVAL)
|
|
|
|
def worker_3():
|
|
peer_queue = []
|
|
|
|
while not stop_event.is_set():
|
|
with members_list_lock:
|
|
current_peers = [member_id for member_id in members_list.keys() if member_id != SELF_ID and members_list[member_id]["state"] == "alive"]
|
|
|
|
members_payload = []
|
|
for m_id, m_data in members_list.items():
|
|
members_payload.append({
|
|
"ip": m_id[0],
|
|
"port": m_id[1],
|
|
"timestamp": m_id[2],
|
|
"heart_beat": m_data["heart_beat"],
|
|
"state": m_data["state"]
|
|
})
|
|
|
|
if current_peers:
|
|
# Clean up peers from queue that might have failed or left
|
|
peer_queue = [p for p in peer_queue if p in current_peers]
|
|
|
|
selected_peers = []
|
|
num_to_select = min(FANOUT, len(current_peers))
|
|
|
|
for _ in range(num_to_select):
|
|
# Reshuffle if the queue is empty
|
|
if not peer_queue:
|
|
peer_queue = list(current_peers)
|
|
random.shuffle(peer_queue)
|
|
|
|
if peer_queue:
|
|
selected_peers.append(peer_queue.pop(0))
|
|
|
|
for peer in selected_peers:
|
|
peer_ip, peer_port, _ = peer
|
|
|
|
payload = {
|
|
"type": "GOSSIP",
|
|
"sender_id": SELF_ID,
|
|
"members": members_payload
|
|
}
|
|
|
|
json_string = json.dumps(payload)
|
|
server_socket.sendto(json_string.encode('utf-8'), (peer_ip, peer_port))
|
|
|
|
time.sleep(GOSSIP_INTERVAL)
|
|
|
|
def worker_4():
|
|
while not stop_event.is_set():
|
|
current_time = utc_time()
|
|
to_remove = []
|
|
|
|
with members_list_lock:
|
|
for member_id, member_data in members_list.items():
|
|
if member_id == SELF_ID:
|
|
continue
|
|
|
|
if member_data["state"] == "alive":
|
|
last_seen = member_data["local_last_seen"]
|
|
time_since_last_seen = current_time - last_seen
|
|
|
|
if time_since_last_seen > FAILURE_TIMEOUT:
|
|
member_data["state"] = "failed"
|
|
member_data["local_failed_since"] = current_time
|
|
|
|
last_failed_since = member_data.get("local_failed_since")
|
|
if last_failed_since is not None and (current_time - last_failed_since) > FAILURE_T_CLEANUP:
|
|
to_remove.append(member_id)
|
|
|
|
for member_id in to_remove:
|
|
dead_ids.add(member_id)
|
|
del members_list[member_id]
|
|
|
|
time.sleep(0.5)
|
|
|
|
def worker_monitoring():
|
|
while not stop_event.is_set():
|
|
with members_list_lock:
|
|
local_ml = {
|
|
member_id: member_data.copy()
|
|
for member_id, member_data in members_list.items()
|
|
}
|
|
|
|
with open("membership_list.txt", "w") as f:
|
|
for member_id, member_data in local_ml.items():
|
|
f.write(f"{member_id}: {member_data}\n")
|
|
time.sleep(0.5)
|
|
|
|
t1 = threading.Thread(target=worker_1)
|
|
t2 = threading.Thread(target=worker_2)
|
|
t3 = threading.Thread(target=worker_3)
|
|
t4 = threading.Thread(target=worker_4)
|
|
t5 = threading.Thread(target=worker_monitoring)
|
|
|
|
t1.start()
|
|
t2.start()
|
|
t3.start()
|
|
t4.start()
|
|
t5.start()
|
|
|
|
try:
|
|
while not stop_event.is_set():
|
|
time.sleep(0.5)
|
|
except KeyboardInterrupt:
|
|
print("\nExiting...")
|
|
stop_event.set()
|
|
|
|
t1.join()
|
|
t2.join()
|
|
t3.join()
|
|
t4.join()
|
|
t5.join()
|
|
server_socket.close()
|