Implement benchmarker framework#1573
Conversation
| max_len = 500 | ||
| if len(s) <= max_len: | ||
| return s | ||
| snippset_len = int(max_len / 2) - 4 |
There was a problem hiding this comment.
nit: -7 to take into account the spaces ?
There was a problem hiding this comment.
Both the start and end get trimmed to this length, so -4 becomes -8. "snippset" is a typo though; fixed :)
| op | ||
| for op in operations | ||
| if op.successful | ||
| and op.completed_at.datetime >= stability_time |
There was a problem hiding this comment.
nit: covered on line 113 ?
| user_specs_map = {u.name: u for u in config.user_types} | ||
| loads_map = {load.name: load for load in config.loads} | ||
|
|
||
| max_io_threads = 400 |
There was a problem hiding this comment.
That seems a value which could impact benchmarking results at some point. Is there a simple way to detect and log that we reached the limit ?
There was a problem hiding this comment.
I agree, though Gemini thinks this isn't trivial. I've added a TODO comment on #1566 for a future PR linking to this comment, and this is the sample code Gemini suggested:
import time
import threading
from concurrent.futures import ThreadPoolExecutor
class MonitoredThreadPoolExecutor(ThreadPoolExecutor):
def __init__(self, max_workers=None, *args, **kwargs):
super().__init__(max_workers=max_workers, *args, **kwargs)
# Store max_workers locally so we don't rely on private attributes
self.max_workers_limit = max_workers or self._max_workers
self._active_tasks = 0
self._lock = threading.Lock()
def submit(self, fn, *args, **kwargs):
with self._lock:
# Check if we are currently at or over capacity before queuing this task
if self._active_tasks >= self.max_workers_limit:
print(f"⚠️ [Alert] Max workers ({self.max_workers_limit}) hit! Task is going to queue.")
# Wrap the function to increment/decrement our active counter
def wrapper(*args_inner, **kwargs_inner):
with self._lock:
self._active_tasks += 1
try:
return fn(*args_inner, **kwargs_inner)
finally:
with self._lock:
self._active_tasks -= 1
return super().submit(wrapper, *args, **kwargs)
@property
def active_tasks_count(self):
"""Returns the number of tasks currently running in a thread."""
with self._lock:
return self._active_tasks
# --- Demonstration ---
if __name__ == "__main__":
# Initialize our custom executor with a limit of 3 workers
with MonitoredThreadPoolExecutor(max_workers=3) as executor:
for i in range(5):
print(f"Submitting task {i}...")
executor.submit(time.sleep, 1)
print(f"Active running tasks: {executor.active_tasks_count}\n")
time.sleep(0.1) # small delay to let threads boot upI think we would probably want to do something more durable than printing/logging though.
This PR continues MVP work for #1566 by adding an implementation of the basic framework. The matplotlib figure generation is stubbed as well as the flight_planner virtual user, but otherwise the PR mostly implements the ability to run the example configuration. The execution hierarchy is roughly: