waitable_event_posix.cc 16 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448
  1. // Copyright (c) 2012 The Chromium Authors. All rights reserved.
  2. // Use of this source code is governed by a BSD-style license that can be
  3. // found in the LICENSE file.
  4. #include <stddef.h>
  5. #include <limits>
  6. #include <vector>
  7. #include "base/check_op.h"
  8. #include "base/debug/activity_tracker.h"
  9. #include "base/memory/raw_ptr.h"
  10. #include "base/ranges/algorithm.h"
  11. #include "base/synchronization/condition_variable.h"
  12. #include "base/synchronization/lock.h"
  13. #include "base/synchronization/waitable_event.h"
  14. #include "base/threading/scoped_blocking_call.h"
  15. #include "base/threading/thread_restrictions.h"
  16. #include "base/time/time.h"
  17. #include "base/time/time_override.h"
  18. #include "third_party/abseil-cpp/absl/types/optional.h"
  19. // -----------------------------------------------------------------------------
  20. // A WaitableEvent on POSIX is implemented as a wait-list. Currently we don't
  21. // support cross-process events (where one process can signal an event which
  22. // others are waiting on). Because of this, we can avoid having one thread per
  23. // listener in several cases.
  24. //
  25. // The WaitableEvent maintains a list of waiters, protected by a lock. Each
  26. // waiter is either an async wait, in which case we have a Task and the
  27. // MessageLoop to run it on, or a blocking wait, in which case we have the
  28. // condition variable to signal.
  29. //
  30. // Waiting involves grabbing the lock and adding oneself to the wait list. Async
  31. // waits can be canceled, which means grabbing the lock and removing oneself
  32. // from the list.
  33. //
  34. // Waiting on multiple events is handled by adding a single, synchronous wait to
  35. // the wait-list of many events. An event passes a pointer to itself when
  36. // firing a waiter and so we can store that pointer to find out which event
  37. // triggered.
  38. // -----------------------------------------------------------------------------
  39. namespace base {
  40. // -----------------------------------------------------------------------------
  41. // This is just an abstract base class for waking the two types of waiters
  42. // -----------------------------------------------------------------------------
  43. WaitableEvent::WaitableEvent(ResetPolicy reset_policy,
  44. InitialState initial_state)
  45. : kernel_(new WaitableEventKernel(reset_policy, initial_state)) {}
  46. WaitableEvent::~WaitableEvent() = default;
  47. void WaitableEvent::Reset() {
  48. base::AutoLock locked(kernel_->lock_);
  49. kernel_->signaled_ = false;
  50. }
  51. void WaitableEvent::Signal() {
  52. base::AutoLock locked(kernel_->lock_);
  53. if (kernel_->signaled_)
  54. return;
  55. if (kernel_->manual_reset_) {
  56. SignalAll();
  57. kernel_->signaled_ = true;
  58. } else {
  59. // In the case of auto reset, if no waiters were woken, we remain
  60. // signaled.
  61. if (!SignalOne())
  62. kernel_->signaled_ = true;
  63. }
  64. }
  65. bool WaitableEvent::IsSignaled() {
  66. base::AutoLock locked(kernel_->lock_);
  67. const bool result = kernel_->signaled_;
  68. if (result && !kernel_->manual_reset_)
  69. kernel_->signaled_ = false;
  70. return result;
  71. }
  72. // -----------------------------------------------------------------------------
  73. // Synchronous waits
  74. // -----------------------------------------------------------------------------
  75. // This is a synchronous waiter. The thread is waiting on the given condition
  76. // variable and the fired flag in this object.
  77. // -----------------------------------------------------------------------------
  78. class SyncWaiter : public WaitableEvent::Waiter {
  79. public:
  80. SyncWaiter()
  81. : fired_(false), signaling_event_(nullptr), lock_(), cv_(&lock_) {}
  82. bool Fire(WaitableEvent* signaling_event) override {
  83. base::AutoLock locked(lock_);
  84. if (fired_)
  85. return false;
  86. fired_ = true;
  87. signaling_event_ = signaling_event;
  88. cv_.Broadcast();
  89. // Unlike AsyncWaiter objects, SyncWaiter objects are stack-allocated on
  90. // the blocking thread's stack. There is no |delete this;| in Fire. The
  91. // SyncWaiter object is destroyed when it goes out of scope.
  92. return true;
  93. }
  94. WaitableEvent* signaling_event() const {
  95. return signaling_event_;
  96. }
  97. // ---------------------------------------------------------------------------
  98. // These waiters are always stack allocated and don't delete themselves. Thus
  99. // there's no problem and the ABA tag is the same as the object pointer.
  100. // ---------------------------------------------------------------------------
  101. bool Compare(void* tag) override { return this == tag; }
  102. // ---------------------------------------------------------------------------
  103. // Called with lock held.
  104. // ---------------------------------------------------------------------------
  105. bool fired() const {
  106. return fired_;
  107. }
  108. // ---------------------------------------------------------------------------
  109. // During a TimedWait, we need a way to make sure that an auto-reset
  110. // WaitableEvent doesn't think that this event has been signaled between
  111. // unlocking it and removing it from the wait-list. Called with lock held.
  112. // ---------------------------------------------------------------------------
  113. void Disable() {
  114. fired_ = true;
  115. }
  116. base::Lock* lock() {
  117. return &lock_;
  118. }
  119. base::ConditionVariable* cv() {
  120. return &cv_;
  121. }
  122. private:
  123. bool fired_;
  124. raw_ptr<WaitableEvent> signaling_event_; // The WaitableEvent which woke us
  125. base::Lock lock_;
  126. base::ConditionVariable cv_;
  127. };
  128. void WaitableEvent::Wait() {
  129. bool result = TimedWait(TimeDelta::Max());
  130. DCHECK(result) << "TimedWait() should never fail with infinite timeout";
  131. }
  132. bool WaitableEvent::TimedWait(const TimeDelta& wait_delta) {
  133. if (wait_delta <= TimeDelta())
  134. return IsSignaled();
  135. // Record the event that this thread is blocking upon (for hang diagnosis) and
  136. // consider it blocked for scheduling purposes. Ignore this for non-blocking
  137. // WaitableEvents.
  138. absl::optional<debug::ScopedEventWaitActivity> event_activity;
  139. absl::optional<internal::ScopedBlockingCallWithBaseSyncPrimitives>
  140. scoped_blocking_call;
  141. if (waiting_is_blocking_) {
  142. event_activity.emplace(this);
  143. scoped_blocking_call.emplace(FROM_HERE, BlockingType::MAY_BLOCK);
  144. }
  145. kernel_->lock_.Acquire();
  146. if (kernel_->signaled_) {
  147. if (!kernel_->manual_reset_) {
  148. // In this case we were signaled when we had no waiters. Now that
  149. // someone has waited upon us, we can automatically reset.
  150. kernel_->signaled_ = false;
  151. }
  152. kernel_->lock_.Release();
  153. return true;
  154. }
  155. SyncWaiter sw;
  156. if (!waiting_is_blocking_)
  157. sw.cv()->declare_only_used_while_idle();
  158. sw.lock()->Acquire();
  159. Enqueue(&sw);
  160. kernel_->lock_.Release();
  161. // We are violating locking order here by holding the SyncWaiter lock but not
  162. // the WaitableEvent lock. However, this is safe because we don't lock |lock_|
  163. // again before unlocking it.
  164. // TimeTicks takes care of overflow but we special case is_max() nonetheless
  165. // to avoid invoking TimeTicksNowIgnoringOverride() unnecessarily (same for
  166. // the increment step of the for loop if the condition variable returns
  167. // early). Ref: https://crbug.com/910524#c7
  168. const TimeTicks end_time =
  169. wait_delta.is_max() ? TimeTicks::Max()
  170. : subtle::TimeTicksNowIgnoringOverride() + wait_delta;
  171. for (TimeDelta remaining = wait_delta; remaining.is_positive() && !sw.fired();
  172. remaining = end_time.is_max()
  173. ? TimeDelta::Max()
  174. : end_time - subtle::TimeTicksNowIgnoringOverride()) {
  175. if (end_time.is_max())
  176. sw.cv()->Wait();
  177. else
  178. sw.cv()->TimedWait(remaining);
  179. }
  180. // Get the SyncWaiter signaled state before releasing the lock.
  181. const bool return_value = sw.fired();
  182. // We can't acquire |lock_| before releasing the SyncWaiter lock (because of
  183. // locking order), however, in between the two a signal could be fired and
  184. // |sw| would accept it, however we will still return false, so the signal
  185. // would be lost on an auto-reset WaitableEvent. Thus we call Disable which
  186. // makes sw::Fire return false.
  187. sw.Disable();
  188. sw.lock()->Release();
  189. // This is a bug that has been enshrined in the interface of WaitableEvent
  190. // now: |Dequeue| is called even when |sw.fired()| is true, even though it'll
  191. // always return false in that case. However, taking the lock ensures that
  192. // |Signal| has completed before we return and means that a WaitableEvent can
  193. // synchronise its own destruction.
  194. kernel_->lock_.Acquire();
  195. kernel_->Dequeue(&sw, &sw);
  196. kernel_->lock_.Release();
  197. return return_value;
  198. }
  199. // -----------------------------------------------------------------------------
  200. // Synchronous waiting on multiple objects.
  201. static bool // StrictWeakOrdering
  202. cmp_fst_addr(const std::pair<WaitableEvent*, unsigned> &a,
  203. const std::pair<WaitableEvent*, unsigned> &b) {
  204. return a.first < b.first;
  205. }
  206. // static
  207. // NO_THREAD_SAFETY_ANALYSIS: Complex control flow.
  208. size_t WaitableEvent::WaitMany(WaitableEvent** raw_waitables,
  209. size_t count) NO_THREAD_SAFETY_ANALYSIS {
  210. DCHECK(count) << "Cannot wait on no events";
  211. internal::ScopedBlockingCallWithBaseSyncPrimitives scoped_blocking_call(
  212. FROM_HERE, BlockingType::MAY_BLOCK);
  213. // Record an event (the first) that this thread is blocking upon.
  214. debug::ScopedEventWaitActivity event_activity(raw_waitables[0]);
  215. // We need to acquire the locks in a globally consistent order. Thus we sort
  216. // the array of waitables by address. We actually sort a pairs so that we can
  217. // map back to the original index values later.
  218. std::vector<std::pair<WaitableEvent*, size_t> > waitables;
  219. waitables.reserve(count);
  220. for (size_t i = 0; i < count; ++i)
  221. waitables.push_back(std::make_pair(raw_waitables[i], i));
  222. DCHECK_EQ(count, waitables.size());
  223. ranges::sort(waitables, cmp_fst_addr);
  224. // The set of waitables must be distinct. Since we have just sorted by
  225. // address, we can check this cheaply by comparing pairs of consecutive
  226. // elements.
  227. for (size_t i = 0; i < waitables.size() - 1; ++i) {
  228. DCHECK(waitables[i].first != waitables[i+1].first);
  229. }
  230. SyncWaiter sw;
  231. const size_t r = EnqueueMany(&waitables[0], count, &sw);
  232. if (r < count) {
  233. // One of the events is already signaled. The SyncWaiter has not been
  234. // enqueued anywhere.
  235. return waitables[r].second;
  236. }
  237. // At this point, we hold the locks on all the WaitableEvents and we have
  238. // enqueued our waiter in them all.
  239. sw.lock()->Acquire();
  240. // Release the WaitableEvent locks in the reverse order
  241. for (size_t i = 0; i < count; ++i) {
  242. waitables[count - (1 + i)].first->kernel_->lock_.Release();
  243. }
  244. for (;;) {
  245. if (sw.fired())
  246. break;
  247. sw.cv()->Wait();
  248. }
  249. sw.lock()->Release();
  250. // The address of the WaitableEvent which fired is stored in the SyncWaiter.
  251. WaitableEvent *const signaled_event = sw.signaling_event();
  252. // This will store the index of the raw_waitables which fired.
  253. size_t signaled_index = 0;
  254. // Take the locks of each WaitableEvent in turn (except the signaled one) and
  255. // remove our SyncWaiter from the wait-list
  256. for (size_t i = 0; i < count; ++i) {
  257. if (raw_waitables[i] != signaled_event) {
  258. raw_waitables[i]->kernel_->lock_.Acquire();
  259. // There's no possible ABA issue with the address of the SyncWaiter here
  260. // because it lives on the stack. Thus the tag value is just the pointer
  261. // value again.
  262. raw_waitables[i]->kernel_->Dequeue(&sw, &sw);
  263. raw_waitables[i]->kernel_->lock_.Release();
  264. } else {
  265. // By taking this lock here we ensure that |Signal| has completed by the
  266. // time we return, because |Signal| holds this lock. This matches the
  267. // behaviour of |Wait| and |TimedWait|.
  268. raw_waitables[i]->kernel_->lock_.Acquire();
  269. raw_waitables[i]->kernel_->lock_.Release();
  270. signaled_index = i;
  271. }
  272. }
  273. return signaled_index;
  274. }
  275. // -----------------------------------------------------------------------------
  276. // If return value == count:
  277. // The locks of the WaitableEvents have been taken in order and the Waiter has
  278. // been enqueued in the wait-list of each. None of the WaitableEvents are
  279. // currently signaled
  280. // else:
  281. // None of the WaitableEvent locks are held. The Waiter has not been enqueued
  282. // in any of them and the return value is the index of the WaitableEvent which
  283. // was signaled with the lowest input index from the original WaitMany call.
  284. // -----------------------------------------------------------------------------
  285. // static
  286. // NO_THREAD_SAFETY_ANALYSIS: Complex control flow.
  287. size_t WaitableEvent::EnqueueMany(std::pair<WaitableEvent*, size_t>* waitables,
  288. size_t count,
  289. Waiter* waiter) NO_THREAD_SAFETY_ANALYSIS {
  290. size_t winner = count;
  291. size_t winner_index = count;
  292. for (size_t i = 0; i < count; ++i) {
  293. auto& kernel = waitables[i].first->kernel_;
  294. kernel->lock_.Acquire();
  295. if (kernel->signaled_ && waitables[i].second < winner) {
  296. winner = waitables[i].second;
  297. winner_index = i;
  298. }
  299. }
  300. // No events signaled. All locks acquired. Enqueue the Waiter on all of them
  301. // and return.
  302. if (winner == count) {
  303. for (size_t i = 0; i < count; ++i)
  304. waitables[i].first->Enqueue(waiter);
  305. return count;
  306. }
  307. // Unlock in reverse order and possibly clear the chosen winner's signal
  308. // before returning its index.
  309. for (auto* w = waitables + count - 1; w >= waitables; --w) {
  310. auto& kernel = w->first->kernel_;
  311. if (w->second == winner) {
  312. if (!kernel->manual_reset_)
  313. kernel->signaled_ = false;
  314. }
  315. kernel->lock_.Release();
  316. }
  317. return winner_index;
  318. }
  319. // -----------------------------------------------------------------------------
  320. // -----------------------------------------------------------------------------
  321. // Private functions...
  322. WaitableEvent::WaitableEventKernel::WaitableEventKernel(
  323. ResetPolicy reset_policy,
  324. InitialState initial_state)
  325. : manual_reset_(reset_policy == ResetPolicy::MANUAL),
  326. signaled_(initial_state == InitialState::SIGNALED) {}
  327. WaitableEvent::WaitableEventKernel::~WaitableEventKernel() = default;
  328. // -----------------------------------------------------------------------------
  329. // Wake all waiting waiters. Called with lock held.
  330. // -----------------------------------------------------------------------------
  331. bool WaitableEvent::SignalAll() {
  332. bool signaled_at_least_one = false;
  333. for (auto* i : kernel_->waiters_) {
  334. if (i->Fire(this))
  335. signaled_at_least_one = true;
  336. }
  337. kernel_->waiters_.clear();
  338. return signaled_at_least_one;
  339. }
  340. // ---------------------------------------------------------------------------
  341. // Try to wake a single waiter. Return true if one was woken. Called with lock
  342. // held.
  343. // ---------------------------------------------------------------------------
  344. bool WaitableEvent::SignalOne() {
  345. for (;;) {
  346. if (kernel_->waiters_.empty())
  347. return false;
  348. const bool r = (*kernel_->waiters_.begin())->Fire(this);
  349. kernel_->waiters_.pop_front();
  350. if (r)
  351. return true;
  352. }
  353. }
  354. // -----------------------------------------------------------------------------
  355. // Add a waiter to the list of those waiting. Called with lock held.
  356. // -----------------------------------------------------------------------------
  357. void WaitableEvent::Enqueue(Waiter* waiter) {
  358. kernel_->waiters_.push_back(waiter);
  359. }
  360. // -----------------------------------------------------------------------------
  361. // Remove a waiter from the list of those waiting. Return true if the waiter was
  362. // actually removed. Called with lock held.
  363. // -----------------------------------------------------------------------------
  364. bool WaitableEvent::WaitableEventKernel::Dequeue(Waiter* waiter, void* tag) {
  365. for (auto i = waiters_.begin(); i != waiters_.end(); ++i) {
  366. if (*i == waiter && (*i)->Compare(tag)) {
  367. waiters_.erase(i);
  368. return true;
  369. }
  370. }
  371. return false;
  372. }
  373. // -----------------------------------------------------------------------------
  374. } // namespace base