Use ConnectEx on Windows, to do async connect of sockets.

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.
 
 #include "platform/globals.h"
 #if defined(TARGET_OS_WINDOWS)
 
 #include "bin/eventhandler.h"
 
 #include <winsock2.h>  // NOLINT
@@ skipping 55 matching lines @@
 OverlappedBuffer* OverlappedBuffer::AllocateSendToBuffer(int buffer_size) {
   return AllocateBuffer(buffer_size, kSendTo);
 }
 
 
 OverlappedBuffer* OverlappedBuffer::AllocateDisconnectBuffer() {
   return AllocateBuffer(0, kDisconnect);
 }
 
 
+OverlappedBuffer* OverlappedBuffer::AllocateConnectBuffer() {
+  return AllocateBuffer(0, kConnect);
+}
+
+
 void OverlappedBuffer::DisposeBuffer(OverlappedBuffer* buffer) {
   delete buffer;
 }
 
 
 OverlappedBuffer* OverlappedBuffer::GetFromOverlapped(OVERLAPPED* overlapped) {
   OverlappedBuffer* buffer =
       CONTAINING_RECORD(overlapped, OverlappedBuffer, overlapped_);
   return buffer;
 }
@@ skipping 381 matching lines @@
   if (!IsClosing()) {
     // Update the accepted socket to support the full range of API calls.
     SOCKET s = socket();
     int rc = setsockopt(buffer->client(),
                         SOL_SOCKET,
                         SO_UPDATE_ACCEPT_CONTEXT,
                         reinterpret_cast<char*>(&s), sizeof(s));
     if (rc == NO_ERROR) {
       // Insert the accepted socket into the list.
       ClientSocket* client_socket = new ClientSocket(buffer->client(), 0);
+      client_socket->mark_connected();
       client_socket->CreateCompletionPort(completion_port);
       if (accepted_head_ == NULL) {
         accepted_head_ = client_socket;
         accepted_tail_ = client_socket;
       } else {
         ASSERT(accepted_tail_ != NULL);
         accepted_tail_->set_next(client_socket);
         accepted_tail_ = client_socket;
       }
     } else {
@@ skipping 317 matching lines @@
     return true;
   }
   OverlappedBuffer::DisposeBuffer(pending_write_);
   pending_write_ = NULL;
   HandleIssueError();
   return false;
 }
 
 
 void ClientSocket::IssueDisconnect() {
-  Dart_Port p = port();
   OverlappedBuffer* buffer = OverlappedBuffer::AllocateDisconnectBuffer();
   BOOL ok = DisconnectEx_(
     socket(), buffer->GetCleanOverlapped(), TF_REUSE_SOCKET, 0);
-  if (!ok && WSAGetLastError() != WSA_IO_PENDING) {
+  // 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);
   }
+  Dart_Port p = port();
   if (p != ILLEGAL_PORT) DartUtils::PostInt32(p, 1 << kDestroyedEvent);
+  port_ = ILLEGAL_PORT;
 }
 
 
 void ClientSocket::DisconnectComplete(OverlappedBuffer* buffer) {
   OverlappedBuffer::DisposeBuffer(buffer);
   closesocket(socket());
   if (data_ready_ != NULL) {
     OverlappedBuffer::DisposeBuffer(data_ready_);
   }
-  // When disconnect is complete get rid of the object.
-  delete this;
+  closed_ = true;
 }
 
 
+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);
+  connected_ = true;
+  Dart_Port p = port();
+  if (p != ILLEGAL_PORT) {
+    // If the port is set, we already listen for this socket in Dart.
+    // Handle the cases here.
+    if (!IsClosedRead()) {
+      IssueRead();
+    }
+    if (!IsClosedWrite()) {
+      DartUtils::PostInt32(p, 1 << kOutEvent);
+    }
+  }
+}
+
+
 void ClientSocket::EnsureInitialized(
     EventHandlerImplementation* event_handler) {
   ScopedLock lock(this);
   if (completion_port_ == INVALID_HANDLE_VALUE) {
     ASSERT(event_handler_ == NULL);
     event_handler_ = event_handler;
     CreateCompletionPort(event_handler_->completion_port());
   }
 }
 
 
 bool ClientSocket::IsClosed() {
-  return false;
+  return closed_;
 }
 
 
 bool DatagramSocket::IssueSendTo(struct sockaddr* sa, socklen_t sa_len) {
   ScopedLock lock(this);
   ASSERT(completion_port_ != INVALID_HANDLE_VALUE);
   ASSERT(pending_write_ != NULL);
   ASSERT(pending_write_->operation() == OverlappedBuffer::kSendTo);
 
   int rc = WSASendTo(socket(),
@@ skipping 66 matching lines @@
   closesocket(socket());
   MarkClosedRead();
   MarkClosedWrite();
 }
 
 
 static void DeleteIfClosed(Handle* handle) {
   if (handle->IsClosed()) {
     Dart_Port port = handle->port();
     delete handle;
-    DartUtils::PostInt32(port, 1 << kDestroyedEvent);
+    if (port != ILLEGAL_PORT) {
+      DartUtils::PostInt32(port, 1 << kDestroyedEvent);
+    }
   }
 }
 
 
 void EventHandlerImplementation::HandleInterrupt(InterruptMessage* msg) {
   if (msg->id == kTimeoutId) {
     // 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) {
@@ skipping 21 matching lines @@
         }
         // Always keep 5 outstanding accepts going, to enhance performance.
         while (listen_socket->pending_accept_count() < 5) {
           bool accept_success = listen_socket->IssueAccept();
           if (!accept_success) {
             HandleError(listen_socket);
             break;
           }
         }
       }
-
-      if ((msg->data & (1 << kCloseCommand)) != 0) {
-        listen_socket->Close();
-      }
     } else {
       handle->EnsureInitialized(this);
 
       Handle::ScopedLock lock(handle);
 
       // Only set mask if we turned on kInEvent or kOutEvent.
       if ((msg->data & ((1 << kInEvent) | (1 << kOutEvent))) != 0) {
         handle->SetPortAndMask(msg->dart_port, msg->data);
       }
 
       // Issue a read.
       if ((msg->data & (1 << kInEvent)) != 0) {
-        handle->SetPortAndMask(msg->dart_port, msg->data);
         if (handle->is_datagram_socket()) {
           handle->IssueRecvFrom();
+        } else if (handle->is_client_socket()) {
+          if (reinterpret_cast<ClientSocket*>(handle)->is_connected()) {
+            handle->IssueRead();
+          }
         } else {
           handle->IssueRead();
         }
       }
 
       // If out events (can write events) have been requested, and there
       // are no pending writes, meaning any writes are already complete,
       // post an out event immediately.
       if ((msg->data & (1 << kOutEvent)) != 0) {
-        handle->SetPortAndMask(msg->dart_port, msg->data);
         if (!handle->HasPendingWrite()) {
-          int event_mask = (1 << kOutEvent);
-          DartUtils::PostInt32(handle->port(), event_mask);
+          if (handle->is_client_socket()) {
+            if (reinterpret_cast<ClientSocket*>(handle)->is_connected()) {
+              DartUtils::PostInt32(handle->port(), 1 << kOutEvent);
+            }
+          } else {
+            DartUtils::PostInt32(handle->port(), 1 << kOutEvent);
+          }
         }
       }
 
       if (handle->is_client_socket()) {
         ClientSocket* client_socket = reinterpret_cast<ClientSocket*>(handle);
         if ((msg->data & (1 << kShutdownReadCommand)) != 0) {
           client_socket->Shutdown(SD_RECEIVE);
         }
 
         if ((msg->data & (1 << kShutdownWriteCommand)) != 0) {
@@ skipping 88 matching lines @@
 
 
 void EventHandlerImplementation::HandleWrite(Handle* handle,
                                              int bytes,
                                              OverlappedBuffer* buffer) {
   handle->WriteComplete(buffer);
 
   if (bytes >= 0) {
     if (!handle->IsError() && !handle->IsClosing()) {
       int event_mask = 1 << kOutEvent;
+      ASSERT(!handle->is_client_socket() ||
+             reinterpret_cast<ClientSocket*>(handle)->is_connected());
       if ((handle->mask() & event_mask) != 0) {
         DartUtils::PostInt32(handle->port(), event_mask);
       }
     }
   } else {
     HandleError(handle);
   }
 
   DeleteIfClosed(handle);
 }
 
 
 void EventHandlerImplementation::HandleDisconnect(
     ClientSocket* client_socket,
     int bytes,
     OverlappedBuffer* buffer) {
   client_socket->DisconnectComplete(buffer);
+  DeleteIfClosed(client_socket);
 }
 
+
+void EventHandlerImplementation::HandleConnect(
+    ClientSocket* client_socket,
+    int bytes,
+    OverlappedBuffer* buffer) {
+  if (bytes < 0) {
+    HandleError(client_socket);
+    OverlappedBuffer::DisposeBuffer(buffer);
+  } else {
+    client_socket->ConnectComplete(buffer);
+  }
+}
+
+
 void EventHandlerImplementation::HandleTimeout() {
   if (!timeout_queue_.HasTimeout()) return;
   DartUtils::PostNull(timeout_queue_.CurrentPort());
   timeout_queue_.RemoveCurrent();
 }
 
 
 void EventHandlerImplementation::HandleIOCompletion(DWORD bytes,
                                                     ULONG_PTR key,
                                                     OVERLAPPED* overlapped) {
@@ skipping 18 matching lines @@
     case OverlappedBuffer::kSendTo: {
       Handle* handle = reinterpret_cast<Handle*>(key);
       HandleWrite(handle, bytes, buffer);
       break;
     }
     case OverlappedBuffer::kDisconnect: {
       ClientSocket* client_socket = reinterpret_cast<ClientSocket*>(key);
       HandleDisconnect(client_socket, bytes, buffer);
       break;
     }
+    case OverlappedBuffer::kConnect: {
+      ClientSocket* client_socket = reinterpret_cast<ClientSocket*>(key);
+      HandleConnect(client_socket, bytes, buffer);
+      break;
+    }
     default:
       UNREACHABLE();
   }
 }
 
 
 EventHandlerImplementation::EventHandlerImplementation() {
   intptr_t result;
   completion_port_ =
       CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, NULL, 1);
@@ skipping 105 matching lines @@
 
 
 void EventHandlerImplementation::Shutdown() {
   SendData(kShutdownId, 0, 0);
 }
 
 }  // namespace bin
 }  // namespace dart
 
 #endif  // defined(TARGET_OS_WINDOWS)