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 519 matching lines @@
   ScopedLock lock(this);
   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;
 }
 
 

[Søren Gjesse, 2013/10/02 08:49:22]

Please add these static variables to the Windows EventHandlerImplementation
class.

[Anders Johnsen, 2013/10/02 10:16:59]

Done.

+static Monitor* write_monitor = new Monitor();
+static bool write_thread_exists = false;
+static bool write_thread_running = false;
+static Handle* write_handle = NULL;
+
 static unsigned int __stdcall WriteFileThread(void* args) {
-  Handle* handle = reinterpret_cast<Handle*>(args);
-  handle->WriteSyncCompleteAsync();
+  MonitorLocker locker(write_monitor);
+  write_thread_running = true;
+  // Notify we have started.
+  locker.Notify();
+
+  while (write_thread_running) {
+    locker.Wait(Monitor::kNoTimeout);
+    if (write_handle) {
+      // We woke up and had a handle. Complete it.
+      write_handle->WriteSyncCompleteAsync();
+      write_handle = NULL;
+    }
+  }
+  write_thread_exists = false;
+  locker.Notify();
   return 0;
 }
 
 
 void Handle::WriteSyncCompleteAsync() {
   ASSERT(pending_write_ != NULL);
 
   DWORD bytes_written = -1;
   BOOL ok = WriteFile(handle_,
                       pending_write_->GetBufferStart(),
@@ skipping 23 matching lines @@
   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;
     return num_bytes;
   } 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.
+    // 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.
-    // 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");
+    // Double-checked locking, to avoid taking the lock in most cases.

[Søren Gjesse, 2013/10/02 08:49:22]

You need to hold the lock before accessing shared variables. There is no
guarantee that these reads are atomic (in almost all cases they are, but the
compiler might fool us).

[Anders Johnsen, 2013/10/02 10:16:59]

Done.

+    if (!write_thread_exists) {
+      MonitorLocker locker(write_monitor);
+      if (!write_thread_exists) {
+        write_thread_exists = true;
+        uint32_t tid;
+        uintptr_t thread_handle =
+            _beginthreadex(NULL, 32 * 1024, WriteFileThread, this, 0, &tid);
+        // Completing asynchronously through thread.
+        if (thread_handle == -1) {
+          FATAL("Failed to start write file thread");
+        }
+        while (!write_thread_running) {
+          // Wait until we the thread is running.
+          locker.Wait(Monitor::kNoTimeout);
+        }
+      }
+    }
+    // Double-checked locking, to avoid taking the lock in most cases.
+    if (write_handle == NULL) {
+      MonitorLocker locker(write_monitor);
+      if (write_handle == NULL) {
+        // Create buffer and notify thread about the new handle.
+        pending_write_ = OverlappedBuffer::AllocateWriteBuffer(num_bytes);
+        pending_write_->Write(buffer, num_bytes);
+        write_handle = this;
+        locker.Notify();
+      }
     }
     return 0;
   }
 }
 
 
 bool ClientSocket::LoadDisconnectEx() {
   // Load the DisconnectEx function into memory using WSAIoctl.
   GUID guid_disconnect_ex = WSAID_DISCONNECTEX;
   DWORD bytes;
@@ skipping 354 matching lines @@
   intptr_t result;
   completion_port_ =
       CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, NULL, 1);
   if (completion_port_ == NULL) {
     FATAL("Completion port creation failed");
   }
   shutdown_ = false;
 }
 
 
 EventHandlerImplementation::~EventHandlerImplementation() {

[Søren Gjesse, 2013/10/02 08:49:22]

Take the lock before looking at shared data.

[Anders Johnsen, 2013/10/02 10:16:59]

Done.

+  if (write_thread_exists) {
+    ASSERT(write_handle == NULL);
+    MonitorLocker locker(write_monitor);
+    write_thread_running = false;
+    locker.Notify();
+    while (write_thread_exists) {
+      locker.Wait(Monitor::kNoTimeout);
+    }
+  }
   CloseHandle(completion_port_);
 }
 
 
 int64_t EventHandlerImplementation::GetTimeout() {
   if (!timeout_queue_.HasTimeout()) {
     return kInfinityTimeout;
   }
   int64_t millis = timeout_queue_.CurrentTimeout() -
       TimerUtils::GetCurrentTimeMilliseconds();
@@ skipping 87 matching lines @@
 
 
 void EventHandlerImplementation::Shutdown() {
   SendData(kShutdownId, 0, 0);
 }
 
 }  // namespace bin
 }  // namespace dart
 
 #endif  // defined(TARGET_OS_WINDOWS)