Make stdout and stderr async.

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 <process.h>  // NOLINT
@@ skipping 89 matching lines @@
 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),
+      thread_wrote_(0),
       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),
+      thread_wrote_(0),
       flags_(0) {
   InitializeCriticalSection(&cs_);
 }
 
 
 Handle::~Handle() {
   DeleteCriticalSection(&cs_);
 }
 
 
@@ skipping 190 matching lines @@
   }
 }
 
 
 bool FileHandle::IsClosed() {
   return IsClosing() && !HasPendingRead() && !HasPendingWrite();
 }
 
 
 void FileHandle::DoClose() {
-  if (GetStdHandle(STD_OUTPUT_HANDLE) == handle_) {
+  if (handle_ == GetStdHandle(STD_OUTPUT_HANDLE)) {
     int fd = _open("NUL", _O_WRONLY);
     ASSERT(fd >= 0);
     _dup2(fd, _fileno(stdout));
     close(fd);
   } else {
     Handle::DoClose();
   }
 }
 
 
 void DirectoryWatchHandle::EnsureInitialized(
     EventHandlerImplementation* event_handler) {
   ScopedLock lock(this);
   event_handler_ = event_handler;
   if (completion_port_ == INVALID_HANDLE_VALUE) {
     CreateCompletionPort(event_handler_->completion_port());
   }
 }
 
 
 bool DirectoryWatchHandle::IsClosed() {
   return IsClosing() && pending_read_ == NULL;
 }
 
 
-void DirectoryWatchHandle::DoClose() {
-  Handle::DoClose();
-}
-
 bool DirectoryWatchHandle::IssueRead() {
   ScopedLock lock(this);
   OverlappedBuffer* buffer = OverlappedBuffer::AllocateReadBuffer(kBufferSize);
 
   ASSERT(completion_port_ != INVALID_HANDLE_VALUE);
 
   BOOL ok = ReadDirectoryChangesW(handle_,
                                   buffer->GetBufferStart(),
                                   buffer->GetBufferSize(),
                                   recursive_,
@@ skipping 180 matching lines @@
   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;
   }
   return num_bytes;
 }
 
 
+static unsigned int __stdcall WriteFileThread(void* args) {
+  Handle* handle = reinterpret_cast<Handle*>(args);
+  handle->WriteSyncCompleteAsync();
+  return 0;
+}
+
+
+void Handle::WriteSyncCompleteAsync() {
+  ASSERT(pending_write_ != NULL);
+
+  DWORD bytes_written = -1;
+  BOOL ok = WriteFile(handle_,
+                      pending_write_->GetBufferStart(),
+                      pending_write_->GetBufferSize(),
+                      &bytes_written,
+                      NULL);
+  if (!ok) {
+    if (GetLastError() != ERROR_BROKEN_PIPE) {
+      Log::PrintErr("WriteFile failed %d\n", GetLastError());
+    }
+    bytes_written = 0;
+  }
+  thread_wrote_ += bytes_written;
+  OVERLAPPED* overlapped = pending_write_->GetCleanOverlapped();
+  ok = PostQueuedCompletionStatus(event_handler_->completion_port(),
+                                  bytes_written,
+                                  reinterpret_cast<ULONG_PTR>(this),
+                                  overlapped);
+  if (!ok) {
+    FATAL("PostQueuedCompletionStatus failed");
+  }
+}
+
+
 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;
   if (SupportsOverlappedIO()) {
-    if (pending_write_ != NULL) return 0;
     if (completion_port_ == INVALID_HANDLE_VALUE) return 0;
-    if (num_bytes > kBufferSize) num_bytes = kBufferSize;
     pending_write_ = OverlappedBuffer::AllocateWriteBuffer(num_bytes);
     pending_write_->Write(buffer, num_bytes);
     if (!IssueWrite()) return -1;
     return num_bytes;
   } else {
-    DWORD bytes_written = -1;
-    BOOL ok = WriteFile(handle_,
-                        buffer,
-                        num_bytes,
-                        &bytes_written,
-                        NULL);
-    if (!ok) {
-      if (GetLastError() != ERROR_BROKEN_PIPE) {
-        Log::PrintErr("WriteFile failed: %d\n", GetLastError());
-      }
-      event_handler_->HandleClosed(this);
+    // In the case of stdout and stderr, OverlappedIO is not supported.
+    // Here we'll instead spawn a new thread for each write, to make it async.
+    // This code is actually never exposed to the user, as stdout and stderr is
+    // not available as a RawSocket, but only wrapped in a Socket.
+    // Note that we return '0', unless a thread have already completed a write.
+    // TODO(ajohnsen): Don't spawn a new thread per write. Issue 13541.
+    if (thread_wrote_ > 0) {
+      if (num_bytes > thread_wrote_) num_bytes = thread_wrote_;
+      thread_wrote_ -= num_bytes;
+      return num_bytes;
     }
-    return bytes_written;
+    pending_write_ = OverlappedBuffer::AllocateWriteBuffer(num_bytes);
+    pending_write_->Write(buffer, num_bytes);
+    // Completing asynchronously through thread.
+    uint32_t tid;
+    uintptr_t thread_handle =
+        _beginthreadex(NULL, 32 * 1024, WriteFileThread, this, 0, &tid);
+    if (thread_handle == -1) {
+      FATAL("Failed to start write file thread");
+    }
+    return 0;
   }
 }
 
 
 bool ClientSocket::LoadDisconnectEx() {
   // Load the DisconnectEx function into memory using WSAIoctl.
   GUID guid_disconnect_ex = WSAID_DISCONNECTEX;
   DWORD bytes;
   int status = WSAIoctl(socket(),
                         SIO_GET_EXTENSION_FUNCTION_POINTER,
@@ skipping 460 matching lines @@
 
 
 void EventHandlerImplementation::Shutdown() {
   SendData(kShutdownId, 0, 0);
 }
 
 }  // namespace bin
 }  // namespace dart
 
 #endif  // defined(TARGET_OS_WINDOWS)