Join embeder threads 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 "bin/eventhandler_win.h"
 
@@ skipping 103 matching lines @@
 
 Handle::Handle(intptr_t handle)
     : DescriptorInfoBase(handle),
       handle_(reinterpret_cast<HANDLE>(handle)),
       completion_port_(INVALID_HANDLE_VALUE),
       event_handler_(NULL),
       data_ready_(NULL),
       pending_read_(NULL),
       pending_write_(NULL),
       last_error_(NOERROR),
-      flags_(0) {
+      flags_(0),
+      read_thread_id_(Thread::kInvalidThreadId),
+      read_thread_running_(false),
+      read_thread_finished_(false),
+      read_thread_monitor_(new Monitor()) {

[Ivan Posva, 2015/08/25 22:21:03]

Can we share the monitor and the critical section?

   InitializeCriticalSection(&cs_);
 }
 
 
 Handle::~Handle() {
+  delete read_thread_monitor_;
   DeleteCriticalSection(&cs_);
 }
 
 
 void Handle::Lock() {
   EnterCriticalSection(&cs_);
 }
 
 
 void Handle::Unlock() {
@@ skipping 39 matching lines @@
   return pending_read_ != NULL;
 }
 
 
 bool Handle::HasPendingWrite() {
   ScopedLock lock(this);
   return pending_write_ != NULL;
 }
 
 
+void Handle::NotifyReadThreadFinished() {
+  MonitorLocker ml(read_thread_monitor_);
+  read_thread_finished_ = true;

[Ivan Posva, 2015/08/25 22:21:03]

Assertions?

[zra, 2015/08/26 05:24:39]

Done.

+  ml.Notify();
+}
+
+
+void Handle::WaitForReadThreadFinished() {
+  // Join the Reader thread if there is one.
+  ThreadId to_join = Thread::kInvalidThreadId;
+  {
+    MonitorLocker ml(read_thread_monitor_);
+    if (read_thread_id_ != Thread::kInvalidThreadId) {
+      while (!read_thread_finished_) {
+        ml.Wait();
+      }
+      read_thread_finished_ = false;
+      read_thread_running_ = false;
+      to_join = read_thread_id_;
+      read_thread_id_ = Thread::kInvalidThreadId;
+    }
+  }
+  if (to_join != Thread::kInvalidThreadId) {
+    Thread::Join(to_join);
+  }
+}
+
+
 void Handle::ReadComplete(OverlappedBuffer* buffer) {
   ScopedLock lock(this);
   // Currently only one outstanding read at the time.
   ASSERT(pending_read_ == buffer);
   ASSERT(data_ready_ == NULL);
   if (!IsClosing() && !buffer->IsEmpty()) {
     data_ready_ = pending_read_;
   } else {
     OverlappedBuffer::DisposeBuffer(buffer);
   }
   pending_read_ = NULL;
+  WaitForReadThreadFinished();
 }
 
 
 void Handle::RecvFromComplete(OverlappedBuffer* buffer) {
   ReadComplete(buffer);
 }
 
 
 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 void ReadFileThread(uword args) {
   Handle* handle = reinterpret_cast<Handle*>(args);
   handle->ReadSyncCompleteAsync();
 }
 
 
+void Handle::NotifyReadThreadStarted() {
+  MonitorLocker ml(read_thread_monitor_);
+  read_thread_id_ = Thread::GetCurrentThreadId();
+  read_thread_running_ = true;
+  ml.Notify();
+}
+
+
+void Handle::WaitForReadThreadStarted() {
+  MonitorLocker ml(read_thread_monitor_);
+  while (!read_thread_running_) {
+    ml.Wait();
+  }
+  ASSERT(read_thread_id_ != Thread::kInvalidThreadId);
+}
+
+
 void Handle::ReadSyncCompleteAsync() {
+  NotifyReadThreadStarted();
   ASSERT(pending_read_ != NULL);
   ASSERT(pending_read_->GetBufferSize() >= kStdOverlappedBufferSize);
 
   DWORD buffer_size = pending_read_->GetBufferSize();
   if (GetFileType(handle_) == FILE_TYPE_CHAR) {
     buffer_size = kStdOverlappedBufferSize;
   }
   DWORD bytes_read = 0;
   BOOL ok = ReadFile(handle_,
                      pending_read_->GetBufferStart(),
                      buffer_size,
                      &bytes_read,
                      NULL);
   if (!ok) {
     bytes_read = 0;
   }
   OVERLAPPED* overlapped = pending_read_->GetCleanOverlapped();
   ok = PostQueuedCompletionStatus(event_handler_->completion_port(),
                                   bytes_read,
                                   reinterpret_cast<ULONG_PTR>(this),
                                   overlapped);
   if (!ok) {
     FATAL("PostQueuedCompletionStatus failed");
   }
+  NotifyReadThreadFinished();
 }
 
 
 bool Handle::IssueRead() {
   ScopedLock lock(this);
   ASSERT(type_ != kListenSocket);
   ASSERT(pending_read_ == NULL);
   OverlappedBuffer* buffer = OverlappedBuffer::AllocateReadBuffer(kBufferSize);
   if (SupportsOverlappedIO()) {
     ASSERT(completion_port_ != INVALID_HANDLE_VALUE);
@@ skipping 12 matching lines @@
     HandleIssueError();
     return false;
   } else {
     // Completing asynchronously through thread.
     pending_read_ = buffer;
     int result = Thread::Start(ReadFileThread,
                                reinterpret_cast<uword>(this));
     if (result != 0) {
       FATAL1("Failed to start read file thread %d", result);
     }
+    WaitForReadThreadStarted();
     return true;
   }
 }
 
 
 bool Handle::IssueRecvFrom() {
   return false;
 }
 
 
@@ skipping 380 matching lines @@
 
 static void WriteFileThread(uword args) {
   StdHandle* handle = reinterpret_cast<StdHandle*>(args);
   handle->RunWriteLoop();
 }
 
 
 void StdHandle::RunWriteLoop() {
   write_monitor_->Enter();
   write_thread_running_ = true;
+  thread_id_ = Thread::GetCurrentThreadId();
   // 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();
     }
   }
@@ skipping 65 matching lines @@
 
 
 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);
     }
+    Thread::Join(thread_id_);
   }
   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(),
@@ skipping 496 matching lines @@
       HandleConnect(client_socket, bytes, buffer);
       break;
     }
     default:
       UNREACHABLE();
   }
 }
 
 
 EventHandlerImplementation::EventHandlerImplementation() {
+  startup_monitor_ = new Monitor();
+  handler_thread_id_ = Thread::kInvalidThreadId;
   completion_port_ =
       CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, NULL, 1);
   if (completion_port_ == NULL) {
     FATAL("Completion port creation failed");
   }
   shutdown_ = false;
 }
 
 
 EventHandlerImplementation::~EventHandlerImplementation() {
+  Thread::Join(handler_thread_id_);
+  delete startup_monitor_;
   CloseHandle(completion_port_);
 }
 
 
 int64_t EventHandlerImplementation::GetTimeout() {
   if (!timeout_queue_.HasTimeout()) {
     return kInfinityTimeout;
   }
   int64_t millis = timeout_queue_.CurrentTimeout() -
       TimerUtils::GetCurrentTimeMilliseconds();
@@ skipping 14 matching lines @@
     FATAL("PostQueuedCompletionStatus failed");
   }
 }
 
 
 void EventHandlerImplementation::EventHandlerEntry(uword args) {
   EventHandler* handler = reinterpret_cast<EventHandler*>(args);
   EventHandlerImplementation* handler_impl = &handler->delegate_;
   ASSERT(handler_impl != NULL);
 
+  {
+    MonitorLocker ml(handler_impl->startup_monitor_);
+    handler_impl->handler_thread_id_ = Thread::GetCurrentThreadId();
+    ml.Notify();
+  }
+
   while (!handler_impl->shutdown_) {
     DWORD bytes;
     ULONG_PTR key;
     OVERLAPPED* overlapped;
     int64_t millis = handler_impl->GetTimeout();
     ASSERT(millis == kInfinityTimeout || millis >= 0);
     if (millis > kMaxInt32) millis = kMaxInt32;
     ASSERT(sizeof(int32_t) == sizeof(DWORD));
     BOOL ok = GetQueuedCompletionStatus(handler_impl->completion_port(),
                                         &bytes,
@@ skipping 40 matching lines @@
 }
 
 
 void EventHandlerImplementation::Start(EventHandler* handler) {
   int result = Thread::Start(EventHandlerEntry,
                              reinterpret_cast<uword>(handler));
   if (result != 0) {
     FATAL1("Failed to start event handler thread %d", result);
   }
 
+  {
+    MonitorLocker ml(startup_monitor_);
+    while (handler_thread_id_ == Thread::kInvalidThreadId) {
+      ml.Wait();
+    }
+  }
+
   // Initialize Winsock32
   if (!Socket::Initialize()) {
     FATAL("Failed to initialized Windows sockets");
   }
 }
 
 
 void EventHandlerImplementation::Shutdown() {
   SendData(kShutdownId, 0, 0);
 }
 
 }  // namespace bin
 }  // namespace dart
 
 #endif  // defined(TARGET_OS_WINDOWS)