(expected_load, load_type)
| 326 | |
| 327 | |
| 328 | def one_test(expected_load, load_type): |
| 329 | try: |
| 330 | task_name = f"Load for {expected_load} threads or % load on {load_type}" |
| 331 | tracker_cpu_load.start_task(task_name + " CPU Load") |
| 332 | tracker_rapl.start_task(task_name + " RAPL") |
| 333 | |
| 334 | # Create and start measurement thread |
| 335 | measure_thread = Thread( |
| 336 | target=measurement_thread, args=(expected_load, load_type) |
| 337 | ) |
| 338 | measure_thread.start() |
| 339 | |
| 340 | stress_cpu(load_type, test_phase_duration, expected_load) |
| 341 | |
| 342 | # Stop measurement thread |
| 343 | # measure_thread.join() |
| 344 | |
| 345 | cpu_load_data = tracker_cpu_load.stop_task() |
| 346 | rapl_data = tracker_rapl.stop_task() |
| 347 | point = measurements[-1] |
| 348 | point.task_name = task_name |
| 349 | point.rapl_power = rapl_data.cpu_power |
| 350 | point.rapl_energy = rapl_data.cpu_energy |
| 351 | point.estimated_power = cpu_load_data.cpu_power |
| 352 | point.estimated_energy = cpu_load_data.cpu_energy |
| 353 | point.duration = rapl_data.duration |
| 354 | |
| 355 | print("=" * 80) |
| 356 | print(measurements[-1].__dict__) |
| 357 | print("=" * 80) |
| 358 | finally: |
| 359 | # Stop measurement thread |
| 360 | measure_thread.join() |
| 361 | |
| 362 | |
| 363 | def measure_power(load_type): |
no test coverage detected
searching dependent graphs…