[dart:io] Create one IOService port per process instead of per isolate.

Nothing cleaned up this port when the creating isolate died, so the port itself and the last unjoined thread in the corresponding thread pool were leaked until VM shutdown.

Cf. 42d796f7c1

TEST=testandalone/io/leak_io_service_test
Change-Id: Icea27804baf9175b1ff5eb9c50970e0f8d07de2c
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/437160
Reviewed-by: Brian Quinlan <bquinlan@google.com>
Commit-Queue: Ryan Macnak <rmacnak@google.com>

runtime/bin/dart_embedder_api_impl.cc

@@ -6,6 +6,7 @@
 
 #include "bin/dartutils.h"
 #include "bin/eventhandler.h"
+#include "bin/io_service.h"
 #include "bin/isolate_data.h"
 #include "bin/process.h"
 #include "bin/secure_socket_filter.h"
@@ -54,6 +55,7 @@ void Cleanup() {
   bin::SSLFilter::Cleanup();
 #endif
   bin::Process::Cleanup();
+  bin::IOService::Cleanup();
 }
 
 Dart_Isolate CreateKernelServiceIsolate(const IsolateCreationData& data,

runtime/bin/io_service.cc

@@ -55,10 +55,22 @@ void IOServiceCallback(Dart_Port dest_port_id, Dart_CObject* message) {
 }
 
 intptr_t IOService::max_concurrency_ = 32;
+std::atomic<Dart_Port> IOService::port_ = ILLEGAL_PORT;
 
 Dart_Port IOService::GetServicePort() {
-  return Dart_NewConcurrentNativePort("IOService", IOServiceCallback,
-                                      max_concurrency_);
+  Dart_Port port = port_;
+  if (port == ILLEGAL_PORT) {
+    port = Dart_NewConcurrentNativePort("IOService", IOServiceCallback,
+                                        max_concurrency_);
+    Dart_Port expected = ILLEGAL_PORT;
+    if (!port_.compare_exchange_strong(expected, port)) {
+      // Lost the initialization race. Use the winner's port and close our port.
+      // The winner's port is eventually implicitly closed by VM shutdown.
+      Dart_CloseNativePort(port);
+      return expected;
+    }
+  }
+  return port;
 }
 
 void FUNCTION_NAME(IOService_NewServicePort)(Dart_NativeArguments args) {

runtime/bin/io_service.h

@@ -9,6 +9,8 @@
 #error "io_service.h can only be included on builds with IO and SSL enabled"
 #endif
 
+#include <atomic>
+
 #include "bin/builtin.h"
 #include "bin/utils.h"
 
@@ -69,12 +71,14 @@ class IOService {
   enum { IO_SERVICE_REQUEST_LIST(DECLARE_REQUEST) };
 
   static Dart_Port GetServicePort();
+  static void Cleanup() { port_ = ILLEGAL_PORT; }
 
   static intptr_t max_concurrency() { return max_concurrency_; }
   static void set_max_concurrency(intptr_t value) { max_concurrency_ = value; }
 
  private:
   static intptr_t max_concurrency_;
+  static std::atomic<Dart_Port> port_;
 
   DISALLOW_ALLOCATION();
   DISALLOW_IMPLICIT_CONSTRUCTORS(IOService);

tests/standalone/io/leak_io_service_test.dart

@@ -0,0 +1,28 @@
+// 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.
+
+import "dart:async";
+import "dart:io";
+import "dart:isolate";
+
+child(replyPort) async {
+  var ops = <Future>[];
+  for (var i = 0; i < 32; i++) {
+    ops.add(File(Platform.executable).stat()); // Uses the IO Service.
+  }
+  await Future.wait(ops);
+  replyPort.send(null);
+}
+
+main() {
+  var pending = 1000;
+  var port = new RawReceivePort();
+  port.handler = (_) {
+    pending--;
+    if (pending == 0) port.close();
+  };
+  for (var i = 0; i < pending; i++) {
+    Isolate.spawn(child, port.sendPort);
+  }
+}