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) {
+  if (ok || WSAGetLastError() != WSA_IO_PENDING) {

[Søren Gjesse, 2014/05/01 07:19:26]

When is it possible to get a success here? Please add a comment on this.

[Anders Johnsen, 2014/05/01 08:05:11]

"On success, the DisconnectEx function returns TRUE. On failure, the function
returns FALSE. Use the WSAGetLastError function to get extended error
information. If a call to the WSAGetLastError function returns ERROR_IO_PENDING,
the operation initiated successfully and is in progress. Under such
circumstances, the call may still fail when the operation completes."

So basically just like the other Overlapped APIs we use. Adding comment.

     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);

[Søren Gjesse, 2014/05/01 07:19:26]

Please add a comment on this call, e.g. " // Update the accepted socket to
support the full range of API calls."

[Anders Johnsen, 2014/05/01 08:05:11]

Done.

+  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)