Method drain_notifications

benches/memory_usage.py:368–378

Read and discard notifications for *seconds*.

(self, seconds: float)

        raise TimeoutError(f"Indexing did not finish within {timeout}s")

    def drain_notifications(self, seconds: float) -> None:
        """Read and discard notifications for *seconds*."""
        deadline = time.monotonic() + seconds
        while time.monotonic() < deadline:
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                break
            try:
                self._notifications.get(timeout=min(remaining, 0.5))
            except queue.Empty:
                pass

    def shutdown(self) -> None:
        """Send shutdown + exit."""