diff --git a/sunnypilot/common/README.md b/sunnypilot/common/README.md index b33e39cd81..fe42a58790 100644 --- a/sunnypilot/common/README.md +++ b/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. diff --git a/sunnypilot/common/param_watcher.py b/sunnypilot/common/param_watcher.py index 9ce712926e..b1c8d0c5f9 100644 --- a/sunnypilot/common/param_watcher.py +++ b/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() diff --git a/sunnypilot/common/tests/test_param_watcher.py b/sunnypilot/common/tests/test_param_watcher.py index 2056c3dabe..04771197e2 100644 --- a/sunnypilot/common/tests/test_param_watcher.py +++ b/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")