Merge branch 'paramwatcher' into watcher

sunnypilot/common/README.md

@@ -35,7 +35,7 @@ The Linux implementation interacts directly with the kernel's `inotify` subsyste
 
 *   **Library Loading**: `libc = ctypes.CDLL('libc.so.6')` loads the standard C library to access system calls.
 *   **Initialization**: `inotify_init()` is called to create a new inotify instance, returning a file descriptor.
-*   **Watch Setup**: `inotify_add_watch(fd, path, mask)` registers the parameters directory. The mask includes `IN_MODIFY | IN_CREATE | IN_DELETE | IN_MOVED_TO | IN_CLOSE_WRITE` to capture all relevant file changes.
+*   **Watch Setup**: `inotify_add_watch(fd, path, mask)` registers the parameters directory. The mask includes `IN_MODIFY | IN_CREATE | IN_DELETE | IN_MOVED_TO | IN_CLOSE_WRITE` (Linux Kernel Organization, 2005) to capture all relevant file changes.
 *   **Event Loop**:
     *   **Polling**: `select.epoll()` is used to efficiently wait for activity on the file descriptor without busy-waiting.
     *   **Reading**: When events occur, `os.read(fd, 1024)` retrieves the raw binary event data.
@@ -48,6 +48,7 @@ The macOS implementation uses the `FSEvents` API from the `CoreServices` framewo
 *   **Framework Loading**: `ctypes.cdll.LoadLibrary` loads `CoreServices` and `CoreFoundation`.
 *   **Callback Definition**: `CFUNCTYPE` is used to define a C-compatible callback function. This function is invoked by the OS whenever a change occurs in the watched directory.
 *   **Stream Creation**: `FSEventStreamCreate` creates a stream for the target directory. The `kFSEventStreamCreateFlagFileEvents` flag is used to request file-level granularity where available.
+*   **Event Filtering**: The callback filters events using flags such as `kFSEventStreamEventFlagItemCreated` and `kFSEventStreamEventFlagItemModified` to ensure only relevant file changes trigger updates (Apple Inc., n.d.-c).
 *   **Scheduling**: `FSEventStreamScheduleWithRunLoop` attaches the stream to the current thread's run loop (Apple Inc., n.d.-b).
 *   **Execution**: `CFRunLoopRun()` starts the event loop. This passes control to the OS, which wakes the thread only when necessary.
 *   **Handling**: Inside the callback, the code iterates through the changed paths provided by the OS. It extracts the filename and calls `_trigger_callbacks` to invalidate the cache for that specific parameter.
@@ -81,14 +82,34 @@ While `ParamWatcher` offers superior performance for UI rendering, it presents s
 *   **Event Latency**: In high-load scenarios, `inotify` events may experience slight delays or coalescing compared to direct reads. However, for user interface applications, this latency (<10ms) is imperceptible.
 *   **Complexity**: The solution (the process singleton approach) requires managing a background thread and OS-specific event loops, increasing code complexity compared to the synchronous `Params::get` function.
 
+## Alternative Architecture Considered: ZeroMQ Service (ZMQ)
+During the development of `ParamWatcher`, a Client-Server architecture using ZMQ was evaluated. In this architecture, a single background service process would monitor file system events and publish changes over a ZMQ PUB socket to multiple client processes (SUB).
+
+### Trade-off Analysis
+| Metric | In-Process (Current) | ZMQ Service (Rejected)                                |
+| :--- | :--- |:------------------------------------------------------|
+| **Memory Usage** | Low (1 thread/process) | High (1 full Python process + ZMQ buffers per client) |
+| **CPU Usage** | Low (Direct callback) | High (Serialization + TCP Stack + Deserialization)    |
+| **Latency** | Instant (<0.1ms) | Variable (TCP Loopback overhead)                      |
+| **Scalability** | Limited by OS file handles | Limited by TCP ports/buffers                          |
+| **Robustness** | Process-isolated failure | Single point of failure (Service crash affects all)   |
+
+### Decision Rationale
+While the ZMQ approach offers better isolation and reduces the total number of OS file watchers (1 vs N), the overhead of inter-process communication (IPC) proved excessive for this use case.
+*   **Efficiency**: Even with 50+ processes, the memory footprint of 50 simple threads is significantly lower than the overhead of a dedicated Python service process plus the ZMQ context in every client.
+*   **Complexity**: The ZMQ architecture introduced synchronization challenges (e.g., service startup race conditions, "Address already in use" errors) that outweighed its benefits.
+*   **Performance**: The latency of serializing messages and passing them through the TCP stack is orders of magnitude higher than a direct function call within the same process memory space.
+
 ## Conclusion
-Replacing polling mechanisms with event-driven caching shifts the computational load from kernel space (syscalls) to user space (RAM). This transition eliminates I/O overhead and UI stutters caused by garbage collection, resulting in a more responsive user experience.
+Replacing polling mechanisms with event-driven caching shifts the computational load from kernel space (syscalls) to user space (RAM). This transition eliminates I/O overhead and UI stutters caused by garbage collection, resulting in a more responsive user experience. The In-Process Singleton approach was selected as the optimal balance between performance, complexity, and resource efficiency.
 
 ## References
 Apple Inc. (n.d.-a). *File System Events*. Retrieved from https://developer.apple.com/documentation/coreservices/file_system_events
 
 Apple Inc. (n.d.-b). *CFRunLoop*. Retrieved from https://developer.apple.com/documentation/corefoundation/cfrunloop
 
+Apple Inc. (n.d.-c). *FSEventStreamEventFlags*. Retrieved from https://developer.apple.com/documentation/coreservices/1455361-fseventstreameventflags
+
 Codezup. (2025). *Efficient File I/O in C++*. Retrieved from https://codezup.com/efficient-file-io-cpp-best-practices/
 
 cppreference.com. (n.d.-a). *std::basic_ifstream*. Retrieved from https://en.cppreference.com/w/cpp/io/basic_ifstream
@@ -101,6 +122,8 @@ Linux man-pages. (2025-b). *read(2)*. Retrieved from https://man7.org/linux/man-
 
 Linux man-pages. (2025 -c). *inotify(7)*. Retrieved from https://man7.org/linux/man-pages/man7/inotify.7.html
 
+Linux Kernel Organization. (2005). *include/uapi/linux/inotify.h*. Retrieved from https://git.kernel.org/pub/scm/linux/kernel/git/torvalds/linux.git/tree/include/uapi/linux/inotify.h
+
 Python Software Foundation. (2025). *ctypes — A foreign function library for Python*. Retrieved from https://docs.python.org/3/library/ctypes.html
 
 Gamma, E., Helm, R., Johnson, R., & Vlissides, J. (1994). *Design Patterns: Elements of Reusable Object-Oriented Software*. Addison-Wesley.

sunnypilot/common/param_watcher.py

@@ -61,10 +61,7 @@ class ParamWatcher(Params):
     if callback not in self._callbacks:
       self._callbacks.append(callback)
 
-  def _trigger_callbacks(self, path, mask):
-    if platform.system() == "Linux" and not (mask & (IN_MODIFY | IN_CREATE | IN_DELETE | IN_MOVED_TO | IN_CLOSE_WRITE)):
-      return
-
+  def _trigger_callbacks(self, path):
     with self._lock:
       if (now := time.monotonic()) - self._last_trigger.get(path, 0) < 0.1:
         return
@@ -74,7 +71,7 @@ class ParamWatcher(Params):
 
     for callback in self._callbacks:
       try:
-        callback(path, mask)
+        callback(path)
       except Exception:
         cloudlog.exception("Param watcher callback failed")
 
@@ -110,13 +107,15 @@ class ParamWatcher(Params):
 
   def _run_watcher(self):
     system = platform.system()
-    try:
-      if system == "Linux":
-        self._run_linux()
-      elif system == "Darwin":
-        self._run_darwin()
-    except Exception:
-      traceback.print_exc()
+    while True:
+      try:
+        if system == "Linux":
+          self._run_linux()
+        elif system == "Darwin":
+          self._run_darwin()
+      except Exception:
+        cloudlog.exception("Param watcher crashed")
+        time.sleep(2)
 
   def _run_linux(self):
     path = Paths.params_root()
@@ -134,7 +133,8 @@ class ParamWatcher(Params):
             i = 0
             while i + 16 <= len(buffer):
               _, mask, _, name_len = struct.unpack_from("iIII", buffer, i)
-              self._trigger_callbacks(buffer[i+16:i+16+name_len].rstrip(b"\0").decode(), mask)
+              if mask & (IN_MODIFY | IN_CREATE | IN_DELETE | IN_MOVED_TO | IN_CLOSE_WRITE):
+                self._trigger_callbacks(buffer[i+16:i+16+name_len].rstrip(b"\0").decode())
               i += 16 + name_len
     finally:
       if 'poll' in locals():
@@ -149,6 +149,10 @@ class ParamWatcher(Params):
     kCFAllocatorDefault = c_void_p(0)
     kCFStringEncodingUTF8 = 0x08000100
     kFSEventStreamCreateFlagFileEvents = 0x00000010
+    kFSEventStreamEventFlagItemCreated = 0x00000100
+    kFSEventStreamEventFlagItemRemoved = 0x00000200
+    kFSEventStreamEventFlagItemRenamed = 0x00000800
+    kFSEventStreamEventFlagItemModified = 0x00001000
 
     CF.CFStringCreateWithCString.restype = c_void_p
     CF.CFStringCreateWithCString.argtypes = [c_void_p, ctypes.c_char_p, c_uint32]
@@ -166,8 +170,9 @@ class ParamWatcher(Params):
         flags_arr = ctypes.cast(flags, POINTER(c_uint32))
         for i in range(num):
           path = ctypes.cast(paths_arr[i], ctypes.c_char_p).value
-          if path:
-            self._trigger_callbacks(os.path.basename(path.decode('utf-8').rstrip('/')), flags_arr[i])
+          if path and (flags_arr[i] & (kFSEventStreamEventFlagItemCreated | kFSEventStreamEventFlagItemRemoved |
+                                       kFSEventStreamEventFlagItemRenamed | kFSEventStreamEventFlagItemModified)):
+            self._trigger_callbacks(os.path.basename(path.decode('utf-8').rstrip('/')))
       except Exception:
         traceback.print_exc()
 

sunnypilot/common/tests/test_param_watcher.py

@@ -6,7 +6,7 @@ import tracemalloc
 from openpilot.common.params import Params
 from openpilot.common.params_pyx import UnknownKeyName
 
-from openpilot.sunnypilot.common.param_watcher import ParamWatcher, IN_MODIFY
+from openpilot.sunnypilot.common.param_watcher import ParamWatcher
 
 
 class TestParamWatcher:
@@ -90,8 +90,8 @@ class TestParamWatcher:
     time.sleep(0.2)
 
     event = threading.Event()
-    param_watcher.add_watcher(lambda key, mask: event.set())
-    param_watcher._trigger_callbacks(self.bytes_key, IN_MODIFY)
+    param_watcher.add_watcher(lambda key: event.set())
+    param_watcher._trigger_callbacks(self.bytes_key)
     assert event.wait(timeout=2), "Callback not triggered"
 
     self.params.put(self.bytes_key, b"new")