Method _ensure_health_check_loop

redis/background.py:198–256

Ensure the shared health check loop and thread are running. Args: timeout: Maximum seconds to wait for the loop to start. Raises: RuntimeError: If the loop fails to start within the timeout.

(self, timeout: float = 5.0)

        future.add_done_callback(on_complete)

    def _ensure_health_check_loop(self, timeout: float = 5.0):
        """
        Ensure the shared health check loop and thread are running.

        Args:
            timeout: Maximum seconds to wait for the loop to start.

        Raises:
            RuntimeError: If the loop fails to start within the timeout.
        """
        # Fast path: if loop is already running, return immediately
        if self._health_check_loop_ready.is_set():
            with self._lock:
                if (
                    self._health_check_loop is not None
                    and self._health_check_loop.is_running()
                ):
                    return

        with self._lock:
            # Double-check after acquiring the lock
            if (
                self._health_check_loop is not None
                and self._health_check_loop.is_running()
            ):
                return

            # Clear the event - we're about to start a new loop
            self._health_check_loop_ready.clear()

            # Create a new event loop for health checks
            self._health_check_loop = asyncio.new_event_loop()
            self._event_loops.append(self._health_check_loop)

            # Start the loop in a background thread
            self._health_check_thread = threading.Thread(
                target=self._run_health_check_loop,
                daemon=True,
            )
            self._health_check_thread.start()

            # Wait for loop to be running INSIDE the lock with a timeout.
            # This prevents other threads from trying to create another loop
            # before this one is fully started, while avoiding permanent deadlock
            # if the background thread fails to start the loop.
            if not self._health_check_loop_ready.wait(timeout=timeout):
                # Timeout expired - the loop failed to start
                # Clean up the failed loop to allow retry
                failed_loop = self._health_check_loop
                self._health_check_loop = None
                if failed_loop in self._event_loops:
                    self._event_loops.remove(failed_loop)
                try:
                    failed_loop.close()
                except Exception:
                    pass
                raise RuntimeError(
                    f"Health check event loop failed to start within {timeout} seconds"