dart-lang/sdk
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

[io] Rewrite Mac OS X FS watcher

Browse Source 
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 <vegorov@google.com>
Reviewed-by: Alexander Aprelev <aam@google.com>
This commit is contained in:
Slava Egorov
2025-10-01 11:28:05 -07:00
committed by Commit Queue
parent ed6bab847b
commit 25df2b3f11
3 changed files with 318 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files
runtime/bin/file_system_watcher_macos.cc
+142 -81
View File
@@ -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 <typename T>
struct SetCObjectValue;
template <>
struct SetCObjectValue<const char*> {
static void Assign(Dart_CObject* o, const char* str) {
o->type = Dart_CObject_kString;
o->value.as_string = str;
}
};
template <>
struct SetCObjectValue<char*> {
static void Assign(Dart_CObject* o, const char* str) {
o->type = Dart_CObject_kString;
o->value.as_string = str;
}
};
template <>
struct SetCObjectValue<int64_t> {
static void Assign(Dart_CObject* o, int64_t v) {
o->type = Dart_CObject_kInt64;
o->value.as_int64 = v;
}
};
template <typename... Ts>
Dart_CObject* CreateCObjectArray(Ts... elements) {
const auto length = sizeof...(elements);
auto array = static_cast<Dart_CObject*>(
malloc(sizeof(Dart_CObject) +
(sizeof(Dart_CObject*) + sizeof(Dart_CObject)) * length));
array->type = Dart_CObject_kArray;
array->value.as_array.values = reinterpret_cast<Dart_CObject**>(array + 1);
array->value.as_array.length = length;
for (uintptr_t i = 0; i < length; i++) {
array->value.as_array.values[i] =
reinterpret_cast<Dart_CObject*>(array->value.as_array.values + length) +
i;
}
int index = 0;
(
[&] {
SetCObjectValue<Ts>::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<Node*>(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<Dart_CObject**>(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<char**>(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<int64_t>(0), path,
/*path_id=*/reinterpret_cast<int64_t>(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<intptr_t>(Node::Watch(path, events, recursive));
return reinterpret_cast<intptr_t>(
Node::Watch(static_cast<Dart_Port>(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<Node*>(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<uint8_t*>(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
runtime/tests/vm/dart/regress_61551_test.dart
+99
View File
@@ -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<Int>(2);
const writeSize = 64 * 1024;
final buf = calloc<Int8>(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<Int Function(Pointer<Int>)>()
external int pipe(Pointer<Int> fds);
@Native<Int Function(Pointer<Int>)>()
external int fnctl(Pointer<Int> fds);
@Native<Int Function(Int, Int, VarArgs<()>)>(symbol: 'fcntl')
external int fcntl0(int fd, int cmd);
@Native<Int Function(Int, Int, VarArgs<(Int,)>)>(symbol: 'fcntl')
external int fcntl1(int fd, int cmd, int val);
@Native<Int Function(Int, Pointer<Void> buf, Size)>()
external int write(int fd, Pointer<Void> buf, int size);
sdk/lib/_internal/vm/bin/file_patch.dart
+77 -31
View File
@@ -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<ffi.Void> 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<int, _WatchedPath> _watchedPaths = <int, _WatchedPath>{};
/// 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<void> _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<void> _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<List<_NativeFSEvent>>? _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<void> _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<void> _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<void> _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(