Skip to content

Commit 3b1d635

Browse files
GaneshPatil7517GaneshPatil7517
committed
fix: prevent NameError in measurement scripts and add missing psutil dependency (fixes #345)
1 parent 0467ce5 commit 3b1d635

4 files changed

Lines changed: 295 additions & 233 deletions

File tree

‎measurements/A.py‎

Lines changed: 93 additions & 76 deletions
Original file line numberDiff line numberDiff line change
@@ -1,76 +1,93 @@
1-
# A.py (Client and Primary Measurement Node)
2-
import concore
3-
import time
4-
import os
5-
import psutil
6-
import sys
7-
8-
# --- ZMQ Initialization ---
9-
# This REQ socket connects to Node B
10-
concore.init_zmq_port(
11-
port_name=PORT_NAME_F1_F2,
12-
port_type="connect",
13-
address="tcp://localhost:" + PORT_F1_F2,
14-
socket_type_str="REQ"
15-
)
16-
17-
print("Node A client started.")
18-
19-
# --- Measurement Initialization ---
20-
min_latency = float('inf')
21-
max_latency = 0.0
22-
total_latency = 0.0
23-
message_count = 0
24-
total_bytes = 0
25-
process = psutil.Process(os.getpid())
26-
overall_start_time = time.monotonic()
27-
loop_start_time = 0
28-
29-
current_value = 0
30-
max_value = 100
31-
32-
while current_value < max_value:
33-
loop_start_time = time.monotonic() # Start timer for round-trip latency
34-
print(f"Node A: Sending value {current_value:.2f} to Node B.")
35-
36-
# 1. Send the current value as a request to the pipeline
37-
concore.write(PORT_NAME_F1_F2, "value", [current_value])
38-
total_bytes += sys.getsizeof([current_value])
39-
40-
# 2. Wait for the final, processed value in reply
41-
received_data = concore.read(PORT_NAME_F1_F2, "value", [0.0])
42-
43-
loop_end_time = time.monotonic()
44-
latency_ms = (loop_end_time - loop_start_time) * 1000
45-
46-
# Update metrics
47-
message_count += 1
48-
min_latency = min(min_latency, latency_ms)
49-
max_latency = max(max_latency, latency_ms)
50-
total_latency += latency_ms
51-
52-
current_value = received_data[0]
53-
print(f"Node A: Received final value {current_value:.2f} from the pipeline. | Latency: {latency_ms:.2f} ms")
54-
print("-" * 20)
55-
56-
# --- Finalize and Report Measurements ---
57-
overall_end_time = time.monotonic()
58-
total_duration = overall_end_time - overall_start_time
59-
cpu_usage = process.cpu_percent() / total_duration if total_duration > 0 else 0
60-
avg_latency = total_latency / message_count if message_count > 0 else 0
61-
62-
print("\n" + "="*35)
63-
print("--- NODE A: END-TO-END RESULTS ---")
64-
print(f"Total pipeline iterations: {message_count}")
65-
print(f"Total data sent: {total_bytes / 1024:.4f} KB")
66-
print(f"Total End-to-End Time: {total_duration:.4f} seconds")
67-
print("-" * 35)
68-
print(f"Min round-trip latency: {min_latency:.2f} ms")
69-
print(f"Avg round-trip latency: {avg_latency:.2f} ms")
70-
print(f"Max round-trip latency: {max_latency:.2f} ms")
71-
print("-" * 35)
72-
print(f"Approximate CPU usage: {cpu_usage:.2f}%")
73-
print("="*35)
74-
75-
print(f"\nNode A: Final value {current_value:.2f} reached the target. Terminating.")
76-
concore.terminate_zmq()
1+
# A.py (Client and Primary Measurement Node)
2+
import concore
3+
import time
4+
import os
5+
import psutil
6+
import sys
7+
8+
# --- Fallback Definitions for Direct Execution ---
9+
# These values are normally injected by copy_with_port_portname.py during study generation.
10+
# When running this script directly, safe defaults are used instead.
11+
_STANDALONE_MODE = False
12+
13+
if 'PORT_NAME_F1_F2' not in globals():
14+
PORT_NAME_F1_F2 = "F1_F2"
15+
_STANDALONE_MODE = True
16+
17+
if 'PORT_F1_F2' not in globals():
18+
PORT_F1_F2 = "5555"
19+
_STANDALONE_MODE = True
20+
21+
if _STANDALONE_MODE:
22+
print("Warning: Port variables not injected. Running in standalone mode with default values.")
23+
print(" For full study behavior, run via study generation (makestudy).")
24+
25+
# --- ZMQ Initialization ---
26+
# This REQ socket connects to Node B
27+
concore.init_zmq_port(
28+
port_name=PORT_NAME_F1_F2,
29+
port_type="connect",
30+
address="tcp://localhost:" + PORT_F1_F2,
31+
socket_type_str="REQ"
32+
)
33+
34+
print("Node A client started.")
35+
36+
# --- Measurement Initialization ---
37+
min_latency = float('inf')
38+
max_latency = 0.0
39+
total_latency = 0.0
40+
message_count = 0
41+
total_bytes = 0
42+
process = psutil.Process(os.getpid())
43+
overall_start_time = time.monotonic()
44+
loop_start_time = 0
45+
46+
current_value = 0
47+
max_value = 100
48+
49+
while current_value < max_value:
50+
loop_start_time = time.monotonic() # Start timer for round-trip latency
51+
print(f"Node A: Sending value {current_value:.2f} to Node B.")
52+
53+
# 1. Send the current value as a request to the pipeline
54+
concore.write(PORT_NAME_F1_F2, "value", [current_value])
55+
total_bytes += sys.getsizeof([current_value])
56+
57+
# 2. Wait for the final, processed value in reply
58+
received_data = concore.read(PORT_NAME_F1_F2, "value", [0.0])
59+
60+
loop_end_time = time.monotonic()
61+
latency_ms = (loop_end_time - loop_start_time) * 1000
62+
63+
# Update metrics
64+
message_count += 1
65+
min_latency = min(min_latency, latency_ms)
66+
max_latency = max(max_latency, latency_ms)
67+
total_latency += latency_ms
68+
69+
current_value = received_data[0]
70+
print(f"Node A: Received final value {current_value:.2f} from the pipeline. | Latency: {latency_ms:.2f} ms")
71+
print("-" * 20)
72+
73+
# --- Finalize and Report Measurements ---
74+
overall_end_time = time.monotonic()
75+
total_duration = overall_end_time - overall_start_time
76+
cpu_usage = process.cpu_percent() / total_duration if total_duration > 0 else 0
77+
avg_latency = total_latency / message_count if message_count > 0 else 0
78+
79+
print("\n" + "="*35)
80+
print("--- NODE A: END-TO-END RESULTS ---")
81+
print(f"Total pipeline iterations: {message_count}")
82+
print(f"Total data sent: {total_bytes / 1024:.4f} KB")
83+
print(f"Total End-to-End Time: {total_duration:.4f} seconds")
84+
print("-" * 35)
85+
print(f"Min round-trip latency: {min_latency:.2f} ms")
86+
print(f"Avg round-trip latency: {avg_latency:.2f} ms")
87+
print(f"Max round-trip latency: {max_latency:.2f} ms")
88+
print("-" * 35)
89+
print(f"Approximate CPU usage: {cpu_usage:.2f}%")
90+
print("="*35)
91+
92+
print(f"\nNode A: Final value {current_value:.2f} reached the target. Terminating.")
93+
concore.terminate_zmq()

‎measurements/B.py‎

Lines changed: 85 additions & 60 deletions
Original file line numberDiff line numberDiff line change
@@ -1,60 +1,85 @@
1-
# B.py (Broker with Measurements)
2-
import concore
3-
import time
4-
5-
# --- ZMQ Initialization ---
6-
# This REP socket binds and waits for requests from Node A
7-
concore.init_zmq_port(
8-
port_name=PORT_NAME_F1_F2,
9-
port_type="bind",
10-
address="tcp://*:" + PORT_F1_F2,
11-
socket_type_str="REP"
12-
)
13-
# This REQ socket connects to Node C
14-
concore.init_zmq_port(
15-
port_name=PORT_NAME_F2_F3,
16-
port_type="connect",
17-
address="tcp://localhost:" + PORT_F2_F3,
18-
socket_type_str="REQ"
19-
)
20-
21-
print("Node B broker started. Waiting for requests...")
22-
23-
# --- Measurement Initialization ---
24-
start_time = time.monotonic()
25-
messages_routed = 0
26-
27-
while True:
28-
# 1. Wait for a request from Node A
29-
value_from_a = concore.read(PORT_NAME_F1_F2, "value", [0.0])
30-
received_value = value_from_a[0]
31-
print(f"Node B: Received {received_value:.2f} from Node A. Forwarding to C...")
32-
33-
# 2. Send the received value as a new request to Node C
34-
concore.write(PORT_NAME_F2_F3, "value", [received_value])
35-
36-
# 3. Wait for the reply from Node C
37-
value_from_c = concore.read(PORT_NAME_F2_F3, "value", [0.0])
38-
processed_value = value_from_c[0]
39-
print(f"Node B: Received {processed_value:.2f} from Node C. Replying to A...")
40-
41-
# 4. Send the processed value back as a reply to Node A
42-
concore.write(PORT_NAME_F1_F2, "value", [processed_value])
43-
messages_routed += 1
44-
45-
# 5. Check termination condition
46-
if processed_value >= 100:
47-
break
48-
49-
# --- Finalize and Report Measurements ---
50-
end_time = time.monotonic()
51-
duration = end_time - start_time
52-
53-
print("\n" + "="*30)
54-
print("--- NODE B: RESULTS ---")
55-
print(f"Total messages routed: {messages_routed}")
56-
print(f"Total execution time: {duration:.4f} seconds")
57-
print("="*30)
58-
59-
print("\nNode B: Terminating.")
60-
concore.terminate_zmq()
1+
# B.py (Broker with Measurements)
2+
import concore
3+
import time
4+
5+
# --- Fallback Definitions for Direct Execution ---
6+
# These values are normally injected by copy_with_port_portname.py during study generation.
7+
# When running this script directly, safe defaults are used instead.
8+
_STANDALONE_MODE = False
9+
10+
if 'PORT_NAME_F1_F2' not in globals():
11+
PORT_NAME_F1_F2 = "F1_F2"
12+
_STANDALONE_MODE = True
13+
14+
if 'PORT_F1_F2' not in globals():
15+
PORT_F1_F2 = "5555"
16+
_STANDALONE_MODE = True
17+
18+
if 'PORT_NAME_F2_F3' not in globals():
19+
PORT_NAME_F2_F3 = "F2_F3"
20+
_STANDALONE_MODE = True
21+
22+
if 'PORT_F2_F3' not in globals():
23+
PORT_F2_F3 = "5556"
24+
_STANDALONE_MODE = True
25+
26+
if _STANDALONE_MODE:
27+
print("Warning: Port variables not injected. Running in standalone mode with default values.")
28+
print(" For full study behavior, run via study generation (makestudy).")
29+
30+
# --- ZMQ Initialization ---
31+
# This REP socket binds and waits for requests from Node A
32+
concore.init_zmq_port(
33+
port_name=PORT_NAME_F1_F2,
34+
port_type="bind",
35+
address="tcp://*:" + PORT_F1_F2,
36+
socket_type_str="REP"
37+
)
38+
# This REQ socket connects to Node C
39+
concore.init_zmq_port(
40+
port_name=PORT_NAME_F2_F3,
41+
port_type="connect",
42+
address="tcp://localhost:" + PORT_F2_F3,
43+
socket_type_str="REQ"
44+
)
45+
46+
print("Node B broker started. Waiting for requests...")
47+
48+
# --- Measurement Initialization ---
49+
start_time = time.monotonic()
50+
messages_routed = 0
51+
52+
while True:
53+
# 1. Wait for a request from Node A
54+
value_from_a = concore.read(PORT_NAME_F1_F2, "value", [0.0])
55+
received_value = value_from_a[0]
56+
print(f"Node B: Received {received_value:.2f} from Node A. Forwarding to C...")
57+
58+
# 2. Send the received value as a new request to Node C
59+
concore.write(PORT_NAME_F2_F3, "value", [received_value])
60+
61+
# 3. Wait for the reply from Node C
62+
value_from_c = concore.read(PORT_NAME_F2_F3, "value", [0.0])
63+
processed_value = value_from_c[0]
64+
print(f"Node B: Received {processed_value:.2f} from Node C. Replying to A...")
65+
66+
# 4. Send the processed value back as a reply to Node A
67+
concore.write(PORT_NAME_F1_F2, "value", [processed_value])
68+
messages_routed += 1
69+
70+
# 5. Check termination condition
71+
if processed_value >= 100:
72+
break
73+
74+
# --- Finalize and Report Measurements ---
75+
end_time = time.monotonic()
76+
duration = end_time - start_time
77+
78+
print("\n" + "="*30)
79+
print("--- NODE B: RESULTS ---")
80+
print(f"Total messages routed: {messages_routed}")
81+
print(f"Total execution time: {duration:.4f} seconds")
82+
print("="*30)
83+
84+
print("\nNode B: Terminating.")
85+
concore.terminate_zmq()

0 commit comments

Comments
 (0)