[io/mac] Ensure FSEventsWatcher::Node is deleted synchronously with Callback that uses it.

This is follow-up to ed82bb6f4c

TEST=tests/standalone/io/file_system_watcher_large_set_test.dart

Change-Id: If02c922eafe1371c6e67196158896b9cb786bfd6
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/202312
Commit-Queue: Alexander Aprelev <aam@google.com>
Reviewed-by: Slava Egorov <vegorov@google.com>

runtime/bin/file_system_watcher.cc

@@ -14,6 +14,8 @@
 namespace dart {
 namespace bin {
 
+bool FileSystemWatcher::delayed_filewatch_callback_ = false;
+
 void FUNCTION_NAME(FileSystemWatcher_IsSupported)(Dart_NativeArguments args) {
   Dart_SetBooleanReturnValue(args, FileSystemWatcher::IsSupported());
 }

runtime/bin/file_system_watcher.h

@@ -48,7 +48,15 @@ class FileSystemWatcher {
   static intptr_t GetSocketId(intptr_t id, intptr_t path_id);
   static Dart_Handle ReadEvents(intptr_t id, intptr_t path_id);
 
+  static void set_delayed_filewatch_callback(bool value) {
+    delayed_filewatch_callback_ = value;
+  }
+  static bool delayed_filewatch_callback() {
+    return delayed_filewatch_callback_;
+  }
+
  private:
+  static bool delayed_filewatch_callback_;
   DISALLOW_COPY_AND_ASSIGN(FileSystemWatcher);
 };
 

runtime/bin/file_system_watcher_macos.cc

@@ -61,7 +61,6 @@ class FSEventsWatcher {
          int write_fd,
          bool recursive)
         : watcher_(watcher),
-          ready_(false),
           base_path_length_(strlen(base_path)),
           path_ref_(CFStringCreateWithCString(NULL,
                                               base_path,
@@ -74,9 +73,14 @@ class FSEventsWatcher {
     }
 
     ~Node() {
-      Stop();
+      // This is invoked outside of [Callback] execution because
+      // [context.release] callback is invoked when [FSEventStream] is
+      // 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_);
+      watcher_ = nullptr;  // this is to catch access-after-free in Callback
     }
 
     void set_ref(FSEventStreamRef ref) { ref_ = ref; }
@@ -85,6 +89,9 @@ class FSEventsWatcher {
       FSEventStreamContext context;
       memset(&context, 0, sizeof(context));
       context.info = reinterpret_cast<void*>(this);
+      context.release = [](const void* info) {
+        delete static_cast<const Node*>(info);
+      };
       CFArrayRef array = CFArrayCreate(
           NULL, reinterpret_cast<const void**>(&path_ref_), 1, NULL);
       FSEventStreamRef ref = FSEventStreamCreate(
@@ -93,7 +100,6 @@ class FSEventsWatcher {
       CFRelease(array);
 
       set_ref(ref);
-      ready_.store(true, std::memory_order_release);
 
       FSEventStreamScheduleWithRunLoop(ref_, watcher_->run_loop_,
                                        kCFRunLoopDefaultMode);
@@ -103,15 +109,12 @@ class FSEventsWatcher {
     }
 
     void Stop() {
-      ASSERT(ready_);
       FSEventStreamStop(ref_);
       FSEventStreamInvalidate(ref_);
       FSEventStreamRelease(ref_);
-      ready_.store(false, std::memory_order_release);
     }
 
     FSEventsWatcher* watcher() const { return watcher_; }
-    bool ready() const { return ready_.load(std::memory_order_acquire); }
     intptr_t base_path_length() const { return base_path_length_; }
     int read_fd() const { return read_fd_; }
     int write_fd() const { return write_fd_; }
@@ -119,7 +122,6 @@ class FSEventsWatcher {
 
    private:
     FSEventsWatcher* watcher_;
-    std::atomic<bool> ready_;
     intptr_t base_path_length_;
     CFStringRef path_ref_;
     int read_fd_;
@@ -218,12 +220,15 @@ class FSEventsWatcher {
                        void* event_paths,
                        const FSEventStreamEventFlags event_flags[],
                        const FSEventStreamEventId event_ids[]) {
-    Node* node = reinterpret_cast<Node*>(client);
+    if (FileSystemWatcher::delayed_filewatch_callback()) {
+      // Used in tests to highlight race between callback invocation
+      // and unwatching the file path, Node destruction
+      TimerUtils::Sleep(1000 /* ms */);
+    }
+    Node* node = static_cast<Node*>(client);
+    RELEASE_ASSERT(node->watcher() != nullptr);
     ASSERT(Thread::Compare(node->watcher()->threadId_,
                            Thread::GetCurrentThreadId()));
-    if (!node->ready()) {
-      return;
-    }
     for (size_t i = 0; i < num_events; i++) {
       char* path = reinterpret_cast<char**>(event_paths)[i];
       FSEvent event;
@@ -274,7 +279,7 @@ intptr_t FileSystemWatcher::WatchPath(intptr_t id,
 
 void FileSystemWatcher::UnwatchPath(intptr_t id, intptr_t path_id) {
   USE(id);
-  delete reinterpret_cast<FSEventsWatcher::Node*>(path_id);
+  reinterpret_cast<FSEventsWatcher::Node*>(path_id)->Stop();
 }
 
 intptr_t FileSystemWatcher::GetSocketId(intptr_t id, intptr_t path_id) {

runtime/bin/main_options.cc

@@ -10,6 +10,7 @@
 
 #include "bin/dartdev_isolate.h"
 #include "bin/error_exit.h"
+#include "bin/file_system_watcher.h"
 #include "bin/options.h"
 #include "bin/platform.h"
 #include "bin/utils.h"
@@ -483,6 +484,9 @@ bool Options::ParseArguments(int argc,
       Options::bypass_trusting_system_roots());
 #endif  // !defined(DART_IO_SECURE_SOCKET_DISABLED)
 
+  FileSystemWatcher::set_delayed_filewatch_callback(
+      Options::delayed_filewatch_callback());
+
   // The arguments to the VM are at positions 1 through i-1 in argv.
   Platform::SetExecutableArguments(i, argv);
 

runtime/bin/main_options.h

@@ -49,7 +49,8 @@ namespace bin {
   V(enable_service_port_fallback, enable_service_port_fallback)                \
   V(disable_dart_dev, disable_dart_dev)                                        \
   V(long_ssl_cert_evaluation, long_ssl_cert_evaluation)                        \
-  V(bypass_trusting_system_roots, bypass_trusting_system_roots)
+  V(bypass_trusting_system_roots, bypass_trusting_system_roots)                \
+  V(delayed_filewatch_callback, delayed_filewatch_callback)
 
 // Boolean flags that have a short form.
 #define SHORT_BOOL_OPTIONS_LIST(V)                                             \

tests/standalone/io/file_system_watcher_large_set_test.dart

@@ -0,0 +1,40 @@
+// Copyright (c) 2013, 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.
+
+// VMOptions=--delayed-filewatch-callback --enable-isolate-groups --experimental-enable-isolate-groups-jit
+// VMOptions=--delayed-filewatch-callback --no-enable-isolate-groups
+
+// Verifies that cancelling subscription from inside of the event handler
+// works as expected, does not result in crash or hang.
+
+import "dart:async";
+import "dart:io";
+
+import "package:path/path.dart";
+
+final completer = Completer<void>();
+late StreamSubscription subscription;
+
+void handleWatchEvent(event) {
+  if (event is FileSystemCreateEvent && event.path.endsWith('txt')) {
+    subscription.cancel();
+    completer.complete();
+  }
+}
+
+void main() async {
+  if (!FileSystemEntity.isWatchSupported) return;
+  final dir = Directory.systemTemp.createTempSync('dart_file_system_watcher');
+  final watcher = dir.watch();
+  subscription = watcher.listen(handleWatchEvent);
+
+  print('watching ${dir.path}');
+  for (int i = 0; i < 1000; i++) {
+    File(join(dir.path, 'file_$i.txt')).createSync();
+  }
+  await completer.future;
+  try {
+    dir.deleteSync(recursive: true);
+  } catch (_) {}
+}

tests/standalone_2/io/file_system_watcher_large_set_test.dart

@@ -0,0 +1,40 @@
+// Copyright (c) 2013, 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.
+
+// VMOptions=--delayed-filewatch-callback --enable-isolate-groups --experimental-enable-isolate-groups-jit
+// VMOptions=--delayed-filewatch-callback --no-enable-isolate-groups
+
+// Verifies that cancelling subscription from inside of the event handler
+// works as expected, does not result in crash or hang.
+
+import "dart:async";
+import "dart:io";
+
+import "package:path/path.dart";
+
+final completer = Completer<void>();
+var subscription;
+
+void handleWatchEvent(event) {
+  if (event is FileSystemCreateEvent && event.path.endsWith('txt')) {
+    subscription.cancel();
+    completer.complete();
+  }
+}
+
+void main() async {
+  if (!FileSystemEntity.isWatchSupported) return;
+  final dir = Directory.systemTemp.createTempSync('dart_file_system_watcher');
+  final watcher = dir.watch();
+  subscription = watcher.listen(handleWatchEvent);
+
+  print('watching ${dir.path}');
+  for (int i = 0; i < 1000; i++) {
+    File(join(dir.path, 'file_$i.txt')).createSync();
+  }
+  await completer.future;
+  try {
+    dir.deleteSync(recursive: true);
+  } catch (_) {}
+}