Wait until the process isn't working very hard. This will compare wall clock time with process time. If the process time is not advancing at the same rate as wall clock time it means the process is idle (i.e. sleeping or waiting for input). When the process is idle it suggests that
(
min_sleep: float = SLEEP_GRANULARITY, max_sleep: float = 1
)
| 6 | |
| 7 | |
| 8 | async def wait_for_idle( |
| 9 | min_sleep: float = SLEEP_GRANULARITY, max_sleep: float = 1 |
| 10 | ) -> None: |
| 11 | """Wait until the process isn't working very hard. |
| 12 | |
| 13 | This will compare wall clock time with process time. If the process time |
| 14 | is not advancing at the same rate as wall clock time it means the process is |
| 15 | idle (i.e. sleeping or waiting for input). |
| 16 | |
| 17 | When the process is idle it suggests that input has been processed and the state |
| 18 | is predictable enough to test. |
| 19 | |
| 20 | Args: |
| 21 | min_sleep: Minimum time to wait. |
| 22 | max_sleep: Maximum time to wait. |
| 23 | """ |
| 24 | start_time = monotonic() |
| 25 | |
| 26 | while True: |
| 27 | cpu_time = process_time() |
| 28 | # Sleep for a predetermined amount of time |
| 29 | await sleep(SLEEP_GRANULARITY) |
| 30 | # Calculate the wall clock elapsed time and the process elapsed time |
| 31 | cpu_elapsed = process_time() - cpu_time |
| 32 | elapsed_time = monotonic() - start_time |
| 33 | |
| 34 | # If we have slept the maximum, we can break |
| 35 | if elapsed_time >= max_sleep: |
| 36 | break |
| 37 | |
| 38 | # If we have slept at least the minimum and the cpu elapsed is significantly less |
| 39 | # than wall clock, then we can assume the process has finished working for now |
| 40 | if elapsed_time > min_sleep and cpu_elapsed < SLEEP_IDLE: |
| 41 | break |
searching dependent graphs…