Fix handling of async IO from pipes on Windows

runtime/bin/eventhandler_win.cc

 // Copyright (c) 2011, 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 <process.h>
 #include <winsock2.h>
 #include <ws2tcpip.h>
 #include <mswsock.h>
 
 #include "bin/builtin.h"
@@ skipping 73 matching lines @@
 
 
 int IOBuffer::GetRemainingLength() {
   ASSERT(operation_ == kRead);
   return data_length_ - index_;
 }
 
 
 Handle::Handle(HANDLE handle)
     : handle_(reinterpret_cast<HANDLE>(handle)),
-      closing_(false),
+      flags_(0),
       port_(0),
       completion_port_(INVALID_HANDLE_VALUE),
       event_handler_(NULL),
       data_ready_(NULL),
       pending_read_(NULL),
       pending_write_(NULL) {
   InitializeCriticalSection(&cs_);
 }
 
 
 Handle::Handle(HANDLE handle, Dart_Port port)
     : handle_(reinterpret_cast<HANDLE>(handle)),
-      closing_(false),
+      flags_(0),
       port_(port),
       completion_port_(INVALID_HANDLE_VALUE),
       event_handler_(NULL),
       data_ready_(NULL),
       pending_read_(NULL),
       pending_write_(NULL) {
   InitializeCriticalSection(&cs_);
 }
 
 
@@ skipping 20 matching lines @@
   if (completion_port_ == NULL) {
     fprintf(stderr, "Error CreateIoCompletionPort: %d\n", GetLastError());
     return false;
   }
   return true;
 }
 
 
 void Handle::close() {
   ScopedLock lock(this);
-  if (!closing_) {
+  if (!IsClosing()) {
     // Close the socket and set the closing state. This close method can be
     // called again if this socket has pending IO operations in flight.
     ASSERT(handle_ != INVALID_HANDLE_VALUE);
-    closing_ = true;
+    MarkClosing();
     // According to the documentation from Microsoft socket handles should
     // not be closed using CloseHandle but using closesocket.
     if (is_socket()) {
       closesocket(reinterpret_cast<SOCKET>(handle_));
     } else {
       CloseHandle(handle_);
     }
     handle_ = INVALID_HANDLE_VALUE;
   }
 
@@ skipping 12 matching lines @@
   ScopedLock lock(this);
   return pending_write_ != NULL;
 }
 
 
 void Handle::ReadComplete(IOBuffer* buffer) {
   ScopedLock lock(this);
   // Currently only one outstanding read at the time.
   ASSERT(pending_read_ == buffer);
   ASSERT(data_ready_ == NULL);
-  if (!closing_ && !buffer->IsEmpty()) {
+  if (!IsClosing() && !buffer->IsEmpty()) {
     data_ready_ = pending_read_;
   } else {
     IOBuffer::DisposeBuffer(buffer);
   }
   pending_read_ = NULL;
 }
 
 
 void Handle::WriteComplete(IOBuffer* buffer) {
   ScopedLock lock(this);
@@ skipping 138 matching lines @@
   }
 
   pending_accept_count_++;
 
   return true;
 }
 
 
 void ListenSocket::AcceptComplete(IOBuffer* buffer, HANDLE completion_port) {
   ScopedLock lock(this);
-  if (!closing_) {
+  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->CreateCompletionPort(completion_port);
@@ skipping 48 matching lines @@
     if (client != NULL) {
       client->close();
     } else {
       break;
     }
   }
 }
 
 
 bool ListenSocket::IsClosed() {
-  return closing_ && !HasPendingAccept();
+  return IsClosing() && !HasPendingAccept();
 }
 
 
 int Handle::Available() {
   ScopedLock lock(this);
   if (data_ready_ == NULL) return 0;
   ASSERT(!data_ready_->IsEmpty());
   return data_ready_->GetRemainingLength();
 }
 
@@ skipping 103 matching lines @@
 void ClientSocket::AfterClose() {
   ScopedLock lock(this);
   if (data_ready_ != NULL) {
     IOBuffer::DisposeBuffer(data_ready_);
     data_ready_ = NULL;
   }
 }
 
 
 bool ClientSocket::IsClosed() {
-  return closing_ && !HasPendingRead() && !HasPendingWrite();
+  return IsClosing() && !HasPendingRead() && !HasPendingWrite();
 }
 
 
 void EventHandlerImplementation::HandleInterrupt(InterruptMessage* msg) {
   if (msg->id == -1) {
     // 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_ = msg->data;
     timeout_port_ = msg->dart_port;
   } else {
-    bool delete_socket = false;
-    Handle* socket_desc =
-        reinterpret_cast<Handle*>(msg->id);
-    ASSERT(socket_desc != NULL);
-    if (socket_desc->is_listen_socket()) {
+    bool delete_handle = false;
+    Handle* handle = reinterpret_cast<Handle*>(msg->id);
+    ASSERT(handle != NULL);
+    if (handle->is_listen_socket()) {
       ListenSocket* listen_socket =
-          reinterpret_cast<ListenSocket*>(socket_desc);
+          reinterpret_cast<ListenSocket*>(handle);
       listen_socket->EnsureInitialized(this);
       listen_socket->SetPortAndMask(msg->dart_port, msg->data);
 
       Handle::ScopedLock lock(listen_socket);
 
       // If incomming connections are requested make sure that pending accepts
       // are issued.
       if ((msg->data & (1 << kInEvent)) != 0) {
         while (listen_socket->pending_accept_count() < 5) {
           listen_socket->IssueAccept();
         }
       }
 
       if ((msg->data & (1 << kCloseCommand)) != 0) {
         listen_socket->close();
         if (listen_socket->IsClosed()) {
-          delete_socket = true;
+          delete_handle = true;
         }
       }
     } else {
-      ClientSocket* client_socket =
-          reinterpret_cast<ClientSocket*>(socket_desc);
-      client_socket->SetPortAndMask(msg->dart_port, msg->data);
-      client_socket->EnsureInitialized(this);
+      handle->SetPortAndMask(msg->dart_port, msg->data);
+      handle->EnsureInitialized(this);
 
-      Handle::ScopedLock lock(client_socket);
+      Handle::ScopedLock lock(handle);
 
       // If the data available callback has been requested and data are
       // available post it immediately. Otherwise make sure that a pending
       // read is issued unless the socket is already closed for read.
       if ((msg->data & (1 << kInEvent)) != 0) {
-        if (client_socket->Available() > 0) {
+        if (handle->Available() > 0) {
           int event_mask = (1 << kInEvent);
-          Dart_PostIntArray(client_socket->port(), 1, &event_mask);
-        } else if (!client_socket->HasPendingRead() &&
-                   !client_socket->IsClosedRead()) {
-          client_socket->IssueRead();
+          Dart_PostIntArray(handle->port(), 1, &event_mask);
+        } else if (!handle->HasPendingRead() &&
+                   !handle->IsClosedRead()) {
+          handle->IssueRead();
         }
       }
 
       // If can send callback had been requested and there is no pending
       // send post it immediately.
       if ((msg->data & (1 << kOutEvent)) != 0) {
-        if (!client_socket->HasPendingWrite()) {
+        if (!handle->HasPendingWrite()) {
           int event_mask = (1 << kOutEvent);
-          Dart_PostIntArray(client_socket->port(), 1, &event_mask);
+          Dart_PostIntArray(handle->port(), 1, &event_mask);
         }
       }
 
-      if ((msg->data & (1 << kShutdownReadCommand)) != 0) {
-        client_socket->Shutdown(SD_RECEIVE);
-      }
+      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) {
-        client_socket->Shutdown(SD_SEND);
+        if ((msg->data & (1 << kShutdownWriteCommand)) != 0) {
+          client_socket->Shutdown(SD_SEND);
+        }
       }
 
       if ((msg->data & (1 << kCloseCommand)) != 0) {
-        client_socket->close();
-        if (client_socket->IsClosed()) {
-          delete_socket = true;
+        handle->close();
+        if (handle->IsClosed()) {
+          delete_handle = true;
         }
       }
     }
-    if (delete_socket) {
-      delete socket_desc;
+    if (delete_handle) {
+      delete handle;
     }
   }
 }
 
 
 void EventHandlerImplementation::HandleAccept(ListenSocket* listen_socket,
                                               IOBuffer* buffer) {
   listen_socket->AcceptComplete(buffer, completion_port_);
 
-  if (!listen_socket->is_closing()) {
+  if (!listen_socket->IsClosing()) {
     int event_mask = 1 << kInEvent;
     if ((listen_socket->mask() & event_mask) != 0) {
       Dart_PostIntArray(listen_socket->port(), 1, &event_mask);
     }
   }
 
   if (listen_socket->IsClosed()) {
     delete listen_socket;
   }
 }
 
 
 void EventHandlerImplementation::HandleClosed(Handle* handle) {
-  if (!handle->is_closing()) {
+  if (!handle->IsClosing()) {
     int event_mask = 1 << kCloseEvent;
     if ((handle->mask() & event_mask) != 0) {
       Dart_PostIntArray(handle->port(), 1, &event_mask);
     }
   }
 }
 
 
-void EventHandlerImplementation::HandleRead(ClientSocket* client_socket,
+void EventHandlerImplementation::HandleRead(Handle* handle,
                                             int bytes,
                                             IOBuffer* buffer) {
   buffer->set_data_length(bytes);
-  client_socket->ReadComplete(buffer);
+  handle->ReadComplete(buffer);
   if (bytes > 0) {
-    if (!client_socket->is_closing()) {
+    if (!handle->IsClosing()) {
       int event_mask = 1 << kInEvent;
-      if ((client_socket->mask() & event_mask) != 0) {
-        Dart_PostIntArray(client_socket->port(), 1, &event_mask);
+      if ((handle->mask() & event_mask) != 0) {
+        Dart_PostIntArray(handle->port(), 1, &event_mask);
       }
     }
   } else {
     ASSERT(bytes == 0);
-    client_socket->MarkClosedRead();
-    HandleClosed(client_socket);
+    handle->MarkClosedRead();
+    HandleClosed(handle);
   }
 
-  if (client_socket->IsClosed()) {
-    delete client_socket;
+  if (handle->IsClosed()) {
+    delete handle;
   }
 }
 
 
-void EventHandlerImplementation::HandleWrite(ClientSocket* client_socket,
+void EventHandlerImplementation::HandleWrite(Handle* handle,
                                              int bytes,
                                              IOBuffer* buffer) {
-  client_socket->WriteComplete(buffer);
+  handle->WriteComplete(buffer);
 
   if (bytes > 0) {
-    if (!client_socket->is_closing()) {
+    if (!handle->IsClosing()) {
       int event_mask = 1 << kOutEvent;
-      if ((client_socket->mask() & event_mask) != 0) {
-        Dart_PostIntArray(client_socket->port(), 1, &event_mask);
+      if ((handle->mask() & event_mask) != 0) {
+        Dart_PostIntArray(handle->port(), 1, &event_mask);
       }
     }
   } else {
     ASSERT(bytes == 0);
-    HandleClosed(client_socket);
+    HandleClosed(handle);
   }
 
-  if (client_socket->IsClosed()) {
-    delete client_socket;
+  if (handle->IsClosed()) {
+    delete handle;
   }
 }
 
 
 void EventHandlerImplementation::HandleTimeout() {
   // TODO(sgjesse) check if there actually is a timeout.
   Dart_PostIntArray(timeout_port_, 0, NULL);
   timeout_ = kInfinityTimeout;
   timeout_port_ = 0;
 }
 
 
 void EventHandlerImplementation::HandleIOCompletion(DWORD bytes,
                                                     ULONG_PTR key,
                                                     OVERLAPPED* overlapped) {
   IOBuffer* buffer = IOBuffer::GetFromOverlapped(overlapped);
   switch (buffer->operation()) {
     case IOBuffer::kAccept: {
       ListenSocket* listen_socket = reinterpret_cast<ListenSocket*>(key);
       HandleAccept(listen_socket, buffer);
       break;
     }
     case IOBuffer::kRead: {
-      ClientSocket* client_socket = reinterpret_cast<ClientSocket*>(key);
-      HandleRead(client_socket, bytes, buffer);
+      Handle* handle = reinterpret_cast<Handle*>(key);
+      HandleRead(handle, bytes, buffer);
       break;
     }
     case IOBuffer::kWrite: {
-      ClientSocket* client_socket = reinterpret_cast<ClientSocket*>(key);
-      HandleWrite(client_socket, bytes, buffer);
+      Handle* handle = reinterpret_cast<Handle*>(key);
+      HandleWrite(handle, bytes, buffer);
       break;
     }
     default:
       UNREACHABLE();
   }
 }
 
 
 EventHandlerImplementation::EventHandlerImplementation() {
   intptr_t result;
@@ skipping 90 matching lines @@
       _beginthreadex(NULL, 32 * 1024, EventHandlerThread, this, 0, &tid);
   if (thread_handle == -1) {
     FATAL("Failed to start event handler thread");
   }
 
   // Initialize Winsock32
   if (!Socket::Initialize()) {
     FATAL("Failed to initialized Windows sockets");
   }
 }