[vm/io] Avoid leaking Handle::data_ready_ on Windows.

Handle::data_ready_ contains bytes which are ready to be sent to Dart
side. Not all code paths fully drain this buffer and delete it
before destroying the handle, for example directory watch implementation
was prone to leaking data_ready_ when subscription was cancelled.

This CL switches the code to use unique_ptr to hold on to data_ready_
which makes sure that it is deleted when Handle is destroyed.

The code would benefit from holding all OverlappedBuffer instances
through unique_ptr but that is a much larger refactoring which
we leave for a later date.

Fixes #52715

TEST=standalone{,_2}/regress_52715_test

Bug: 52715
Cq-Include-Trybots: luci.dart.try:vm-win-release-x64-try,vm-win-debug-x64-try,vm-aot-win-release-x64-try,analyzer-win-release-try
Change-Id: Ie8d728b823de7e8f9de1489898e270580c2af269
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/311841
Commit-Queue: Slava Egorov <vegorov@google.com>
Commit-Queue: Martin Kustermann <kustermann@google.com>
Reviewed-by: Martin Kustermann <kustermann@google.com>
Auto-Submit: Slava Egorov <vegorov@google.com>

runtime/bin/eventhandler_win.cc

@@ -119,7 +119,7 @@ Handle::Handle(intptr_t handle)
       handle_(reinterpret_cast<HANDLE>(handle)),
       completion_port_(INVALID_HANDLE_VALUE),
       event_handler_(nullptr),
-      data_ready_(nullptr),
+      data_ready_(),
       pending_read_(nullptr),
       pending_write_(nullptr),
       last_error_(NOERROR),
@@ -221,7 +221,7 @@ void Handle::ReadComplete(OverlappedBuffer* buffer) {
     ASSERT(pending_read_ == buffer);
     ASSERT(data_ready_ == nullptr);
     if (!IsClosing()) {
-      data_ready_ = pending_read_;
+      data_ready_.reset(pending_read_);
     } else {
       OverlappedBuffer::DisposeBuffer(buffer);
     }
@@ -665,7 +665,6 @@ intptr_t Handle::Read(void* buffer, intptr_t num_bytes) {
   num_bytes =
       data_ready_->Read(buffer, Utils::Minimum<intptr_t>(num_bytes, INT_MAX));
   if (data_ready_->IsEmpty()) {
-    OverlappedBuffer::DisposeBuffer(data_ready_);
     data_ready_ = nullptr;
     if (!IsClosing() && !IsClosedRead()) {
       IssueRead();
@@ -694,7 +693,6 @@ intptr_t Handle::RecvFrom(void* buffer,
   }
   // Always dispose of the buffer, as UDP messages must be read in their
   // entirety to match how recvfrom works in a socket.
-  OverlappedBuffer::DisposeBuffer(data_ready_);
   data_ready_ = nullptr;
   if (!IsClosing() && !IsClosedRead()) {
     IssueRecvFrom();
@@ -975,9 +973,7 @@ void ClientSocket::IssueDisconnect() {
 void ClientSocket::DisconnectComplete(OverlappedBuffer* buffer) {
   OverlappedBuffer::DisposeBuffer(buffer);
   closesocket(socket());
-  if (data_ready_ != nullptr) {
-    OverlappedBuffer::DisposeBuffer(data_ready_);
-  }
+  data_ready_ = nullptr;
   mark_closed();
 #if defined(DEBUG)
   disconnecting_--;

runtime/bin/eventhandler_win.h

@@ -100,6 +100,8 @@ class OverlappedBuffer {
 
   void set_data_length(int data_length) { data_length_ = data_length; }
 
+  void operator delete(void* buffer) { free(buffer); }
+
  private:
   OverlappedBuffer(int buffer_size, Operation operation)
       : operation_(operation), buflen_(buffer_size) {
@@ -130,8 +132,6 @@ class OverlappedBuffer {
     return malloc(size + buffer_size);
   }
 
-  void operator delete(void* buffer) { free(buffer); }
-
   // Allocate an overlapped buffer for thse specified amount of data and
   // operation. Some operations need additional buffer space, which is
   // handled by this method.
@@ -272,7 +272,8 @@ class Handle : public ReferenceCounted<Handle>, public DescriptorInfoBase {
   HANDLE completion_port_;
   EventHandlerImplementation* event_handler_;
 
-  OverlappedBuffer* data_ready_;     // Buffer for data ready to be read.
+  std::unique_ptr<OverlappedBuffer>
+      data_ready_;                   // Buffer for data ready to be read.
   OverlappedBuffer* pending_read_;   // Buffer for pending read.
   OverlappedBuffer* pending_write_;  // Buffer for pending write
 

tests/standalone/regress_52715_test.dart

@@ -0,0 +1,24 @@
+// 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.
+
+// Verify that cancelled [Directory.watch] subscriptions do not waste memory.
+
+import 'dart:io';
+
+import 'package:expect/expect.dart';
+
+void main() async {
+  final startRss = ProcessInfo.currentRss;
+
+  for (var i = 0; i < 1024; i++) {
+    final subscription = Directory.systemTemp.watch().listen((event) {});
+    await subscription.cancel();
+  }
+
+  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}');
+}

tests/standalone_2/regress_52715_test.dart

@@ -0,0 +1,26 @@
+// 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.
+
+// Verify that cancelled [Directory.watch] subscriptions do not waste memory.
+
+// @dart=2.9
+
+import 'dart:io';
+
+import 'package:expect/expect.dart';
+
+void main() async {
+  final startRss = ProcessInfo.currentRss;
+
+  for (var i = 0; i < 1024; i++) {
+    final subscription = Directory.systemTemp.watch().listen((event) {});
+    await subscription.cancel();
+  }
+
+  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}');
+}