Implement windows support for having multiple Dart_Port's registered on one OS socket

dart/runtime/bin/eventhandler.h

 // 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.
 
 #ifndef BIN_EVENTHANDLER_H_
 #define BIN_EVENTHANDLER_H_
 
 #include "bin/builtin.h"
 #include "bin/dartutils.h"
 #include "bin/isolate_data.h"
@@ skipping 168 matching lines @@
           prev->next_ = next;
           next->prev_ = prev;
           delete current;
           return;
         }
         current = current->next_;
       } while (current != head_);
     }
   }
 
+  void RemoveAll() {
+    while (HasHead()) {
+      RemoveHead();
+    }
+  }
 
   T head() const { return head_->t; }
 
   bool HasHead() const {
     return head_ != NULL;
   }
 
   void Rotate() {
     if (head_ != NULL) {
       ASSERT(head_->next_ != NULL);
@@ skipping 27 matching lines @@
   // Whether this descriptor refers to an underlying listening OS socket.
   virtual bool IsListeningSocket() const = 0;
 
   // Inserts or updates a new Dart_Port which is interested in events specified
   // in `mask`.
   virtual void SetPortAndMask(Dart_Port port, intptr_t mask) = 0;
 
   // Removes a port from the interested listeners.
   virtual void RemovePort(Dart_Port port) = 0;
 
+  // Removes all ports from the interested listeners.
+  virtual void RemoveAllPorts() = 0;
+
   // Returns a port to which `events_ready` can be sent to. It will also
   // decrease the token count by 1 for this port.
   virtual Dart_Port NextNotifyDartPort(intptr_t events_ready) = 0;
 
   // Will post `data` to all known Dart_Ports. It will also decrease the token
   // count by 1 for all ports.
   virtual void NotifyAllDartPorts(uintptr_t events) = 0;
 
   // Returns `count` tokens for the given port.
   virtual void ReturnTokens(Dart_Port port, int count) = 0;
@@ skipping 13 matching lines @@
 // Describes a OS descriptor (e.g. file descriptor on linux or HANDLE on
 // windows) which is connected to a single Dart_Port.
 //
 // Subclasses of this class can be e.g. connected tcp sockets.
 template<typename DI>
 class DescriptorInfoSingleMixin : public DI {
  private:
   static const int kTokenCount = 16;
 
  public:
-  explicit DescriptorInfoSingleMixin(intptr_t fd)
-      : DI(fd), port_(0), tokens_(kTokenCount), mask_(0) {}
+  explicit DescriptorInfoSingleMixin(intptr_t fd, bool disable_tokens)
+      : DI(fd), port_(0), tokens_(kTokenCount), mask_(0),
+        disable_tokens_(disable_tokens) {}
 
   virtual ~DescriptorInfoSingleMixin() { }
 
   virtual bool IsListeningSocket() const { return false; }
 
   virtual void SetPortAndMask(Dart_Port port, intptr_t mask) {
     ASSERT(port_ == 0 || port == port_);
     port_ = port;
     mask_ = mask;
   }
 
   virtual void RemovePort(Dart_Port port) {
     // TODO(dart:io): Find out where we call RemovePort() with the invalid
     // port. Afterwards remove the part in the ASSERT here.
     ASSERT(port_ == 0 || port_ == port);
     port_ = 0;
     mask_ = 0;
   }
 
+  virtual void RemoveAllPorts() {
+    port_ = 0;
+    mask_ = 0;
+  }
+
   virtual Dart_Port NextNotifyDartPort(intptr_t events_ready) {
     ASSERT(IS_IO_EVENT(events_ready) ||
            IS_EVENT(events_ready, kDestroyedEvent));
-    tokens_--;
+    if (!disable_tokens_) {
+      tokens_--;
+    }
     return port_;
   }
 
   virtual void NotifyAllDartPorts(uintptr_t events) {
-    // Unexpected close or error events are the only ones we broadcast to all
-    // listeners and are the only ones where we do not count tokens.
+    // Unexpected close, asynchronous destroy or error events are the only
+    // ones we broadcast to all listeners.
     ASSERT(IS_EVENT(events, kCloseEvent) ||
-           IS_EVENT(events, kErrorEvent));
+           IS_EVENT(events, kErrorEvent) ||
+           IS_EVENT(events, kDestroyedEvent));
 
     if (port_ != 0) {
       DartUtils::PostInt32(port_, events);
     }
-    tokens_--;
+    if (!disable_tokens_) {
+      tokens_--;
+    }
   }
 
   virtual void ReturnTokens(Dart_Port port, int count) {
     ASSERT(port_ == port);
     ASSERT(tokens_ >= 0);
-    tokens_ += count;
+    if (!disable_tokens_) {
+      tokens_ += count;
+    }
     ASSERT(tokens_ <= kTokenCount);
   }
 
   virtual intptr_t Mask() {
     if (tokens_ <= 0) {
       return 0;
     }
     return mask_;
   }
 
   virtual void Close() {
     DI::Close();
   }
 
  private:
   Dart_Port port_;
   int tokens_;
   intptr_t mask_;
+  bool disable_tokens_;
 };
 
 
 // Describes a OS descriptor (e.g. file descriptor on linux or HANDLE on
 // windows) which is connected to multiple Dart_Port's.
 //
 // Subclasses of this class can be e.g. a listening socket which multiple
 // isolates are listening on.
 template<typename DI>
 class DescriptorInfoMultipleMixin : public DI {
@@ skipping 24 matching lines @@
 
   struct PortEntry {
     Dart_Port dart_port;
     intptr_t is_reading;
     intptr_t token_count;
 
     bool IsReady() { return token_count > 0 && is_reading; }
   };
 
  public:
-  explicit DescriptorInfoMultipleMixin(intptr_t fd)
-      : DI(fd), tokens_map_(&SamePortValue, kTokenCount) {}
+  explicit DescriptorInfoMultipleMixin(intptr_t fd, bool disable_tokens)
+      : DI(fd), tokens_map_(&SamePortValue, kTokenCount),
+        disable_tokens_(disable_tokens) {}
 
   virtual ~DescriptorInfoMultipleMixin() {}
 
   virtual bool IsListeningSocket() const { return true; }
 
   virtual void SetPortAndMask(Dart_Port port, intptr_t mask) {
     HashMap::Entry* entry = tokens_map_.Lookup(
         GetHashmapKeyFromPort(port), GetHashmapHashFromPort(port), true);
     PortEntry* pentry;
     if (entry->value == NULL) {
@@ skipping 14 matching lines @@
 
       if (was_ready && !is_ready) {
         active_readers_.Remove(pentry);
       } else if (!was_ready && is_ready) {
         active_readers_.Add(pentry);
       }
     }
 
 #ifdef DEBUG
     // To ensure that all readers are ready.
-    PortEntry* root = reinterpret_cast<PortEntry*>(active_readers_.head());
+    int ready_count = 0;
 
-    int ready_count = 0;
-    if (root != NULL) {
+    if (active_readers_.HasHead()) {
+      PortEntry* root = reinterpret_cast<PortEntry*>(active_readers_.head());
       PortEntry* current = root;
       do {
         ASSERT(current->IsReady());
         ready_count++;
         active_readers_.Rotate();
         current = active_readers_.head();
       } while (current != root);
     }
+
     for (HashMap::Entry *entry = tokens_map_.Start();
          entry != NULL;
          entry = tokens_map_.Next(entry)) {
       PortEntry* pentry = reinterpret_cast<PortEntry*>(entry->value);
       if (pentry->IsReady()) {
         ready_count--;
       }
     }
     // Ensure all ready items are in `active_readers_`.
     ASSERT(ready_count == 0);
@@ skipping 17 matching lines @@
       // If a listening socket is not listened on, the event handler does not
       // know about it beforehand. So the first time the event handler knows
       // about it, is when it is supposed to be closed. We therefore do nothing
       // here.
       //
       // But whether to close it, depends on whether other isolates have it open
       // as well or not.
     }
   }
 
+  virtual void RemoveAllPorts() {
+    active_readers_.RemoveAll();
+    for (HashMap::Entry *entry = tokens_map_.Start();
+         entry != NULL;
+         entry = tokens_map_.Next(entry)) {
+      PortEntry* pentry = reinterpret_cast<PortEntry*>(entry->value);
+      delete pentry;
+    }
+    tokens_map_.Clear();
+  }
+
   virtual Dart_Port NextNotifyDartPort(intptr_t events_ready) {
     // We're only sending `kInEvents` if there are multiple listeners (which is
     // listening socktes).
     ASSERT(IS_EVENT(events_ready, kInEvent) ||
            IS_EVENT(events_ready, kDestroyedEvent));
 
     if (active_readers_.HasHead()) {
       PortEntry* pentry = reinterpret_cast<PortEntry*>(active_readers_.head());
 
       // Update token count.
-      pentry->token_count--;
+      if (!disable_tokens_) {
+        pentry->token_count--;
+      }
       if (pentry->token_count <= 0) {
         active_readers_.RemoveHead();
       } else {
         active_readers_.Rotate();
       }
 
       return pentry->dart_port;
     }
     return 0;
   }
 
   virtual void NotifyAllDartPorts(uintptr_t events) {
-    // Unexpected close or error events are the only ones we broadcast to all
-    // listeners and are the only ones where we do not count tokens.
+    // Unexpected close, asynchronous destroy or error events are the only
+    // ones we broadcast to all listeners.
     ASSERT(IS_EVENT(events, kCloseEvent) ||
-           IS_EVENT(events, kErrorEvent));
+           IS_EVENT(events, kErrorEvent) ||
+           IS_EVENT(events, kDestroyedEvent));
 
     for (HashMap::Entry *entry = tokens_map_.Start();
          entry != NULL;
          entry = tokens_map_.Next(entry)) {
       PortEntry* pentry = reinterpret_cast<PortEntry*>(entry->value);
       DartUtils::PostInt32(pentry->dart_port, events);
 
       // Update token count.
       bool was_ready = pentry->IsReady();
-      pentry->token_count--;
+      if (!disable_tokens_) {
+        pentry->token_count--;
+      }
 
       if (was_ready && pentry->token_count <= 0) {
         active_readers_.Remove(pentry);
       }
     }
   }
 
   virtual void ReturnTokens(Dart_Port port, int count) {
     HashMap::Entry* entry = tokens_map_.Lookup(
         GetHashmapKeyFromPort(port), GetHashmapHashFromPort(port), false);
     ASSERT(entry != NULL);
 
     PortEntry* pentry = reinterpret_cast<PortEntry*>(entry->value);
     bool was_ready = pentry->IsReady();
     ASSERT(pentry->token_count >= 0);
-    pentry->token_count += count;
+    if (!disable_tokens_) {
+      pentry->token_count += count;
+    }
     ASSERT(pentry->token_count <= kTokenCount);
     bool is_ready = pentry->token_count > 0 && pentry->IsReady();
     if (!was_ready && is_ready) {
       active_readers_.Add(pentry);
     }
   }
 
   virtual intptr_t Mask() {
     if (active_readers_.HasHead()) {
       return 1 << kInEvent;
     }
     return 0;
   }
 
   virtual void Close() {
     DI::Close();
   }
 
  private:
   // The [Dart_Port]s which are not paused (i.e. are interested in read events,
   // i.e. `mask == (1 << kInEvent)`) and we have enough tokens to communicate
   // with them.
   CircularLinkedList<PortEntry *> active_readers_;
 
   // A convenience mapping:
   //   Dart_Port -> struct PortEntry { dart_port, mask, token_count }
   HashMap tokens_map_;
+
+  bool disable_tokens_;
 };
 
 
 }  // namespace bin
 }  // namespace dart
 
 // The event handler delegation class is OS specific.
 #if defined(TARGET_OS_ANDROID)
 #include "bin/eventhandler_android.h"
 #elif defined(TARGET_OS_LINUX)
@@ skipping 35 matching lines @@
 
  private:
   friend class EventHandlerImplementation;
   EventHandlerImplementation delegate_;
 };
 
 }  // namespace bin
 }  // namespace dart
 
 #endif  // BIN_EVENTHANDLER_H_