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

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 "bin/eventhandler_win.h"
 
 #include <winsock2.h>  // NOLINT
 #include <ws2tcpip.h>  // NOLINT
 #include <mswsock.h>  // NOLINT
 #include <io.h>  // NOLINT
 #include <fcntl.h>  // NOLINT
 
 #include "bin/builtin.h"
 #include "bin/dartutils.h"
 #include "bin/lockers.h"
 #include "bin/log.h"
 #include "bin/socket.h"
 #include "bin/thread.h"
 #include "bin/utils.h"
 
 #include "platform/utils.h"
 
 namespace dart {
 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 =
       new(buffer_size) OverlappedBuffer(buffer_size, operation);
   return buffer;
 }
 
 
 OverlappedBuffer* OverlappedBuffer::AllocateAcceptBuffer(int buffer_size) {
   OverlappedBuffer* buffer = AllocateBuffer(buffer_size, kAccept);
@@ skipping 62 matching lines @@
   data_length_ = num_bytes;
   return num_bytes;
 }
 
 
 int OverlappedBuffer::GetRemainingLength() {
   ASSERT(operation_ == kRead || operation_ == kRecvFrom);
   return data_length_ - index_;
 }
 
-
-Handle::Handle(HANDLE handle)
-    : handle_(reinterpret_cast<HANDLE>(handle)),
-      port_(0),
-      mask_(0),
+Handle::Handle(intptr_t handle)
+    : DescriptorInfoBase(handle),
+      handle_(reinterpret_cast<HANDLE>(handle)),
       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),
-      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() {
   DeleteCriticalSection(&cs_);
@@ skipping 135 matching lines @@
       pending_read_ = buffer;
       return true;
     }
     OverlappedBuffer::DisposeBuffer(buffer);
     HandleIssueError();
     return false;
   } else {
     // 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);
     }
     return true;
   }
 }
 
 
 bool Handle::IssueRecvFrom() {
   return false;
@@ skipping 25 matching lines @@
 
 
 bool Handle::IssueSendTo(struct sockaddr* sa, socklen_t sa_len) {
   return false;
 }
 
 
 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);
   }
 }
 
 
 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);
   }
 }
 
 
 void Handle::HandleIssueError() {
   DWORD error = GetLastError();
   if (error == ERROR_BROKEN_PIPE) {
     HandleClosed(this);
   } else {
     HandleError(this);
@@ skipping 95 matching lines @@
                         NULL);
   if (status == SOCKET_ERROR) {
     return false;
   }
   return true;
 }
 
 
 bool ListenSocket::IssueAccept() {
   ScopedLock lock(this);
+
   // For AcceptEx there needs to be buffer storage for address
   // information for two addresses (local and remote address). The
   // AcceptEx documentation says: "This value must be at least 16
   // bytes more than the maximum address length for the transport
   // protocol in use."
   static const int kAcceptExAddressAdditionalBytes = 16;
   static const int kAcceptExAddressStorageSize =
       sizeof(SOCKADDR_STORAGE) + kAcceptExAddressAdditionalBytes;
   OverlappedBuffer* buffer =
       OverlappedBuffer::AllocateAcceptBuffer(2 * kAcceptExAddressStorageSize);
@@ skipping 28 matching lines @@
   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) {
       // 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) {
         accepted_head_ = client_socket;
         accepted_tail_ = client_socket;
       } else {
         ASSERT(accepted_tail_ != NULL);
         accepted_tail_->set_next(client_socket);
         accepted_tail_ = client_socket;
       }
+      accepted_count_++;
     } else {
       closesocket(buffer->client());
     }
   } else {
     // Close the socket, as it's already accepted.
     closesocket(buffer->client());
   }
 
   pending_accept_count_--;
   OverlappedBuffer::DisposeBuffer(buffer);
 }
 
 
 static void DeleteIfClosed(Handle* handle) {
   if (handle->IsClosed()) {
-    Dart_Port port = handle->port();
+    handle->SendToAll(1 << kDestroyedEvent);
     delete handle;
-    if (port != ILLEGAL_PORT) {
-      DartUtils::PostInt32(port, 1 << kDestroyedEvent);
-    }
   }
 }
 
 
 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();
       DeleteIfClosed(client);
     } 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);
+
+  ClientSocket *result = NULL;
+
+  if (accepted_head_ != NULL) {
+    result = accepted_head_;
+    accepted_head_ = accepted_head_->next();
+    if (accepted_head_ == NULL) accepted_tail_ = NULL;
+    result->set_next(NULL);
+    accepted_count_--;
+  }
+
   if (!IsClosing()) {
     if (!IssueAccept()) {
       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 276 matching lines @@
 
 void ClientSocket::IssueDisconnect() {
   OverlappedBuffer* buffer = OverlappedBuffer::AllocateDisconnectBuffer();
   BOOL ok = DisconnectEx_(
     socket(), buffer->GetCleanOverlapped(), TF_REUSE_SOCKET, 0);
   // DisconnectEx works like other OverlappedIO APIs, where we can get either an
   // immediate success or delayed operation by WSA_IO_PENDING being set.
   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);
+  }
 }
 
 
 void ClientSocket::DisconnectComplete(OverlappedBuffer* buffer) {
   OverlappedBuffer::DisposeBuffer(buffer);
   closesocket(socket());
   if (data_ready_ != NULL) {
     OverlappedBuffer::DisposeBuffer(data_ready_);
   }
   closed_ = true;
 }
 
 
 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);
     }
   }
 }
 
 
 void ClientSocket::EnsureInitialized(
     EventHandlerImplementation* event_handler) {
   ScopedLock lock(this);
   if (completion_port_ == INVALID_HANDLE_VALUE) {
     ASSERT(event_handler_ == NULL);
@@ skipping 82 matching lines @@
   // Just close the socket. This will cause any queued requests to be aborted.
   closesocket(socket());
   MarkClosedRead();
   MarkClosedWrite();
   handle_ = INVALID_HANDLE_VALUE;
 }
 
 
 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 (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 incoming 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);
+        listen_socket->SetPortAndMask(msg->dart_port, msg->data & EVENT_MASK);
+        TryDispatchingPendingAccepts(listen_socket);
+      }
+
+      if (IS_COMMAND(msg->data, kReturnTokenCommand)) {
+        int count = TOKEN_COUNT(msg->data);
+        listen_socket->ReturnTokens(msg->dart_port, count);
+        TryDispatchingPendingAccepts(listen_socket);
+        return;
+      } else if (IS_COMMAND(msg->data, kCloseCommand)) {
+        Dart_Port port = msg->dart_port;
+        listen_socket->RemovePort(port);
+
+        MutexLocker locker(globalTcpListeningSocketRegistry.mutex());
+        if (globalTcpListeningSocketRegistry.CloseSafe(
+            reinterpret_cast<intptr_t>(listen_socket))) {
+          handle->Close();
         }
+        DartUtils::PostInt32(port, 1 << kDestroyedEvent);
       }
     } else {
       handle->EnsureInitialized(this);
 
       Handle::ScopedLock lock(handle);
 
+      if (IS_COMMAND(msg->data, kReturnTokenCommand)) {
+        int count = TOKEN_COUNT(msg->data);
+        handle->ReturnTokens(msg->dart_port, count);
+        // TODO(kustermann): How can we continue with sending events
+        // to dart from here?
+        return;
+      }
+
       // 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->SetPortAndMask(msg->dart_port, msg->data & EVENT_MASK);
       }
 
       // Issue a read.
       if ((msg->data & (1 << kInEvent)) != 0) {
         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) {
         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);
           }
         }
       }
 
       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 & EVENT_MASK);
+        handle->Close();
+      }
     }
-
-    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()) {
-    int event_mask = 1 << kInEvent;
-    if ((listen_socket->mask() & event_mask) != 0) {
-      DartUtils::PostInt32(listen_socket->port(), event_mask);
+  TryDispatchingPendingAccepts(listen_socket);
+
+  DeleteIfClosed(listen_socket);
+}
+
+
+void EventHandlerImplementation::TryDispatchingPendingAccepts(
+    ListenSocket *listen_socket) {
+  Handle::ScopedLock lock(listen_socket);
+
+  if (!listen_socket->IsClosing() && listen_socket->CanAccept()) {
+    for (int i = 0;
+         i < listen_socket->accepted_count() && listen_socket->HasNextPort();
+         i++) {
+      Dart_Port port = listen_socket->NextPort();
+      DartUtils::PostInt32(port, 1 << kInEvent);
+      if (listen_socket->TakeToken()) {
+        break;
+      }
     }
   }
-
-  DeleteIfClosed(listen_socket);
 }
 
 
 void EventHandlerImplementation::HandleRead(Handle* handle,
                                             int bytes,
                                             OverlappedBuffer* buffer) {
   buffer->set_data_length(bytes);
   handle->ReadComplete(buffer);
   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 {
     handle->MarkClosedRead();
     if (bytes == 0) {
       HandleClosed(handle);
     } else {
       HandleError(handle);
     }
   }
 
   DeleteIfClosed(handle);
 }
 
 
 void EventHandlerImplementation::HandleRecvFrom(Handle* handle,
                                                 int bytes,
                                                 OverlappedBuffer* buffer) {
   ASSERT(handle->is_datagram_socket());
   buffer->set_data_length(bytes);
   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);
     }
   }
 
   DeleteIfClosed(handle);
 }
 
 
 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);
+      if ((handle->Mask() & event_mask) != 0) {
+        DartUtils::PostInt32(handle->NextPort(), event_mask);
       }
     }
   } else {
     HandleError(handle);
   }
 
   DeleteIfClosed(handle);
 }
 
 
@@ skipping 162 matching lines @@
     } else {
       handler_impl->HandleIOCompletion(bytes, key, overlapped);
     }
   }
   delete handler;
 }
 
 
 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);
   }
 
   // Initialize Winsock32
   if (!Socket::Initialize()) {
     FATAL("Failed to initialized Windows sockets");
   }
 }
 
 
 void EventHandlerImplementation::Shutdown() {
   SendData(kShutdownId, 0, 0);
 }
 
 }  // namespace bin
 }  // namespace dart
 
 #endif  // defined(TARGET_OS_WINDOWS)