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

runtime/bin/eventhandler_win.cc

 // 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.
 
 #if !defined(DART_IO_DISABLED)
 
 #include "platform/globals.h"
 #if defined(HOST_OS_WINDOWS)
 
 #include "bin/eventhandler.h"
@@ skipping 97 matching lines @@
   data_length_ = num_bytes;
   return num_bytes;
 }
 
 
 int OverlappedBuffer::GetRemainingLength() {
   ASSERT(operation_ == kRead || operation_ == kRecvFrom);
   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),
       data_ready_(NULL),
       pending_read_(NULL),
       pending_write_(NULL),
       last_error_(NOERROR),
       flags_(0),
       read_thread_id_(Thread::kInvalidThreadId),
       read_thread_handle_(NULL),
       read_thread_starting_(false),
       read_thread_finished_(false),
       monitor_(new Monitor()) {}
 
 
 Handle::~Handle() {
   delete monitor_;
 }
 
 
 bool Handle::CreateCompletionPort(HANDLE completion_port) {
+  ASSERT(completion_port_ == INVALID_HANDLE_VALUE);
+  // A reference to the Handle is Retained by the IO completion port.
+  // It is Released by DeleteIfClosed.
+  Retain();
   completion_port_ = CreateIoCompletionPort(
       handle(), completion_port, reinterpret_cast<ULONG_PTR>(this), 0);
   return (completion_port_ != NULL);
 }
 
 
 void Handle::Close() {
   MonitorLocker ml(monitor_);
   if (!SupportsOverlappedIO()) {
     // If the handle uses synchronous I/O (e.g. stdin), cancel any pending
@@ skipping 234 matching lines @@
   } else {
     HandleError(this);
   }
   SetLastError(error);
 }
 
 
 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.
+      // It is Released by DeleteIfClosed after ReadSyncCompleteAsync
+      // manually puts an event on the IO completion port.
+      Retain();
+      completion_port_ = event_handler_->completion_port();
+    }
   }
 }
 
 
 bool FileHandle::IsClosed() {
   return IsClosing() && !HasPendingRead() && !HasPendingWrite();
 }
 
 
 void DirectoryWatchHandle::EnsureInitialized(
@@ skipping 132 matching lines @@
     closesocket(buffer->client());
   }
 
   pending_accept_count_--;
   OverlappedBuffer::DisposeBuffer(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;
+    // Once the Handle is closed, no further events on the IO completion port
+    // will mention it. Thus, we can drop the reference here.
+    handle->Release();
   }
 }
 
 
 void ListenSocket::DoClose() {
   closesocket(socket());
   handle_ = INVALID_HANDLE_VALUE;
   while (CanAccept()) {
     // Get rid of connections already accepted.
     ClientSocket* client = Accept();
     if (client != NULL) {
       client->Close();
+      // Release the reference from the list.
+      // 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;
     }
   }
+  // To finish resetting the state of the ListenSocket back to what it was
+  // before EnsureInitialized was called, we have to reset the AcceptEx_
+  // function pointer.
+  AcceptEx_ = NULL;
 }
 
 
 bool ListenSocket::CanAccept() {
   MonitorLocker ml(monitor_);
   return accepted_head_ != NULL;
 }
 
 
 ClientSocket* ListenSocket::Accept() {
@@ skipping 206 matching lines @@
   // Note that we return '0', unless a thread have already completed a write.
   if (thread_wrote_ > 0) {
     if (num_bytes > thread_wrote_) {
       num_bytes = thread_wrote_;
     }
     thread_wrote_ -= num_bytes;
     return num_bytes;
   }
   if (!write_thread_exists_) {
     write_thread_exists_ = true;
+    // 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.
+    Retain();
     int result = Thread::Start(WriteFileThread, reinterpret_cast<uword>(this));
     if (result != 0) {
       FATAL1("Failed to start write file thread %d", result);
     }
     while (!write_thread_running_) {
       // Wait until we the thread is running.
       ml.Wait(Monitor::kNoTimeout);
     }
   }
   // Only queue up to INT_MAX bytes.
@@ skipping 16 matching lines @@
     }
     // Join the thread.
     DWORD res = WaitForSingleObject(thread_handle_, INFINITE);
     CloseHandle(thread_handle_);
     ASSERT(res == WAIT_OBJECT_0);
   }
   Handle::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;
   DWORD bytes;
   int status =
       WSAIoctl(socket(), SIO_GET_EXTENSION_FUNCTION_POINTER,
                &guid_disconnect_ex, sizeof(guid_disconnect_ex), &DisconnectEx_,
                sizeof(DisconnectEx_), &bytes, NULL, NULL);
   return (status != SOCKET_ERROR);
 }
@@ skipping 66 matching lines @@
 
 void ClientSocket::IssueDisconnect() {
   OverlappedBuffer* buffer = OverlappedBuffer::AllocateDisconnectBuffer();
   BOOL ok =
       DisconnectEx_(socket(), buffer->GetCleanOverlapped(), TF_REUSE_SOCKET, 0);
   // DisconnectEx works like other OverlappedIO APIs, where we can get either an
   // immediate success or delayed operation by WSA_IO_PENDING being set.
   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
 }
 
 
 void ClientSocket::DisconnectComplete(OverlappedBuffer* buffer) {
   OverlappedBuffer::DisposeBuffer(buffer);
   closesocket(socket());
   if (data_ready_ != NULL) {
     OverlappedBuffer::DisposeBuffer(data_ready_);
   }
   mark_closed();
+#if defined(DEBUG)
+  disconnecting_--;
+#endif
 }
 
 
 void ClientSocket::ConnectComplete(OverlappedBuffer* buffer) {
   OverlappedBuffer::DisposeBuffer(buffer);
   // Update socket to support full socket API, after ConnectEx completed.
   setsockopt(socket(), SOL_SOCKET, SO_UPDATE_CONNECT_CONTEXT, NULL, 0);
   // If the port is set, we already listen for this socket in Dart.
   // Handle the cases here.
   if (!IsClosedRead() && ((Mask() & (1 << kInEvent)) != 0)) {
@@ skipping 91 matching lines @@
 
 void EventHandlerImplementation::HandleInterrupt(InterruptMessage* msg) {
   ASSERT(this != NULL);
   if (msg->id == kTimerId) {
     // Change of timeout request. Just set the new timeout and port as the
     // completion thread will use the new timeout value for its next wait.
     timeout_queue_.UpdateTimeout(msg->dart_port, msg->data);
   } 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()) {
       ListenSocket* listen_socket = reinterpret_cast<ListenSocket*>(handle);
       listen_socket->EnsureInitialized(this);
 
       MonitorLocker ml(listen_socket->monitor_);
 
       if (IS_COMMAND(msg->data, kReturnTokenCommand)) {
         listen_socket->ReturnTokens(msg->dart_port, TOKEN_COUNT(msg->data));
       } else if (IS_COMMAND(msg->data, kSetEventMaskCommand)) {
         // `events` can only have kInEvent/kOutEvent flags set.
         intptr_t events = msg->data & EVENT_MASK;
         ASSERT(0 == (events & ~(1 << kInEvent | 1 << kOutEvent)));
         listen_socket->SetPortAndMask(msg->dart_port, events);
         TryDispatchingPendingAccepts(listen_socket);
       } else if (IS_COMMAND(msg->data, kCloseCommand)) {
         listen_socket->RemovePort(msg->dart_port);
 
         // We only close the socket file descriptor from the operating
         // system if there are no other dart socket objects which
         // 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);
       } else {
         UNREACHABLE();
       }
     } else {
       handle->EnsureInitialized(this);
       MonitorLocker ml(handle->monitor_);
 
@@ skipping 47 matching lines @@
         ClientSocket* client_socket = reinterpret_cast<ClientSocket*>(handle);
         client_socket->Shutdown(SD_RECEIVE);
       } else if (IS_COMMAND(msg->data, kShutdownWriteCommand)) {
         ASSERT(handle->is_client_socket());
 
         ClientSocket* client_socket = reinterpret_cast<ClientSocket*>(handle);
         client_socket->Shutdown(SD_SEND);
       } else if (IS_COMMAND(msg->data, kCloseCommand)) {
         handle->SetPortAndMask(msg->dart_port, 0);
         handle->Close();
+        socket->SetClosedFd();
       } else {
         UNREACHABLE();
       }
     }
 
     DeleteIfClosed(handle);
   }
 }
 
 
@@ skipping 161 matching lines @@
       ClientSocket* client_socket = reinterpret_cast<ClientSocket*>(key);
       HandleConnect(client_socket, bytes, buffer);
       break;
     }
     default:
       UNREACHABLE();
   }
 }
 
 
+void EventHandlerImplementation::HandleCompletionOrInterrupt(
+    BOOL ok,
+    DWORD bytes,
+    ULONG_PTR key,
+    OVERLAPPED* overlapped) {
+  if (!ok) {
+    // Treat ERROR_CONNECTION_ABORTED as connection closed.
+    // The error ERROR_OPERATION_ABORTED is set for pending
+    // accept requests for a listen socket which is closed.
+    // ERROR_NETNAME_DELETED occurs when the client closes
+    // the socket it is reading from.
+    DWORD last_error = GetLastError();
+    if ((last_error == ERROR_CONNECTION_ABORTED) ||
+        (last_error == ERROR_OPERATION_ABORTED) ||
+        (last_error == ERROR_NETNAME_DELETED) ||
+        (last_error == ERROR_BROKEN_PIPE)) {
+      ASSERT(bytes == 0);
+      HandleIOCompletion(bytes, key, overlapped);
+    } else if (last_error == ERROR_MORE_DATA) {
+      // Don't ASSERT no bytes in this case. This can happen if the receive
+      // buffer for datagram sockets is too small to contain a full datagram,
+      // and in this case bytes hold the bytes that was read.
+      HandleIOCompletion(-1, key, overlapped);
+    } else {
+      ASSERT(bytes == 0);
+      HandleIOCompletion(-1, key, overlapped);
+    }
+  } else if (key == NULL) {
+    // A key of NULL signals an interrupt message.
+    InterruptMessage* msg = reinterpret_cast<InterruptMessage*>(overlapped);
+    HandleInterrupt(msg);
+    delete msg;
+  } else {
+    HandleIOCompletion(bytes, key, overlapped);
+  }
+}
+
+
 EventHandlerImplementation::EventHandlerImplementation() {
   startup_monitor_ = new Monitor();
   handler_thread_id_ = Thread::kInvalidThreadId;
   handler_thread_handle_ = NULL;
   completion_port_ =
       CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, NULL, 1);
   if (completion_port_ == NULL) {
     FATAL("Completion port creation failed");
   }
   shutdown_ = false;
@@ skipping 41 matching lines @@
   ASSERT(handler_impl != NULL);
 
   {
     MonitorLocker ml(handler_impl->startup_monitor_);
     handler_impl->handler_thread_id_ = Thread::GetCurrentThreadId();
     handler_impl->handler_thread_handle_ =
         OpenThread(SYNCHRONIZE, false, handler_impl->handler_thread_id_);
     ml.Notify();
   }
 
+  DWORD bytes;
+  ULONG_PTR key;
+  OVERLAPPED* overlapped;
+  BOOL ok;
   while (!handler_impl->shutdown_) {
-    DWORD bytes;
-    ULONG_PTR key;
-    OVERLAPPED* overlapped;
     int64_t millis = handler_impl->GetTimeout();
     ASSERT(millis == kInfinityTimeout || millis >= 0);
     if (millis > kMaxInt32) {
       millis = kMaxInt32;
     }
     ASSERT(sizeof(int32_t) == sizeof(DWORD));
-    BOOL ok =
-        GetQueuedCompletionStatus(handler_impl->completion_port(), &bytes, &key,
-                                  &overlapped, static_cast<DWORD>(millis));
+    DWORD timeout = static_cast<DWORD>(millis);
+    ok = GetQueuedCompletionStatus(handler_impl->completion_port(), &bytes,
+                                   &key, &overlapped, timeout);
 
     if (!ok && (overlapped == NULL)) {
       if (GetLastError() == ERROR_ABANDONED_WAIT_0) {
         // The completion port should never be closed.
         Log::Print("Completion port closed\n");
         UNREACHABLE();
       } else {
         // Timeout is signalled by false result and NULL in overlapped.
         handler_impl->HandleTimeout();
       }
-    } else if (!ok) {
-      // Treat ERROR_CONNECTION_ABORTED as connection closed.
-      // The error ERROR_OPERATION_ABORTED is set for pending
-      // accept requests for a listen socket which is closed.
-      // ERROR_NETNAME_DELETED occurs when the client closes
-      // the socket it is reading from.
-      DWORD last_error = GetLastError();
-      if ((last_error == ERROR_CONNECTION_ABORTED) ||
-          (last_error == ERROR_OPERATION_ABORTED) ||
-          (last_error == ERROR_NETNAME_DELETED) ||
-          (last_error == ERROR_BROKEN_PIPE)) {
-        ASSERT(bytes == 0);
-        handler_impl->HandleIOCompletion(bytes, key, overlapped);
-      } else if (last_error == ERROR_MORE_DATA) {
-        // Don't ASSERT no bytes in this case. This can happen if the receive
-        // buffer for datagram sockets is to small to contain a full datagram,
-        // and in this case bytes hold the bytes that was read.
-        handler_impl->HandleIOCompletion(-1, key, overlapped);
-      } else {
-        ASSERT(bytes == 0);
-        handler_impl->HandleIOCompletion(-1, key, overlapped);
-      }
-    } else if (key == NULL) {
-      // A key of NULL signals an interrupt message.
-      InterruptMessage* msg = reinterpret_cast<InterruptMessage*>(overlapped);
-      handler_impl->HandleInterrupt(msg);
-      delete msg;
     } else {
-      handler_impl->HandleIOCompletion(bytes, key, overlapped);
+      handler_impl->HandleCompletionOrInterrupt(ok, bytes, key, overlapped);
     }
   }
+
+// In a Debug build, drain the IO completion port to make sure we aren't
+// leaking any (non-disconnecting) Handles. In a Release build, we don't care
+// because the VM is going down, and the asserts below are Debug-only.
+#if defined(DEBUG)
+  while (true) {
+    ok = GetQueuedCompletionStatus(handler_impl->completion_port(), &bytes,
+                                   &key, &overlapped, 0);
+    if (!ok && (overlapped == NULL)) {
+      // There was an error or nothing is ready. Assume the port is drained.
+      break;
+    }
+    handler_impl->HandleCompletionOrInterrupt(ok, bytes, key, overlapped);
+  }
+#endif
+
+  // 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();
 }
 
 
 void EventHandlerImplementation::Start(EventHandler* handler) {
   int result =
       Thread::Start(EventHandlerEntry, reinterpret_cast<uword>(handler));
   if (result != 0) {
     FATAL1("Failed to start event handler thread %d", result);
   }
@@ skipping 15 matching lines @@
 void EventHandlerImplementation::Shutdown() {
   SendData(kShutdownId, 0, 0);
 }
 
 }  // namespace bin
 }  // namespace dart
 
 #endif  // defined(HOST_OS_WINDOWS)
 
 #endif  // !defined(DART_IO_DISABLED)