Merge IO v2 branch to bleeding edge

runtime/bin/eventhandler_win.cc

 // Copyright (c) 2012, 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 <process.h>  // NOLINT
 #include <winsock2.h>  // NOLINT
 #include <ws2tcpip.h>  // NOLINT
 #include <mswsock.h>  // NOLINT
 
 #include "bin/builtin.h"
 #include "bin/dartutils.h"
 #include "bin/log.h"
 #include "bin/socket.h"
 #include "bin/utils.h"
 #include "platform/thread.h"
 
 
 static const int kInfinityTimeout = -1;
 static const int kTimeoutId = -1;
 static const int kShutdownId = -2;
 
-
 IOBuffer* IOBuffer::AllocateBuffer(int buffer_size, Operation operation) {
   IOBuffer* buffer = new(buffer_size) IOBuffer(buffer_size, operation);
   return buffer;
 }
 
 
 IOBuffer* IOBuffer::AllocateAcceptBuffer(int buffer_size) {
   IOBuffer* buffer = AllocateBuffer(buffer_size, kAccept);
   return buffer;
 }
 
 
 IOBuffer* IOBuffer::AllocateReadBuffer(int buffer_size) {
   return AllocateBuffer(buffer_size, kRead);
 }
 
 
 IOBuffer* IOBuffer::AllocateWriteBuffer(int buffer_size) {
   return AllocateBuffer(buffer_size, kWrite);
 }
 
 
+IOBuffer* IOBuffer::AllocateDisconnectBuffer() {
+  return AllocateBuffer(0, kDisconnect);
+}
+
+
 void IOBuffer::DisposeBuffer(IOBuffer* buffer) {
   delete buffer;
 }
 
 
 IOBuffer* IOBuffer::GetFromOverlapped(OVERLAPPED* overlapped) {
   IOBuffer* buffer = CONTAINING_RECORD(overlapped, IOBuffer, overlapped_);
   return buffer;
 }
 
@@ skipping 73 matching lines @@
                                             reinterpret_cast<ULONG_PTR>(this),
                                             0);
   if (completion_port_ == NULL) {
     Log::PrintErr("Error CreateIoCompletionPort: %d\n", GetLastError());
     return false;
   }
   return true;
 }
 
 
-void Handle::close() {
+void Handle::Close() {
   ScopedLock lock(this);
   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);
     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;
+    // Perform handle type specific closing.
+    DoClose();
   }
-
-  // Perform socket type specific close handling.
-  AfterClose();
 }
 
 
+void Handle::DoClose() {
+  CloseHandle(handle_);
+  handle_ = INVALID_HANDLE_VALUE;
+}
+
+
 bool Handle::HasPendingRead() {
   ScopedLock lock(this);
   return pending_read_ != NULL;
 }
 
 
 bool Handle::HasPendingWrite() {
   ScopedLock lock(this);
   return pending_write_ != NULL;
 }
@@ skipping 131 matching lines @@
     CreateCompletionPort(event_handler_->completion_port());
   }
 }
 
 
 bool FileHandle::IsClosed() {
   return false;
 }
 
 
-void FileHandle::AfterClose() {
-}
-
-
 void SocketHandle::HandleIssueError() {
   int error = WSAGetLastError();
   if (error == WSAECONNRESET) {
     event_handler_->HandleClosed(this);
   } else {
     event_handler_->HandleError(this);
   }
   WSASetLastError(error);
 }
 
 
 bool ListenSocket::LoadAcceptEx() {
   // Load the AcceptEx function into memory using WSAIoctl.
-  // The WSAIoctl function is an extension of the ioctlsocket()
-  // function that can use overlapped I/O. The function's 3rd
-  // through 6th parameters are input and output buffers where
-  // we pass the pointer to our AcceptEx function. This is used
-  // so that we can call the AcceptEx function directly, rather
-  // than refer to the Mswsock.lib library.
   GUID guid_accept_ex = WSAID_ACCEPTEX;
   DWORD bytes;
   int status = WSAIoctl(socket(),
                         SIO_GET_EXTENSION_FUNCTION_POINTER,
                         &guid_accept_ex,
                         sizeof(guid_accept_ex),
                         &AcceptEx_,
                         sizeof(AcceptEx_),
                         &bytes,
                         NULL,
@@ skipping 47 matching lines @@
 void ListenSocket::AcceptComplete(IOBuffer* buffer, HANDLE completion_port) {
   ScopedLock lock(this);
   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) {
-      linger l;
-      l.l_onoff = 1;
-      l.l_linger = 10;
-      int status = setsockopt(buffer->client(),
-                              SOL_SOCKET,
-                              SO_LINGER,
-                              reinterpret_cast<char*>(&l),
-                              sizeof(l));
-      if (status != NO_ERROR) {
-        FATAL("Failed setting SO_LINGER on socket");
-      }
       // Insert the accepted socket into the list.
       ClientSocket* client_socket = new ClientSocket(buffer->client(), 0);
       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 {
       Log::PrintErr("setsockopt failed: %d\n", WSAGetLastError());
       closesocket(buffer->client());
     }
   }
 
   pending_accept_count_--;
   IOBuffer::DisposeBuffer(buffer);
 }
 
 
+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();
+    } else {
+      break;
+    }
+  }
+}
+
+
+bool ListenSocket::CanAccept() {
+  ScopedLock lock(this);
+  return accepted_head_ != NULL;
+}
+
+
 ClientSocket* ListenSocket::Accept() {
   ScopedLock lock(this);
   if (accepted_head_ == NULL) return NULL;
   ClientSocket* result = accepted_head_;
   accepted_head_ = accepted_head_->next();
   if (accepted_head_ == NULL) accepted_tail_ = NULL;
   result->set_next(NULL);
   return result;
 }
 
 
 void ListenSocket::EnsureInitialized(
     EventHandlerImplementation* event_handler) {
   ScopedLock lock(this);
   if (AcceptEx_ == NULL) {
     ASSERT(completion_port_ == INVALID_HANDLE_VALUE);
     ASSERT(event_handler_ == NULL);
     event_handler_ = event_handler;
     CreateCompletionPort(event_handler_->completion_port());
     LoadAcceptEx();
   }
 }
 
 
-void ListenSocket::AfterClose() {
-  ScopedLock lock(this);
-  while (true) {
-    // Get rid of connections already accepted.
-    ClientSocket *client = Accept();
-    if (client != NULL) {
-      client->close();
-    } else {
-      break;
-    }
-  }
-}
-
-
 bool ListenSocket::IsClosed() {
   return IsClosing() && !HasPendingAccept();
 }
 
 
 int Handle::Available() {
   ScopedLock lock(this);
   if (data_ready_ == NULL) return 0;
   ASSERT(!data_ready_->IsEmpty());
   return data_ready_->GetRemainingLength();
@@ skipping 33 matching lines @@
       if (GetLastError() != ERROR_BROKEN_PIPE) {
         Log::PrintErr("WriteFile failed: %d\n", GetLastError());
       }
       event_handler_->HandleClosed(this);
     }
     return bytes_written;
   }
 }
 
 
+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);
+  if (status == SOCKET_ERROR) {
+    Log::PrintErr("Error WSAIoctl failed: %d\n", WSAGetLastError());
+    return false;
+  }
+  return true;
+}
+
+
 void ClientSocket::Shutdown(int how) {
   int rc = shutdown(socket(), how);
-  if (rc == SOCKET_ERROR) {
-    Log::PrintErr("shutdown failed: %d %d\n", socket(), WSAGetLastError());
-  }
   if (how == SD_RECEIVE) MarkClosedRead();
   if (how == SD_SEND) MarkClosedWrite();
   if (how == SD_BOTH) {
     MarkClosedRead();
     MarkClosedWrite();
   }
 }
 
 
+void ClientSocket::DoClose() {
+  // Always do a suhtdown before initiating a disconnect.
+  shutdown(socket(), SD_BOTH);
+  IssueDisconnect();
+}
+
+
 bool ClientSocket::IssueRead() {
   ScopedLock lock(this);
   ASSERT(completion_port_ != INVALID_HANDLE_VALUE);
   ASSERT(pending_read_ == NULL);
 
   IOBuffer* buffer = IOBuffer::AllocateReadBuffer(1024);
 
   DWORD flags;
   flags = 0;
   int rc = WSARecv(socket(),
@@ skipping 30 matching lines @@
   if (rc == NO_ERROR || WSAGetLastError() == WSA_IO_PENDING) {
     return true;
   }
   IOBuffer::DisposeBuffer(pending_write_);
   pending_write_ = NULL;
   HandleIssueError();
   return false;
 }
 
 
+void ClientSocket::IssueDisconnect() {
+  IOBuffer* buffer = IOBuffer::AllocateDisconnectBuffer();
+  BOOL ok = DisconnectEx_(
+    socket(), buffer->GetCleanOverlapped(), TF_REUSE_SOCKET, 0);
+  if (!ok && WSAGetLastError() != WSA_IO_PENDING) {
+    DisconnectComplete(buffer);
+  }
+}
+
+
+void ClientSocket::DisconnectComplete(IOBuffer* buffer) {
+  IOBuffer::DisposeBuffer(buffer);
+  closesocket(socket());
+  if (data_ready_ != NULL) {
+    IOBuffer::DisposeBuffer(data_ready_);
+  }
+  // When disconnect is complete get rid of the object.
+  delete this;
+}
+
+
 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());
   }
 }
 
 
-void ClientSocket::AfterClose() {
-  ScopedLock lock(this);
-  if (data_ready_ != NULL) {
-    IOBuffer::DisposeBuffer(data_ready_);
-    data_ready_ = NULL;
-  }
-}
-
-
 bool ClientSocket::IsClosed() {
-  return IsClosing() && !HasPendingRead() && !HasPendingWrite();
+  return false;
 }
 
 
 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_ = msg->data;
     timeout_port_ = msg->dart_port;
   } else if (msg->id == kShutdownId) {
     shutdown_ = true;
   } else {
     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*>(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 incomming connections are requested make sure to post already
+      // accepted connections.
       if ((msg->data & (1 << kInEvent)) != 0) {
+        if (listen_socket->CanAccept()) {
+          int event_mask = (1 << kInEvent);
+          handle->set_mask(handle->mask() & ~event_mask);
+          DartUtils::PostInt32(handle->port(), event_mask);
+        }
+        // Always keep 5 outstanding accepts going, to enhance performance.
         while (listen_socket->pending_accept_count() < 5) {
           listen_socket->IssueAccept();
         }
       }
 
       if ((msg->data & (1 << kCloseCommand)) != 0) {
-        listen_socket->close();
+        listen_socket->Close();
         if (listen_socket->IsClosed()) {
           delete_handle = true;
         }
       }
     } else {
-      handle->SetPortAndMask(msg->dart_port, msg->data);
       handle->EnsureInitialized(this);
 
       Handle::ScopedLock lock(handle);
 
       if (!handle->IsError()) {
-        // If in events (data available events) have been requested, and data
-        // is available, post an in event 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 (handle->Available() > 0) {
-            int event_mask = (1 << kInEvent);
-            handle->set_mask(handle->mask() & ~event_mask);
-            DartUtils::PostInt32(handle->port(), event_mask);
-          } else if (!handle->HasPendingRead() &&
-                     !handle->IsClosedRead()) {
-            handle->IssueRead();
+        if ((msg->data & ((1 << kInEvent) | (1 << kOutEvent))) != 0) {
+          // Only set mask if we turned on kInEvent or kOutEvent.
+          handle->SetPortAndMask(msg->dart_port, msg->data);
+
+          // If in events (data available events) have been requested, and data
+          // is available, post an in event 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 (handle->Available() > 0) {
+              int event_mask = (1 << kInEvent);
+              handle->set_mask(handle->mask() & ~event_mask);
+              DartUtils::PostInt32(handle->port(), event_mask);
+            } else if (handle->IsClosedRead()) {
+              int event_mask = (1 << kCloseEvent);
+              DartUtils::PostInt32(handle->port(), event_mask);
+            } else if (!handle->HasPendingRead()) {
+              handle->IssueRead();
+            }
+          }
+
+          // If out events (can write events) have been requested, and there
+          // are no pending writes, post an out event immediately.
+          if ((msg->data & (1 << kOutEvent)) != 0) {
+            if (!handle->HasPendingWrite()) {
+              int event_mask = (1 << kOutEvent);
+              handle->set_mask(handle->mask() & ~event_mask);
+              DartUtils::PostInt32(handle->port(), event_mask);
+            }
           }
         }
 
-        // If out events (can write events) have been requested, and there
-        // are no pending writes, post an out event immediately.
-        if ((msg->data & (1 << kOutEvent)) != 0) {
-          if (!handle->HasPendingWrite()) {
-            int event_mask = (1 << kOutEvent);
-            handle->set_mask(handle->mask() & ~event_mask);
-            DartUtils::PostInt32(handle->port(), event_mask);
-          }
-        }
-
         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 << kCloseCommand)) != 0) {
-        handle->close();
+        handle->Close();
         if (handle->IsClosed()) {
           delete_handle = true;
         }
       }
     }
     if (delete_handle) {
       delete handle;
     }
   }
 }
@@ skipping 78 matching lines @@
   } else {
     HandleError(handle);
   }
 
   if (handle->IsClosed()) {
     delete handle;
   }
 }
 
 
+void EventHandlerImplementation::HandleDisconnect(
+    ClientSocket* client_socket,
+    int bytes,
+    IOBuffer* buffer) {
+  client_socket->DisconnectComplete(buffer);
+}
+
 void EventHandlerImplementation::HandleTimeout() {
   // TODO(sgjesse) check if there actually is a timeout.
   DartUtils::PostNull(timeout_port_);
   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: {
       Handle* handle = reinterpret_cast<Handle*>(key);
       HandleRead(handle, bytes, buffer);
       break;
     }
     case IOBuffer::kWrite: {
       Handle* handle = reinterpret_cast<Handle*>(key);
       HandleWrite(handle, bytes, buffer);
       break;
     }
+    case IOBuffer::kDisconnect: {
+      ClientSocket* client_socket = reinterpret_cast<ClientSocket*>(key);
+      HandleDisconnect(client_socket, bytes, buffer);
+      break;
+    }
     default:
       UNREACHABLE();
   }
 }
 
 
 EventHandlerImplementation::EventHandlerImplementation() {
   intptr_t result;
   completion_port_ =
       CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, NULL, 1);
@@ skipping 94 matching lines @@
     FATAL("Failed to initialized Windows sockets");
   }
 }
 
 
 void EventHandlerImplementation::Shutdown() {
   SendData(kShutdownId, 0, 0);
 }
 
 #endif  // defined(TARGET_OS_WINDOWS)