Reuse thread when writing stdout/stderr on Windows.

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
 #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/log.h"
 #include "bin/socket.h"
 #include "bin/utils.h"
-#include "platform/thread.h"
+#include "vm/thread.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;
@@ skipping 67 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 67 matching lines @@
 
 void Handle::WriteComplete(OverlappedBuffer* buffer) {
   ScopedLock lock(this);
   // Currently only one outstanding write at the time.
   ASSERT(pending_write_ == buffer);
   OverlappedBuffer::DisposeBuffer(buffer);
   pending_write_ = NULL;
 }
 
 
-static unsigned int __stdcall ReadFileThread(void* args) {
+static void ReadFileThread(uword args) {
   Handle* handle = reinterpret_cast<Handle*>(args);
   handle->ReadSyncCompleteAsync();
-  return 0;
 }
 
 
 void Handle::ReadSyncCompleteAsync() {
   ASSERT(pending_read_ != NULL);
   ASSERT(pending_read_->GetBufferSize() >= kStdOverlappedBufferSize);
 
   DWORD buffer_size = pending_read_->GetBufferSize();
   if (GetFileType(handle_) == FILE_TYPE_CHAR) {
     buffer_size = kStdOverlappedBufferSize;
@@ skipping 38 matching lines @@
       // Completing asynchronously.
       pending_read_ = buffer;
       return true;
     }
     OverlappedBuffer::DisposeBuffer(buffer);
     HandleIssueError();
     return false;
   } else {
     // Completing asynchronously through thread.
     pending_read_ = buffer;
-    uint32_t tid;
-    uintptr_t thread_handle =
-        _beginthreadex(NULL, 32 * 1024, ReadFileThread, this, 0, &tid);
-    if (thread_handle == -1) {
-      FATAL("Failed to start read file thread");
+    int result = dart::Thread::Start(ReadFileThread,
+                                     reinterpret_cast<uword>(this));
+    if (result != 0) {
+      FATAL("Failed to start read file thread %d", result);
     }
     return true;
   }
 }
 
 
 bool Handle::IssueWrite() {
   ScopedLock lock(this);
   ASSERT(type_ != kListenSocket);
   ASSERT(completion_port_ != INVALID_HANDLE_VALUE);
@@ skipping 35 matching lines @@
     CreateCompletionPort(event_handler_->completion_port());
   }
 }
 
 
 bool FileHandle::IsClosed() {
   return IsClosing() && !HasPendingRead() && !HasPendingWrite();
 }
 
 
-void FileHandle::DoClose() {
-  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());
   }
 }
 
 
@@ skipping 195 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;
+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;
+  ASSERT(SupportsOverlappedIO());
+  if (completion_port_ == INVALID_HANDLE_VALUE) return 0;
+  pending_write_ = OverlappedBuffer::AllocateWriteBuffer(num_bytes);
+  pending_write_->Write(buffer, num_bytes);
+  if (!IssueWrite()) return -1;
+  return num_bytes;
 }
 
 
-void Handle::WriteSyncCompleteAsync() {
+static void WriteFileThread(uword args) {
+  StdHandle* handle = reinterpret_cast<StdHandle*>(args);
+  handle->RunWriteLoop();
+}
+
+
+void StdHandle::RunWriteLoop() {
+  write_monitor_->Enter();
+  write_thread_running_ = true;
+  // Notify we have started.
+  write_monitor_->Notify();
+
+  while (write_thread_running_) {
+    write_monitor_->Wait(Monitor::kNoTimeout);
+    if (pending_write_ != NULL) {
+      // We woke up and had a pending write. Execute it.
+      WriteSyncCompleteAsync();
+    }
+  }
+
+  write_thread_exists_ = false;
+  write_monitor_->Notify();
+  write_monitor_->Exit();
+}
+
+
+void StdHandle::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) {
+int StdHandle::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 (completion_port_ == INVALID_HANDLE_VALUE) return 0;
-    pending_write_ = OverlappedBuffer::AllocateWriteBuffer(num_bytes);
-    pending_write_->Write(buffer, num_bytes);
-    if (!IssueWrite()) return -1;
+  // In the case of stdout and stderr, OverlappedIO is not supported.
+  // Here we'll instead use a thread, 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.
+  MonitorLocker locker(write_monitor_);
+  if (thread_wrote_ > 0) {
+    if (num_bytes > thread_wrote_) num_bytes = thread_wrote_;
+    thread_wrote_ -= num_bytes;
     return num_bytes;
+  }
+  if (!write_thread_exists_) {
+    write_thread_exists_ = true;
+    int result = dart::Thread::Start(WriteFileThread,
+                                     reinterpret_cast<uword>(this));
+    if (result != 0) {
+      FATAL("Failed to start write file thread %d", result);
+    }
+    while (!write_thread_running_) {
+      // Wait until we the thread is running.
+      locker.Wait(Monitor::kNoTimeout);
+    }
+  }
+  // Create buffer and notify thread about the new handle.
+  pending_write_ = OverlappedBuffer::AllocateWriteBuffer(num_bytes);
+  pending_write_->Write(buffer, num_bytes);
+  locker.Notify();
+  return 0;
+}
+
+
+void StdHandle::DoClose() {
+  MonitorLocker locker(write_monitor_);
+  if (write_thread_exists_) {
+    write_thread_running_ = false;
+    locker.Notify();
+    while (write_thread_exists_) {
+      locker.Wait(Monitor::kNoTimeout);
+    }
+  }
+  if (handle_ == GetStdHandle(STD_OUTPUT_HANDLE)) {
+    int fd = _open("NUL", _O_WRONLY);
+    ASSERT(fd >= 0);
+    _dup2(fd, _fileno(stdout));
+    close(fd);
   } else {
-    // 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;
-    }
-    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;
+    Handle::DoClose();
   }
 }
 
 
 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)