added total bytes r/s + updated teh view
This commit is contained in:
1 parent
af0e9d423d
commit
6822da6e34
2 files changed
+39
-4
No files matched your search
+29
-4
@@ -25,6 +25,10 @@ def log(message):
|
|||||||
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
|
||||||
@@ -116,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
|
||||||
|
|
||||||
@@ -127,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()
|
||||||
@@ -137,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:
|
||||||
@@ -195,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
|
||||||
@@ -227,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}")
|
||||||
|
|
||||||
@@ -377,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:
|
||||||
@@ -407,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}")
|
||||||
|
|
||||||
@@ -530,6 +549,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():
|
||||||
@@ -537,7 +560,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:
|
||||||
|
|||||||
@@ -73,6 +73,8 @@ def render(lines):
|
|||||||
|
|
||||||
mode = "?"
|
mode = "?"
|
||||||
drop_rate = "?"
|
drop_rate = "?"
|
||||||
|
sent_bytes = "?"
|
||||||
|
received_bytes = "?"
|
||||||
|
|
||||||
now = time.time()
|
now = time.time()
|
||||||
|
|
||||||
@@ -94,6 +96,10 @@ def render(lines):
|
|||||||
|
|
||||||
mode = pieces[0].split(":", 1)[1].strip()
|
mode = pieces[0].split(":", 1)[1].strip()
|
||||||
drop_rate = pieces[1].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:
|
except Exception:
|
||||||
pass
|
pass
|
||||||
@@ -166,6 +172,10 @@ def render(lines):
|
|||||||
f"{GRAY}|{RESET} "
|
f"{GRAY}|{RESET} "
|
||||||
f"{BOLD}{CYAN}DROP:{RESET} "
|
f"{BOLD}{CYAN}DROP:{RESET} "
|
||||||
f"{WHITE}{drop_rate}%{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}"
|
f"{CLEAR_LINE}"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|||||||
Reference in new issue
Block a user