Coverage for adhoc-cicd-odoo-odoo / odoo / tools / gc.py: 68%

1""" Garbage collector tools 

2 

3## Reference 

4https://github.com/python/cpython/blob/main/InternalDocs/garbage_collector.md 

5 

6## TLDR cpython 

7 

8Objects have reference counts, but we need garbage collection for cyclic 

9references. All allocated objects are split into collections (aka generations). 

10There is also one permanent generation that is never collected (see 

11``gc.freeze``). 

12 

13The GC is triggered by the number of created objects. For the first collection, 

14at every allocation and deallocation, a counter is respectively increased and 

15decreased. Once it reaches a threshold, that collection is automatically 

16collected. 

17Before 3.14, other thresolds indicate that every X collections, the next 

18collection is collected. Since the, there is only one additional collection 

19which is collected inrementally; `1 / threshold1` percent of the heap is 

20collected. 

21""" 

22import contextlib 

23import gc 

24import logging 

25from time import thread_time_ns as _gc_time 

26 

27_logger = logging.getLogger('gc') 

28_gc_start: int = 0 

29_gc_init_stats = gc.get_stats() 

30_gc_timings = [0, 0, 0] 

31 

32 

33def _to_ms(ns): 

34 return round(ns / 1_000_000, 2) 

35 

36 

37def _timing_gc_callback(event, info): 

38 """Called before and after each run of the gc, see gc_set_timing.""" 

39 global _gc_start # noqa: PLW0603 

40 gen = info['generation'] 

41 if event == 'start': 

42 _gc_start = _gc_time() 

43 # python 3.14; gen2 is only collected when calling gc.collect() manually 

44 if gen == 2 and _logger.isEnabledFor(logging.DEBUG): 44 ↛ 45line 44 didn't jump to line 45 because the condition on line 44 was never true

45 _logger.debug("info %s, starting collection of gen2", gc_info()) 

46 else: 

47 timing = _gc_time() - _gc_start 

48 _gc_timings[gen] += timing 

49 _gc_start = 0 

50 if gen > 0: 

51 _logger.debug("collected %s in %.2fms", info, _to_ms(timing)) 

52 

53 

54def gc_set_timing(*, enable: bool): 

55 """Enable or disable timing callback. 

56 

57 This collects information about how much time is spent by the GC. 

58 It logs GC times (at debug level) for collections bigger than 0. 

59 The overhead is under a microsecond. 

60 """ 

61 if _timing_gc_callback in gc.callbacks: 61 ↛ 62line 61 didn't jump to line 62 because the condition on line 61 was never true

62 if enable: 

63 return 

64 gc.callbacks.remove(_timing_gc_callback) 

65 elif enable: 65 ↛ exitline 65 didn't return from function 'gc_set_timing' because the condition on line 65 was always true

66 global _gc_init_stats, _gc_timings # noqa: PLW0603 

67 _gc_init_stats = gc.get_stats() 

68 _gc_timings = [0, 0, 0] 

69 gc.callbacks.append(_timing_gc_callback) 

70 

71 

72def gc_info(): 

73 """Return a dict with stats about the garbage collector.""" 

74 stats = gc.get_stats() 

75 times = [] 

76 cumulative_time = sum(_gc_timings) or 1 

77 for info, info_init, time in zip(stats, _gc_init_stats, _gc_timings): 

78 count = info['collections'] - info_init['collections'] 

79 times.append({ 

80 'avg_time': time // count if count > 0 else 0, 

81 'time': _to_ms(time), 

82 'pct': round(time / cumulative_time, 3) 

83 }) 

84 return { 

85 'cumulative_time': _to_ms(cumulative_time), 

86 'time': times if _timing_gc_callback in gc.callbacks else (), 

87 'count': stats, 

88 'thresholds': (gc.get_count(), gc.get_threshold()), 

89 } 

90 

91 

92@contextlib.contextmanager 

93def disabling_gc(): 

94 """Disable gc in the context manager.""" 

95 if not gc.isenabled(): 95 ↛ 96line 95 didn't jump to line 96 because the condition on line 95 was never true

96 yield False 

97 return 

98 gc.disable() 

99 _logger.debug('disabled, counts %s', gc.get_count()) 

100 yield True 

101 counts = gc.get_count() 

102 gc.enable() 

103 _logger.debug('enabled, counts %s', counts)