modification_waiter.py (1615B)
1 # Copyright 2024 The Chromium Authors 2 # Use of this source code is governed by a BSD-style license that can be 3 # found in the LICENSE file. 4 """ An AbstractContextManager to wait the modifications to finish during exit. 5 """ 6 7 import os 8 import time 9 from contextlib import AbstractContextManager 10 11 12 class ModificationWaiter(AbstractContextManager): 13 """ Exits if there is no modifications for a certain time period, or the 14 timeout has been reached. """ 15 16 def __init__(self, path: str) -> None: 17 self._path = path 18 # Waits at most 60 seconds. 19 self._timeout = 60 20 # Exits early if no modification happened during last 5 seconds. 21 self._quiet_time = 5 22 23 def __enter__(self) -> None: 24 # Do nothing, the logic happens in __exit__ 25 return 26 27 def __exit__(self, exc_type, exc_value, traceback) -> bool: 28 # The default log.dir is /tmp and it's not a good idea to monitor it. 29 if not self._path: 30 return False 31 # Always consider the last modification happening now to avoid an 32 # unexpected early return. 33 last_mod_time = time.time() 34 start_time = last_mod_time 35 while True: 36 cur_time = time.time() 37 if cur_time - start_time >= self._timeout: 38 break 39 cur_mod_time = os.path.getmtime(self._path) 40 if cur_mod_time > last_mod_time: 41 last_mod_time = cur_mod_time 42 elif cur_time - last_mod_time >= self._quiet_time: 43 break 44 time.sleep(1) 45 46 # Do not suppress exceptions. 47 return False