Join embeder threads on Windows.

runtime/bin/eventhandler_win.cc

Index: runtime/bin/eventhandler_win.cc
diff --git a/runtime/bin/eventhandler_win.cc b/runtime/bin/eventhandler_win.cc
index db698a3c8831e637cb17e15fbeef1c242548d152..b10c406314474a5a2b7d53dadb88e293562b2cd4 100644
--- a/runtime/bin/eventhandler_win.cc
+++ b/runtime/bin/eventhandler_win.cc
@@ -121,12 +121,17 @@ Handle::Handle(intptr_t handle)
       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_);
 }
 
@@ -186,6 +191,34 @@ bool Handle::HasPendingWrite() {
 }
 
 
+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.
@@ -197,6 +230,7 @@ void Handle::ReadComplete(OverlappedBuffer* buffer) {
     OverlappedBuffer::DisposeBuffer(buffer);
   }
   pending_read_ = NULL;
+  WaitForReadThreadFinished();
 }
 
 
@@ -220,7 +254,25 @@ static void ReadFileThread(uword args) {
 }
 
 
+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);
 
@@ -245,6 +297,7 @@ void Handle::ReadSyncCompleteAsync() {
   if (!ok) {
     FATAL("PostQueuedCompletionStatus failed");
   }
+  NotifyReadThreadFinished();
 }
 
 
@@ -277,6 +330,7 @@ bool Handle::IssueRead() {
     if (result != 0) {
       FATAL1("Failed to start read file thread %d", result);
     }
+    WaitForReadThreadStarted();
     return true;
   }
 }
@@ -677,6 +731,7 @@ static void WriteFileThread(uword args) {
 void StdHandle::RunWriteLoop() {
   write_monitor_->Enter();
   write_thread_running_ = true;
+  thread_id_ = Thread::GetCurrentThreadId();
   // Notify we have started.
   write_monitor_->Notify();
 
@@ -762,6 +817,7 @@ void StdHandle::DoClose() {
     while (write_thread_exists_) {
       locker.Wait(Monitor::kNoTimeout);
     }
+    Thread::Join(thread_id_);
   }
   Handle::DoClose();
 }
@@ -1278,6 +1334,8 @@ void EventHandlerImplementation::HandleIOCompletion(DWORD bytes,
 
 
 EventHandlerImplementation::EventHandlerImplementation() {
+  startup_monitor_ = new Monitor();
+  handler_thread_id_ = Thread::kInvalidThreadId;
   completion_port_ =
       CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, NULL, 1);
   if (completion_port_ == NULL) {
@@ -1288,6 +1346,8 @@ EventHandlerImplementation::EventHandlerImplementation() {
 
 
 EventHandlerImplementation::~EventHandlerImplementation() {
+  Thread::Join(handler_thread_id_);
+  delete startup_monitor_;
   CloseHandle(completion_port_);
 }
 
@@ -1322,6 +1382,12 @@ void EventHandlerImplementation::EventHandlerEntry(uword 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;
@@ -1382,6 +1448,13 @@ void EventHandlerImplementation::Start(EventHandler* handler) {
     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");