mirror of
https://github.com/migatu/vtt_work.git
synced 2026-07-14 13:34:42 +00:00
127 lines
4.5 KiB
Python
127 lines
4.5 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test script to validate the LOW priority scheduler optimization for reducing redundant
|
|
get_value_cb calls in ActionScheduler.run_task_once().
|
|
|
|
The optimization caches get_stats_for() results within a single task execution to avoid:
|
|
1. Redundant calls when the same source is used for multiple axes (x_src == y_src)
|
|
2. Unnecessary calls for None sources
|
|
3. Multiple lock acquisitions and deque copying for the same data
|
|
|
|
Performance improvement: Reduces scheduler overhead by 33-66% for common scenarios.
|
|
"""
|
|
|
|
import sys
|
|
import time
|
|
|
|
# Add the monitor directory to path
|
|
sys.path.insert(0, 'nucleares_monitor')
|
|
|
|
def test_scheduler_optimization():
|
|
"""Test that the scheduler optimization reduces get_value_cb calls."""
|
|
from control_board_monitor import ActionScheduler
|
|
|
|
print("Testing ActionScheduler call optimization...")
|
|
|
|
class MockTask:
|
|
def __init__(self, x_src='var1', y_src='var1', z_src='var2', expr='x + y + z'):
|
|
self.x_src = x_src
|
|
self.y_src = y_src
|
|
self.z_src = z_src
|
|
self.expr = expr
|
|
self.value = '0'
|
|
self.name = 'test'
|
|
self.x_mode = 'raw'
|
|
self.y_mode = 'raw'
|
|
self.z_mode = 'raw'
|
|
|
|
call_count = 0
|
|
call_log = []
|
|
|
|
def mock_get_value_cb(src):
|
|
nonlocal call_count, call_log
|
|
call_count += 1
|
|
call_log.append(src)
|
|
# Simulate some work (lock acquisition, deque copying, etc.)
|
|
time.sleep(0.001) # 1ms per call
|
|
return {'x': 10.0, 'x_avg': 8.0, 'dx': 1.0, 'dx_avg': 0.5}
|
|
|
|
def mock_get_base_url():
|
|
return 'http://localhost:8080'
|
|
|
|
scheduler = ActionScheduler(mock_get_base_url, mock_get_value_cb)
|
|
|
|
test_cases = [
|
|
("Same x and y source", MockTask('var1', 'var1', 'var2'), 2),
|
|
("All same source", MockTask('var1', 'var1', 'var1'), 1),
|
|
("One None source", MockTask('var1', None, 'var2'), 2),
|
|
("All different sources", MockTask('var1', 'var2', 'var3'), 3),
|
|
("Two None sources", MockTask('var1', None, None), 1),
|
|
]
|
|
|
|
results = []
|
|
|
|
for description, task, expected_calls in test_cases:
|
|
call_count = 0
|
|
call_log = []
|
|
|
|
start_time = time.time()
|
|
try:
|
|
scheduler.run_task_once(task)
|
|
except Exception:
|
|
pass # Expected due to HTTP/expression errors in test
|
|
end_time = time.time()
|
|
|
|
execution_time = (end_time - start_time) * 1000 # Convert to ms
|
|
|
|
results.append({
|
|
'description': description,
|
|
'expected_calls': expected_calls,
|
|
'actual_calls': call_count,
|
|
'call_log': call_log,
|
|
'execution_time_ms': execution_time,
|
|
'optimized': call_count == expected_calls
|
|
})
|
|
|
|
print(f"\n{description}:")
|
|
print(f" Sources: x={task.x_src}, y={task.y_src}, z={task.z_src}")
|
|
print(f" Expected calls: {expected_calls}")
|
|
print(f" Actual calls: {call_count}")
|
|
print(f" Called for: {call_log}")
|
|
print(f" Execution time: {execution_time:.1f}ms")
|
|
print(" ✅ Optimized" if call_count == expected_calls else " ❌ Not optimized")
|
|
|
|
# Summary
|
|
print(f"\n{'='*60}")
|
|
print("OPTIMIZATION RESULTS:")
|
|
print(f"{'='*60}")
|
|
|
|
optimized_count = sum(1 for r in results if r['optimized'])
|
|
total_count = len(results)
|
|
|
|
print(f"Tests passed: {optimized_count}/{total_count}")
|
|
|
|
if optimized_count == total_count:
|
|
print("✅ ALL TESTS PASSED - Scheduler optimization working correctly!")
|
|
|
|
# Calculate potential savings
|
|
unoptimized_calls = sum(3 for _ in results) # Old version always called 3 times
|
|
optimized_calls = sum(r['actual_calls'] for r in results)
|
|
savings_percent = ((unoptimized_calls - optimized_calls) / unoptimized_calls) * 100
|
|
|
|
print("\nPerformance improvement:")
|
|
print(f" Old version: {unoptimized_calls} total calls")
|
|
print(f" Optimized version: {optimized_calls} total calls")
|
|
print(f" Savings: {savings_percent:.1f}% reduction in get_value_cb calls")
|
|
|
|
else:
|
|
print("❌ Some tests failed - optimization needs review")
|
|
for r in results:
|
|
if not r['optimized']:
|
|
print(f" Failed: {r['description']}")
|
|
|
|
return optimized_count == total_count
|
|
|
|
if __name__ == '__main__':
|
|
success = test_scheduler_optimization()
|
|
sys.exit(0 if success else 1) |