[vm] Fix leakage of file system watcher related data structures

Closes #52703

TEST=runtime/tests/vm/dart/regress_52703_test.dart

Change-Id: Ia76c721a3d3a1466ed1c2e7bd847c331cee95ed7
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/309862
Reviewed-by: Lasse Nielsen <lrn@google.com>
Auto-Submit: Martin Kustermann <kustermann@google.com>
Commit-Queue: Martin Kustermann <kustermann@google.com>

runtime/tests/vm/dart/regress_52703_test.dart

@@ -0,0 +1,60 @@
+// Copyright (c) 2023, 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.
+
+import 'dart:async';
+import 'dart:developer';
+import 'dart:io';
+
+import 'package:expect/expect.dart';
+import 'package:path/path.dart' as path;
+import 'package:vm_service/vm_service.dart';
+
+import 'use_flag_test_helper.dart';
+import '../../../tools/heapsnapshot/lib/src/analysis.dart';
+
+void main() async {
+  if (buildDir.contains('Product')) return;
+
+  final dir = Directory('sdk');
+
+  StreamSubscription? current;
+  Future<void> iterate() async {
+    final temp = dir.watch().listen((event) {});
+    await current?.cancel();
+    current = dir.watch().listen((event) {});
+    await temp.cancel();
+  }
+
+  Future<void> finish() async {
+    await current?.cancel();
+  }
+
+  await withTempDir('ama-test', (String tempDir) async {
+    await iterate();
+
+    final before = countInstances(tempDir, '_BroadcastSubscription');
+    print('Before = $before');
+    for (int i = 0; i < 100; i++) {
+      await iterate();
+    }
+    final after = countInstances(tempDir, '_BroadcastSubscription');
+    print('After = $after');
+
+    await finish();
+
+    Expect.equals(0, (after - before));
+  });
+}
+
+int countInstances(String tempDir, String className) {
+  final snapshotFile = path.join(tempDir, 'foo.heapsnapshot');
+  NativeRuntime.writeHeapSnapshotToFile(snapshotFile);
+  final bytes = File(snapshotFile).readAsBytesSync();
+  File(snapshotFile).deleteSync();
+  final graph = HeapSnapshotGraph.fromChunks(
+      [bytes.buffer.asByteData(bytes.offsetInBytes, bytes.length)]);
+  final analysis = Analysis(graph);
+  return analysis
+      .filterByClassPatterns(analysis.reachableObjects, [className]).length;
+}

sdk/lib/_internal/vm/bin/file_patch.dart

@@ -137,6 +137,12 @@ abstract class _FileSystemWatcher {
   final StreamController<FileSystemEvent> _broadcastController =
       new 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) {
@@ -193,7 +199,9 @@ abstract class _FileSystemWatcher {
     }
     _watcherPath = _idMap[pathId];
     _watcherPath!.count++;
-    _pathWatched().pipe(_broadcastController);
+    _sourceSubscription = _pathWatched().listen(_broadcastController.add,
+        onError: _broadcastController.addError,
+        onDone: _broadcastController.close);
   }
 
   void _cancel() {
@@ -222,14 +230,16 @@ abstract class _FileSystemWatcher {
       _doneWatcher();
       _id = null;
     }
+    _sourceSubscription?.cancel();
+    _sourceSubscription = null;
   }
 
   // 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 _pathWatched();
+  Stream<FileSystemEvent> _pathWatched();
   // Called when a path is no longer being watched.
   void _donePathWatched() {}
 
@@ -387,7 +397,7 @@ abstract class _FileSystemWatcher {
 }
 
 class _InotifyFileSystemWatcher extends _FileSystemWatcher {
-  static final Map<int, StreamController> _idMap = {};
+  static final Map<int, StreamController<FileSystemEvent>> _idMap = {};
   static late StreamSubscription _subscription;
 
   _InotifyFileSystemWatcher(path, events, recursive)
@@ -411,7 +421,7 @@ class _InotifyFileSystemWatcher extends _FileSystemWatcher {
     _subscription.cancel();
   }
 
-  Stream _pathWatched() {
+  Stream<FileSystemEvent> _pathWatched() {
     var pathId = _watcherPath!.pathId;
     if (!_idMap.containsKey(pathId)) {
       _idMap[pathId] = new StreamController<FileSystemEvent>.broadcast();
@@ -429,12 +439,12 @@ class _InotifyFileSystemWatcher extends _FileSystemWatcher {
 
 class _Win32FileSystemWatcher extends _FileSystemWatcher {
   late StreamSubscription _subscription;
-  late StreamController _controller;
+  late StreamController<FileSystemEvent> _controller;
 
   _Win32FileSystemWatcher(path, events, recursive)
       : super._(path, events, recursive);
 
-  Stream _pathWatched() {
+  Stream<FileSystemEvent> _pathWatched() {
     var pathId = _watcherPath!.pathId;
     _controller = new StreamController<FileSystemEvent>();
     _subscription =
@@ -457,12 +467,12 @@ class _Win32FileSystemWatcher extends _FileSystemWatcher {
 
 class _FSEventStreamFileSystemWatcher extends _FileSystemWatcher {
   late StreamSubscription _subscription;
-  late StreamController _controller;
+  late StreamController<FileSystemEvent> _controller;
 
   _FSEventStreamFileSystemWatcher(path, events, recursive)
       : super._(path, events, recursive);
 
-  Stream _pathWatched() {
+  Stream<FileSystemEvent> _pathWatched() {
     var pathId = _watcherPath!.pathId;
     var socketId = _FileSystemWatcher._getSocketId(0, pathId);
     _controller = new StreamController<FileSystemEvent>();