cengal.performance_test_lib.versions.v_1.performance_test_lib
1#!/usr/bin/env python 2# coding=utf-8 3 4# Copyright © 2012-2024 ButenkoMS. All rights reserved. Contacts: <gtalk@butenkoms.space> 5# 6# Licensed under the Apache License, Version 2.0 (the "License"); 7# you may not use this file except in compliance with the License. 8# You may obtain a copy of the License at 9# 10# http://www.apache.org/licenses/LICENSE-2.0 11# 12# Unless required by applicable law or agreed to in writing, software 13# distributed under the License is distributed on an "AS IS" BASIS, 14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 15# See the License for the specific language governing permissions and 16# limitations under the License. 17 18import copy 19import gc 20from cengal.code_flow_control.gc import DisableGC 21from contextlib import contextmanager 22from cengal.code_flow_control.smart_values.versions.v_2 import ValueExistence 23from cengal.parallel_execution.coroutines.coro_standard_services.lazy_print.versions.v_0.lazy_print import lprint 24from cengal.time_management.cpu_clock_cycles import perf_counter 25from cengal.time_management.load_best_timer import process_time 26from cengal.time_management.repeat_for_a_time import Tracer, ClockType 27from cengal.math.numbers import RationalNumber 28from cengal.introspection.inspect import is_async 29from cengal.introspection.inspect import func_name, entity_properties 30from cengal.code_inspection.auto_line_tracer import alt, LineType, OutputFields, AutoLineTracer 31 32from typing import Union, Callable, Awaitable, List, Optional, Type, Set 33 34""" 35Module Docstring 36Docstrings: http://www.python.org/dev/peps/pep-0257/ 37""" 38 39__author__ = "ButenkoMS <gtalk@butenkoms.space>" 40__copyright__ = "Copyright © 2012-2024 ButenkoMS. All rights reserved. Contacts: <gtalk@butenkoms.space>" 41__credits__ = ["ButenkoMS <gtalk@butenkoms.space>", ] 42__license__ = "Apache License, Version 2.0" 43__version__ = "4.4.1" 44__maintainer__ = "ButenkoMS <gtalk@butenkoms.space>" 45__email__ = "gtalk@butenkoms.space" 46# __status__ = "Prototype" 47__status__ = "Development" 48# __status__ = "Production" 49 50 51class PerformanceTestResult(Exception): 52 def __init__(self, result): 53 super(PerformanceTestResult, self).__init__() 54 self.result = result 55 56 57@contextmanager 58def test_run_time(test_name: str, number_of_iterations: int, throw_result: bool=False, throw_result_anyway: bool=True, ignore_index=False): 59 index = ValueExistence(True, copy.copy(number_of_iterations)) 60 start_time = perf_counter() 61 exception_occures = False 62 try: 63 yield index 64 except: 65 if not throw_result_anyway: 66 exception_occures = True 67 raise 68 finally: 69 if not ignore_index: 70 number_of_iterations -= index.value 71 72 end_time = perf_counter() 73 result_time = end_time - start_time 74 if result_time > 0: 75 text_result = f'>>> "{test_name}"\n\tIt was used {result_time} seconds to process {number_of_iterations} iterations.\n\tThere is {number_of_iterations / result_time} iterations per second\n' 76 else: 77 text_result = f'>>> "{test_name}"\n\tIt was used {result_time} seconds to process {number_of_iterations} iterations.\n' 78 79 lprint(text_result) 80 81 if (not exception_occures) and throw_result: 82 result_data = dict() 83 result_data['test_name'] = test_name 84 result_data['result_time'] = result_time 85 if result_time > 0: 86 result_data['iterations_per_time_unit'] = number_of_iterations / result_time 87 else: 88 result_data['iterations_per_time_unit'] = None 89 raise PerformanceTestResult(result_data) 90 91 92def measure_time(test_name: str = str()): 93 return test_run_time(test_name, 1, ignore_index=True) 94 95 96def test_function_run_time( 97 test_name: str = None, 98 iterations_qnt: int = None, 99 throw_result: bool = None, 100 ): 101 """ 102 Use 'performance_test_lib__iterations_qnt=1000000' parameter to pass number of iterations 103 :param testable_function: function 104 :return: 105 """ 106 def wrapper(testable_function): 107 if is_async(testable_function): 108 async def sub_wrapper(*args, **kwargs): 109 test_name_ = '' if test_name is None else test_name 110 test_full_name = '{}: {}'.format(str(testable_function), test_name_) 111 number_of_iterations = 1 if iterations_qnt is None else iterations_qnt 112 throw_result_ = False if throw_result is None else throw_result 113 with test_run_time(test_full_name, number_of_iterations, throw_result_) as index: 114 while index.value > 0: 115 await testable_function(*args, **kwargs) 116 index.value -= 1 117 118 return sub_wrapper 119 else: 120 def sub_wrapper(*args, **kwargs): 121 test_name_ = '' if test_name is None else test_name 122 test_full_name = '{}: {}'.format(str(testable_function), test_name_) 123 number_of_iterations = 1 if iterations_qnt is None else iterations_qnt 124 throw_result_ = False if throw_result is None else throw_result 125 with test_run_time(test_full_name, number_of_iterations, throw_result_) as index: 126 while index.value > 0: 127 testable_function(*args, **kwargs) 128 index.value -= 1 129 130 return sub_wrapper 131 return wrapper 132 133 134def process_performance_test_results(tracer: Tracer, test_name: str, throw_result: bool=False): 135 number_of_iterations = tracer.iterations_made 136 result_time = tracer.time_spent 137 iterations_per_time_unit = tracer.iter_per_time_unit 138 print('>>> "{}"'.format(test_name)) 139 print('\t' + 'It was used', result_time, 'seconds to process', number_of_iterations, 'iterations.') 140 print('\t' + 'There is', iterations_per_time_unit, 'iterations per second') 141 142 if throw_result: 143 result_data = (test_name, result_time, iterations_per_time_unit) 144 raise PerformanceTestResult(result_data) 145 146 147@contextmanager 148def test_performance(test_name: str, run_time: float, throw_result: bool=False, clock_type=ClockType.perf_counter): 149 tracer = Tracer(run_time, clock_type) 150 try: 151 yield tracer 152 except: 153 raise 154 finally: 155 process_performance_test_results(tracer, test_name, throw_result) 156 157 158def test_function_performance( 159 test_name: str = None, 160 run_time: RationalNumber = None, 161 throw_result: bool = None, 162 clock_type: ClockType = None, 163 ): 164 """ 165 Use 'performance_test_lib__run_time=1.5' parameter to pass number of seconds to test 166 :param testable_function: function 167 :return: 168 """ 169 def wrapper(testable_function): 170 if is_async(testable_function): 171 async def sub_wrapper(*args, **kwargs): 172 test_name_ = '' if test_name is None else test_name 173 full_test_name = '{}: {}'.format(str(testable_function), test_name_) 174 run_time_ = 1.0 if run_time is None else run_time 175 throw_result_ = False if throw_result is None else throw_result 176 clock_type_ = ClockType.perf_counter if clock_type is None else clock_type 177 with test_performance(full_test_name, run_time_, throw_result_, clock_type_) as tracer: 178 while tracer.iter(): 179 await testable_function(*args, **kwargs) 180 181 return sub_wrapper 182 else: 183 def sub_wrapper(*args, **kwargs): 184 test_name_ = '' if test_name is None else test_name 185 full_test_name = '{}: {}'.format(str(testable_function), test_name_) 186 run_time_ = 1.0 if run_time is None else run_time 187 throw_result_ = False if throw_result is None else throw_result 188 clock_type_ = ClockType.perf_counter if clock_type is None else clock_type 189 with test_performance(full_test_name, run_time_, throw_result_, clock_type_) as tracer: 190 while tracer.iter(): 191 testable_function(*args, **kwargs) 192 193 return sub_wrapper 194 return wrapper 195 196 197class PrecisePerformanceTestTracer(Tracer): 198 """ 199 Precise tracer. 200 At first you need to use it as a usual Tracer. After tracing was done - use it as a fast `for i in range(...)` block 201 202 Example of use: 203 204 tr = PrecisePerformanceTestTracer(10.0) 205 while tr.iter(): 206 i = '456' 207 k = int('1243' + i) 208 209 with tr as fast_iter: 210 for i in fast_iter: 211 i = '456' 212 k = int('1243' + i) 213 214 print('{} iter/s; {} seconds; {} iterations'.format(tr.iter_per_time_unit, tr.time_spent, tr.iterations_made)) 215 216 """ 217 218 def __init__(self, 219 run_time: float, 220 clock_type: ClockType=ClockType.perf_counter, 221 suppress_exceptions: bool=False, 222 turn_off_gc: bool=False 223 ): 224 super().__init__(run_time, clock_type) 225 self.suppress_exceptions = suppress_exceptions 226 self.turn_off_gc = turn_off_gc 227 self.gc_was_enabled = None 228 229 def __enter__(self): 230 self._relevant_start_time = self._start_time = self._relevant_stop_time = self._end_time = self._clock() 231 self._relevant_number_of_iterations_at_start = 0 232 if self.turn_off_gc: 233 self.gc_was_enabled = gc.isenabled() 234 gc.disable() 235 236 return range(self._last_tracked_number_of_iterations) 237 238 def __exit__(self, exc_type, exc_val, exc_tb): 239 self._relevant_stop_time = self._end_time = self._clock() 240 if self.turn_off_gc and self.gc_was_enabled: 241 gc.enable() 242 243 if self.suppress_exceptions: 244 return True 245 246 247class MeasureTime: 248 def __init__(self, name: str=None, iterations: int = 1, do_print: Union[bool, Callable] = True, raise_exceptions: bool = True): 249 self.name: str = name 250 self.iterations: int = 1 if iterations < 1 else iterations 251 self.do_print: Union[bool, Callable] = do_print 252 self.raise_exceptions: bool = raise_exceptions 253 self.start_time = None 254 self.stop_time = None 255 self.time_spent = None 256 self.exc_type = None 257 self.exc_value = None 258 self.exc_tb = None 259 260 def __enter__(self): 261 self.start_time = perf_counter() 262 return self 263 264 def __exit__(self, exc_type, exc_val, exc_tb): 265 self.exc_type = exc_type 266 self.exc_value = exc_val 267 self.exc_tb = exc_tb 268 269 self.stop_time = perf_counter() 270 self.time_spent = self.stop_time - self.start_time 271 if self.do_print: 272 if isinstance(self.do_print, bool): 273 if self.name is not None: 274 print(f'>>> "{self.name}"') 275 else: 276 print('>>>') 277 278 if self.exc_type is not None: 279 print(f'\t Exception: {self.exc_type}') 280 print(f'\t Exception value: {self.exc_value}') 281 print(f'\t Exception traceback: {self.exc_tb}') 282 283 if self.iterations > 1: 284 print(f'\t It was used {self.time_spent} secondss to process {self.iterations} iterations') 285 else: 286 print(f'\t It was used {self.time_spent} seconds') 287 288 if self.time_spent: 289 print(f'\t There is {self.iterations / self.time_spent} iterations/seconds') 290 print() 291 else: 292 self.do_print(self) 293 294 return not self.raise_exceptions 295 296 297class MeasureTimeTraceLine: 298 def __init__(self, name: str=None, iterations: int = 1, raise_exceptions: bool = True, measuring_class: Type = MeasureTime, 299 line_type: LineType = LineType.current_line, line_num: Optional[int] = None, output_fields: Optional[Set[OutputFields]] = None, 300 do_print: Union[bool, Callable] = True, auto_line_tracer: AutoLineTracer = None, depth: int = 1): 301 self.measuring_class: Type = measuring_class 302 self.measuring_obj: MeasureTime = self.measuring_class(name, iterations, self.printer if do_print is True else do_print, raise_exceptions) 303 self.depth: int = depth 304 self.output_fields: Set[OutputFields] = { 305 OutputFields.trace_name, 306 OutputFields.file_name, 307 OutputFields.line, 308 OutputFields.func_name, 309 OutputFields.code_line, 310 OutputFields.new_line_after_end, 311 } if output_fields is None else output_fields 312 self.line_type: LineType = line_type 313 self.line_num: Optional[int] = line_num 314 self.auto_line_tracer: AutoLineTracer = alt if auto_line_tracer is None else auto_line_tracer 315 316 def printer(self, measuring_obj: MeasureTime) -> None: 317 print('='*40) 318 if self.measuring_obj.name is not None: 319 print(f'>>> "{measuring_obj.name}"') 320 321 self.auto_line_tracer.pc('', line_type=self.line_type, line_num=self.line_num, output_fields=self.output_fields, depth=self.depth + 3) 322 323 if measuring_obj.exc_type is not None: 324 print(f'\t Exception: {measuring_obj.exc_type}') 325 print(f'\t Exception value: {measuring_obj.exc_value}') 326 print(f'\t Exception traceback: {measuring_obj.exc_tb}') 327 328 if measuring_obj.iterations > 1: 329 print(f'It was used {measuring_obj.time_spent} secondss to process {measuring_obj.iterations} iterations') 330 else: 331 print(f'It was used {measuring_obj.time_spent} seconds') 332 333 if measuring_obj.time_spent: 334 print(f'There is {measuring_obj.iterations / measuring_obj.time_spent} iterations/seconds') 335 336 print() 337 338 def __enter__(self): 339 return self.measuring_obj.__enter__() 340 341 def __exit__(self, exc_type, exc_val, exc_tb): 342 return self.measuring_obj.__exit__(exc_type, exc_val, exc_tb) 343 344 345class MeasureProcessTime: 346 def __init__(self, name: str=None, iterations: int = 1, do_print: Union[bool, Callable] = False, raise_exceptions: bool = True, depth: int = 1): 347 self.name: str = name 348 self.iterations: int = 1 if iterations < 1 else iterations 349 self.do_print: Union[bool, Callable] = do_print 350 self.raise_exceptions: bool = raise_exceptions 351 self.depth: int = depth 352 self.start_time = None 353 self.stop_time = None 354 self.time_spent = None 355 self.exc_type = None 356 self.exc_value = None 357 self.exc_tb = None 358 359 def __enter__(self): 360 self.start_time = process_time() 361 return self 362 363 def __exit__(self, exc_type, exc_val, exc_tb): 364 self.exc_type = exc_type 365 self.exc_value = exc_val 366 self.exc_tb = exc_tb 367 368 self.stop_time = process_time() 369 self.time_spent = self.stop_time - self.start_time 370 if self.do_print: 371 if isinstance(self.do_print, bool): 372 if self.name is not None: 373 print(f'>>> "{self.name}"') 374 else: 375 print('>>>') 376 377 if self.exc_type is not None: 378 print(f'\t Exception: {self.exc_type}') 379 print(f'\t Exception value: {self.exc_value}') 380 print(f'\t Exception traceback: {self.exc_tb}') 381 382 if self.iterations > 1: 383 print(f'\t It was used {self.time_spent} secondss to process {self.iterations} iterations') 384 else: 385 print(f'\t It was used {self.time_spent} seconds') 386 387 if self.time_spent: 388 print(f'\t There is {self.iterations / self.time_spent} iterations/seconds') 389 print() 390 else: 391 self.do_print(self) 392 393 return not self.raise_exceptions 394 395 396def measure_func_performance(func: Callable, 397 run_time: float, 398 name: str=None, 399 do_print: Union[bool, Callable] = False, 400 clock_type: ClockType=ClockType.perf_counter, 401 suppress_exceptions: bool=False, 402 turn_off_gc: bool=False 403 ): 404 tr = PrecisePerformanceTestTracer(run_time, clock_type, suppress_exceptions, turn_off_gc) 405 try: 406 if turn_off_gc: 407 gc_was_enabled = gc.isenabled() 408 gc.disable() 409 410 while tr.iter(): 411 func() 412 413 with tr as fast_iter: 414 for i in fast_iter: 415 func() 416 finally: 417 if turn_off_gc and gc_was_enabled: 418 gc.enable() 419 420 if do_print: 421 if isinstance(do_print, bool): 422 if name is not None: 423 print(f'>>> "{name}": {func_name(func)}()') 424 else: 425 print(f'>>> {func_name(func)}()') 426 427 print(f'\t It was used {tr.time_spent} seconds to make {tr.iterations_made} iterations. Performance: {tr.iter_per_time_unit} iterations/seconds') 428 else: 429 do_print(func, run_time, name, clock_type, tr) 430 431 432async def measure_afunc_performance(afunc: Awaitable, 433 run_time: float, 434 name: str=None, 435 do_print: Union[bool, Callable] = False, 436 clock_type: ClockType=ClockType.perf_counter, 437 suppress_exceptions: bool=False, 438 turn_off_gc: bool=False 439 ): 440 tr = PrecisePerformanceTestTracer(run_time, clock_type, suppress_exceptions, turn_off_gc) 441 try: 442 if turn_off_gc: 443 gc_was_enabled = gc.isenabled() 444 gc.disable() 445 446 while tr.iter(): 447 await afunc 448 449 with tr as fast_iter: 450 for i in fast_iter: 451 await afunc 452 finally: 453 if turn_off_gc and gc_was_enabled: 454 gc.enable() 455 456 if do_print: 457 if isinstance(do_print, bool): 458 if name is not None: 459 print(f'>>> "{name}": {func_name(afunc)}()') 460 else: 461 print(f'>>> {func_name(afunc)}()') 462 463 print(f'\t It was used {tr.time_spent} seconds to make {tr.iterations_made} iterations. Performance: {tr.iter_per_time_unit} iterations/seconds') 464 else: 465 do_print(afunc, run_time, name, clock_type, tr) 466 467 468def measure_func_isolated_performance(func: Callable, 469 run_time: float, 470 name: str=None, 471 do_print: Union[bool, Callable] = False, 472 clock_type: ClockType=ClockType.perf_counter, 473 suppress_exceptions: bool=False, 474 turn_off_gc: bool=False 475 ): 476 """Measure functions best/top/isolated performance. Hopefully isolated from an interference with other threads executed on the same CPU core 477 478 Args: 479 func (Callable): _description_ 480 run_time (float): _description_ 481 name (str, optional): _description_. Defaults to None. 482 do_print (Union[bool, Callable], optional): _description_. Defaults to False. 483 clock_type (ClockType, optional): _description_. Defaults to ClockType.perf_counter. 484 suppress_exceptions (bool, optional): _description_. Defaults to False. 485 turn_off_gc (bool, optional): _description_. Defaults to False. 486 """ 487 tr = PrecisePerformanceTestTracer(run_time, clock_type, suppress_exceptions, turn_off_gc) 488 measurements: List[float] = list() 489 try: 490 if turn_off_gc: 491 gc_was_enabled = gc.isenabled() 492 gc.disable() 493 494 while tr.iter(): 495 func() 496 497 with tr as fast_iter: 498 for i in fast_iter: 499 start_time = perf_counter() 500 func() 501 measurements.append(perf_counter() - start_time) 502 finally: 503 if turn_off_gc and gc_was_enabled: 504 gc.enable() 505 506 if do_print: 507 best_measurement: float = min(measurements) 508 best_performance: Optional[float] = (1 / best_measurement) if best_measurement else None 509 if isinstance(do_print, bool): 510 if name is not None: 511 print(f'>>> "{name}": {func_name(func)}()') 512 else: 513 print(f'>>> {func_name(func)}()') 514 515 print(f'\t It was used {tr.time_spent} seconds to make {tr.iterations_made} iterations. Performance: {tr.iter_per_time_unit} iterations/seconds') 516 print(f'\t Isolated run time: {best_measurement} seconds; Isolated performance: {best_performance} iterations/seconds') 517 else: 518 do_print(func, run_time, name, clock_type, tr, best_measurement, best_performance) 519 520 521async def measure_afunc_isolated_performance(afunc: Awaitable, 522 run_time: float, 523 name: str=None, 524 do_print: Union[bool, Callable] = False, 525 clock_type: ClockType=ClockType.perf_counter, 526 suppress_exceptions: bool=False, 527 turn_off_gc: bool=False 528 ): 529 """Measure functions best/top/isolated performance. Hopefully isolated from an interference with other threads executed on the same CPU core 530 531 Args: 532 afunc (Awaitable): _description_ 533 run_time (float): _description_ 534 name (str, optional): _description_. Defaults to None. 535 do_print (Union[bool, Callable], optional): _description_. Defaults to False. 536 clock_type (ClockType, optional): _description_. Defaults to ClockType.perf_counter. 537 suppress_exceptions (bool, optional): _description_. Defaults to False. 538 turn_off_gc (bool, optional): _description_. Defaults to False. 539 """ 540 tr = PrecisePerformanceTestTracer(run_time, clock_type, suppress_exceptions, turn_off_gc) 541 measurements: List[float] = list() 542 try: 543 if turn_off_gc: 544 gc_was_enabled = gc.isenabled() 545 gc.disable() 546 547 while tr.iter(): 548 await afunc 549 550 with tr as fast_iter: 551 for i in fast_iter: 552 start_time = perf_counter() 553 await afunc 554 measurements.append(perf_counter() - start_time) 555 finally: 556 if turn_off_gc and gc_was_enabled: 557 gc.enable() 558 559 if do_print: 560 best_measurement: float = min(measurements) 561 best_performance: Optional[float] = (1 / best_measurement) if best_measurement else None 562 if isinstance(do_print, bool): 563 if name is not None: 564 print(f'>>> "{name}": {func_name(afunc)}()') 565 else: 566 print(f'>>> {func_name(afunc)}()') 567 568 print(f'\t It was used {tr.time_spent} seconds to make {tr.iterations_made} iterations. Performance: {tr.iter_per_time_unit} iterations/seconds') 569 print(f'\t Isolated run time: {best_measurement} seconds; Isolated performance: {best_performance} iterations/seconds') 570 else: 571 do_print(afunc, run_time, name, clock_type, tr, best_measurement, best_performance)
52class PerformanceTestResult(Exception): 53 def __init__(self, result): 54 super(PerformanceTestResult, self).__init__() 55 self.result = result
Common base class for all non-exit exceptions.
Inherited Members
- builtins.BaseException
- with_traceback
- args
58@contextmanager 59def test_run_time(test_name: str, number_of_iterations: int, throw_result: bool=False, throw_result_anyway: bool=True, ignore_index=False): 60 index = ValueExistence(True, copy.copy(number_of_iterations)) 61 start_time = perf_counter() 62 exception_occures = False 63 try: 64 yield index 65 except: 66 if not throw_result_anyway: 67 exception_occures = True 68 raise 69 finally: 70 if not ignore_index: 71 number_of_iterations -= index.value 72 73 end_time = perf_counter() 74 result_time = end_time - start_time 75 if result_time > 0: 76 text_result = f'>>> "{test_name}"\n\tIt was used {result_time} seconds to process {number_of_iterations} iterations.\n\tThere is {number_of_iterations / result_time} iterations per second\n' 77 else: 78 text_result = f'>>> "{test_name}"\n\tIt was used {result_time} seconds to process {number_of_iterations} iterations.\n' 79 80 lprint(text_result) 81 82 if (not exception_occures) and throw_result: 83 result_data = dict() 84 result_data['test_name'] = test_name 85 result_data['result_time'] = result_time 86 if result_time > 0: 87 result_data['iterations_per_time_unit'] = number_of_iterations / result_time 88 else: 89 result_data['iterations_per_time_unit'] = None 90 raise PerformanceTestResult(result_data)
97def test_function_run_time( 98 test_name: str = None, 99 iterations_qnt: int = None, 100 throw_result: bool = None, 101 ): 102 """ 103 Use 'performance_test_lib__iterations_qnt=1000000' parameter to pass number of iterations 104 :param testable_function: function 105 :return: 106 """ 107 def wrapper(testable_function): 108 if is_async(testable_function): 109 async def sub_wrapper(*args, **kwargs): 110 test_name_ = '' if test_name is None else test_name 111 test_full_name = '{}: {}'.format(str(testable_function), test_name_) 112 number_of_iterations = 1 if iterations_qnt is None else iterations_qnt 113 throw_result_ = False if throw_result is None else throw_result 114 with test_run_time(test_full_name, number_of_iterations, throw_result_) as index: 115 while index.value > 0: 116 await testable_function(*args, **kwargs) 117 index.value -= 1 118 119 return sub_wrapper 120 else: 121 def sub_wrapper(*args, **kwargs): 122 test_name_ = '' if test_name is None else test_name 123 test_full_name = '{}: {}'.format(str(testable_function), test_name_) 124 number_of_iterations = 1 if iterations_qnt is None else iterations_qnt 125 throw_result_ = False if throw_result is None else throw_result 126 with test_run_time(test_full_name, number_of_iterations, throw_result_) as index: 127 while index.value > 0: 128 testable_function(*args, **kwargs) 129 index.value -= 1 130 131 return sub_wrapper 132 return wrapper
Use 'performance_test_lib__iterations_qnt=1000000' parameter to pass number of iterations :param testable_function: function :return:
135def process_performance_test_results(tracer: Tracer, test_name: str, throw_result: bool=False): 136 number_of_iterations = tracer.iterations_made 137 result_time = tracer.time_spent 138 iterations_per_time_unit = tracer.iter_per_time_unit 139 print('>>> "{}"'.format(test_name)) 140 print('\t' + 'It was used', result_time, 'seconds to process', number_of_iterations, 'iterations.') 141 print('\t' + 'There is', iterations_per_time_unit, 'iterations per second') 142 143 if throw_result: 144 result_data = (test_name, result_time, iterations_per_time_unit) 145 raise PerformanceTestResult(result_data)
148@contextmanager 149def test_performance(test_name: str, run_time: float, throw_result: bool=False, clock_type=ClockType.perf_counter): 150 tracer = Tracer(run_time, clock_type) 151 try: 152 yield tracer 153 except: 154 raise 155 finally: 156 process_performance_test_results(tracer, test_name, throw_result)
159def test_function_performance( 160 test_name: str = None, 161 run_time: RationalNumber = None, 162 throw_result: bool = None, 163 clock_type: ClockType = None, 164 ): 165 """ 166 Use 'performance_test_lib__run_time=1.5' parameter to pass number of seconds to test 167 :param testable_function: function 168 :return: 169 """ 170 def wrapper(testable_function): 171 if is_async(testable_function): 172 async def sub_wrapper(*args, **kwargs): 173 test_name_ = '' if test_name is None else test_name 174 full_test_name = '{}: {}'.format(str(testable_function), test_name_) 175 run_time_ = 1.0 if run_time is None else run_time 176 throw_result_ = False if throw_result is None else throw_result 177 clock_type_ = ClockType.perf_counter if clock_type is None else clock_type 178 with test_performance(full_test_name, run_time_, throw_result_, clock_type_) as tracer: 179 while tracer.iter(): 180 await testable_function(*args, **kwargs) 181 182 return sub_wrapper 183 else: 184 def sub_wrapper(*args, **kwargs): 185 test_name_ = '' if test_name is None else test_name 186 full_test_name = '{}: {}'.format(str(testable_function), test_name_) 187 run_time_ = 1.0 if run_time is None else run_time 188 throw_result_ = False if throw_result is None else throw_result 189 clock_type_ = ClockType.perf_counter if clock_type is None else clock_type 190 with test_performance(full_test_name, run_time_, throw_result_, clock_type_) as tracer: 191 while tracer.iter(): 192 testable_function(*args, **kwargs) 193 194 return sub_wrapper 195 return wrapper
Use 'performance_test_lib__run_time=1.5' parameter to pass number of seconds to test :param testable_function: function :return:
198class PrecisePerformanceTestTracer(Tracer): 199 """ 200 Precise tracer. 201 At first you need to use it as a usual Tracer. After tracing was done - use it as a fast `for i in range(...)` block 202 203 Example of use: 204 205 tr = PrecisePerformanceTestTracer(10.0) 206 while tr.iter(): 207 i = '456' 208 k = int('1243' + i) 209 210 with tr as fast_iter: 211 for i in fast_iter: 212 i = '456' 213 k = int('1243' + i) 214 215 print('{} iter/s; {} seconds; {} iterations'.format(tr.iter_per_time_unit, tr.time_spent, tr.iterations_made)) 216 217 """ 218 219 def __init__(self, 220 run_time: float, 221 clock_type: ClockType=ClockType.perf_counter, 222 suppress_exceptions: bool=False, 223 turn_off_gc: bool=False 224 ): 225 super().__init__(run_time, clock_type) 226 self.suppress_exceptions = suppress_exceptions 227 self.turn_off_gc = turn_off_gc 228 self.gc_was_enabled = None 229 230 def __enter__(self): 231 self._relevant_start_time = self._start_time = self._relevant_stop_time = self._end_time = self._clock() 232 self._relevant_number_of_iterations_at_start = 0 233 if self.turn_off_gc: 234 self.gc_was_enabled = gc.isenabled() 235 gc.disable() 236 237 return range(self._last_tracked_number_of_iterations) 238 239 def __exit__(self, exc_type, exc_val, exc_tb): 240 self._relevant_stop_time = self._end_time = self._clock() 241 if self.turn_off_gc and self.gc_was_enabled: 242 gc.enable() 243 244 if self.suppress_exceptions: 245 return True
Precise tracer.
At first you need to use it as a usual Tracer. After tracing was done - use it as a fast for i in range(...)
block
Example of use:
tr = PrecisePerformanceTestTracer(10.0)
while tr.iter():
i = '456'
k = int('1243' + i)
with tr as fast_iter:
for i in fast_iter:
i = '456'
k = int('1243' + i)
print('{} iter/s; {} seconds; {} iterations'.format(tr.iter_per_time_unit, tr.time_spent, tr.iterations_made))
219 def __init__(self, 220 run_time: float, 221 clock_type: ClockType=ClockType.perf_counter, 222 suppress_exceptions: bool=False, 223 turn_off_gc: bool=False 224 ): 225 super().__init__(run_time, clock_type) 226 self.suppress_exceptions = suppress_exceptions 227 self.turn_off_gc = turn_off_gc 228 self.gc_was_enabled = None
Inherited Members
- cengal.time_management.repeat_for_a_time.versions.v_0.repeat_for_a_time__cython.Tracer
- iter
- iter_per_time_unit
- cengal.time_management.repeat_for_a_time.versions.v_0.repeat_for_a_time__cython.BaseTracer
- iterations_made
- total_number_of_iterations_made
- time_spent
- total_amount_of_time_spent
- clock_type
248class MeasureTime: 249 def __init__(self, name: str=None, iterations: int = 1, do_print: Union[bool, Callable] = True, raise_exceptions: bool = True): 250 self.name: str = name 251 self.iterations: int = 1 if iterations < 1 else iterations 252 self.do_print: Union[bool, Callable] = do_print 253 self.raise_exceptions: bool = raise_exceptions 254 self.start_time = None 255 self.stop_time = None 256 self.time_spent = None 257 self.exc_type = None 258 self.exc_value = None 259 self.exc_tb = None 260 261 def __enter__(self): 262 self.start_time = perf_counter() 263 return self 264 265 def __exit__(self, exc_type, exc_val, exc_tb): 266 self.exc_type = exc_type 267 self.exc_value = exc_val 268 self.exc_tb = exc_tb 269 270 self.stop_time = perf_counter() 271 self.time_spent = self.stop_time - self.start_time 272 if self.do_print: 273 if isinstance(self.do_print, bool): 274 if self.name is not None: 275 print(f'>>> "{self.name}"') 276 else: 277 print('>>>') 278 279 if self.exc_type is not None: 280 print(f'\t Exception: {self.exc_type}') 281 print(f'\t Exception value: {self.exc_value}') 282 print(f'\t Exception traceback: {self.exc_tb}') 283 284 if self.iterations > 1: 285 print(f'\t It was used {self.time_spent} secondss to process {self.iterations} iterations') 286 else: 287 print(f'\t It was used {self.time_spent} seconds') 288 289 if self.time_spent: 290 print(f'\t There is {self.iterations / self.time_spent} iterations/seconds') 291 print() 292 else: 293 self.do_print(self) 294 295 return not self.raise_exceptions
249 def __init__(self, name: str=None, iterations: int = 1, do_print: Union[bool, Callable] = True, raise_exceptions: bool = True): 250 self.name: str = name 251 self.iterations: int = 1 if iterations < 1 else iterations 252 self.do_print: Union[bool, Callable] = do_print 253 self.raise_exceptions: bool = raise_exceptions 254 self.start_time = None 255 self.stop_time = None 256 self.time_spent = None 257 self.exc_type = None 258 self.exc_value = None 259 self.exc_tb = None
298class MeasureTimeTraceLine: 299 def __init__(self, name: str=None, iterations: int = 1, raise_exceptions: bool = True, measuring_class: Type = MeasureTime, 300 line_type: LineType = LineType.current_line, line_num: Optional[int] = None, output_fields: Optional[Set[OutputFields]] = None, 301 do_print: Union[bool, Callable] = True, auto_line_tracer: AutoLineTracer = None, depth: int = 1): 302 self.measuring_class: Type = measuring_class 303 self.measuring_obj: MeasureTime = self.measuring_class(name, iterations, self.printer if do_print is True else do_print, raise_exceptions) 304 self.depth: int = depth 305 self.output_fields: Set[OutputFields] = { 306 OutputFields.trace_name, 307 OutputFields.file_name, 308 OutputFields.line, 309 OutputFields.func_name, 310 OutputFields.code_line, 311 OutputFields.new_line_after_end, 312 } if output_fields is None else output_fields 313 self.line_type: LineType = line_type 314 self.line_num: Optional[int] = line_num 315 self.auto_line_tracer: AutoLineTracer = alt if auto_line_tracer is None else auto_line_tracer 316 317 def printer(self, measuring_obj: MeasureTime) -> None: 318 print('='*40) 319 if self.measuring_obj.name is not None: 320 print(f'>>> "{measuring_obj.name}"') 321 322 self.auto_line_tracer.pc('', line_type=self.line_type, line_num=self.line_num, output_fields=self.output_fields, depth=self.depth + 3) 323 324 if measuring_obj.exc_type is not None: 325 print(f'\t Exception: {measuring_obj.exc_type}') 326 print(f'\t Exception value: {measuring_obj.exc_value}') 327 print(f'\t Exception traceback: {measuring_obj.exc_tb}') 328 329 if measuring_obj.iterations > 1: 330 print(f'It was used {measuring_obj.time_spent} secondss to process {measuring_obj.iterations} iterations') 331 else: 332 print(f'It was used {measuring_obj.time_spent} seconds') 333 334 if measuring_obj.time_spent: 335 print(f'There is {measuring_obj.iterations / measuring_obj.time_spent} iterations/seconds') 336 337 print() 338 339 def __enter__(self): 340 return self.measuring_obj.__enter__() 341 342 def __exit__(self, exc_type, exc_val, exc_tb): 343 return self.measuring_obj.__exit__(exc_type, exc_val, exc_tb)
299 def __init__(self, name: str=None, iterations: int = 1, raise_exceptions: bool = True, measuring_class: Type = MeasureTime, 300 line_type: LineType = LineType.current_line, line_num: Optional[int] = None, output_fields: Optional[Set[OutputFields]] = None, 301 do_print: Union[bool, Callable] = True, auto_line_tracer: AutoLineTracer = None, depth: int = 1): 302 self.measuring_class: Type = measuring_class 303 self.measuring_obj: MeasureTime = self.measuring_class(name, iterations, self.printer if do_print is True else do_print, raise_exceptions) 304 self.depth: int = depth 305 self.output_fields: Set[OutputFields] = { 306 OutputFields.trace_name, 307 OutputFields.file_name, 308 OutputFields.line, 309 OutputFields.func_name, 310 OutputFields.code_line, 311 OutputFields.new_line_after_end, 312 } if output_fields is None else output_fields 313 self.line_type: LineType = line_type 314 self.line_num: Optional[int] = line_num 315 self.auto_line_tracer: AutoLineTracer = alt if auto_line_tracer is None else auto_line_tracer
317 def printer(self, measuring_obj: MeasureTime) -> None: 318 print('='*40) 319 if self.measuring_obj.name is not None: 320 print(f'>>> "{measuring_obj.name}"') 321 322 self.auto_line_tracer.pc('', line_type=self.line_type, line_num=self.line_num, output_fields=self.output_fields, depth=self.depth + 3) 323 324 if measuring_obj.exc_type is not None: 325 print(f'\t Exception: {measuring_obj.exc_type}') 326 print(f'\t Exception value: {measuring_obj.exc_value}') 327 print(f'\t Exception traceback: {measuring_obj.exc_tb}') 328 329 if measuring_obj.iterations > 1: 330 print(f'It was used {measuring_obj.time_spent} secondss to process {measuring_obj.iterations} iterations') 331 else: 332 print(f'It was used {measuring_obj.time_spent} seconds') 333 334 if measuring_obj.time_spent: 335 print(f'There is {measuring_obj.iterations / measuring_obj.time_spent} iterations/seconds') 336 337 print()
346class MeasureProcessTime: 347 def __init__(self, name: str=None, iterations: int = 1, do_print: Union[bool, Callable] = False, raise_exceptions: bool = True, depth: int = 1): 348 self.name: str = name 349 self.iterations: int = 1 if iterations < 1 else iterations 350 self.do_print: Union[bool, Callable] = do_print 351 self.raise_exceptions: bool = raise_exceptions 352 self.depth: int = depth 353 self.start_time = None 354 self.stop_time = None 355 self.time_spent = None 356 self.exc_type = None 357 self.exc_value = None 358 self.exc_tb = None 359 360 def __enter__(self): 361 self.start_time = process_time() 362 return self 363 364 def __exit__(self, exc_type, exc_val, exc_tb): 365 self.exc_type = exc_type 366 self.exc_value = exc_val 367 self.exc_tb = exc_tb 368 369 self.stop_time = process_time() 370 self.time_spent = self.stop_time - self.start_time 371 if self.do_print: 372 if isinstance(self.do_print, bool): 373 if self.name is not None: 374 print(f'>>> "{self.name}"') 375 else: 376 print('>>>') 377 378 if self.exc_type is not None: 379 print(f'\t Exception: {self.exc_type}') 380 print(f'\t Exception value: {self.exc_value}') 381 print(f'\t Exception traceback: {self.exc_tb}') 382 383 if self.iterations > 1: 384 print(f'\t It was used {self.time_spent} secondss to process {self.iterations} iterations') 385 else: 386 print(f'\t It was used {self.time_spent} seconds') 387 388 if self.time_spent: 389 print(f'\t There is {self.iterations / self.time_spent} iterations/seconds') 390 print() 391 else: 392 self.do_print(self) 393 394 return not self.raise_exceptions
347 def __init__(self, name: str=None, iterations: int = 1, do_print: Union[bool, Callable] = False, raise_exceptions: bool = True, depth: int = 1): 348 self.name: str = name 349 self.iterations: int = 1 if iterations < 1 else iterations 350 self.do_print: Union[bool, Callable] = do_print 351 self.raise_exceptions: bool = raise_exceptions 352 self.depth: int = depth 353 self.start_time = None 354 self.stop_time = None 355 self.time_spent = None 356 self.exc_type = None 357 self.exc_value = None 358 self.exc_tb = None
397def measure_func_performance(func: Callable, 398 run_time: float, 399 name: str=None, 400 do_print: Union[bool, Callable] = False, 401 clock_type: ClockType=ClockType.perf_counter, 402 suppress_exceptions: bool=False, 403 turn_off_gc: bool=False 404 ): 405 tr = PrecisePerformanceTestTracer(run_time, clock_type, suppress_exceptions, turn_off_gc) 406 try: 407 if turn_off_gc: 408 gc_was_enabled = gc.isenabled() 409 gc.disable() 410 411 while tr.iter(): 412 func() 413 414 with tr as fast_iter: 415 for i in fast_iter: 416 func() 417 finally: 418 if turn_off_gc and gc_was_enabled: 419 gc.enable() 420 421 if do_print: 422 if isinstance(do_print, bool): 423 if name is not None: 424 print(f'>>> "{name}": {func_name(func)}()') 425 else: 426 print(f'>>> {func_name(func)}()') 427 428 print(f'\t It was used {tr.time_spent} seconds to make {tr.iterations_made} iterations. Performance: {tr.iter_per_time_unit} iterations/seconds') 429 else: 430 do_print(func, run_time, name, clock_type, tr)
433async def measure_afunc_performance(afunc: Awaitable, 434 run_time: float, 435 name: str=None, 436 do_print: Union[bool, Callable] = False, 437 clock_type: ClockType=ClockType.perf_counter, 438 suppress_exceptions: bool=False, 439 turn_off_gc: bool=False 440 ): 441 tr = PrecisePerformanceTestTracer(run_time, clock_type, suppress_exceptions, turn_off_gc) 442 try: 443 if turn_off_gc: 444 gc_was_enabled = gc.isenabled() 445 gc.disable() 446 447 while tr.iter(): 448 await afunc 449 450 with tr as fast_iter: 451 for i in fast_iter: 452 await afunc 453 finally: 454 if turn_off_gc and gc_was_enabled: 455 gc.enable() 456 457 if do_print: 458 if isinstance(do_print, bool): 459 if name is not None: 460 print(f'>>> "{name}": {func_name(afunc)}()') 461 else: 462 print(f'>>> {func_name(afunc)}()') 463 464 print(f'\t It was used {tr.time_spent} seconds to make {tr.iterations_made} iterations. Performance: {tr.iter_per_time_unit} iterations/seconds') 465 else: 466 do_print(afunc, run_time, name, clock_type, tr)
469def measure_func_isolated_performance(func: Callable, 470 run_time: float, 471 name: str=None, 472 do_print: Union[bool, Callable] = False, 473 clock_type: ClockType=ClockType.perf_counter, 474 suppress_exceptions: bool=False, 475 turn_off_gc: bool=False 476 ): 477 """Measure functions best/top/isolated performance. Hopefully isolated from an interference with other threads executed on the same CPU core 478 479 Args: 480 func (Callable): _description_ 481 run_time (float): _description_ 482 name (str, optional): _description_. Defaults to None. 483 do_print (Union[bool, Callable], optional): _description_. Defaults to False. 484 clock_type (ClockType, optional): _description_. Defaults to ClockType.perf_counter. 485 suppress_exceptions (bool, optional): _description_. Defaults to False. 486 turn_off_gc (bool, optional): _description_. Defaults to False. 487 """ 488 tr = PrecisePerformanceTestTracer(run_time, clock_type, suppress_exceptions, turn_off_gc) 489 measurements: List[float] = list() 490 try: 491 if turn_off_gc: 492 gc_was_enabled = gc.isenabled() 493 gc.disable() 494 495 while tr.iter(): 496 func() 497 498 with tr as fast_iter: 499 for i in fast_iter: 500 start_time = perf_counter() 501 func() 502 measurements.append(perf_counter() - start_time) 503 finally: 504 if turn_off_gc and gc_was_enabled: 505 gc.enable() 506 507 if do_print: 508 best_measurement: float = min(measurements) 509 best_performance: Optional[float] = (1 / best_measurement) if best_measurement else None 510 if isinstance(do_print, bool): 511 if name is not None: 512 print(f'>>> "{name}": {func_name(func)}()') 513 else: 514 print(f'>>> {func_name(func)}()') 515 516 print(f'\t It was used {tr.time_spent} seconds to make {tr.iterations_made} iterations. Performance: {tr.iter_per_time_unit} iterations/seconds') 517 print(f'\t Isolated run time: {best_measurement} seconds; Isolated performance: {best_performance} iterations/seconds') 518 else: 519 do_print(func, run_time, name, clock_type, tr, best_measurement, best_performance)
Measure functions best/top/isolated performance. Hopefully isolated from an interference with other threads executed on the same CPU core
Args: func (Callable): _description_ run_time (float): _description_ name (str, optional): _description_. Defaults to None. do_print (Union[bool, Callable], optional): _description_. Defaults to False. clock_type (ClockType, optional): _description_. Defaults to ClockType.perf_counter. suppress_exceptions (bool, optional): _description_. Defaults to False. turn_off_gc (bool, optional): _description_. Defaults to False.
522async def measure_afunc_isolated_performance(afunc: Awaitable, 523 run_time: float, 524 name: str=None, 525 do_print: Union[bool, Callable] = False, 526 clock_type: ClockType=ClockType.perf_counter, 527 suppress_exceptions: bool=False, 528 turn_off_gc: bool=False 529 ): 530 """Measure functions best/top/isolated performance. Hopefully isolated from an interference with other threads executed on the same CPU core 531 532 Args: 533 afunc (Awaitable): _description_ 534 run_time (float): _description_ 535 name (str, optional): _description_. Defaults to None. 536 do_print (Union[bool, Callable], optional): _description_. Defaults to False. 537 clock_type (ClockType, optional): _description_. Defaults to ClockType.perf_counter. 538 suppress_exceptions (bool, optional): _description_. Defaults to False. 539 turn_off_gc (bool, optional): _description_. Defaults to False. 540 """ 541 tr = PrecisePerformanceTestTracer(run_time, clock_type, suppress_exceptions, turn_off_gc) 542 measurements: List[float] = list() 543 try: 544 if turn_off_gc: 545 gc_was_enabled = gc.isenabled() 546 gc.disable() 547 548 while tr.iter(): 549 await afunc 550 551 with tr as fast_iter: 552 for i in fast_iter: 553 start_time = perf_counter() 554 await afunc 555 measurements.append(perf_counter() - start_time) 556 finally: 557 if turn_off_gc and gc_was_enabled: 558 gc.enable() 559 560 if do_print: 561 best_measurement: float = min(measurements) 562 best_performance: Optional[float] = (1 / best_measurement) if best_measurement else None 563 if isinstance(do_print, bool): 564 if name is not None: 565 print(f'>>> "{name}": {func_name(afunc)}()') 566 else: 567 print(f'>>> {func_name(afunc)}()') 568 569 print(f'\t It was used {tr.time_spent} seconds to make {tr.iterations_made} iterations. Performance: {tr.iter_per_time_unit} iterations/seconds') 570 print(f'\t Isolated run time: {best_measurement} seconds; Isolated performance: {best_performance} iterations/seconds') 571 else: 572 do_print(afunc, run_time, name, clock_type, tr, best_measurement, best_performance)
Measure functions best/top/isolated performance. Hopefully isolated from an interference with other threads executed on the same CPU core
Args: afunc (Awaitable): _description_ run_time (float): _description_ name (str, optional): _description_. Defaults to None. do_print (Union[bool, Callable], optional): _description_. Defaults to False. clock_type (ClockType, optional): _description_. Defaults to ClockType.perf_counter. suppress_exceptions (bool, optional): _description_. Defaults to False. turn_off_gc (bool, optional): _description_. Defaults to False.