From 25df2b3f111bc7975776226183e1bb07cfa8ae08 Mon Sep 17 00:00:00 2001 From: Slava Egorov Date: Wed, 1 Oct 2025 11:28:05 -0700 Subject: [PATCH] [io] Rewrite Mac OS X FS watcher Watcher used pipes to fit into the "socket"-like implementation used by other OSes. However it turns out that pipe buffer on Mac OS X is not necessarily large enough to fit the whole FSEvent structure meaning that we can't expect to be able to read FSEvents atomically from a pipe. This situation happens because Mac OS X inherits BSD behavior of limiting total number of kernel memory reserved for pipe buffers. Once kern.ipc.maxpipekva limit is crossed pipe buffers no longer grow to 64KB and remain 512 bytes large (which is absolute minimum allowed by POSIX which requires writes smaller than PIPE_BUF to be atomic). So our code needs to be prepared that only half of FSEvent structure (which is 1032 bytes large) will fit into pipe buffer. We could fix this by reading each FSEvent in chunks but this seems ridiculously ineffecient. Instead we rewrite the code to pass file system events via SendPort instead. Fixes https://github.com/dart-lang/sdk/issues/61551 TEST=runtime/tests/vm/dart/regress_61551_test.dart CoreLibraryReviewExempt: VM only changes. Change-Id: I6a6a696499044832cfb61998bc6a519bbaadf8bc Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/451260 Commit-Queue: Slava Egorov Reviewed-by: Alexander Aprelev --- runtime/bin/file_system_watcher_macos.cc | 223 +++++++++++------- runtime/tests/vm/dart/regress_61551_test.dart | 99 ++++++++ sdk/lib/_internal/vm/bin/file_patch.dart | 108 ++++++--- 3 files changed, 318 insertions(+), 112 deletions(-) create mode 100644 runtime/tests/vm/dart/regress_61551_test.dart diff --git a/runtime/bin/file_system_watcher_macos.cc b/runtime/bin/file_system_watcher_macos.cc index 50c4f9aca78..ee8c84c5766 100644 --- a/runtime/bin/file_system_watcher_macos.cc +++ b/runtime/bin/file_system_watcher_macos.cc @@ -36,15 +36,75 @@ union FSEvent { uint8_t bytes[PATH_MAX + 8]; }; +// A helper for creating Dart_CObject array using a single allocation. +// +// We can't use CObject helpers because those rely on Dart_ScopeAllocate. +namespace { + +template +struct SetCObjectValue; + +template <> +struct SetCObjectValue { + static void Assign(Dart_CObject* o, const char* str) { + o->type = Dart_CObject_kString; + o->value.as_string = str; + } +}; + +template <> +struct SetCObjectValue { + static void Assign(Dart_CObject* o, const char* str) { + o->type = Dart_CObject_kString; + o->value.as_string = str; + } +}; + +template <> +struct SetCObjectValue { + static void Assign(Dart_CObject* o, int64_t v) { + o->type = Dart_CObject_kInt64; + o->value.as_int64 = v; + } +}; + +template +Dart_CObject* CreateCObjectArray(Ts... elements) { + const auto length = sizeof...(elements); + auto array = static_cast( + malloc(sizeof(Dart_CObject) + + (sizeof(Dart_CObject*) + sizeof(Dart_CObject)) * length)); + array->type = Dart_CObject_kArray; + array->value.as_array.values = reinterpret_cast(array + 1); + array->value.as_array.length = length; + for (uintptr_t i = 0; i < length; i++) { + array->value.as_array.values[i] = + reinterpret_cast(array->value.as_array.values + length) + + i; + } + + int index = 0; + ( + [&] { + SetCObjectValue::Assign(array->value.as_array.values[index], + elements); + ++index; + }(), + ...); + + return array; +} + +} // namespace + class Node { public: - Node(char* base_path, int read_fd, int write_fd, bool recursive) - : base_path_length_(strlen(base_path)), + Node(Dart_Port port, char* base_path, bool recursive) + : port_(port), + base_path_length_(strlen(base_path)), path_ref_(CFStringCreateWithCString(nullptr, base_path, kCFStringEncodingUTF8)), - read_fd_(read_fd), - write_fd_(write_fd), recursive_(recursive), ref_(nullptr) { Start(); @@ -56,7 +116,6 @@ class Node { // deallocated, the same [FSEventStream] that [Callback] gets a reference // to during its execution. [Callback] holding a reference prevents stream // from deallocation. - close(write_fd_); CFRelease(path_ref_); } @@ -96,20 +155,16 @@ class Node { } intptr_t base_path_length() const { return base_path_length_; } - int read_fd() const { return read_fd_; } - int write_fd() const { return write_fd_; } bool recursive() const { return recursive_; } - static Node* Watch(const char* path, int events, bool recursive) { - int fds[2]; - VOID_NO_RETRY_EXPECTED(pipe(fds)); - FDUtils::SetNonBlocking(fds[0]); - FDUtils::SetBlocking(fds[1]); - + static Node* Watch(Dart_Port port, + const char* path, + int events, + bool recursive) { char base_path[PATH_MAX]; realpath(path, base_path); - return new Node(base_path, fds[0], fds[1], recursive); + return new Node(port, base_path, recursive); } static void Unwatch(Node* node) { @@ -147,11 +202,16 @@ class Node { TimerUtils::Sleep(1000 /* ms */); } Node* node = static_cast(client); + + // Can't use CObject helpers because they expect Dart_ScopeAllocate to work + // and this thread is not attached to any isolate or native message handler. + Dart_CObject events; + events.type = Dart_CObject_kArray; + events.value.as_array.values = + static_cast(malloc(sizeof(Dart_CObject*) * num_events)); + events.value.as_array.length = 0; for (size_t i = 0; i < num_events; i++) { char* path = reinterpret_cast(event_paths)[i]; - FSEvent event; - event.data.exists = - File::GetType(nullptr, path, false) != File::kDoesNotExist; path += node->base_path_length(); // If path is longer the base, skip next character ('/'). if (path[0] != '\0') { @@ -160,18 +220,72 @@ class Node { if (!node->recursive() && (strstr(path, "/") != nullptr)) { continue; } - event.data.flags = event_flags[i]; - memmove(event.data.path, path, strlen(path) + 1); - write(node->write_fd(), event.bytes, sizeof(event)); + + const bool is_path_empty = path[0] == '\0'; + const bool path_exists = + !is_path_empty && + File::GetType(nullptr, path, false) != File::kDoesNotExist; + + events.value.as_array.values[events.value.as_array.length++] = + CreateCObjectArray( + /*flags=*/ConvertEventFlags(event_flags[i], is_path_empty, + path_exists), + /*cookie=*/static_cast(0), path, + /*path_id=*/reinterpret_cast(node)); } + + if (events.value.as_array.length != 0) { + Dart_PostCObject(node->port_, &events); + } + + for (int i = 0; i < events.value.as_array.length; i++) { + free(events.value.as_array.values[i]); + } + free(events.value.as_array.values); + } + + static int64_t ConvertEventFlags(FSEventStreamEventFlags flags, + bool is_path_empty, + bool path_exists) { + int64_t mask = 0; + if ((flags & kFSEventStreamEventFlagItemRenamed) != 0) { + if (is_path_empty) { + // The moved path is the path being watched. + mask |= FileSystemWatcher::kDeleteSelf; + } else if (path_exists) { + mask |= FileSystemWatcher::kCreate; + } else { + mask |= FileSystemWatcher::kDelete; + } + } + if ((flags & kFSEventStreamEventFlagItemModified) != 0) { + mask |= FileSystemWatcher::kModifyContent; + } + if ((flags & kFSEventStreamEventFlagItemXattrMod) != 0) { + mask |= FileSystemWatcher::kModifyAttribute; + } + if ((flags & kFSEventStreamEventFlagItemCreated) != 0) { + mask |= FileSystemWatcher::kCreate; + } + if ((flags & kFSEventStreamEventFlagItemIsDir) != 0) { + mask |= FileSystemWatcher::kIsDir; + } + if ((flags & kFSEventStreamEventFlagItemRemoved) != 0) { + if (is_path_empty) { + // The removed path is the path being watched. + mask |= FileSystemWatcher::kDeleteSelf; + } else { + mask |= FileSystemWatcher::kDelete; + } + } + return mask; } static dispatch_queue_t notification_queue_; + Dart_Port port_; intptr_t base_path_length_; CFStringRef path_ref_; - int read_fd_; - int write_fd_; bool recursive_; FSEventStreamRef ref_; Monitor monitor_; @@ -203,7 +317,8 @@ intptr_t FileSystemWatcher::WatchPath(intptr_t id, const char* path, int events, bool recursive) { - return reinterpret_cast(Node::Watch(path, events, recursive)); + return reinterpret_cast( + Node::Watch(static_cast(id), path, events, recursive)); } void FileSystemWatcher::UnwatchPath(intptr_t id, intptr_t path_id) { @@ -216,67 +331,13 @@ void FileSystemWatcher::DestroyWatch(intptr_t path_id) { } intptr_t FileSystemWatcher::GetSocketId(intptr_t id, intptr_t path_id) { - return reinterpret_cast(path_id)->read_fd(); + // This API should not be called. We are communicating over ports instead. + return -1; } Dart_Handle FileSystemWatcher::ReadEvents(intptr_t id, intptr_t path_id) { - intptr_t fd = GetSocketId(id, path_id); - intptr_t avail = FDUtils::AvailableBytes(fd); - int count = avail / sizeof(FSEvent); - if (count <= 0) { - return Dart_NewList(0); - } - Dart_Handle events = Dart_NewList(count); - FSEvent e; - for (int i = 0; i < count; i++) { - intptr_t bytes = TEMP_FAILURE_RETRY(read(fd, e.bytes, sizeof(e))); - if (bytes < 0) { - return DartUtils::NewDartOSError(); - } - size_t path_len = strlen(e.data.path); - Dart_Handle event = Dart_NewList(kEventNumElements); - int flags = e.data.flags; - int mask = 0; - if ((flags & kFSEventStreamEventFlagItemRenamed) != 0) { - if (path_len == 0) { - // The moved path is the path being watched. - mask |= kDeleteSelf; - } else { - mask |= e.data.exists ? kCreate : kDelete; - } - } - if ((flags & kFSEventStreamEventFlagItemModified) != 0) { - mask |= kModifyContent; - } - if ((flags & kFSEventStreamEventFlagItemXattrMod) != 0) { - mask |= kModifyAttribute; - } - if ((flags & kFSEventStreamEventFlagItemCreated) != 0) { - mask |= kCreate; - } - if ((flags & kFSEventStreamEventFlagItemIsDir) != 0) { - mask |= kIsDir; - } - if ((flags & kFSEventStreamEventFlagItemRemoved) != 0) { - if (path_len == 0) { - // The removed path is the path being watched. - mask |= kDeleteSelf; - } else { - mask |= kDelete; - } - } - Dart_ListSetAt(event, kEventFlagsIndex, Dart_NewInteger(mask)); - Dart_ListSetAt(event, kEventCookieIndex, Dart_NewInteger(0)); - Dart_Handle name = Dart_NewStringFromUTF8( - reinterpret_cast(e.data.path), path_len); - if (Dart_IsError(name)) { - return name; - } - Dart_ListSetAt(event, kEventPathIndex, name); - Dart_ListSetAt(event, kEventPathIdIndex, Dart_NewInteger(path_id)); - Dart_ListSetAt(events, i, event); - } - return events; + // This API should not be called. We are communicating over ports instead. + return DartUtils::NewDartOSError(); } } // namespace bin diff --git a/runtime/tests/vm/dart/regress_61551_test.dart b/runtime/tests/vm/dart/regress_61551_test.dart new file mode 100644 index 00000000000..f86b6180d9a --- /dev/null +++ b/runtime/tests/vm/dart/regress_61551_test.dart @@ -0,0 +1,99 @@ +// Copyright (c) 2025, the Dart project authors. Please see the AUTHORS file +// for details. All rights reserved. Use of this source code is governed by a +// BSD-style license that can be found in the LICENSE file. +// +// Make sure that file system watcher works when Mac OS X restricts pipe buffer +// size to 512 bytes (see https://github.com/dart-lang/sdk/issues/61551). + +import 'dart:async'; +import 'dart:io'; +import 'dart:ffi'; + +import 'package:expect/expect.dart'; +import 'package:ffi/ffi.dart'; +import 'package:path/path.dart' as p; + +void main() async { + if (!Platform.isMacOS) { + return; + } + + exhaustMaxPipeKVA(); + + var tempDir = await Directory.systemTemp.createTemp('fsevents_test_'); + try { + var watcher = tempDir.watch(); + var eventsReceived = 0; + + var subscription = watcher.listen((event) { + eventsReceived++; + }); + + // Wait for the watcher to become active. + await Future.delayed(Duration(milliseconds: 500)); + + // Make some changes in the directory. + final testFile = File(p.join(tempDir.path, 'test_file.txt')); + await testFile.writeAsString('test content'); + await testFile.writeAsString('modified content', mode: FileMode.append); + await testFile.delete(); + + // Wait a bit for events to arrive. + await Future.delayed(Duration(seconds: 2)); + + // Cancel the watcher. + await subscription.cancel(); + + // We should have received at least some events. + Expect.isTrue(eventsReceived > 0); + } finally { + await tempDir.delete(recursive: true); + } +} + +// Create pipes and force their buffers to grow to 64KB until we reach +// kern.ipc.maxpipekva. This should not take more than 256 pipes because +// the limit is 16 MB. +void exhaustMaxPipeKVA() { + final fds = calloc(2); + + const writeSize = 64 * 1024; + final buf = calloc(writeSize); + + for (int i = 0; i < 256; i++) { + final pipeRes = pipe(fds); + // We do not expect to run out of file descriptors before we run out of space + // in kernel for pipebuffers. + Expect.equals(0, pipeRes, 'Failed to create a pipe'); + var writeFd = (fds + 1).value; + + const F_GETFL = 3; + const F_SETFL = 4; + const O_NONBLOCK = 0x00000004; + + final flags = fcntl0(writeFd, F_GETFL); + Expect.isTrue(flags != -1, 'Failed to call fcntl($writeFd, F_GETFL)'); + + final setFlagsRes = fcntl1(writeFd, F_SETFL, flags | O_NONBLOCK); + Expect.isTrue(setFlagsRes != -1, 'Failed to call fcntl($writeFd, F_SETFL)'); + + if (write(writeFd, buf.cast(), writeSize) != writeSize) { + break; + } + } +} + +@Native)>() +external int pipe(Pointer fds); + +@Native)>() +external int fnctl(Pointer fds); + +@Native)>(symbol: 'fcntl') +external int fcntl0(int fd, int cmd); + +@Native)>(symbol: 'fcntl') +external int fcntl1(int fd, int cmd, int val); + +@Native buf, Size)>() +external int write(int fd, Pointer buf, int size); diff --git a/sdk/lib/_internal/vm/bin/file_patch.dart b/sdk/lib/_internal/vm/bin/file_patch.dart index 271b2953f33..2df5886baed 100644 --- a/sdk/lib/_internal/vm/bin/file_patch.dart +++ b/sdk/lib/_internal/vm/bin/file_patch.dart @@ -450,11 +450,11 @@ abstract class _FileSystemWatcher { } if (Platform.isWindows) { - return _SocketPerPathFileSystemWatcher(); + return _Win32FileSystemWatcher(); } if (Platform.isMacOS) { - return _SocketPerPathFileSystemWatcher(); + return _FSEventStreamFileSystemWatcher(); } } @@ -497,16 +497,54 @@ abstract class _FileSystemWatcher { external static void _destroyWatch(ffi.Pointer pathId); } -class _InotifyFileSystemWatcher extends _FileSystemWatcher { +/// A watcher that receives events for multiple `pathId` on a single channel. +abstract class _MultiplexingFileSystemWatcher extends _FileSystemWatcher { + /// Map of [_watchedPath] indexed by `pathId` values. final Map _watchedPaths = {}; + /// Perform necessary initialization of the native state for the watcher. + void _ensureWatcherIsRunning(); + + /// Shutdown the watcher when there is no actively watched paths. + /// + /// If shutdown requires asynchronous actions returns [Future] which will + /// complete when shutdown is finished. + FutureOr _stopWatcher(); + + @override + _WatchedPath _startWatching(String path, int events, bool recursive) { + _ensureWatcherIsRunning(); + final pathId = super._watchPath(path, events, recursive); + // On Linux inotify_add_watch will return an existing watch descriptor + // for the inode if there is already one associated with it. Thus we + // need accept the possibility that calling _watchPath twice will + // return the same pathId. Other OSes do not reuse pathId values. + assert(Platform.isLinux || !_watchedPaths.containsKey(pathId)); + return _watchedPaths[pathId] ??= _WatchedPath(pathId, path, events); + } + + @override + Future _stopWatching(_WatchedPath wp) async { + assert(_watchedPaths[wp.pathId] == wp); + _watchedPaths.remove(wp.pathId); + await super._stopWatching(wp); + + // If there are no more active watcher close inotify descriptor. + if (_watchedPaths.isEmpty) { + await _stopWatcher(); + } + } +} + +class _InotifyFileSystemWatcher extends _MultiplexingFileSystemWatcher { int? _inotifyFd; StreamSubscription>? _inotifySubscription; @override int get _watcherId => _inotifyFd!; - void _ensureInotifyFD() { + @override + void _ensureWatcherIsRunning() { if (_inotifyFd != null) { return; } @@ -518,22 +556,6 @@ class _InotifyFileSystemWatcher extends _FileSystemWatcher { ).listen(_handleEvents); } - @override - _WatchedPath _startWatching(String path, int events, bool recursive) { - _ensureInotifyFD(); - // On Linux inotify_add_watch will return an existing watch descriptor - // for the inode if there is already one associated with it. Thus we - // need accept the possibility that calling _watchPath twice will - // return the same pathId. - final pathId = super._watchPath(path, events, recursive); - final watchedPath = _watchedPaths[pathId] ??= _WatchedPath( - pathId, - path, - events, - ); - return watchedPath; - } - void _handleEvents(List<_NativeFSEvent> events) { Set<_WatchedPath>? dirty; @@ -556,22 +578,46 @@ class _InotifyFileSystemWatcher extends _FileSystemWatcher { } @override - Future _stopWatching(_WatchedPath wp) async { - assert(_watchedPaths[wp.pathId] == wp); - _watchedPaths.remove(wp.pathId); - await super._stopWatching(wp); - - // If there are no more active watcher close inotify descriptor. + FutureOr _stopWatcher() { final subscription = _inotifySubscription; - if (_watchedPaths.isEmpty && subscription != null) { - _inotifyFd = null; - _inotifySubscription = null; - return subscription.cancel(); + _inotifyFd = null; + _inotifySubscription = null; + return subscription?.cancel(); + } +} + +class _FSEventStreamFileSystemWatcher extends _MultiplexingFileSystemWatcher { + final _port = RawReceivePort()..keepIsolateAlive = false; + + @override + late final _watcherId = ffi.NativePort(_port.sendPort).nativePort; + + @override + void _ensureWatcherIsRunning() { + _port.keepIsolateAlive = true; + _port.handler = _handleEvents; + } + + @override + FutureOr _stopWatcher() { + _port.keepIsolateAlive = false; + _port.handler = null; + } + + void _handleEvents(List events) { + // All events in a bundle have the same pathId, and we never get an empty + // bundle. + final pathId = (events[0] as _NativeFSEvent).pathId; + if (_watchedPaths[pathId] case final watchedPath?) { + for (_NativeFSEvent event in events) { + watchedPath.addEvent(event); + } + watchedPath.flushUnmatchedMoves(); } } } -class _SocketPerPathFileSystemWatcher extends _FileSystemWatcher { +class _Win32FileSystemWatcher extends _FileSystemWatcher { @override _WatchedPath _startWatching(String path, int events, bool recursive) { final watchedPath = _WatchedPath(