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)
class PerformanceTestResult(builtins.Exception):
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.

PerformanceTestResult(result)
53    def __init__(self, result):
54        super(PerformanceTestResult, self).__init__()
55        self.result = result
result
Inherited Members
builtins.BaseException
with_traceback
args
@contextmanager
def test_run_time( test_name: str, number_of_iterations: int, throw_result: bool = False, throw_result_anyway: bool = True, ignore_index=False):
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)
def measure_time(test_name: str = ''):
93def measure_time(test_name: str = str()):
94    return test_run_time(test_name, 1, ignore_index=True)
def test_function_run_time( test_name: str = None, iterations_qnt: int = None, throw_result: bool = None):
 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:

def process_performance_test_results( tracer: cengal.time_management.repeat_for_a_time.versions.v_0.repeat_for_a_time__cython.Tracer, test_name: str, throw_result: bool = False):
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)
@contextmanager
def test_performance( test_name: str, run_time: float, throw_result: bool = False, clock_type=<ClockType.perf_counter: 2>):
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)
def test_function_performance( test_name: str = None, run_time: Union[int, float] = None, throw_result: bool = None, clock_type: cengal.time_management.repeat_for_a_time.versions.v_0.repeat_for_a_time__cython.ClockType = None):
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:

class PrecisePerformanceTestTracer(cengal.time_management.repeat_for_a_time.versions.v_0.repeat_for_a_time__cython.Tracer):
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))
PrecisePerformanceTestTracer( run_time: float, clock_type: cengal.time_management.repeat_for_a_time.versions.v_0.repeat_for_a_time__cython.ClockType = <ClockType.perf_counter: 2>, suppress_exceptions: bool = False, turn_off_gc: bool = False)
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
suppress_exceptions
turn_off_gc
gc_was_enabled
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
class MeasureTime:
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
MeasureTime( name: str = None, iterations: int = 1, do_print: Union[bool, Callable] = True, raise_exceptions: bool = True)
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
name: str
iterations: int
do_print: Union[bool, Callable]
raise_exceptions: bool
start_time
stop_time
time_spent
exc_type
exc_value
exc_tb
class MeasureTimeTraceLine:
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)
MeasureTimeTraceLine( name: str = None, iterations: int = 1, raise_exceptions: bool = True, measuring_class: Type = <class 'MeasureTime'>, line_type: cengal.code_inspection.auto_line_tracer.versions.v_0.auto_line_tracer.LineType = <LineType.current_line: 1>, line_num: Union[int, NoneType] = None, output_fields: Union[Set[cengal.code_inspection.auto_line_tracer.versions.v_0.auto_line_tracer.OutputFields], NoneType] = None, do_print: Union[bool, Callable] = True, auto_line_tracer: cengal.code_inspection.auto_line_tracer.versions.v_0.auto_line_tracer.AutoLineTracer = None, depth: int = 1)
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
measuring_class: Type
measuring_obj: MeasureTime
depth: int
output_fields: Set[cengal.code_inspection.auto_line_tracer.versions.v_0.auto_line_tracer.OutputFields]
line_type: cengal.code_inspection.auto_line_tracer.versions.v_0.auto_line_tracer.LineType
line_num: Union[int, NoneType]
auto_line_tracer: cengal.code_inspection.auto_line_tracer.versions.v_0.auto_line_tracer.AutoLineTracer
def printer( self, measuring_obj: MeasureTime) -> None:
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()
class MeasureProcessTime:
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
MeasureProcessTime( name: str = None, iterations: int = 1, do_print: Union[bool, Callable] = False, raise_exceptions: bool = True, depth: int = 1)
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
name: str
iterations: int
do_print: Union[bool, Callable]
raise_exceptions: bool
depth: int
start_time
stop_time
time_spent
exc_type
exc_value
exc_tb
def measure_func_performance( func: Callable, run_time: float, name: str = None, do_print: Union[bool, Callable] = False, clock_type: cengal.time_management.repeat_for_a_time.versions.v_0.repeat_for_a_time__cython.ClockType = <ClockType.perf_counter: 2>, suppress_exceptions: bool = False, turn_off_gc: bool = False):
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)
async def measure_afunc_performance( afunc: Awaitable, run_time: float, name: str = None, do_print: Union[bool, Callable] = False, clock_type: cengal.time_management.repeat_for_a_time.versions.v_0.repeat_for_a_time__cython.ClockType = <ClockType.perf_counter: 2>, suppress_exceptions: bool = False, turn_off_gc: bool = False):
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)
def measure_func_isolated_performance( func: Callable, run_time: float, name: str = None, do_print: Union[bool, Callable] = False, clock_type: cengal.time_management.repeat_for_a_time.versions.v_0.repeat_for_a_time__cython.ClockType = <ClockType.perf_counter: 2>, suppress_exceptions: bool = False, turn_off_gc: bool = False):
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.

async def measure_afunc_isolated_performance( afunc: Awaitable, run_time: float, name: str = None, do_print: Union[bool, Callable] = False, clock_type: cengal.time_management.repeat_for_a_time.versions.v_0.repeat_for_a_time__cython.ClockType = <ClockType.perf_counter: 2>, suppress_exceptions: bool = False, turn_off_gc: bool = 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.