idle detection

Author: luke-beamcloud

fastloop/config.default.yaml

@@ -4,8 +4,6 @@ prettyPrintLogs: true
 loopDelayS: 0.1
 ssePollIntervalS: 0.1
 sseKeepAliveS: 10.0
-shutdownIdle: true
-maxIdleCycles: 10
 shutdownTimeoutS: 10.0
 cors:
   enabled: true

fastloop/constants.py

@@ -8,3 +8,4 @@
 LEASE_TTL_S = 30
 LEASE_HEARTBEAT_INTERVAL_S = 10
 MAX_EVENT_HISTORY = 1000
+MEANINGFUL_WORK_THRESHOLD_S = 0.01

fastloop/context.py

@@ -53,13 +53,18 @@ def __init__(
         self.initial_event: LoopEvent | None = initial_event
         self.state_manager: StateManager = state_manager
         self.event_this_cycle: bool = False
+        self._wait_time_this_cycle: float = 0.0
 
         integrations = integrations or []
         self.integrations: dict[str, Integration] = {i.type(): i for i in integrations}
         self.integration_events: dict[str, list[Any]] = {
             i.type(): i.events() for i in integrations
         }
 
+    def _reset_cycle_tracking(self) -> None:
+        self.event_this_cycle = False
+        self._wait_time_this_cycle = 0.0
+
     def stop(self):
         self._stop_requested = True
         raise LoopStoppedError()
@@ -132,7 +137,6 @@ async def wait_for(
                 if self.should_stop:
                     raise LoopStoppedError()
 
-                # Try to get event immediately
                 event_result = await self.state_manager.pop_event(
                     self.loop_id,
                     event,  # type: ignore
@@ -142,18 +146,18 @@ async def wait_for(
                     self.event_this_cycle = True
                     return cast(E, event_result)  # noqa
 
-                # Wait for notification or timeout
                 remaining_timeout = timeout - (asyncio.get_event_loop().time() - start)
                 if remaining_timeout <= 0:
                     break
 
-                # Wait for event notification or poll interval
                 poll_timeout = min(
                     EVENT_POLL_INTERVAL_S, remaining_timeout or EVENT_POLL_INTERVAL_S
                 )
+                wait_start = time.monotonic()
                 await self.state_manager.wait_for_event_notification(
                     pubsub, timeout=poll_timeout
                 )
+                self._wait_time_this_cycle += time.monotonic() - wait_start
 
         finally:
             if pubsub is not None:

fastloop/fastloop.py

@@ -210,8 +210,14 @@ def loop(
         on_stop: Callable[..., Any] | None = None,
         integrations: list[Integration] | None = None,
         stop_on_disconnect: bool = False,
+        stop_after_idle_seconds: float | None = None,
+        pause_after_idle_seconds: float | None = None,
     ) -> Callable[[Callable[..., Any] | type[Loop]], Callable[..., Any] | type[Loop]]:
         """Decorator to register a loop function or class."""
+        if stop_after_idle_seconds is not None and pause_after_idle_seconds is not None:
+            raise ValueError(
+                "Cannot set both stop_after_idle_seconds and pause_after_idle_seconds"
+            )
 
         def _decorator(
             func_or_class: Callable[..., Any] | type[Loop],
@@ -250,6 +256,8 @@ def _decorator(
                     "loop_delay": self.config.loop_delay_s,
                     "integrations": integrations,
                     "stop_on_disconnect": stop_on_disconnect,
+                    "stop_after_idle_seconds": stop_after_idle_seconds,
+                    "pause_after_idle_seconds": pause_after_idle_seconds,
                     "loop_instance": loop_instance,
                 }
             else:
@@ -361,6 +369,8 @@ async def _event_handler(request: dict[str, Any], func: Any = func):
                     context=context,
                     loop=loop,
                     loop_delay=self.config.loop_delay_s,
+                    stop_after_idle_seconds=stop_after_idle_seconds,
+                    pause_after_idle_seconds=pause_after_idle_seconds,
                 )
                 if started:
                     logger.info(
@@ -700,6 +710,8 @@ async def restart_loop(self, loop_id: str) -> bool:
                 context=context,
                 loop=loop,
                 loop_delay=metadata["loop_delay"],
+                stop_after_idle_seconds=metadata.get("stop_after_idle_seconds"),
+                pause_after_idle_seconds=metadata.get("pause_after_idle_seconds"),
             )
             if started:
                 logger.info("Restarted loop", extra={"loop_id": loop.loop_id})

fastloop/loop.py

@@ -8,6 +8,7 @@
 
 import asyncio
 import json
+import time
 import traceback
 import uuid
 from collections.abc import Callable
@@ -16,7 +17,7 @@
 from fastapi import HTTPException
 from fastapi.responses import StreamingResponse
 
-from .constants import CANCEL_GRACE_PERIOD_S
+from .constants import CANCEL_GRACE_PERIOD_S, MEANINGFUL_WORK_THRESHOLD_S
 from .exceptions import (
     EventTimeoutError,
     LoopClaimError,
@@ -76,6 +77,8 @@ async def _run(
         delay: float,
         loop_start_func: Callable[..., Any] | None,
         loop_stop_func: Callable[..., Any] | None,
+        stop_after_idle_seconds: float | None = None,
+        pause_after_idle_seconds: float | None = None,
     ) -> None:
         try:
             async with self.state_manager.with_claim(loop_id):  # type: ignore
@@ -85,10 +88,12 @@ async def _run(
                         await loop_start_func(context)
                     else:
                         loop_start_func(context)  # type: ignore
-                idle_cycles = 0
+
+                last_active_time = time.time()
 
                 while not context.should_stop and not context.should_pause:
-                    context.event_this_cycle = False
+                    context._reset_cycle_tracking()
+                    cycle_start = time.monotonic()
 
                     try:
                         if asyncio.iscoroutinefunction(func):
@@ -122,15 +127,41 @@ async def _run(
                             },
                         )
 
-                    if not context.event_this_cycle:
-                        idle_cycles += 1
+                    cycle_duration = time.monotonic() - cycle_start
+                    work_time = cycle_duration - context._wait_time_this_cycle
+                    is_active = (
+                        work_time > MEANINGFUL_WORK_THRESHOLD_S
+                        or context.event_this_cycle
+                    )
+
+                    if is_active:
+                        last_active_time = time.time()
+                    else:
+                        idle_seconds = time.time() - last_active_time
                         if (
-                            idle_cycles >= self.config.max_idle_cycles
-                            and self.config.shutdown_idle
+                            stop_after_idle_seconds is not None
+                            and idle_seconds >= stop_after_idle_seconds
                         ):
+                            logger.info(
+                                "Loop idle timeout reached, stopping",
+                                extra={
+                                    "loop_id": loop_id,
+                                    "idle_seconds": idle_seconds,
+                                },
+                            )
+                            raise LoopStoppedError()
+                        if (
+                            pause_after_idle_seconds is not None
+                            and idle_seconds >= pause_after_idle_seconds
+                        ):
+                            logger.info(
+                                "Loop idle timeout reached, pausing",
+                                extra={
+                                    "loop_id": loop_id,
+                                    "idle_seconds": idle_seconds,
+                                },
+                            )
                             raise LoopPausedError()
-                    else:
-                        idle_cycles = 0
 
                     try:
                         await asyncio.sleep(delay)
@@ -180,13 +211,22 @@ async def start(
         context: Any,
         loop: LoopState,
         loop_delay: float = 0.1,
+        stop_after_idle_seconds: float | None = None,
+        pause_after_idle_seconds: float | None = None,
     ) -> bool:
         if loop.loop_id in self.loop_tasks:
             return False
 
         self.loop_tasks[loop.loop_id] = asyncio.create_task(
             self._run(
-                func, context, loop.loop_id, loop_delay, loop_start_func, loop_stop_func
+                func,
+                context,
+                loop.loop_id,
+                loop_delay,
+                loop_start_func,
+                loop_stop_func,
+                stop_after_idle_seconds,
+                pause_after_idle_seconds,
             )
         )
 

fastloop/types.py

@@ -148,8 +148,6 @@ class BaseConfig(BaseModel):
     loop_delay_s: float = 0.1
     sse_poll_interval_s: float = 0.1
     sse_keep_alive_s: float = 10.0
-    shutdown_idle: bool = True
-    max_idle_cycles: int = 10
     shutdown_timeout_s: float = 10.0
     port: int = 8000
     host: str = "localhost"

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "fastloop"
-version = "0.1.101"
+version = "0.1.102"
 description = "A Python package for deploying stateful loops"
 readme = "README.md"
 requires-python = ">=3.12"