[io] Rewrite _FileSystemWatcher implementation
Existing implementation is an entangled mess which consists of shared code residing in the base class which in random places invokes a number of undocumented poorly named methods overloaded in OS specific subclasses. Some of these methods mutate static state. There are no clear lifetime guarantees for different parts of the system (including comments saying that some values might or might not be valid at certain points). The rewrite aims to clean most of this up - sharing everything that can be shared and moving OS specific logic to clearly documented methods. Furthermore, we change the code to ensure proper lifetime guarantees - so we no longer find ourself in situations where we don't know whether pathId is valid or not. This refactoring by itself fixes a number of issues, most specifically a bug where watcher would stop receiving events on Windows because DirectoryWatchHandle ends up allocated at precisely the same address as a previous destroyed one - which confuses Dart side to think that newly created handle is the same as the old one (due to a race between event handler thread and Dart thread). We fix Windows lifetime issue by a) not keeping pathId based mapping in the watcher anymore and b) keeping DirectoryWatchHandler alive until it is stoped by the Dart side - this is achieved by retaining it after it is created and releasing it once path is unwatched. This way Dart side is always sure that pathId values are valid until they are explicitly released via _unwatchPath - which makes code very uniform. To make sure that native objects created by _watchPath are released when surrounding isolate exists abruptly (e.g. via Isolate.exit - without letting Dart code to shutdown and call _unwatchPath naturally) we attach NativeFinalizer to them. This fixes the existing leak of file watchers on Mac OS X - as Node objects it created were not freed if surrounding isolate exited. Note that inotify descriptors did not leak in the same way because they were wrapped into sockets. Finally, this refactoring also make sure that the last subscriber cancelling subscription on filesystem event stream will get a proper cancellation future back and can wait for the watcher to shutdown. Previously implementation used broadcast streams which simply return an already completed future when subscriber cancels. New implementation uses Stream.multi instead which gives a better result. Now doing watch().listen().cancel() returns a future which will only complete once watcher is fully disposed (e.g. inotify descriptor is closed). Bad behavior was revealed by analysing standalone/regress_52715 - which revealed that repeatedly watching and cancelling might flakely cause us to hit fd limit depending on whether eventhandler thread can keep up closing file descriptors created by the main thread or not. Fixes https://github.com/dart-lang/sdk/issues/61378 TEST=standalone/{regress_61378,file_system_watcher_isolate_exit_leak} CoreLibraryReviewExempt: VM only changes. Change-Id: I6a6a69642b1f2673f2be78434bc64270846ad8c5 Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/450921 Reviewed-by: Lasse Nielsen <lrn@google.com>
This commit is contained in:
Slava Egorov
committed by
Commit Queue
Commit Queue
parent
f434804f6b
commit
ed6bab847b
@@ -37,6 +37,9 @@ void Builtin::SetNativeResolver(BuiltinLibraryId id) {
|
||||
Dart_Handle result =
|
||||
Dart_SetNativeResolver(library, NativeLookup, NativeSymbol);
|
||||
ASSERT(!Dart_IsError(result));
|
||||
// Setup the ffi native resolver for built in library functions.
|
||||
result = Dart_SetFfiNativeResolver(library, FfiNativeLookup);
|
||||
ASSERT(!Dart_IsError(result));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -46,6 +46,8 @@ class Builtin {
|
||||
int argument_count,
|
||||
bool* auto_setup_scope);
|
||||
|
||||
static void* FfiNativeLookup(const char* name, uintptr_t argument_count);
|
||||
|
||||
static const uint8_t* NativeSymbol(Dart_NativeFunction nf);
|
||||
|
||||
static const int num_libs_;
|
||||
|
||||
@@ -47,6 +47,10 @@ Dart_NativeFunction Builtin::NativeLookup(Dart_Handle name,
|
||||
return IONativeLookup(name, argument_count, auto_setup_scope);
|
||||
}
|
||||
|
||||
void* Builtin::FfiNativeLookup(const char* name, uintptr_t argument_count) {
|
||||
return nullptr;
|
||||
}
|
||||
|
||||
const uint8_t* Builtin::NativeSymbol(Dart_NativeFunction nf) {
|
||||
int num_entries = sizeof(BuiltinEntries) / sizeof(struct NativeEntries);
|
||||
for (int i = 0; i < num_entries; i++) {
|
||||
|
||||
@@ -69,6 +69,10 @@ Dart_NativeFunction Builtin::NativeLookup(Dart_Handle name,
|
||||
return result;
|
||||
}
|
||||
|
||||
void* Builtin::FfiNativeLookup(const char* name, uintptr_t argument_count) {
|
||||
return IOFfiNativeLookup(name, argument_count);
|
||||
}
|
||||
|
||||
const uint8_t* Builtin::NativeSymbol(Dart_NativeFunction nf) {
|
||||
int num_entries = sizeof(BuiltinEntries) / sizeof(struct NativeEntries);
|
||||
for (int i = 0; i < num_entries; i++) {
|
||||
|
||||
@@ -86,6 +86,10 @@ Dart_NativeFunction LookupIONative(Dart_Handle name,
|
||||
return IONativeLookup(name, argument_count, auto_setup_scope);
|
||||
}
|
||||
|
||||
void* LookupIOFfiNative(const char* name, uintptr_t argument_count) {
|
||||
return IOFfiNativeLookup(name, argument_count);
|
||||
}
|
||||
|
||||
const uint8_t* LookupIONativeSymbol(Dart_NativeFunction nf) {
|
||||
return IONativeSymbol(nf);
|
||||
}
|
||||
|
||||
@@ -60,7 +60,7 @@ void FUNCTION_NAME(FileSystemWatcher_ReadEvents)(Dart_NativeArguments args) {
|
||||
void FUNCTION_NAME(FileSystemWatcher_GetSocketId)(Dart_NativeArguments args) {
|
||||
intptr_t id = DartUtils::GetIntptrValue(Dart_GetNativeArgument(args, 0));
|
||||
intptr_t path_id = DartUtils::GetIntptrValue(Dart_GetNativeArgument(args, 1));
|
||||
int socket_id = FileSystemWatcher::GetSocketId(id, path_id);
|
||||
intptr_t socket_id = FileSystemWatcher::GetSocketId(id, path_id);
|
||||
Dart_SetIntegerReturnValue(args, socket_id);
|
||||
}
|
||||
|
||||
|
||||
@@ -26,14 +26,20 @@ class FileSystemWatcher {
|
||||
kMove = 1 << 3,
|
||||
kModifyAttribute = 1 << 4,
|
||||
kDeleteSelf = 1 << 5,
|
||||
kIsDir = 1 << 6
|
||||
kIsDir = 1 << 6,
|
||||
kMovedTo = 1 << 7,
|
||||
};
|
||||
|
||||
struct Event {
|
||||
intptr_t path_id;
|
||||
int event;
|
||||
const char* filename;
|
||||
int link;
|
||||
// ReadEvents returns array of arrays encoding individual events.
|
||||
// Each event has the following elements.
|
||||
//
|
||||
// Keep in sync with _EventConverter in file_patch.dart.
|
||||
enum {
|
||||
kEventFlagsIndex = 0, // See EventType
|
||||
kEventCookieIndex = 1, // Integer used to link move from and move to events
|
||||
kEventPathIndex = 2, // String path
|
||||
kEventPathIdIndex = 3, // Integer pathId
|
||||
kEventNumElements = 4,
|
||||
};
|
||||
|
||||
static void InitOnce();
|
||||
@@ -51,6 +57,10 @@ class FileSystemWatcher {
|
||||
static intptr_t GetSocketId(intptr_t id, intptr_t path_id);
|
||||
static Dart_Handle ReadEvents(intptr_t id, intptr_t path_id);
|
||||
|
||||
#if defined(DART_HOST_OS_MACOS) || defined(DART_HOST_OS_WINDOWS)
|
||||
static void DestroyWatch(intptr_t path_id);
|
||||
#endif
|
||||
|
||||
static void set_delayed_filewatch_callback(bool value) {
|
||||
delayed_filewatch_callback_ = value;
|
||||
}
|
||||
|
||||
@@ -88,6 +88,9 @@ static int InotifyEventToMask(struct inotify_event* e) {
|
||||
if ((e->mask & IN_MOVE) != 0) {
|
||||
mask |= FileSystemWatcher::kMove;
|
||||
}
|
||||
if ((e->mask & IN_MOVED_TO) != 0) {
|
||||
mask |= FileSystemWatcher::kMovedTo;
|
||||
}
|
||||
if ((e->mask & IN_DELETE) != 0) {
|
||||
mask |= FileSystemWatcher::kDelete;
|
||||
}
|
||||
@@ -118,28 +121,31 @@ Dart_Handle FileSystemWatcher::ReadEvents(intptr_t id, intptr_t path_id) {
|
||||
struct inotify_event* e =
|
||||
reinterpret_cast<struct inotify_event*>(buffer + offset);
|
||||
if ((e->mask & IN_IGNORED) == 0) {
|
||||
Dart_Handle event = Dart_NewList(5);
|
||||
Dart_Handle event = Dart_NewList(kEventNumElements);
|
||||
int mask = InotifyEventToMask(e);
|
||||
Dart_ListSetAt(event, 0, Dart_NewInteger(mask));
|
||||
Dart_ListSetAt(event, 1, Dart_NewInteger(e->cookie));
|
||||
Dart_ListSetAt(event, kEventFlagsIndex, Dart_NewInteger(mask));
|
||||
Dart_ListSetAt(event, kEventCookieIndex, Dart_NewInteger(e->cookie));
|
||||
if (e->len > 0) {
|
||||
Dart_Handle name = Dart_NewStringFromUTF8(
|
||||
reinterpret_cast<uint8_t*>(e->name), strlen(e->name));
|
||||
if (Dart_IsError(name)) {
|
||||
return name;
|
||||
}
|
||||
Dart_ListSetAt(event, 2, name);
|
||||
Dart_ListSetAt(event, kEventPathIndex, name);
|
||||
} else {
|
||||
Dart_ListSetAt(event, 2, Dart_Null());
|
||||
Dart_ListSetAt(event, kEventPathIndex, Dart_Null());
|
||||
}
|
||||
Dart_ListSetAt(event, 3, Dart_NewBoolean((e->mask & IN_MOVED_TO) != 0u));
|
||||
Dart_ListSetAt(event, 4, Dart_NewInteger(e->wd));
|
||||
Dart_ListSetAt(event, kEventPathIdIndex, Dart_NewInteger(e->wd));
|
||||
Dart_ListSetAt(events, i, event);
|
||||
i++;
|
||||
}
|
||||
offset += kEventSize + e->len;
|
||||
}
|
||||
ASSERT(offset == bytes);
|
||||
if (i == 0) {
|
||||
// No events in the chunk.
|
||||
return Dart_NewList(0);
|
||||
}
|
||||
return events;
|
||||
}
|
||||
|
||||
|
||||
@@ -211,6 +211,10 @@ void FileSystemWatcher::UnwatchPath(intptr_t id, intptr_t path_id) {
|
||||
Node::Unwatch(reinterpret_cast<Node*>(path_id));
|
||||
}
|
||||
|
||||
void FileSystemWatcher::DestroyWatch(intptr_t path_id) {
|
||||
FileSystemWatcher::UnwatchPath(0, path_id);
|
||||
}
|
||||
|
||||
intptr_t FileSystemWatcher::GetSocketId(intptr_t id, intptr_t path_id) {
|
||||
return reinterpret_cast<Node*>(path_id)->read_fd();
|
||||
}
|
||||
@@ -230,7 +234,7 @@ Dart_Handle FileSystemWatcher::ReadEvents(intptr_t id, intptr_t path_id) {
|
||||
return DartUtils::NewDartOSError();
|
||||
}
|
||||
size_t path_len = strlen(e.data.path);
|
||||
Dart_Handle event = Dart_NewList(5);
|
||||
Dart_Handle event = Dart_NewList(kEventNumElements);
|
||||
int flags = e.data.flags;
|
||||
int mask = 0;
|
||||
if ((flags & kFSEventStreamEventFlagItemRenamed) != 0) {
|
||||
@@ -261,16 +265,15 @@ Dart_Handle FileSystemWatcher::ReadEvents(intptr_t id, intptr_t path_id) {
|
||||
mask |= kDelete;
|
||||
}
|
||||
}
|
||||
Dart_ListSetAt(event, 0, Dart_NewInteger(mask));
|
||||
Dart_ListSetAt(event, 1, Dart_NewInteger(1));
|
||||
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, 2, name);
|
||||
Dart_ListSetAt(event, 3, Dart_NewBoolean(true));
|
||||
Dart_ListSetAt(event, 4, Dart_NewInteger(path_id));
|
||||
Dart_ListSetAt(event, kEventPathIndex, name);
|
||||
Dart_ListSetAt(event, kEventPathIdIndex, Dart_NewInteger(path_id));
|
||||
Dart_ListSetAt(events, i, event);
|
||||
}
|
||||
return events;
|
||||
@@ -299,6 +302,8 @@ bool FileSystemWatcher::IsSupported() {
|
||||
|
||||
void FileSystemWatcher::UnwatchPath(intptr_t id, intptr_t path_id) {}
|
||||
|
||||
void FileSystemWatcher::DestroyWatch(intptr_t path_id) {}
|
||||
|
||||
void FileSystemWatcher::InitOnce() {}
|
||||
|
||||
void FileSystemWatcher::Cleanup() {}
|
||||
|
||||
@@ -58,14 +58,20 @@ intptr_t FileSystemWatcher::WatchPath(intptr_t id,
|
||||
DirectoryWatchHandle* handle =
|
||||
new DirectoryWatchHandle(dir, list_events, recursive);
|
||||
handle->Start();
|
||||
handle->Retain();
|
||||
return reinterpret_cast<intptr_t>(handle);
|
||||
}
|
||||
|
||||
void FileSystemWatcher::DestroyWatch(intptr_t path_id) {
|
||||
FileSystemWatcher::UnwatchPath(0, path_id);
|
||||
}
|
||||
|
||||
void FileSystemWatcher::UnwatchPath(intptr_t id, intptr_t path_id) {
|
||||
USE(id);
|
||||
DirectoryWatchHandle* handle =
|
||||
reinterpret_cast<DirectoryWatchHandle*>(path_id);
|
||||
handle->Stop();
|
||||
handle->Release();
|
||||
}
|
||||
|
||||
intptr_t FileSystemWatcher::GetSocketId(intptr_t id, intptr_t path_id) {
|
||||
@@ -91,7 +97,7 @@ Dart_Handle FileSystemWatcher::ReadEvents(intptr_t id, intptr_t path_id) {
|
||||
FILE_NOTIFY_INFORMATION* e =
|
||||
reinterpret_cast<FILE_NOTIFY_INFORMATION*>(buffer + offset);
|
||||
|
||||
Dart_Handle event = Dart_NewList(5);
|
||||
Dart_Handle event = Dart_NewList(kEventNumElements);
|
||||
int mask = 0;
|
||||
if (e->Action == FILE_ACTION_ADDED) {
|
||||
mask |= kCreate;
|
||||
@@ -106,17 +112,16 @@ Dart_Handle FileSystemWatcher::ReadEvents(intptr_t id, intptr_t path_id) {
|
||||
mask |= kMove;
|
||||
}
|
||||
if (e->Action == FILE_ACTION_RENAMED_NEW_NAME) {
|
||||
mask |= kMove;
|
||||
mask |= kMove | kMovedTo;
|
||||
}
|
||||
Dart_ListSetAt(event, 0, Dart_NewInteger(mask));
|
||||
Dart_ListSetAt(event, kEventFlagsIndex, Dart_NewInteger(mask));
|
||||
// Move events come in pairs. Just 'enable' by default.
|
||||
Dart_ListSetAt(event, 1, Dart_NewInteger(1));
|
||||
Dart_ListSetAt(event, kEventCookieIndex, Dart_NewInteger(1));
|
||||
Dart_ListSetAt(
|
||||
event, 2,
|
||||
event, kEventPathIndex,
|
||||
Dart_NewStringFromUTF16(reinterpret_cast<uint16_t*>(e->FileName),
|
||||
e->FileNameLength / 2));
|
||||
Dart_ListSetAt(event, 3, Dart_NewBoolean(true));
|
||||
Dart_ListSetAt(event, 4, Dart_NewInteger(path_id));
|
||||
Dart_ListSetAt(event, kEventPathIdIndex, Dart_NewInteger(path_id));
|
||||
Dart_ListSetAt(events, i, event);
|
||||
i++;
|
||||
if (e->NextEntryOffset == 0) {
|
||||
|
||||
@@ -9,6 +9,7 @@
|
||||
|
||||
#include "bin/builtin.h"
|
||||
#include "bin/dartutils.h"
|
||||
#include "bin/file_system_watcher.h"
|
||||
#include "bin/socket_base.h"
|
||||
#include "include/dart_api.h"
|
||||
#include "platform/assert.h"
|
||||
@@ -205,6 +206,12 @@ namespace bin {
|
||||
V(X509_StartValidity, 1) \
|
||||
V(X509_EndValidity, 1)
|
||||
|
||||
#if defined(DART_HOST_OS_MACOS) || defined(DART_HOST_OS_WINDOWS)
|
||||
#define IO_FFI_NATIVE_LIST(V) V(FileSystemWatcher::DestroyWatch)
|
||||
#else
|
||||
#define IO_FFI_NATIVE_LIST(V)
|
||||
#endif
|
||||
|
||||
IO_NATIVE_LIST(DECLARE_FUNCTION);
|
||||
|
||||
static const struct NativeEntries {
|
||||
@@ -244,5 +251,23 @@ const uint8_t* IONativeSymbol(Dart_NativeFunction nf) {
|
||||
return nullptr;
|
||||
}
|
||||
|
||||
#define REGISTER_FFI_NATIVE_ENTRY(name) {#name, reinterpret_cast<void*>(&name)},
|
||||
|
||||
static const struct FfiNativeEntries {
|
||||
const char* const name_;
|
||||
void* const function_;
|
||||
} IOFfiEntries[] = {IO_FFI_NATIVE_LIST(REGISTER_FFI_NATIVE_ENTRY)};
|
||||
|
||||
void* IOFfiNativeLookup(const char* name, uintptr_t argument_count) {
|
||||
int num_entries = sizeof(IOFfiEntries) / sizeof(struct FfiNativeEntries);
|
||||
for (int i = 0; i < num_entries; i++) {
|
||||
const struct FfiNativeEntries* entry = &(IOFfiEntries[i]);
|
||||
if (strcmp(name, entry->name_) == 0) {
|
||||
return entry->function_;
|
||||
}
|
||||
}
|
||||
return nullptr;
|
||||
}
|
||||
|
||||
} // namespace bin
|
||||
} // namespace dart
|
||||
|
||||
@@ -16,6 +16,8 @@ Dart_NativeFunction IONativeLookup(Dart_Handle name,
|
||||
|
||||
const uint8_t* IONativeSymbol(Dart_NativeFunction nf);
|
||||
|
||||
void* IOFfiNativeLookup(const char* name, uintptr_t argument_count);
|
||||
|
||||
} // namespace bin
|
||||
} // namespace dart
|
||||
|
||||
|
||||
@@ -59,6 +59,10 @@ Dart_NativeFunction LookupIONative(Dart_Handle name,
|
||||
int argument_count,
|
||||
bool* auto_setup_scope);
|
||||
|
||||
// Performs a lookup of the I/O function with a specified 'name' and
|
||||
// 'argument_count'. Returns nullptr if matching function is not found.
|
||||
void* LookupIOFfiNative(const char* name, uintptr_t argument_count);
|
||||
|
||||
// Returns the symbol for I/O native function 'nf'. Returns NULL if 'nf' is not
|
||||
// a valid I/O native function.
|
||||
const uint8_t* LookupIONativeSymbol(Dart_NativeFunction nf);
|
||||
|
||||
@@ -1166,6 +1166,7 @@ void KernelLoader::LoadLibraryImportsAndExports(Library* library,
|
||||
library->url() != Symbols::DartCore().ptr() &&
|
||||
library->url() != Symbols::DartConcurrent().ptr() &&
|
||||
library->url() != Symbols::DartInternal().ptr() &&
|
||||
library->url() != Symbols::DartIo().ptr() &&
|
||||
library->url() != Symbols::DartFfi().ptr()) {
|
||||
H.ReportError(
|
||||
"import of dart:ffi is not supported in the current Dart runtime");
|
||||
|
||||
@@ -63,6 +63,7 @@ class ObjectPointerVisitor;
|
||||
V(DartFfi, "dart:ffi") \
|
||||
V(DartInternal, "dart:_internal") \
|
||||
V(DartIsVM, "dart.isVM") \
|
||||
V(DartIo, "dart:io") \
|
||||
V(DartIsolate, "dart:isolate") \
|
||||
V(DartLibrary, "dart.library.") \
|
||||
V(DartLibraryFfi, "dart.library.ffi") \
|
||||
|
||||
@@ -13,6 +13,7 @@ import "dart:async"
|
||||
show
|
||||
Completer,
|
||||
Future,
|
||||
MultiStreamController,
|
||||
Stream,
|
||||
StreamConsumer,
|
||||
StreamController,
|
||||
@@ -29,6 +30,8 @@ import "dart:developer" show registerExtension;
|
||||
|
||||
import "dart:isolate" show RawReceivePort, ReceivePort, SendPort;
|
||||
|
||||
import "dart:ffi" as ffi;
|
||||
|
||||
import "dart:math" show min;
|
||||
|
||||
import "dart:nativewrappers" show NativeFieldWrapperClass1;
|
||||
|
||||
@@ -131,290 +131,338 @@ base class _RandomAccessFileOpsImpl extends NativeFieldWrapperClass1
|
||||
external lock(int lock, int start, int end);
|
||||
}
|
||||
|
||||
class _WatcherPath {
|
||||
class _WatchedPath implements ffi.Finalizable {
|
||||
/// Path ID returned by [_FileSystemWatcher._watchPath].
|
||||
///
|
||||
/// Will remain valid until either [_unwatchPath] is called.
|
||||
final int pathId;
|
||||
|
||||
final String path;
|
||||
final int events;
|
||||
int count = 0;
|
||||
_WatcherPath(this.pathId, this.path, this.events);
|
||||
|
||||
/// Listeners subscribed to [FileSystemEvent] occuring at [path].
|
||||
///
|
||||
/// Once the last listener is unsubscribed the underlying watcher will
|
||||
/// stop monitoring changes at [path].
|
||||
final List<MultiStreamController<FileSystemEvent>> listeners =
|
||||
<MultiStreamController<FileSystemEvent>>[];
|
||||
|
||||
bool isClosed = false;
|
||||
|
||||
/// Source of [_NativeFSEvent] for this path.
|
||||
///
|
||||
/// This subscription will be cancelled once the last listener is gone.
|
||||
///
|
||||
/// Might be `null` if events for this path are delivered over multiplexed
|
||||
/// stream (see [_InotifyFileSystemWatcher]).
|
||||
StreamSubscription<List<_NativeFSEvent>>? source;
|
||||
|
||||
/// Finalizer associated with [_WatchedPath] instances.
|
||||
///
|
||||
/// [_FileSystemWatcher._watchPathImpl] returns `pathId` values which behave
|
||||
/// slightly differently on different OSes.
|
||||
///
|
||||
/// * On Mac OS X `pathId` values which are actually pointers to `Node`
|
||||
/// objects. These objects need to be released by calling
|
||||
/// [_FileSystemWatcher._unwatchPath] - otherwise they will leak.
|
||||
/// Attaching a [NativeFinalizer] ensures that even if Isolate exits
|
||||
/// abruptly via [Isolate.exit] we will still free these objects.
|
||||
/// * On Linux `pathId` is a _watch descriptor_ (returned by
|
||||
/// `inotify_add_watch`) associated with a specific inotify instance.
|
||||
/// Inotify instances are created by [_FileSystemWatcher._initWatcher] (see
|
||||
/// `inotify_init`) which returns a file descriptor. This file descriptor is
|
||||
/// wrapped in a [_NativeSocket] by
|
||||
/// [_FileSystemWatcher._eventsStreamFromSocket]. Socket takes ownership of
|
||||
/// the file descriptor and has a finalizer attached to itself. This
|
||||
/// finalizer will close the descriptor when socket is garbage collected.
|
||||
/// Thus there is no need to associate a separate finalizer with
|
||||
/// [_WatchedPath].
|
||||
/// * Windows is a mixture of Linux and Mac OS X: `pathId` itself is a
|
||||
/// pointer to a socket-like `DirectoryWatchHandle`. It is wrapped in a
|
||||
/// socket by [_FileSystemWatcher._eventsStreamFromSocket] by we do not
|
||||
/// allow the socket to take full ownership of the `pathId` handle, because
|
||||
/// we want to guarantee that invariant that `pathId` remains valid
|
||||
/// until we explicitly call `_FileSystemWatcher._unwatchPath`. To ensure
|
||||
/// this we explicitly retain `DirectoryWatchHandle` before returning it
|
||||
/// from [_FileSystemWatcher._initWatcher]. This will keep the handle
|
||||
/// alive even after event handler is done with it. This however means that
|
||||
/// [_FileSystemWatcher._unwatchPath] must be called to release the handle.
|
||||
/// Thus we need to attach a finalizer to [_WatchedPath] to guarantee that.
|
||||
static final ffi.NativeFinalizer? finalizer =
|
||||
Platform.isMacOS || Platform.isWindows
|
||||
? ffi.NativeFinalizer(
|
||||
ffi.Native.addressOf(_FileSystemWatcher._destroyWatch),
|
||||
)
|
||||
: null;
|
||||
|
||||
_WatchedPath(this.pathId, this.path, this.events) {
|
||||
finalizer?.attach(this, .fromAddress(pathId), detach: this);
|
||||
}
|
||||
|
||||
FutureOr<void> dispose() {
|
||||
assert(listeners.isEmpty);
|
||||
isClosed = true;
|
||||
finalizer?.detach(this);
|
||||
return source?.cancel();
|
||||
}
|
||||
|
||||
void add(FileSystemEvent event) {
|
||||
if (isClosed) {
|
||||
return;
|
||||
}
|
||||
|
||||
if ((event.type & events) == 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
for (var listener in listeners) {
|
||||
listener.add(event);
|
||||
}
|
||||
}
|
||||
|
||||
void close({Object? error}) {
|
||||
if (isClosed) {
|
||||
return;
|
||||
}
|
||||
|
||||
isClosed = true;
|
||||
for (var listener in listeners) {
|
||||
if (error != null) {
|
||||
listener.addError(error);
|
||||
}
|
||||
listener.close();
|
||||
}
|
||||
}
|
||||
|
||||
/// Emit the given [_NativeFSEvent] to listeners.
|
||||
///
|
||||
/// A single [_NativeFSEvent] is expanded into a sequence of appropriate
|
||||
/// [FileSystemEvent].
|
||||
///
|
||||
/// Note: this might modify [unmatchedMoves] - the caller is responsible for
|
||||
/// calling [flushUnmatchedMoves] once it emitted all events from a chunk
|
||||
/// of events.
|
||||
void addEvent(_NativeFSEvent event) {
|
||||
if (isClosed) {
|
||||
return;
|
||||
}
|
||||
|
||||
final flags = event.flags;
|
||||
final fullPath = fullPathOf(event);
|
||||
final isDir = _NativeFSEvent.isDirectory(event, fullPath);
|
||||
|
||||
if ((flags & FileSystemEvent.create) != 0) {
|
||||
add(FileSystemCreateEvent(fullPath, isDir));
|
||||
}
|
||||
|
||||
if ((flags & FileSystemEvent.modify) != 0) {
|
||||
add(FileSystemModifyEvent(fullPath, isDir, true));
|
||||
}
|
||||
|
||||
if ((flags & FileSystemEvent._modifyAttributes) != 0) {
|
||||
add(FileSystemModifyEvent(fullPath, isDir, false));
|
||||
}
|
||||
|
||||
if ((flags & FileSystemEvent.move) != 0) {
|
||||
// Use cookie to merge pairs of move from and move to events.
|
||||
final int cookie = event.cookie;
|
||||
if (cookie > 0) {
|
||||
if (unmatchedMoves.remove(cookie) case final linkedEvent?) {
|
||||
add(FileSystemMoveEvent(fullPathOf(linkedEvent), isDir, fullPath));
|
||||
} else {
|
||||
unmatchedMoves[cookie] = event;
|
||||
}
|
||||
} else {
|
||||
addMove(event, fullPath, isDir);
|
||||
}
|
||||
}
|
||||
|
||||
if ((flags & FileSystemEvent.delete) != 0) {
|
||||
add(FileSystemDeleteEvent(fullPath, false));
|
||||
}
|
||||
|
||||
if ((flags & FileSystemEvent._deleteSelf) != 0) {
|
||||
add(FileSystemDeleteEvent(fullPath, false));
|
||||
// Emit all unmatched moves before emitting the stop event to avoid
|
||||
// loosing these events.
|
||||
flushUnmatchedMoves();
|
||||
close();
|
||||
}
|
||||
}
|
||||
|
||||
/// Flush unmatched move events accumulated in [unmatchedMoves].
|
||||
void flushUnmatchedMoves() {
|
||||
for (var move in unmatchedMoves.values) {
|
||||
final fullPathOfMove = fullPathOf(move);
|
||||
addMove(
|
||||
move,
|
||||
fullPathOfMove,
|
||||
_NativeFSEvent.isDirectory(move, fullPathOf(move)),
|
||||
);
|
||||
}
|
||||
unmatchedMoves.clear();
|
||||
}
|
||||
|
||||
/// Most recently encountered move events by their [_NativeFSEvent.cookie].
|
||||
///
|
||||
/// When converting a chunk of _NativeFSEvent to corresponding FileSystemEvents
|
||||
/// we try to match and merge pairs of events which correspond to a single
|
||||
/// move operation (e.g. FILE_ACTION_RENAMED_{OLD|NEW}_NAME on Windows and
|
||||
/// IN_MOVED_{FROM|TO} on Linux). We use this as a temporary storage for
|
||||
/// matching. The caller feeding native events must call [flushUnmatchedMoves]
|
||||
/// at the end of the chunk to flush all unmatched moves.
|
||||
final Map<int, _NativeFSEvent> unmatchedMoves = {};
|
||||
|
||||
String fullPathOf(_NativeFSEvent event) {
|
||||
assert(event.pathId == pathId);
|
||||
if (event.relativePath case final eventPath? when eventPath.isNotEmpty) {
|
||||
return '${path}${Platform.pathSeparator}${eventPath}';
|
||||
} else {
|
||||
return path;
|
||||
}
|
||||
}
|
||||
|
||||
void addMove(_NativeFSEvent event, String fullPath, bool isDir) {
|
||||
if ((event.flags & FileSystemEvent._movedTo) != 0) {
|
||||
add(FileSystemCreateEvent(fullPath, isDir));
|
||||
} else {
|
||||
add(FileSystemDeleteEvent(fullPath, false));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@patch
|
||||
abstract class _FileSystemWatcher {
|
||||
void _pathWatchedEnd();
|
||||
|
||||
static int? _id;
|
||||
static final Map<int, _WatcherPath> _idMap = {};
|
||||
|
||||
final String _path;
|
||||
final int _events;
|
||||
final bool _recursive;
|
||||
|
||||
_WatcherPath? _watcherPath;
|
||||
|
||||
final StreamController<FileSystemEvent> _broadcastController =
|
||||
StreamController<FileSystemEvent>.broadcast();
|
||||
|
||||
/// Subscription on the stream returned by [_watchPath].
|
||||
///
|
||||
/// Stored while piping events from that stream into [_broadcastController],
|
||||
/// so it can be cancelled when [_broadcastController] is cancelled.
|
||||
StreamSubscription? _sourceSubscription;
|
||||
|
||||
@patch
|
||||
static Stream<FileSystemEvent> _watch(
|
||||
String path,
|
||||
int events,
|
||||
bool recursive,
|
||||
) {
|
||||
if (Platform.isLinux || Platform.isAndroid) {
|
||||
return _InotifyFileSystemWatcher(path, events, recursive)._stream;
|
||||
}
|
||||
if (Platform.isWindows) {
|
||||
return _Win32FileSystemWatcher(path, events, recursive)._stream;
|
||||
}
|
||||
if (Platform.isMacOS) {
|
||||
return _FSEventStreamFileSystemWatcher(path, events, recursive)._stream;
|
||||
}
|
||||
throw FileSystemException(
|
||||
"File system watching is not supported on this platform",
|
||||
);
|
||||
return _watcher._watchImpl(path, events, recursive);
|
||||
}
|
||||
|
||||
_FileSystemWatcher._(this._path, this._events, this._recursive) {
|
||||
if (!isSupported) {
|
||||
throw FileSystemException(
|
||||
"File system watching is not supported on this platform",
|
||||
_path,
|
||||
);
|
||||
}
|
||||
_broadcastController
|
||||
..onListen = _listen
|
||||
..onCancel = _cancel;
|
||||
}
|
||||
|
||||
Stream<FileSystemEvent> get _stream => _broadcastController.stream;
|
||||
|
||||
void _listen() {
|
||||
if (_id == null) {
|
||||
Stream<FileSystemEvent> _watchImpl(String path, int events, bool recursive) {
|
||||
_WatchedPath? watchedPath;
|
||||
final stream = Stream<FileSystemEvent>.multi((controller) {
|
||||
_WatchedPath wp;
|
||||
try {
|
||||
_id = _initWatcher();
|
||||
_newWatcher();
|
||||
} on dynamic catch (e) {
|
||||
_broadcastController.addError(
|
||||
FileSystemException._fromOSError(
|
||||
e,
|
||||
"Failed to initialize file system entity watcher",
|
||||
_path,
|
||||
),
|
||||
);
|
||||
_broadcastController.close();
|
||||
wp = watchedPath ??= _watcher._startWatching(path, events, recursive);
|
||||
} on FileSystemException catch (e, st) {
|
||||
controller.addError(e, st);
|
||||
controller.close();
|
||||
return;
|
||||
}
|
||||
}
|
||||
var pathId;
|
||||
try {
|
||||
pathId = _watchPath(
|
||||
_id!,
|
||||
_Namespace._namespace,
|
||||
_path,
|
||||
_events,
|
||||
_recursive,
|
||||
);
|
||||
} on dynamic catch (e) {
|
||||
_broadcastController.addError(
|
||||
FileSystemException._fromOSError(e, "Failed to watch path", _path),
|
||||
);
|
||||
_broadcastController.close();
|
||||
return;
|
||||
}
|
||||
if (!_idMap.containsKey(pathId)) {
|
||||
_idMap[pathId] = _WatcherPath(pathId, _path, _events);
|
||||
}
|
||||
_watcherPath = _idMap[pathId];
|
||||
_watcherPath!.count++;
|
||||
_sourceSubscription = _pathWatched().listen(
|
||||
_broadcastController.add,
|
||||
onError: _broadcastController.addError,
|
||||
onDone: _broadcastController.close,
|
||||
);
|
||||
}
|
||||
|
||||
void _cancel() {
|
||||
final watcherPath = _watcherPath;
|
||||
if (watcherPath != null) {
|
||||
assert(watcherPath.count > 0);
|
||||
watcherPath.count--;
|
||||
if (watcherPath.count == 0) {
|
||||
var pathId = watcherPath.pathId;
|
||||
// DirectoryWatchHandle(aka pathId) might be closed already initiated
|
||||
// by issueReadEvent for example. When that happens, appropriate closeEvent
|
||||
// will arrive to us and we will remove this pathId from _idMap. If that
|
||||
// happens we should not try to close it again as pathId is no
|
||||
// longer usable(the memory it points to might be released)
|
||||
if (_idMap.containsKey(pathId)) {
|
||||
_unwatchPath(_id!, pathId);
|
||||
_pathWatchedEnd();
|
||||
_idMap.remove(pathId);
|
||||
}
|
||||
if (wp.isClosed) {
|
||||
controller.addError(
|
||||
FileSystemException('Directory watcher is already closed', path),
|
||||
);
|
||||
controller.close();
|
||||
return;
|
||||
}
|
||||
_watcherPath = null;
|
||||
}
|
||||
final id = _id;
|
||||
if (_idMap.isEmpty && id != null) {
|
||||
_doneWatcher();
|
||||
_id = null;
|
||||
}
|
||||
_sourceSubscription?.cancel();
|
||||
_sourceSubscription = null;
|
||||
|
||||
wp.listeners.add(controller);
|
||||
controller.onCancel = () {
|
||||
wp.listeners.remove(controller);
|
||||
if (wp.listeners.isEmpty) {
|
||||
watchedPath = null;
|
||||
return _stopWatching(wp);
|
||||
}
|
||||
};
|
||||
});
|
||||
|
||||
return stream;
|
||||
}
|
||||
|
||||
// Called when (and after) a new watcher instance is created and available.
|
||||
void _newWatcher() {}
|
||||
// Called when a watcher is no longer needed.
|
||||
void _doneWatcher() {}
|
||||
// Called when a new path is being watched.
|
||||
Stream<FileSystemEvent> _pathWatched();
|
||||
// Called when a path is no longer being watched.
|
||||
void _donePathWatched() {}
|
||||
|
||||
static _WatcherPath _pathFromPathId(int pathId) {
|
||||
return _idMap[pathId]!;
|
||||
}
|
||||
|
||||
static Stream _listenOnSocket(int socketId, int id, int pathId) {
|
||||
Stream<List<_NativeFSEvent>> _eventsStreamFromSocket(
|
||||
int socketId,
|
||||
int pathId,
|
||||
) {
|
||||
final nativeSocket = _NativeSocket._watch(socketId);
|
||||
final rawSocket = _RawSocket(nativeSocket);
|
||||
return rawSocket.expand((event) {
|
||||
var stops = [];
|
||||
var events = [];
|
||||
var pair = {};
|
||||
return _RawSocket(nativeSocket).map((event) {
|
||||
if (event == RawSocketEvent.read) {
|
||||
String getPath(event) {
|
||||
var path = _pathFromPathId(event[4]).path;
|
||||
if (event[2] != null && event[2].isNotEmpty) {
|
||||
path += Platform.pathSeparator;
|
||||
path += event[2];
|
||||
}
|
||||
return path;
|
||||
}
|
||||
final result = <_NativeFSEvent>[];
|
||||
|
||||
bool getIsDir(event) {
|
||||
if (Platform.isWindows) {
|
||||
// Windows does not get 'isDir' as part of the event.
|
||||
// Links should also be skipped.
|
||||
return FileSystemEntity.isDirectorySync(getPath(event)) &&
|
||||
!FileSystemEntity.isLinkSync(getPath(event));
|
||||
}
|
||||
return (event[0] & FileSystemEvent._isDir) != 0;
|
||||
}
|
||||
|
||||
void add(id, event) {
|
||||
if ((event.type & _pathFromPathId(id).events) == 0) return;
|
||||
events.add([id, event]);
|
||||
}
|
||||
|
||||
void rewriteMove(event, isDir) {
|
||||
if (event[3]) {
|
||||
add(event[4], FileSystemCreateEvent(getPath(event), isDir));
|
||||
} else {
|
||||
add(event[4], FileSystemDeleteEvent(getPath(event), false));
|
||||
}
|
||||
}
|
||||
|
||||
int eventCount;
|
||||
int totalEvents;
|
||||
do {
|
||||
eventCount = 0;
|
||||
for (var event in _readEvents(id, pathId)) {
|
||||
if (event == null) continue;
|
||||
eventCount++;
|
||||
int pathId = event[4];
|
||||
if (!_idMap.containsKey(pathId)) {
|
||||
// Path is no longer being wathed.
|
||||
continue;
|
||||
}
|
||||
bool isDir = getIsDir(event);
|
||||
var path = getPath(event);
|
||||
if ((event[0] & FileSystemEvent.create) != 0) {
|
||||
add(event[4], FileSystemCreateEvent(path, isDir));
|
||||
}
|
||||
if ((event[0] & FileSystemEvent.modify) != 0) {
|
||||
add(event[4], FileSystemModifyEvent(path, isDir, true));
|
||||
}
|
||||
if ((event[0] & FileSystemEvent._modifyAttributes) != 0) {
|
||||
add(event[4], FileSystemModifyEvent(path, isDir, false));
|
||||
}
|
||||
if ((event[0] & FileSystemEvent.move) != 0) {
|
||||
int link = event[1];
|
||||
if (link > 0) {
|
||||
pair.putIfAbsent(pathId, () => {});
|
||||
if (pair[pathId].containsKey(link)) {
|
||||
add(
|
||||
event[4],
|
||||
FileSystemMoveEvent(
|
||||
getPath(pair[pathId][link]),
|
||||
isDir,
|
||||
path,
|
||||
),
|
||||
);
|
||||
pair[pathId].remove(link);
|
||||
} else {
|
||||
pair[pathId][link] = event;
|
||||
}
|
||||
} else {
|
||||
rewriteMove(event, isDir);
|
||||
}
|
||||
}
|
||||
if ((event[0] & FileSystemEvent.delete) != 0) {
|
||||
add(event[4], FileSystemDeleteEvent(path, false));
|
||||
}
|
||||
if ((event[0] & FileSystemEvent._deleteSelf) != 0) {
|
||||
add(event[4], FileSystemDeleteEvent(path, false));
|
||||
// Signal done event.
|
||||
stops.add([event[4], null]);
|
||||
totalEvents = result.length;
|
||||
for (_NativeFSEvent? e in _readEvents(_watcherId, pathId)) {
|
||||
if (e == null) {
|
||||
break;
|
||||
}
|
||||
result.add(e);
|
||||
}
|
||||
} while (eventCount > 0);
|
||||
} while (result.length > totalEvents);
|
||||
|
||||
// Be sure to clear this manually, as the sockets are not read through
|
||||
// the _NativeSocket interface.
|
||||
nativeSocket.available = 0;
|
||||
for (var map in pair.values) {
|
||||
for (var event in map.values) {
|
||||
rewriteMove(event, getIsDir(event));
|
||||
}
|
||||
}
|
||||
} else if (event == RawSocketEvent.closed) {
|
||||
// After this point we should not try to do anything with pathId as
|
||||
// the handle it represented is closed and gone now.
|
||||
if (_idMap.containsKey(pathId)) {
|
||||
_idMap.remove(pathId);
|
||||
if (_idMap.isEmpty && _id != null) {
|
||||
_id = null;
|
||||
}
|
||||
}
|
||||
} else if (event == RawSocketEvent.readClosed) {
|
||||
// If Directory watcher buffer overflows, it will send an readClosed event.
|
||||
// Normal closing will cancel stream subscription so that path is
|
||||
// no longer being watched, not present in _idMap.
|
||||
if (_idMap.containsKey(pathId)) {
|
||||
var path = _pathFromPathId(pathId).path;
|
||||
_idMap.remove(pathId);
|
||||
if (_idMap.isEmpty && _id != null) {
|
||||
_id = null;
|
||||
}
|
||||
throw FileSystemException(
|
||||
'Directory watcher closed unexpectedly',
|
||||
path,
|
||||
);
|
||||
}
|
||||
} else {
|
||||
assert(false);
|
||||
|
||||
return result;
|
||||
}
|
||||
events.addAll(stops);
|
||||
return events;
|
||||
return [];
|
||||
});
|
||||
}
|
||||
|
||||
/// Native ID associated with the watcher.
|
||||
///
|
||||
/// Passed as a parameter to [_watchPathImpl] and other native functions.
|
||||
int get _watcherId => 0;
|
||||
|
||||
/// Start watching the given path for the specified events.
|
||||
///
|
||||
/// Returns [_WatchedPath] instances representing the watch.
|
||||
_WatchedPath _startWatching(String path, int events, bool recursive);
|
||||
|
||||
/// Stop watching the given path.
|
||||
///
|
||||
/// If this causes the watcher to free some native resources (e.g. because
|
||||
/// this was the last active filesystem watch) this function will return
|
||||
/// an instance of [Future] which will complete after native cleanup is
|
||||
/// complete.
|
||||
FutureOr<void> _stopWatching(_WatchedPath wp) {
|
||||
_unwatchPath(_watcherId, wp.pathId);
|
||||
return wp.dispose();
|
||||
}
|
||||
|
||||
/// Wrapper over [_watchPathImpl] which takes care of converting
|
||||
/// [OSError] into [FileSystemException].
|
||||
int _watchPath(String path, int events, bool recursive) {
|
||||
try {
|
||||
return _watchPathImpl(
|
||||
_watcherId,
|
||||
_Namespace._namespace,
|
||||
path,
|
||||
events,
|
||||
recursive,
|
||||
);
|
||||
} on OSError catch (e) {
|
||||
throw FileSystemException._fromOSError(e, "Failed to watch path", path);
|
||||
}
|
||||
}
|
||||
|
||||
/// Singleton [_FileSystemWatcher] which takes care of watching file system.
|
||||
static _FileSystemWatcher _watcher = () {
|
||||
if (isSupported) {
|
||||
if (Platform.isLinux || Platform.isAndroid) {
|
||||
return _InotifyFileSystemWatcher();
|
||||
}
|
||||
|
||||
if (Platform.isWindows) {
|
||||
return _SocketPerPathFileSystemWatcher();
|
||||
}
|
||||
|
||||
if (Platform.isMacOS) {
|
||||
return _SocketPerPathFileSystemWatcher();
|
||||
}
|
||||
}
|
||||
|
||||
throw FileSystemException(
|
||||
"File system watching is not supported on this platform",
|
||||
);
|
||||
}();
|
||||
|
||||
@patch
|
||||
@pragma("vm:external-name", "FileSystemWatcher_IsSupported")
|
||||
external static bool get isSupported;
|
||||
@@ -423,116 +471,185 @@ abstract class _FileSystemWatcher {
|
||||
external static int _initWatcher();
|
||||
|
||||
@pragma("vm:external-name", "FileSystemWatcher_WatchPath")
|
||||
external static int _watchPath(
|
||||
int id,
|
||||
external static int _watchPathImpl(
|
||||
int watcherId,
|
||||
_Namespace namespace,
|
||||
String path,
|
||||
int events,
|
||||
bool recursive,
|
||||
);
|
||||
@pragma("vm:external-name", "FileSystemWatcher_UnwatchPath")
|
||||
external static void _unwatchPath(int id, int path_id);
|
||||
external static void _unwatchPath(int watcherId, int pathId);
|
||||
|
||||
/// Returns a list each element of which is [_NativeFSEvents] or `null`.
|
||||
///
|
||||
/// After the first `null` only `null` entries will follow, in other words
|
||||
/// all non-`null` entries form the prefix of the list.
|
||||
@pragma("vm:external-name", "FileSystemWatcher_ReadEvents")
|
||||
external static List _readEvents(int id, int path_id);
|
||||
external static List _readEvents(int watcherId, int pathId);
|
||||
|
||||
@pragma("vm:external-name", "FileSystemWatcher_GetSocketId")
|
||||
external static int _getSocketId(int id, int path_id);
|
||||
external static int _getSocketId(int watcherId, int pathId);
|
||||
|
||||
@ffi.Native<ffi.Void Function(ffi.Pointer<ffi.Void>)>(
|
||||
symbol: "FileSystemWatcher::DestroyWatch",
|
||||
)
|
||||
external static void _destroyWatch(ffi.Pointer<ffi.Void> pathId);
|
||||
}
|
||||
|
||||
class _InotifyFileSystemWatcher extends _FileSystemWatcher {
|
||||
static final Map<int, StreamController<FileSystemEvent>> _idMap = {};
|
||||
static late StreamSubscription _subscription;
|
||||
final Map<int, _WatchedPath> _watchedPaths = <int, _WatchedPath>{};
|
||||
|
||||
_InotifyFileSystemWatcher(path, events, recursive)
|
||||
: super._(path, events, recursive);
|
||||
int? _inotifyFd;
|
||||
StreamSubscription<List<_NativeFSEvent>>? _inotifySubscription;
|
||||
|
||||
void _newWatcher() {
|
||||
int id = _FileSystemWatcher._id!;
|
||||
_subscription = _FileSystemWatcher._listenOnSocket(id, id, 0).listen((
|
||||
event,
|
||||
) {
|
||||
if (_idMap.containsKey(event[0])) {
|
||||
if (event[1] != null) {
|
||||
_idMap[event[0]]!.add(event[1]);
|
||||
} else {
|
||||
_idMap[event[0]]!.close();
|
||||
@override
|
||||
int get _watcherId => _inotifyFd!;
|
||||
|
||||
void _ensureInotifyFD() {
|
||||
if (_inotifyFd != null) {
|
||||
return;
|
||||
}
|
||||
|
||||
final inotifyFd = _inotifyFd = _FileSystemWatcher._initWatcher();
|
||||
_inotifySubscription = _eventsStreamFromSocket(
|
||||
inotifyFd,
|
||||
0,
|
||||
).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;
|
||||
|
||||
// Distribute events to corresponding _WatchedPath objects based on
|
||||
// pathId.
|
||||
for (_NativeFSEvent event in events) {
|
||||
if (_watchedPaths[event.pathId] case final watchedPath?) {
|
||||
watchedPath.addEvent(event);
|
||||
if (watchedPath.unmatchedMoves.isNotEmpty) {
|
||||
(dirty ??= {}).add(watchedPath);
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
void _doneWatcher() {
|
||||
_subscription.cancel();
|
||||
}
|
||||
|
||||
Stream<FileSystemEvent> _pathWatched() {
|
||||
var pathId = _watcherPath!.pathId;
|
||||
if (!_idMap.containsKey(pathId)) {
|
||||
_idMap[pathId] = StreamController<FileSystemEvent>.broadcast();
|
||||
}
|
||||
return _idMap[pathId]!.stream;
|
||||
|
||||
if (dirty != null) {
|
||||
for (var watchedPath in dirty) {
|
||||
watchedPath.flushUnmatchedMoves();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void _pathWatchedEnd() {
|
||||
var pathId = _watcherPath!.pathId;
|
||||
if (!_idMap.containsKey(pathId)) return;
|
||||
_idMap[pathId]!.close();
|
||||
_idMap.remove(pathId);
|
||||
@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.
|
||||
final subscription = _inotifySubscription;
|
||||
if (_watchedPaths.isEmpty && subscription != null) {
|
||||
_inotifyFd = null;
|
||||
_inotifySubscription = null;
|
||||
return subscription.cancel();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
class _Win32FileSystemWatcher extends _FileSystemWatcher {
|
||||
late StreamSubscription _subscription;
|
||||
late StreamController<FileSystemEvent> _controller;
|
||||
|
||||
_Win32FileSystemWatcher(path, events, recursive)
|
||||
: super._(path, events, recursive);
|
||||
|
||||
Stream<FileSystemEvent> _pathWatched() {
|
||||
var pathId = _watcherPath!.pathId;
|
||||
_controller = StreamController<FileSystemEvent>();
|
||||
_subscription = _FileSystemWatcher._listenOnSocket(pathId, 0, pathId)
|
||||
.listen((event) {
|
||||
assert(event[0] == pathId);
|
||||
if (event[1] != null) {
|
||||
_controller.add(event[1]);
|
||||
} else {
|
||||
_controller.close();
|
||||
}
|
||||
});
|
||||
return _controller.stream;
|
||||
}
|
||||
|
||||
void _pathWatchedEnd() {
|
||||
_subscription.cancel();
|
||||
_controller.close();
|
||||
class _SocketPerPathFileSystemWatcher extends _FileSystemWatcher {
|
||||
@override
|
||||
_WatchedPath _startWatching(String path, int events, bool recursive) {
|
||||
final watchedPath = _WatchedPath(
|
||||
_watchPath(path, events, recursive),
|
||||
path,
|
||||
events,
|
||||
);
|
||||
watchedPath.source =
|
||||
_eventsStreamFromSocket(
|
||||
_FileSystemWatcher._getSocketId(0, watchedPath.pathId),
|
||||
watchedPath.pathId,
|
||||
).listen(
|
||||
(events) {
|
||||
for (var e in events) {
|
||||
watchedPath.addEvent(e);
|
||||
}
|
||||
watchedPath.flushUnmatchedMoves();
|
||||
},
|
||||
onError: (error) {
|
||||
if (watchedPath.listeners.isNotEmpty) {
|
||||
watchedPath.close(
|
||||
error: FileSystemException(
|
||||
'Directory watcher failed due to: $error',
|
||||
watchedPath.path,
|
||||
),
|
||||
);
|
||||
}
|
||||
},
|
||||
onDone: () {
|
||||
if (watchedPath.listeners.isNotEmpty) {
|
||||
watchedPath.close(
|
||||
error: FileSystemException(
|
||||
'Directory watcher closed unexpectedly',
|
||||
watchedPath.path,
|
||||
),
|
||||
);
|
||||
}
|
||||
},
|
||||
cancelOnError: true,
|
||||
);
|
||||
return watchedPath;
|
||||
}
|
||||
}
|
||||
|
||||
class _FSEventStreamFileSystemWatcher extends _FileSystemWatcher {
|
||||
late StreamSubscription _subscription;
|
||||
late StreamController<FileSystemEvent> _controller;
|
||||
extension type _NativeFSEvent(List<dynamic> _) {
|
||||
// See FileSystemWatcher::kEvent*Index constants.
|
||||
static const int flagsIndex = 0;
|
||||
static const int cookieIndex = 1;
|
||||
static const int pathIndex = 2;
|
||||
static const int pathIdIndex = 3;
|
||||
|
||||
_FSEventStreamFileSystemWatcher(path, events, recursive)
|
||||
: super._(path, events, recursive);
|
||||
int get flags => this._[flagsIndex];
|
||||
|
||||
Stream<FileSystemEvent> _pathWatched() {
|
||||
var pathId = _watcherPath!.pathId;
|
||||
var socketId = _FileSystemWatcher._getSocketId(0, pathId);
|
||||
_controller = StreamController<FileSystemEvent>();
|
||||
_subscription = _FileSystemWatcher._listenOnSocket(socketId, 0, pathId)
|
||||
.listen((event) {
|
||||
if (event[1] != null) {
|
||||
_controller.add(event[1]);
|
||||
} else {
|
||||
_controller.close();
|
||||
}
|
||||
});
|
||||
return _controller.stream;
|
||||
}
|
||||
/// A unique identifier (32-bit unsigned integer) for matching related events.
|
||||
///
|
||||
/// On Linux (inotify) associates unique cookie values with pairs of
|
||||
/// `IN_MOVED_FROM` and `IN_MOVED_TO` events.
|
||||
///
|
||||
/// On Windows we set cookie to `1` on pairs of `FILE_ACTION_RENAMED_OLD_NAME`
|
||||
/// and `FILE_ACTION_RENAMED_NEW_NAME`.
|
||||
///
|
||||
/// Not used on Mac OS X (because `FSEventStream` does not generate move
|
||||
/// events).
|
||||
int get cookie => this._[cookieIndex];
|
||||
|
||||
void _pathWatchedEnd() {
|
||||
_subscription.cancel();
|
||||
_controller.close();
|
||||
String? get relativePath => this._[pathIndex];
|
||||
|
||||
int get pathId => this._[pathIdIndex];
|
||||
|
||||
static bool isDirectory(_NativeFSEvent event, String fullPath) {
|
||||
if (Platform.isWindows) {
|
||||
// Windows does not get FileSystemEvent._isDir bit as part of the event
|
||||
// so we need to compute it by checking the file-system. We ignore links
|
||||
// when computing isDirectory.
|
||||
return FileSystemEntity._isDirectoryIgnoringLinksSync(fullPath);
|
||||
} else {
|
||||
return (event.flags & FileSystemEvent._isDir) != 0;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -774,6 +774,10 @@ abstract class FileSystemEntity {
|
||||
(_getTypeSync(_toUtf8Array(path), true) ==
|
||||
FileSystemEntityType.directory);
|
||||
|
||||
static bool _isDirectoryIgnoringLinksSync(String path) =>
|
||||
(_getTypeSync(_toUtf8Array(path), false) ==
|
||||
FileSystemEntityType.directory);
|
||||
|
||||
external static _getTypeNative(
|
||||
_Namespace namespace,
|
||||
Uint8List rawPath,
|
||||
@@ -946,6 +950,7 @@ sealed class FileSystemEvent {
|
||||
static const int _modifyAttributes = 1 << 4;
|
||||
static const int _deleteSelf = 1 << 5;
|
||||
static const int _isDir = 1 << 6;
|
||||
static const int _movedTo = 1 << 7;
|
||||
|
||||
/// The type of event. See [FileSystemEvent] for a list of events.
|
||||
final int type;
|
||||
|
||||
@@ -0,0 +1,37 @@
|
||||
// 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.
|
||||
|
||||
// Verify that active [Directory.watch] do not leak when created inside an
|
||||
// isolate which then exits.
|
||||
//
|
||||
// On Mac OS X FileSystemWatcher::Cleanup will hang until all watchers are
|
||||
// destroyed.
|
||||
|
||||
import 'dart:io';
|
||||
import 'dart:isolate';
|
||||
|
||||
import 'package:expect/expect.dart';
|
||||
|
||||
void main() async {
|
||||
// Wait to give `vm-service` isolate time to start up and initialize,
|
||||
// because starting vm-service will consume memory and increase RSS.
|
||||
await Future.delayed(const Duration(seconds: 1));
|
||||
|
||||
final startRss = ProcessInfo.currentRss;
|
||||
for (var i = 0; i < 500; i++) {
|
||||
await Isolate.run(() async {
|
||||
Directory.systemTemp.watch().listen((event) {});
|
||||
// Give the watcher a chance to start before exiting the isolate.
|
||||
await Future.delayed(const Duration(milliseconds: 10));
|
||||
});
|
||||
}
|
||||
|
||||
final endRss = ProcessInfo.currentRss;
|
||||
final allocatedBytes = (endRss - startRss);
|
||||
final limit = 10 * 1024 * 1024;
|
||||
Expect.isTrue(
|
||||
allocatedBytes < limit,
|
||||
'expected VM RSS growth to be below ${limit} but got ${allocatedBytes}',
|
||||
);
|
||||
}
|
||||
@@ -0,0 +1,70 @@
|
||||
// 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.
|
||||
//
|
||||
// Verify that on Windows FileSystemEntity.watch does not enter incorrect state
|
||||
// after overflow (https://dartbug.com/61378). On other platforms we just
|
||||
// expect watcher to never experience overflow receive events without issues.
|
||||
//
|
||||
// This was happening because Dart side of things was using raw addresses of
|
||||
// DirectoryWatchHandle's created by native allocator as keys in a map
|
||||
// inside _FileSystemWatcher implementation without ensuring that map is cleared
|
||||
// before native memory is freed and reused for another similar object -
|
||||
// thus making Dart side confuse two unrelated entities.
|
||||
|
||||
import 'dart:async';
|
||||
import 'dart:io';
|
||||
|
||||
import 'package:expect/async_helper.dart';
|
||||
import 'package:expect/expect.dart';
|
||||
|
||||
var eventsSeen = 0;
|
||||
var restartedTimes = 0;
|
||||
|
||||
Future<void> main() async {
|
||||
asyncStart();
|
||||
final temp = Directory.systemTemp.createTempSync('regress-61378');
|
||||
// Long file name to consume more space in the OS buffer which contains
|
||||
// file system events.
|
||||
final file = File('${temp.path}/file'.padRight(255, 'a'));
|
||||
try {
|
||||
startWatcher(temp);
|
||||
// Iteration numbers are selected based on experiments to maximize the
|
||||
// chance of error without significantly increase test running time.
|
||||
for (var times = 0; times < 20; ++times) {
|
||||
eventsSeen = 0;
|
||||
// Cause the watcher buffer to fill in a sync block.
|
||||
for (var i = 0; i < 200; ++i) {
|
||||
file.writeAsStringSync('$i');
|
||||
}
|
||||
// Allow async processing so the error has chance to happen.
|
||||
await Future.delayed(Duration(milliseconds: 100));
|
||||
if (Platform.isWindows) {
|
||||
Expect.isTrue(restartedTimes > 0, 'Expected some restarts to happen');
|
||||
}
|
||||
Expect.isTrue(eventsSeen > 0, 'Watcher did not get any events');
|
||||
}
|
||||
await subscription.cancel();
|
||||
asyncEnd();
|
||||
} finally {
|
||||
temp.deleteSync(recursive: true);
|
||||
}
|
||||
}
|
||||
|
||||
late StreamSubscription subscription;
|
||||
|
||||
void startWatcher(Directory temp) {
|
||||
subscription = temp.watch().listen(
|
||||
(e) {
|
||||
++eventsSeen;
|
||||
},
|
||||
onError: (e) async {
|
||||
restartedTimes++;
|
||||
await subscription.cancel();
|
||||
startWatcher(temp);
|
||||
},
|
||||
onDone: () {
|
||||
Expect.fail('Not expecting DONE.');
|
||||
},
|
||||
);
|
||||
}
|
||||
Reference in New Issue
Block a user