Reimplement Windows Monitors.

runtime/platform/thread_win.cc

 // Copyright (c) 2012, 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/thread.h"
 
 #include <process.h>
 
 #include "platform/assert.h"
 
@@ skipping 118 matching lines @@
 
 
 void Mutex::Unlock() {
   BOOL result = ReleaseSemaphore(data_.semaphore_, 1, NULL);
   if (result == 0) {
     FATAL("Mutex unlock failed");
   }
 }
 
 
+ThreadLocalKey MonitorWaitData::monitor_wait_data_key_ =
+    Thread::kUnsetThreadLocalKey;
+
+
 Monitor::Monitor() {
   InitializeCriticalSection(&data_.cs_);
-  // Create auto-reset event used to implement Notify. Auto-reset
-  // events only wake one thread waiting for them on SetEvent.
-  data_.notify_event_ = CreateEvent(NULL, FALSE, FALSE, NULL);
-  // Create manual-reset event used to implement
-  // NotifyAll. Manual-reset events wake all threads waiting for them
-  // on SetEvent.
-  data_.notify_all_event_ = CreateEvent(NULL, TRUE, FALSE, NULL);
-  if ((data_.notify_event_ == NULL) || (data_.notify_all_event_ == NULL)) {
-    FATAL("Failed allocating event object for monitor");
-  }
   InitializeCriticalSection(&data_.waiters_cs_);
+  data_.waiters_head_ = NULL;
+  data_.waiters_tail_ = NULL;
   data_.waiters_ = 0;
 }
 
 
 Monitor::~Monitor() {
   DeleteCriticalSection(&data_.cs_);
-  CloseHandle(data_.notify_event_);
-  CloseHandle(data_.notify_all_event_);
   DeleteCriticalSection(&data_.waiters_cs_);
 }
 
 
 void Monitor::Enter() {
   EnterCriticalSection(&data_.cs_);
 }
 
 
 void Monitor::Exit() {
   LeaveCriticalSection(&data_.cs_);
 }
 
 
+void MonitorData::AddWaiter(MonitorWaitData* wait_data) {
+  // Add the MonitorWaitData object to the list of objects waiting for
+  // this monitor.
+  EnterCriticalSection(&waiters_cs_);
+  if (waiters_tail_ == NULL) {

[Søren Gjesse, 2012/02/21 16:00:48]

ASSERT(waiters_head_ == NULL)

[Mads Ager (google), 2012/02/22 12:32:51]

Done.

+    waiters_head_ = waiters_tail_ = wait_data;
+  } else {
+    waiters_tail_->next_ = wait_data;
+    waiters_tail_ = wait_data;
+  }
+  waiters_++;
+  LeaveCriticalSection(&waiters_cs_);
+}
+
+
+void MonitorData::RemoveWaiter(MonitorWaitData* wait_data) {
+  // Remove the MonitorWaitData object from the list of objects
+  // waiting for this monitor.
+  EnterCriticalSection(&waiters_cs_);
+  MonitorWaitData* previous = NULL;
+  MonitorWaitData* current = waiters_head_;
+  while (current != NULL) {
+    if (current == wait_data) {
+      if (waiters_head_ == waiters_tail_) {
+        waiters_head_ = waiters_tail_ = NULL;
+      } else if (current == waiters_head_) {
+        waiters_head_ = waiters_head_->next_;
+      } else if (current == waiters_tail_) {
+        ASSERT(previous != NULL);
+        waiters_tail_ = previous;
+        previous->next_ = NULL;
+      } else {
+        ASSERT(previous != NULL);
+        previous->next_ = current->next_;
+      }
+      waiters_--;
+      break;
+    }
+    previous = current;
+    current = current->next_;
+  }
+  LeaveCriticalSection(&waiters_cs_);
+}
+
+
+MonitorWaitData* MonitorData::RemoveFirstWaiter() {
+  EnterCriticalSection(&waiters_cs_);
+  MonitorWaitData* first = NULL;
+  if (waiters_ > 0) {
+    first = waiters_head_;
+    if (waiters_ == 1) {
+      waiters_tail_ = waiters_head_ = NULL;
+    } else {
+      waiters_head_ = waiters_head_->next_;
+    }
+    waiters_--;
+  }
+  LeaveCriticalSection(&waiters_cs_);
+  return first;
+}
+
+
+MonitorWaitData** MonitorData::RemoveAllWaiters() {
+  EnterCriticalSection(&waiters_cs_);
+  MonitorWaitData** result =
+    reinterpret_cast<MonitorWaitData**>(calloc(waiters_ + 1, kWordSize));
+  MonitorWaitData* current = waiters_head_;
+  intptr_t i = 0;
+  while (current != NULL) {
+    result[i++] = current;
+    current = current->next_;
+  }
+  ASSERT(i == waiters_);
+  result[i] = NULL;
+  waiters_head_ = waiters_tail_ = NULL;
+  waiters_ = 0;
+  LeaveCriticalSection(&waiters_cs_);
+  return result;
+}
+
+
+MonitorWaitData* MonitorData::GetMonitorWaitDataForThread() {
+  // Ensure that the thread local key for monitor wait data objects is
+  // initialized.
+  EnterCriticalSection(&waiters_cs_);
+  if (MonitorWaitData::monitor_wait_data_key_ == Thread::kUnsetThreadLocalKey) {
+    MonitorWaitData::monitor_wait_data_key_ = Thread::CreateThreadLocal();
+  }
+  LeaveCriticalSection(&waiters_cs_);
+
+  // Get the MonitorWaitData object containing the event for this
+  // thread from thread local storage. Create it if it does not exist.
+  uword raw_wait_data =
+    Thread::GetThreadLocal(MonitorWaitData::monitor_wait_data_key_);
+  MonitorWaitData* wait_data = NULL;
+  if (raw_wait_data == 0) {
+    HANDLE event = CreateEvent(NULL, FALSE, FALSE, NULL);
+    wait_data = new MonitorWaitData(event);
+    Thread::SetThreadLocal(MonitorWaitData::monitor_wait_data_key_,
+                           reinterpret_cast<uword>(wait_data));
+  } else {
+    wait_data = reinterpret_cast<MonitorWaitData*>(raw_wait_data);
+    wait_data->next_ = NULL;
+  }
+  return wait_data;
+}
+
+
 Monitor::WaitResult Monitor::Wait(int64_t millis) {
   Monitor::WaitResult retval = kNotified;
 
-  // Record the fact that we will start waiting. This is used to only
-  // reset the notify all event when all waiting threads have dealt
-  // with the event.
-  EnterCriticalSection(&data_.waiters_cs_);
-  data_.waiters_++;
-  LeaveCriticalSection(&data_.waiters_cs_);
+  // Get the wait data object containing the event to wait for.
+  MonitorWaitData* wait_data = data_.GetMonitorWaitDataForThread();
+
+  // Start waiting by adding the MonitorWaitData to the list of
+  // waiters.
+  data_.AddWaiter(wait_data);
 
   // Leave the monitor critical section while waiting.
   LeaveCriticalSection(&data_.cs_);
 
-  // Perform the actual wait using wait for multiple objects on both
-  // the notify and the notify all events.
-  static const intptr_t kNotifyEventIndex = 0;
-  static const intptr_t kNotifyAllEventIndex = 1;
-  static const intptr_t kNumberOfEvents = 2;
-  HANDLE events[kNumberOfEvents];
-  events[kNotifyEventIndex] = data_.notify_event_;
-  events[kNotifyAllEventIndex] = data_.notify_all_event_;
-
+  // Perform the actual wait on the event.
   DWORD result = WAIT_FAILED;
   if (millis == 0) {
     // Wait forever for a Notify or a NotifyAll event.
-    result = WaitForMultipleObjects(2, events, FALSE, INFINITE);
+    result = WaitForSingleObject(wait_data->event_, INFINITE);
     if (result == WAIT_FAILED) {
       FATAL("Monitor::Wait failed");
     }
   } else {
     // Wait for the given period of time for a Notify or a NotifyAll
     // event.
-    result = WaitForMultipleObjects(2, events, FALSE, millis);
+    result = WaitForSingleObject(wait_data->event_, millis);
     if (result == WAIT_FAILED) {
       FATAL("Monitor::Wait with timeout failed");
     }
     if (result == WAIT_TIMEOUT) {
+      // No longer waiting. Remove from the list of waiters.
+      data_.RemoveWaiter(wait_data);
       retval = kTimedOut;
     }
   }
 
-  // Check if we are the last waiter on a notify all. If we are, reset
-  // the notify all event.
-  EnterCriticalSection(&data_.waiters_cs_);
-  data_.waiters_--;
-  if ((data_.waiters_ == 0) &&
-      (result == (WAIT_OBJECT_0 + kNotifyAllEventIndex))) {
-    ResetEvent(data_.notify_all_event_);
-  }
-  LeaveCriticalSection(&data_.waiters_cs_);
-
   // Reacquire the monitor critical section before continuing.
   EnterCriticalSection(&data_.cs_);
 
   return retval;
 }
 
 
 void Monitor::Notify() {
-  // Signal one waiter through the notify auto-reset event if there
-  // are any waiters.
-  EnterCriticalSection(&data_.waiters_cs_);
-  if (data_.waiters_ > 0) {
-    SetEvent(data_.notify_event_);
+  // Extract the first element in the list of waiters.
+  MonitorWaitData* head = data_.RemoveFirstWaiter();
+
+  // Signal the first element if there was one.
+  if (head != NULL) {
+    BOOL result = SetEvent(head->event_);
+    if (result == 0) {
+      FATAL("Monitor::Notify failed to signal event");
+    }
   }
-  LeaveCriticalSection(&data_.waiters_cs_);
 }
 
 
 void Monitor::NotifyAll() {
-  // Signal all waiters through the notify all manual-reset event if
-  // there are any waiters.
-  EnterCriticalSection(&data_.waiters_cs_);
-  if (data_.waiters_ > 0) {
-    SetEvent(data_.notify_all_event_);
+  // Extract the entire list of waiters into a separate array. Do not
+  // use the linked-list structure embedded in the MonitorWaitData
+  // objects. If a timeout occurs on an object in the array while we
+  // are processing them the MonitorWaitData object can be reused and
+  // the link structure can change. That should not influence which
+  // events are signaled here so we use a separate array.
+  //

[cshapiro, 2012/02/21 18:49:36]

When a thread waits on a monitor and the wait times out, the thread must
reacquire the monitor internal lock to unlink itself from the wait queue.  By
following this protocol, the link structure should not change if a thread wakes
up during a notify.

[Mads Ager (google), 2012/02/22 12:32:51]

I don't understand. The notification loop over the structure is done without
having any lock on the list. Therefore, the wait can time out and change the
link structure. Therefore, we have to not use that link structure as it could
lead to notification being lost or the wrong waiters being signaled.

The alternative is to keep the internal lock during the loop below at which
point you are right that a waiting thread that times out cannot continue and
change the link structure before moving on. That would simplify at the cost of
holding the lock during the loop. I'll do that so we avoid the allocation and
copying out here.

+  // If one of the objects in the array wakes because of a timeout
+  // before we signal it, that object will get an extra signal. This
+  // will be treated as a spurious wake-up and is OK since all uses of
+  // monitors should recheck the condition after a Wait.
+  MonitorWaitData** waiters = data_.RemoveAllWaiters();
+
+  // Signal all the waiters.
+  for (intptr_t i = 0; waiters[i] != NULL; i++) {

[cshapiro, 2012/02/21 18:49:36]

As noted in the header, this should be written as an appending of the wait queue
onto the entry queue.  Waking up each of the threads can become very, very
expensive.

[Mads Ager (google), 2012/02/22 12:32:51]

This could be expensive, I agree. However, it is much better than what we had
before. There are multiple things we can do here. We can follow the design that
you described, we can attempt to load the Vista ConditionVariables so this is
only in play for pre-Vista users. All of that I would like to do in a separate
changelist. I agree that this solution is not ideal and I will keep the bug
open. However, I don't think we need to spent more time on this at this point.

+    BOOL result = SetEvent(waiters[i]->event_);
+    if (result == 0) {
+      FATAL("Monitor::NotifyAll failed to signal event");
+    }
   }
-  LeaveCriticalSection(&data_.waiters_cs_);
+
+  free(waiters);
 }
 
 }  // namespace dart