Fix handling of async IO from pipes on Windows

runtime/bin/eventhandler_win.h

 // Copyright (c) 2011, 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_WIN_H_
 #define BIN_EVENTHANDLER_WIN_H_
 
 #include <winsock2.h>
 #include <mswsock.h>
 
@@ skipping 134 matching lines @@
   int Write(const void* buffer, int num_bytes);
 
   // Internal interface used by the event handler.
   virtual bool IssueRead();
   virtual bool IssueWrite();
   bool HasPendingRead();
   bool HasPendingWrite();
   void ReadComplete(IOBuffer* buffer);
   void WriteComplete(IOBuffer* buffer);
 
+  bool IsClosing() { return (flags_ & (1 << kClosing)) != 0; }
+  bool IsClosedRead() { return (flags_ & (1 << kCloseRead)) != 0; }
+  bool IsClosedWrite() { return (flags_ & (1 << kCloseWrite)) != 0; }
+  void MarkClosing() { flags_ |= (1 << kClosing); }
+  void MarkClosedRead() { flags_ |= (1 << kCloseRead); }
+  void MarkClosedWrite() { flags_ |= (1 << kCloseWrite); }
+
   virtual void EnsureInitialized(
     EventHandlerImplementation* event_handler) = 0;
 
   HANDLE handle() { return handle_; }
   Dart_Port port() { return port_; }
   EventHandlerImplementation* event_handler() { return event_handler_; }
 
   void Lock();
   void Unlock();
 
   bool CreateCompletionPort(HANDLE completion_port);
 
   void close();
   virtual bool IsClosed() = 0;
 
   void SetPortAndMask(Dart_Port port, intptr_t mask) {
     port_ = port;
     mask_ = mask;
   }
   Type type() { return type_; }
   bool is_file() { return type_ == kFile; }
   bool is_socket() { return type_ == kListenSocket || type_ == kClientSocket; }
   bool is_listen_socket() { return type_ == kListenSocket; }
   bool is_client_socket() { return type_ == kClientSocket; }
   intptr_t mask() { return mask_; }
 
-  bool is_closing() { return closing_; }
+ protected:
+  enum Flags {
+    kClosing = 0,
+    kCloseRead = 1,
+    kCloseWrite = 2
+  };
 
- protected:
   explicit Handle(HANDLE handle);
   Handle(HANDLE handle, Dart_Port port);
 
   virtual void AfterClose() = 0;
 
   Type type_;
   HANDLE handle_;
-  bool closing_;  // Is this handle in the process of closing?
   Dart_Port port_;  // Dart port to communicate events for this socket.
   intptr_t mask_;  // Mask of events to report through the port.
   HANDLE completion_port_;
-  CRITICAL_SECTION cs_;  // Critical section protecting this object.
   EventHandlerImplementation* event_handler_;
 
   IOBuffer* data_ready_;  // IO buffer for data ready to be read.
   IOBuffer* pending_read_;  // IO buffer for pending read.
   IOBuffer* pending_write_;  // IO buffer for pending write
+
+ private:
+  int flags_;
+  CRITICAL_SECTION cs_;  // Critical section protecting this object.
 };
 
 
 class FileHandle : public Handle {
  public:
   explicit FileHandle(HANDLE handle)
       : Handle(reinterpret_cast<HANDLE>(handle)) { type_ = kFile; }
   FileHandle(HANDLE handle, Dart_Port port)
       : Handle(reinterpret_cast<HANDLE>(handle), port) { type_ = kFile; }
 
@@ skipping 55 matching lines @@
   // Linked list of accepted connections provided by completion code. Ready to
   // be handed over through accept.
   ClientSocket* accepted_head_;
   ClientSocket* accepted_tail_;
 };
 
 
 // Information on connected sockets.
 class ClientSocket : public SocketHandle {
  public:
-  explicit ClientSocket(SOCKET s)
-      : SocketHandle(s),
-        next_(NULL),
-        flags_(0) { type_ = kClientSocket; }
+  explicit ClientSocket(SOCKET s) : SocketHandle(s), next_(NULL) {
+    type_ = kClientSocket;
+  }
 
-  ClientSocket(SOCKET s, Dart_Port port)
-      : SocketHandle(s, port),
-        next_(NULL),
-        flags_(0) { type_ = kClientSocket; }
+  ClientSocket(SOCKET s, Dart_Port port) : SocketHandle(s, port), next_(NULL) {
+    type_ = kClientSocket;
+  }
 
   virtual ~ClientSocket() {
     // Don't delete this object until all pending requests have been handled.
     ASSERT(!HasPendingRead());
     ASSERT(!HasPendingWrite());
     ASSERT(next_ == NULL);
   };
 
   void Shutdown(int how);
-  bool IsClosedRead() { return (flags_ & (1 << kCloseRead)) != 0; }
-  bool IsClosedWrite() { return (flags_ & (1 << kCloseWrite)) != 0; }
-
-  void MarkClosedRead() { flags_ |= (1 << kCloseRead); }
-  void MarkClosedWrite() { flags_ |= (1 << kCloseWrite); }
 
   // Internal interface used by the event handler.
   virtual bool IssueRead();
   virtual bool IssueWrite();
 
   virtual void EnsureInitialized(
     EventHandlerImplementation* event_handler);
   virtual bool IsClosed();
 
   ClientSocket* next() { return next_; }
   void set_next(ClientSocket* next) { next_ = next; }
 
  private:
-  enum Flags {
-    kCloseRead = 0,
-    kCloseWrite = 1
-  };
-
   virtual void AfterClose();
 
   ClientSocket* next_;
-  int flags_;
 };
 
 
 // Event handler.
 class EventHandlerImplementation {
  public:
   EventHandlerImplementation();
   virtual ~EventHandlerImplementation() {}
 
   void SendData(intptr_t id, Dart_Port dart_port, intptr_t data);
   void StartEventHandler();
 
   DWORD GetTimeout();
   void HandleInterrupt(InterruptMessage* msg);
   void HandleTimeout();
   void HandleAccept(ListenSocket* listen_socket, IOBuffer* buffer);
   void HandleClosed(Handle* handle);
-  void HandleRead(ClientSocket* client_socket, int bytes, IOBuffer* buffer);
-  void HandleWrite(ClientSocket* client_socket, int bytes, IOBuffer* buffer);
+  void HandleRead(Handle* handle, int bytes, IOBuffer* buffer);
+  void HandleWrite(Handle* handle, int bytes, IOBuffer* buffer);
   void HandleClose(ClientSocket* client_socket);
   void HandleIOCompletion(DWORD bytes, ULONG_PTR key, OVERLAPPED* overlapped);
 
   HANDLE completion_port() { return completion_port_; }
 
  private:
   ClientSocket* client_sockets_head_;
 
   int64_t timeout_;  // Time for next timeout.
   Dart_Port timeout_port_;
   HANDLE completion_port_;
 };
 
 
 #endif  // BIN_EVENTHANDLER_WIN_H_