Introduce optional 'bool shared' parameter to ServerSocket.bind() ...

dart/runtime/bin/eventhandler_win.cc

Index: dart/runtime/bin/eventhandler_win.cc
diff --git a/dart/runtime/bin/eventhandler_win.cc b/dart/runtime/bin/eventhandler_win.cc
index fff9d87d9ae8f1db639c3eb777639452c42517e0..c357a736e650bc5de5b00361b30c0d43712d0965 100644
--- a/dart/runtime/bin/eventhandler_win.cc
+++ b/dart/runtime/bin/eventhandler_win.cc
@@ -6,6 +6,7 @@
 #if defined(TARGET_OS_WINDOWS)
 
 #include "bin/eventhandler.h"
+#include "bin/eventhandler_win.h"
 
 #include <winsock2.h>  // NOLINT
 #include <ws2tcpip.h>  // NOLINT
@@ -29,10 +30,6 @@ namespace bin {
 static const int kBufferSize = 64 * 1024;
 static const int kStdOverlappedBufferSize = 16 * 1024;
 
-static const int kInfinityTimeout = -1;
-static const int kTimeoutId = -1;
-static const int kShutdownId = -2;
-
 OverlappedBuffer* OverlappedBuffer::AllocateBuffer(int buffer_size,
                                                    Operation operation) {
   OverlappedBuffer* buffer =
@@ -115,26 +112,9 @@ int OverlappedBuffer::GetRemainingLength() {
   return data_length_ - index_;
 }
 
-
-Handle::Handle(HANDLE handle)
-    : handle_(reinterpret_cast<HANDLE>(handle)),
-      port_(0),
-      mask_(0),
-      completion_port_(INVALID_HANDLE_VALUE),
-      event_handler_(NULL),
-      data_ready_(NULL),
-      pending_read_(NULL),
-      pending_write_(NULL),
-      last_error_(NOERROR),
-      flags_(0) {
-  InitializeCriticalSection(&cs_);
-}
-
-
-Handle::Handle(HANDLE handle, Dart_Port port)
-    : handle_(reinterpret_cast<HANDLE>(handle)),
-      port_(port),
-      mask_(0),
+Handle::Handle(intptr_t handle)
+    : DescriptorInfo(handle),
+      handle_(reinterpret_cast<HANDLE>(handle)),
       completion_port_(INVALID_HANDLE_VALUE),
       event_handler_(NULL),
       data_ready_(NULL),
@@ -293,7 +273,7 @@ bool Handle::IssueRead() {
     // Completing asynchronously through thread.
     pending_read_ = buffer;
     int result = Thread::Start(ReadFileThread,
-                                     reinterpret_cast<uword>(this));
+                               reinterpret_cast<uword>(this));
     if (result != 0) {
       FATAL1("Failed to start read file thread %d", result);
     }
@@ -339,7 +319,7 @@ bool Handle::IssueSendTo(struct sockaddr* sa, socklen_t sa_len) {
 static void HandleClosed(Handle* handle) {
   if (!handle->IsClosing()) {
     int event_mask = 1 << kCloseEvent;
-    DartUtils::PostInt32(handle->port(), event_mask);
+    DartUtils::PostInt32(handle->NextPort(), event_mask);
   }
 }
 
@@ -347,11 +327,8 @@ static void HandleClosed(Handle* handle) {
 static void HandleError(Handle* handle) {
   handle->set_last_error(WSAGetLastError());
   handle->MarkError();
-  if (!handle->IsClosing()) {
-    Dart_Port port = handle->port();
-    if (port != ILLEGAL_PORT) {
-      DartUtils::PostInt32(port, 1 << kErrorEvent);
-    }
+  if (!handle->IsClosing() && handle->HasNextPort()) {
+    DartUtils::PostInt32(handle->NextPort(), 1 << kErrorEvent);
   }
 }
 
@@ -515,7 +492,7 @@ void ListenSocket::AcceptComplete(OverlappedBuffer* buffer,
                         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);
+      ClientSocket* client_socket = new ClientSocket(buffer->client());
       client_socket->mark_connected();
       client_socket->CreateCompletionPort(completion_port);
       if (accepted_head_ == NULL) {
@@ -541,11 +518,12 @@ void ListenSocket::AcceptComplete(OverlappedBuffer* buffer,
 
 static void DeleteIfClosed(Handle* handle) {
   if (handle->IsClosed()) {

[Søren Gjesse, 2015/02/02 10:56:14]

Indentation.

[kustermann, 2015/02/03 10:45:35]

Done.

-    Dart_Port port = handle->port();
-    delete handle;
-    if (port != ILLEGAL_PORT) {
-      DartUtils::PostInt32(port, 1 << kDestroyedEvent);
-    }
+  while (handle->HasNextPort()) {
+    Dart_Port port = handle->NextPort();
+    DartUtils::PostInt32(port, 1 << kDestroyedEvent);
+    handle->RemovePort(port);
+  }
+  delete handle;
   }
 }
 
@@ -880,9 +858,11 @@ void ClientSocket::IssueDisconnect() {
   if (ok || WSAGetLastError() != WSA_IO_PENDING) {
     DisconnectComplete(buffer);
   }
-  Dart_Port p = port();
-  if (p != ILLEGAL_PORT) DartUtils::PostInt32(p, 1 << kDestroyedEvent);
-  port_ = ILLEGAL_PORT;
+  if (HasNextPort()) {
+    Dart_Port p = NextPort();
+    DartUtils::PostInt32(p, 1 << kDestroyedEvent);
+    RemovePort(p);
+  }
 }
 
 
@@ -900,15 +880,14 @@ 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);
-  Dart_Port p = port();
-  if (p != ILLEGAL_PORT) {
+  if (HasNextPort()) {
     // 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);
+      DartUtils::PostInt32(NextPort(), 1 << kOutEvent);
     }
   }
 }
@@ -1011,22 +990,33 @@ void DatagramSocket::DoClose() {
 
 void EventHandlerImplementation::HandleInterrupt(InterruptMessage* msg) {
   ASSERT(this != NULL);
-  if (msg->id == kTimeoutId) {
+  if (msg->id == kTimerId) {
     // 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) {
     shutdown_ = true;
   } else {
-    // No tokens to return on Windows.
-    if ((msg->data & (1 << kReturnTokenCommand)) != 0) return;
     Handle* handle = reinterpret_cast<Handle*>(msg->id);
     ASSERT(handle != NULL);
+
+    if (IS_COMMAND(msg->data, kReturnTokenCommand)) {
+      int count = TOKEN_COUNT(msg->data);
+      bool become_active = handle->ReturnTokens(msg->dart_port, count);
+      // TODO(kustermann): on linux we do

[Søren Gjesse, 2015/02/02 10:56:14]

On Windows we always have 5 outstanding accept requests. These are either
pending or completed and sitting in the linked list of accepted connections
ready to pass on to Dart.

+      // if (become_active) {
+      //   ReturnToEpollSet();
+      // }
+      return;
+    }
+
     if (handle->is_listen_socket()) {
       ListenSocket* listen_socket =
           reinterpret_cast<ListenSocket*>(handle);
       listen_socket->EnsureInitialized(this);
-      listen_socket->SetPortAndMask(msg->dart_port, msg->data);
+
+      listen_socket->SetMask(msg->data);
+      listen_socket->AddPort(msg->dart_port);
 
       Handle::ScopedLock lock(listen_socket);
 
@@ -1034,9 +1024,17 @@ void EventHandlerImplementation::HandleInterrupt(InterruptMessage* msg) {
       // accepted connections.
       if ((msg->data & (1 << kInEvent)) != 0) {
         if (listen_socket->CanAccept()) {
+          Dart_Port port = handle->NextPort();
           int event_mask = (1 << kInEvent);
-          handle->set_mask(handle->mask() & ~event_mask);
-          DartUtils::PostInt32(handle->port(), event_mask);
+          DartUtils::PostInt32(port, event_mask);
+          handle->SetMask(handle->Mask() & ~event_mask);
+          if (listen_socket->TakeToken()) {
+            // TODO(kustermann): On linux we remove listening sockeet from
+            // epoll set. here
+            // FIXME(kustermann): Should we do this only if there are no more
+            // listeners?
+            // handle->SetMask(handle->Mask() & ~event_mask);
+          }
         }
       }
     } else {
@@ -1046,7 +1044,8 @@ void EventHandlerImplementation::HandleInterrupt(InterruptMessage* msg) {
 
       // 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);
+        handle->SetMask(msg->data);
+        handle->AddPort(msg->dart_port);
       }
 
       // Issue a read.
@@ -1069,10 +1068,10 @@ void EventHandlerImplementation::HandleInterrupt(InterruptMessage* msg) {
         if (!handle->HasPendingWrite()) {
           if (handle->is_client_socket()) {
             if (reinterpret_cast<ClientSocket*>(handle)->is_connected()) {
-              DartUtils::PostInt32(handle->port(), 1 << kOutEvent);
+              DartUtils::PostInt32(handle->NextPort(), 1 << kOutEvent);
             }
           } else {
-            DartUtils::PostInt32(handle->port(), 1 << kOutEvent);
+            DartUtils::PostInt32(handle->NextPort(), 1 << kOutEvent);
           }
         }
       }
@@ -1087,11 +1086,12 @@ void EventHandlerImplementation::HandleInterrupt(InterruptMessage* msg) {
           client_socket->Shutdown(SD_SEND);
         }
       }
-    }
 
-    if ((msg->data & (1 << kCloseCommand)) != 0) {
-      handle->SetPortAndMask(msg->dart_port, msg->data);
-      handle->Close();
+      if ((msg->data & (1 << kCloseCommand)) != 0) {
+        handle->SetMask(msg->data);
+        handle->AddPort(msg->dart_port);
+        handle->Close();
+      }
     }
 
     DeleteIfClosed(handle);
@@ -1104,9 +1104,8 @@ void EventHandlerImplementation::HandleAccept(ListenSocket* listen_socket,
   listen_socket->AcceptComplete(buffer, completion_port_);
 
   if (!listen_socket->IsClosing()) {
-    int event_mask = 1 << kInEvent;
-    if ((listen_socket->mask() & event_mask) != 0) {
-      DartUtils::PostInt32(listen_socket->port(), event_mask);
+    if ((listen_socket->Mask() & (1 << kInEvent)) != 0) {
+      DartUtils::PostInt32(listen_socket->NextPort(), 1 << kInEvent);
     }
   }
 
@@ -1122,8 +1121,8 @@ void EventHandlerImplementation::HandleRead(Handle* handle,
   if (bytes > 0) {
     if (!handle->IsClosing()) {
       int event_mask = 1 << kInEvent;
-      if ((handle->mask() & event_mask) != 0) {
-        DartUtils::PostInt32(handle->port(), event_mask);
+      if ((handle->Mask() & event_mask) != 0) {
+        DartUtils::PostInt32(handle->NextPort(), event_mask);
       }
     }
   } else {
@@ -1147,8 +1146,8 @@ void EventHandlerImplementation::HandleRecvFrom(Handle* handle,
   handle->ReadComplete(buffer);
   if (!handle->IsClosing()) {
     int event_mask = 1 << kInEvent;
-    if ((handle->mask() & event_mask) != 0) {
-      DartUtils::PostInt32(handle->port(), event_mask);
+    if ((handle->Mask() & event_mask) != 0) {
+      DartUtils::PostInt32(handle->NextPort(), event_mask);
     }
   }
 
@@ -1166,8 +1165,8 @@ void EventHandlerImplementation::HandleWrite(Handle* handle,
       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);
+      if ((handle->Mask() & event_mask) != 0) {
+        DartUtils::PostInt32(handle->NextPort(), event_mask);
       }
     }
   } else {
@@ -1350,7 +1349,7 @@ void EventHandlerImplementation::EventHandlerEntry(uword args) {
 
 void EventHandlerImplementation::Start(EventHandler* handler) {
   int result = Thread::Start(EventHandlerEntry,
-                                   reinterpret_cast<uword>(handler));
+                             reinterpret_cast<uword>(handler));
   if (result != 0) {
     FATAL1("Failed to start event handler thread %d", result);
   }