Make event-handlers edge-triggered and move socket-state to Dart.

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 355 matching lines @@
 }
 
 
 bool DirectoryWatchHandle::IsClosed() {
   return IsClosing() && pending_read_ == NULL;
 }
 
 
 bool DirectoryWatchHandle::IssueRead() {
   ScopedLock lock(this);
+  // It may have been started before, as we start the directory-handler when
+  // we create it.
+  if (pending_read_ != NULL) return true;
   OverlappedBuffer* buffer = OverlappedBuffer::AllocateReadBuffer(kBufferSize);
 
   ASSERT(completion_port_ != INVALID_HANDLE_VALUE);
 
   BOOL ok = ReadDirectoryChangesW(handle_,
                                   buffer->GetBufferStart(),
                                   buffer->GetBufferSize(),
                                   recursive_,
                                   events_,
                                   NULL,
@@ skipping 136 matching lines @@
 }
 
 
 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);
+  if (!IsClosing()) {
+    while (pending_accept_count() < 5) {
+      if (!IssueAccept()) {
+        event_handler_->HandleError(this);
+      }
+    }
+  }
   return result;
 }
 
 
 void ListenSocket::EnsureInitialized(
     EventHandlerImplementation* event_handler) {
   ScopedLock lock(this);
   if (AcceptEx_ == NULL) {
     ASSERT(completion_port_ == INVALID_HANDLE_VALUE);
     ASSERT(event_handler_ == NULL);
@@ skipping 17 matching lines @@
 }
 
 
 int Handle::Read(void* buffer, int num_bytes) {
   ScopedLock lock(this);
   if (data_ready_ == NULL) return 0;
   num_bytes = data_ready_->Read(buffer, num_bytes);
   if (data_ready_->IsEmpty()) {
     OverlappedBuffer::DisposeBuffer(data_ready_);
     data_ready_ = NULL;
+    if (!IsClosing() && !IsClosedRead()) IssueRead();
   }
   return num_bytes;
 }
 
 
 int Handle::RecvFrom(
     void* buffer, int num_bytes, struct sockaddr* sa, socklen_t sa_len) {
   ScopedLock lock(this);
   if (data_ready_ == NULL) return 0;
   num_bytes = data_ready_->Read(buffer, num_bytes);
   if (data_ready_->from()->sa_family == AF_INET) {
     ASSERT(sa_len >= sizeof(struct sockaddr_in));
     memmove(sa, data_ready_->from(), sizeof(struct sockaddr_in));
   } else {
     ASSERT(data_ready_->from()->sa_family == AF_INET6);
     ASSERT(sa_len >= sizeof(struct sockaddr_in6));
     memmove(sa, data_ready_->from(), sizeof(struct sockaddr_in6));
   }
   // Always dispose of the buffer, as UDP messages must be read in their
   // entirety to match how recvfrom works in a socket.
   OverlappedBuffer::DisposeBuffer(data_ready_);
   data_ready_ = NULL;
+  if (!IsClosing() && !IsClosedRead()) IssueRecvFrom();
   return num_bytes;
 }
 
 
 int Handle::Write(const void* buffer, int num_bytes) {
   ScopedLock lock(this);
   if (pending_write_ != NULL) return 0;
   if (num_bytes > kBufferSize) num_bytes = kBufferSize;
   ASSERT(SupportsOverlappedIO());
   if (completion_port_ == INVALID_HANDLE_VALUE) return 0;
@@ skipping 373 matching lines @@
       }
 
       if ((msg->data & (1 << kCloseCommand)) != 0) {
         listen_socket->Close();
       }
     } else {
       handle->EnsureInitialized(this);
 
       Handle::ScopedLock lock(handle);
 
-      if (handle->IsError()) {
-        DartUtils::PostInt32(msg->dart_port, 1 << kErrorEvent);
-      } else {
-        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);
+      // 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);
+      }
 
-          // 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()) {
-              if (handle->is_datagram_socket()) {
-                handle->IssueRecvFrom();
-              } else {
-                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 (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);
-          }
+      // Issue a read.
+      if ((msg->data & (1 << kInEvent)) != 0) {
+        handle->SetPortAndMask(msg->dart_port, msg->data);
+        if (handle->is_datagram_socket()) {
+          handle->IssueRecvFrom();
+        } else {
+          handle->IssueRead();
         }
       }
 
-      if ((msg->data & (1 << kCloseCommand)) != 0) {
+      // 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);
-        handle->Close();
+        if (!handle->HasPendingWrite()) {
+          int event_mask = (1 << kOutEvent);
+          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->SetPortAndMask(msg->dart_port, msg->data);
+      handle->Close();
+    }
+
     DeleteIfClosed(handle);
   }
 }
 
 
 void EventHandlerImplementation::HandleAccept(ListenSocket* listen_socket,
                                               OverlappedBuffer* buffer) {
   listen_socket->AcceptComplete(buffer, completion_port_);
 
   if (!listen_socket->IsClosing()) {
@@ skipping 252 matching lines @@
 
 
 void EventHandlerImplementation::Shutdown() {
   SendData(kShutdownId, 0, 0);
 }
 
 }  // namespace bin
 }  // namespace dart
 
 #endif  // defined(TARGET_OS_WINDOWS)