[dart:io] Adds a finalizer to _NativeSocket to avoid socket leaks

runtime/bin/eventhandler_win.cc

Index: runtime/bin/eventhandler_win.cc
diff --git a/runtime/bin/eventhandler_win.cc b/runtime/bin/eventhandler_win.cc
index eedcf167da2c6db3b478c657c1d2db5660453d31..805e67803a987af73f067999cbbabfdb475c1928 100644
--- a/runtime/bin/eventhandler_win.cc
+++ b/runtime/bin/eventhandler_win.cc
@@ -115,8 +115,10 @@ int OverlappedBuffer::GetRemainingLength() {
   return data_length_ - index_;
 }
 
+
 Handle::Handle(intptr_t handle)
-    : DescriptorInfoBase(handle),
+    : ReferenceCounted(),
+      DescriptorInfoBase(handle),
       handle_(reinterpret_cast<HANDLE>(handle)),
       completion_port_(INVALID_HANDLE_VALUE),
       event_handler_(NULL),
@@ -138,6 +140,8 @@ Handle::~Handle() {
 
 
 bool Handle::CreateCompletionPort(HANDLE completion_port) {
+  ASSERT(completion_port_ == INVALID_HANDLE_VALUE);
+  Retain();

[zra, 2017/03/25 22:39:50]

// A reference to the Handle is Retained by the IO completion port.
// It is Released by DeleteIfClosed.

   completion_port_ = CreateIoCompletionPort(
       handle(), completion_port, reinterpret_cast<ULONG_PTR>(this), 0);
   return (completion_port_ != NULL);
@@ -392,8 +396,14 @@ void Handle::HandleIssueError() {
 void FileHandle::EnsureInitialized(EventHandlerImplementation* event_handler) {
   MonitorLocker ml(monitor_);
   event_handler_ = event_handler;
-  if (SupportsOverlappedIO() && (completion_port_ == INVALID_HANDLE_VALUE)) {
-    CreateCompletionPort(event_handler_->completion_port());
+  if (completion_port_ == INVALID_HANDLE_VALUE) {
+    if (SupportsOverlappedIO()) {
+      CreateCompletionPort(event_handler_->completion_port());
+    } else {
+      // We need to retain the Handle even if overlapped IO is not supported.
+      Retain();

[zra, 2017/03/25 22:39:50]

// It is Released by DeleteIfClosed after ReadSyncCompleteAsync
// manually puts an event on the IO completion port.

+      completion_port_ = event_handler_->completion_port();
+    }
   }
 }
 
@@ -546,9 +556,11 @@ void ListenSocket::AcceptComplete(OverlappedBuffer* buffer,
 
 static void DeleteIfClosed(Handle* handle) {
   if (handle->IsClosed()) {
+    handle->set_completion_port(INVALID_HANDLE_VALUE);
+    handle->set_event_handler(NULL);
     handle->NotifyAllDartPorts(1 << kDestroyedEvent);
     handle->RemoveAllPorts();
-    delete handle;
+    handle->Release();

[zra, 2017/03/25 22:39:50]

// Once the Handle is closed, no further events on the IO completion port
// will mention it. Thus, we can drop the reference here.

   }
 }
 
@@ -561,11 +573,14 @@ void ListenSocket::DoClose() {
     ClientSocket* client = Accept();
     if (client != NULL) {
       client->Close();
+      // Release the reference from the list.

[zra, 2017/03/25 22:39:50]

// When an accept completes, we make a new ClientSocket (1 reference),
// and add it to the IO completion port (1 more reference). If an
// accepted connection is never requested by the Dart code, then
// this list owns a reference (first Release), and the IO completion
// port owns a reference, (second Release in DeleteIfClosed).

+      client->Release();
       DeleteIfClosed(client);
     } else {
       break;
     }
   }
+  AcceptEx_ = NULL;
 }
 
 
@@ -792,6 +807,7 @@ intptr_t StdHandle::Write(const void* buffer, intptr_t num_bytes) {
   }
   if (!write_thread_exists_) {
     write_thread_exists_ = true;
+    Retain();

[zra, 2017/03/25 22:39:50]

// The write thread gets a reference to the Handle, which it places in
// the events it puts on the IO completion port. The reference is
// Released by DeleteIfClosed.

     int result = Thread::Start(WriteFileThread, reinterpret_cast<uword>(this));
     if (result != 0) {
       FATAL1("Failed to start write file thread %d", result);
@@ -828,6 +844,11 @@ void StdHandle::DoClose() {
 }
 
 
+#if defined(DEBUG)
+intptr_t ClientSocket::disconnecting_ = 0;
+#endif
+
+
 bool ClientSocket::LoadDisconnectEx() {
   // Load the DisconnectEx function into memory using WSAIoctl.
   GUID guid_disconnect_ex = WSAID_DISCONNECTEX;
@@ -914,8 +935,18 @@ void ClientSocket::IssueDisconnect() {
   if (ok || (WSAGetLastError() != WSA_IO_PENDING)) {
     DisconnectComplete(buffer);
   }
+  // When the Dart side receives this event, it may decide to close its Dart
+  // ports. When all ports are closed, the VM will shut down. The EventHandler
+  // will then shut down. If the EventHandler shuts down before this
+  // asynchronous disconnect finishes, this ClientSocket will be leaked.
+  // TODO(dart:io): Retain a list of client sockets that are in the process of
+  // disconnecting. Disconnect them forcefully, and clean up their resources
+  // when the EventHandler shuts down.
   NotifyAllDartPorts(1 << kDestroyedEvent);
   RemoveAllPorts();
+#if defined(DEBUG)
+  disconnecting_++;
+#endif
 }
 
 
@@ -926,6 +957,9 @@ void ClientSocket::DisconnectComplete(OverlappedBuffer* buffer) {
     OverlappedBuffer::DisposeBuffer(data_ready_);
   }
   mark_closed();
+#if defined(DEBUG)
+  disconnecting_--;
+#endif
 }
 
 
@@ -1037,7 +1071,12 @@ void EventHandlerImplementation::HandleInterrupt(InterruptMessage* msg) {
   } else if (msg->id == kShutdownId) {
     shutdown_ = true;
   } else {
-    Handle* handle = reinterpret_cast<Handle*>(msg->id);
+    Socket* socket = reinterpret_cast<Socket*>(msg->id);
+    RefCntReleaseScope<Socket> rs(socket);
+    if (socket->fd() == -1) {
+      return;
+    }
+    Handle* handle = reinterpret_cast<Handle*>(socket->fd());
     ASSERT(handle != NULL);
 
     if (handle->is_listen_socket()) {
@@ -1062,9 +1101,10 @@ void EventHandlerImplementation::HandleInterrupt(InterruptMessage* msg) {
         // are listening on the same (address, port) combination.
         ListeningSocketRegistry* registry = ListeningSocketRegistry::Instance();
         MutexLocker locker(registry->mutex());
-        if (registry->CloseSafe(reinterpret_cast<intptr_t>(listen_socket))) {
+        if (registry->CloseSafe(socket)) {
           ASSERT(listen_socket->Mask() == 0);
           listen_socket->Close();
+          socket->SetClosedFd();
         }
 
         DartUtils::PostInt32(msg->dart_port, 1 << kDestroyedEvent);
@@ -1132,6 +1172,7 @@ void EventHandlerImplementation::HandleInterrupt(InterruptMessage* msg) {
       } else if (IS_COMMAND(msg->data, kCloseCommand)) {
         handle->SetPortAndMask(msg->dart_port, 0);
         handle->Close();
+        socket->SetClosedFd();
       } else {
         UNREACHABLE();
       }
@@ -1428,6 +1469,16 @@ void EventHandlerImplementation::EventHandlerEntry(uword args) {
       handler_impl->HandleIOCompletion(bytes, key, overlapped);
     }
   }
+
+  // The eventhandler thread is going down so there should be no more live
+  // Handles or Sockets.
+  // TODO(dart:io): It would be nice to be able to assert here that:
+  //     ReferenceCounted<Handle>::instances() == 0;
+  // However, we cannot at the moment. See the TODO on:
+  //     ClientSocket::IssueDisconnect()
+  DEBUG_ASSERT(ReferenceCounted<Handle>::instances() ==
+               ClientSocket::disconnecting());
+  DEBUG_ASSERT(ReferenceCounted<Socket>::instances() == 0);
   handler->NotifyShutdownDone();
 }