Ensure consistent return ordering in base::WaitableEvent::WaitMany

base/synchronization/waitable_event_posix.cc

 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include <stddef.h>
 
 #include <algorithm>
+#include <limits>
 #include <vector>
 
 #include "base/debug/activity_tracker.h"
 #include "base/logging.h"
 #include "base/synchronization/condition_variable.h"
 #include "base/synchronization/lock.h"
 #include "base/synchronization/waitable_event.h"
 #include "base/threading/thread_restrictions.h"
 
 // -----------------------------------------------------------------------------
@@ skipping 241 matching lines @@
   // The set of waitables must be distinct. Since we have just sorted by
   // address, we can check this cheaply by comparing pairs of consecutive
   // elements.
   for (size_t i = 0; i < waitables.size() - 1; ++i) {
     DCHECK(waitables[i].first != waitables[i+1].first);
   }
 
   SyncWaiter sw;
 
   const size_t r = EnqueueMany(&waitables[0], count, &sw);
-  if (r) {
+  if (r < count) {
     // One of the events is already signaled. The SyncWaiter has not been
-    // enqueued anywhere. EnqueueMany returns the count of remaining waitables
-    // when the signaled one was seen, so the index of the signaled event is
-    // @count - @r.
-    return waitables[count - r].second;
+    // enqueued anywhere.
+    return waitables[r].second;
   }
 
   // At this point, we hold the locks on all the WaitableEvents and we have
   // enqueued our waiter in them all.
   sw.lock()->Acquire();
     // Release the WaitableEvent locks in the reverse order
     for (size_t i = 0; i < count; ++i) {
       waitables[count - (1 + i)].first->kernel_->lock_.Release();
     }
 
@@ skipping 27 matching lines @@
       raw_waitables[i]->kernel_->lock_.Acquire();
       raw_waitables[i]->kernel_->lock_.Release();
       signaled_index = i;
     }
   }
 
   return signaled_index;
 }
 
 // -----------------------------------------------------------------------------
-// If return value == 0:
+// If return value == count:
 //   The locks of the WaitableEvents have been taken in order and the Waiter has
 //   been enqueued in the wait-list of each. None of the WaitableEvents are
 //   currently signaled
 // else:
 //   None of the WaitableEvent locks are held. The Waiter has not been enqueued
-//   in any of them and the return value is the index of the first WaitableEvent
-//   which was signaled, from the end of the array.
+//   in any of them and the return value is the index of the WaitableEvent which
+//   was signaled with the lowest input index from the original WaitMany call.
 // -----------------------------------------------------------------------------
 // static
-size_t WaitableEvent::EnqueueMany
-    (std::pair<WaitableEvent*, size_t>* waitables,
-     size_t count, Waiter* waiter) {
-  if (!count)
-    return 0;
+size_t WaitableEvent::EnqueueMany(std::pair<WaitableEvent*, size_t>* waitables,
+                                  size_t count,
+                                  Waiter* waiter) {
+  size_t winner = count;
+  size_t winner_index = count;
+  for (size_t i = 0; i < count; ++i) {
+    auto& kernel = waitables[i].first->kernel_;
+    kernel->lock_.Acquire();
+    if (kernel->signaled_ && waitables[i].second < winner) {
+      winner = waitables[i].second;
+      winner_index = i;
+    }
+  }
 
-  waitables[0].first->kernel_->lock_.Acquire();
-    if (waitables[0].first->kernel_->signaled_) {
-      if (!waitables[0].first->kernel_->manual_reset_)
-        waitables[0].first->kernel_->signaled_ = false;
-      waitables[0].first->kernel_->lock_.Release();
-      return count;
+  // No events signaled. All locks acquired. Enqueue the Waiter on all of them
+  // and return.
+  if (winner == count) {
+    for (size_t i = 0; i < count; ++i)
+      waitables[i].first->Enqueue(waiter);
+    return count;
+  }
+
+  // Unlock in reverse order and possibly clear the chosen winner's signal
+  // before returning its index.
+  DCHECK_GE(count, 1u);
+  DCHECK_LE(count, static_cast<size_t>(std::numeric_limits<int>::max()));
+  for (int i = static_cast<int>(count - 1); i >= 0; --i) {

[dcheng, 2017/03/17 22:53:47]

One alternative that doesn't require DCHECKing and hoping indices are in range:

for (auto* waitable = waitables + count - 1; waitable >= waitables[0];
--waitable)

[Ken Rockot(use gerrit already), 2017/03/18 00:05:25]

Oh right, of course. Done.

+    auto& kernel = waitables[i].first->kernel_;
+    if (waitables[i].second == winner) {
+      if (!kernel->manual_reset_)
+        kernel->signaled_ = false;
     }
+    kernel->lock_.Release();
+  }
 
-    const size_t r = EnqueueMany(waitables + 1, count - 1, waiter);
-    if (r) {
-      waitables[0].first->kernel_->lock_.Release();
-    } else {
-      waitables[0].first->Enqueue(waiter);
-    }
-
-    return r;
+  return winner_index;
 }
 
 // -----------------------------------------------------------------------------
 
 
 // -----------------------------------------------------------------------------
 // Private functions...
 
 WaitableEvent::WaitableEventKernel::WaitableEventKernel(
     ResetPolicy reset_policy,
@@ skipping 54 matching lines @@
       return true;
     }
   }
 
   return false;
 }
 
 // -----------------------------------------------------------------------------
 
 }  // namespace base