Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(376)

Side by Side Diff: webrtc/base/asyncinvoker.cc

Issue 2904543003: Revert of Fixing potential AsyncInvoker deadlock that occurs for "reentrant" invocations. (Closed)
Patch Set: Created 3 years, 7 months ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View unified diff | Download patch
« no previous file with comments | « webrtc/base/asyncinvoker.h ('k') | webrtc/base/asyncinvoker-inl.h » ('j') | no next file with comments »
Toggle Intra-line Diffs ('i') | Expand Comments ('e') | Collapse Comments ('c') | Show Comments Hide Comments ('s')
OLDNEW
1 /* 1 /*
2 * Copyright 2014 The WebRTC Project Authors. All rights reserved. 2 * Copyright 2014 The WebRTC Project Authors. All rights reserved.
3 * 3 *
4 * Use of this source code is governed by a BSD-style license 4 * Use of this source code is governed by a BSD-style license
5 * that can be found in the LICENSE file in the root of the source 5 * that can be found in the LICENSE file in the root of the source
6 * tree. An additional intellectual property rights grant can be found 6 * tree. An additional intellectual property rights grant can be found
7 * in the file PATENTS. All contributing project authors may 7 * in the file PATENTS. All contributing project authors may
8 * be found in the AUTHORS file in the root of the source tree. 8 * be found in the AUTHORS file in the root of the source tree.
9 */ 9 */
10 10
11 #include "webrtc/base/asyncinvoker.h" 11 #include "webrtc/base/asyncinvoker.h"
12 12
13 #include "webrtc/base/atomicops.h" 13 #include "webrtc/base/atomicops.h"
14 #include "webrtc/base/checks.h" 14 #include "webrtc/base/checks.h"
15 #include "webrtc/base/logging.h" 15 #include "webrtc/base/logging.h"
16 16
17 namespace rtc { 17 namespace rtc {
18 18
19 AsyncInvoker::AsyncInvoker() : invocation_complete_(false, false) {} 19 AsyncInvoker::AsyncInvoker() : invocation_complete_(false, false) {}
20 20
21 AsyncInvoker::~AsyncInvoker() { 21 AsyncInvoker::~AsyncInvoker() {
22 AtomicOps::Increment(&destroying_); 22 destroying_ = true;
23 // Messages for this need to be cleared *before* our destructor is complete. 23 // Messages for this need to be cleared *before* our destructor is complete.
24 MessageQueueManager::Clear(this); 24 MessageQueueManager::Clear(this);
25 // And we need to wait for any invocations that are still in progress on 25 // And we need to wait for any invocations that are still in progress on
26 // other threads. 26 // other threads.
27 while (AtomicOps::AcquireLoad(&pending_invocations_)) { 27 while (AtomicOps::AcquireLoad(&pending_invocations_)) {
28 // If the destructor was called while AsyncInvoke was being called by 28 // If the destructor was called while AsyncInvoke was being called by
29 // another thread, WITHIN an AsyncInvoked functor, it may do another 29 // another thread, WITHIN an AsyncInvoked functor, it may do another
30 // Thread::Post even after we called MessageQueueManager::Clear(this). So 30 // Thread::Post even after we called MessageQueueManager::Clear(this). So
31 // we need to keep calling Clear to discard these posts. 31 // we need to keep calling Clear to discard these posts.
32 MessageQueueManager::Clear(this); 32 MessageQueueManager::Clear(this);
33 invocation_complete_.Wait(Event::kForever); 33 invocation_complete_.Wait(Event::kForever);
34 } 34 }
35 } 35 }
36 36
37 void AsyncInvoker::OnMessage(Message* msg) { 37 void AsyncInvoker::OnMessage(Message* msg) {
38 // Get the AsyncClosure shared ptr from this message's data. 38 // Get the AsyncClosure shared ptr from this message's data.
39 ScopedMessageData<AsyncClosure>* data = 39 ScopedMessageData<AsyncClosure>* data =
40 static_cast<ScopedMessageData<AsyncClosure>*>(msg->pdata); 40 static_cast<ScopedMessageData<AsyncClosure>*>(msg->pdata);
41 // Execute the closure and trigger the return message if needed. 41 // Execute the closure and trigger the return message if needed.
42 data->inner_data().Execute(); 42 data->inner_data().Execute();
43 delete data; 43 delete data;
44 } 44 }
45 45
46 void AsyncInvoker::Flush(Thread* thread, uint32_t id /*= MQID_ANY*/) { 46 void AsyncInvoker::Flush(Thread* thread, uint32_t id /*= MQID_ANY*/) {
47 if (AtomicOps::AcquireLoad(&destroying_)) 47 if (destroying_) return;
48 return;
49 48
50 // Run this on |thread| to reduce the number of context switches. 49 // Run this on |thread| to reduce the number of context switches.
51 if (Thread::Current() != thread) { 50 if (Thread::Current() != thread) {
52 thread->Invoke<void>(RTC_FROM_HERE, 51 thread->Invoke<void>(RTC_FROM_HERE,
53 Bind(&AsyncInvoker::Flush, this, thread, id)); 52 Bind(&AsyncInvoker::Flush, this, thread, id));
54 return; 53 return;
55 } 54 }
56 55
57 MessageList removed; 56 MessageList removed;
58 thread->Clear(this, id, &removed); 57 thread->Clear(this, id, &removed);
59 for (MessageList::iterator it = removed.begin(); it != removed.end(); ++it) { 58 for (MessageList::iterator it = removed.begin(); it != removed.end(); ++it) {
60 // This message was pending on this thread, so run it now. 59 // This message was pending on this thread, so run it now.
61 thread->Send(it->posted_from, it->phandler, it->message_id, it->pdata); 60 thread->Send(it->posted_from, it->phandler, it->message_id, it->pdata);
62 } 61 }
63 } 62 }
64 63
65 void AsyncInvoker::DoInvoke(const Location& posted_from, 64 void AsyncInvoker::DoInvoke(const Location& posted_from,
66 Thread* thread, 65 Thread* thread,
67 std::unique_ptr<AsyncClosure> closure, 66 std::unique_ptr<AsyncClosure> closure,
68 uint32_t id) { 67 uint32_t id) {
69 if (AtomicOps::AcquireLoad(&destroying_)) { 68 if (destroying_) {
70 LOG(LS_WARNING) << "Tried to invoke while destroying the invoker."; 69 LOG(LS_WARNING) << "Tried to invoke while destroying the invoker.";
71 return; 70 return;
72 } 71 }
72 AtomicOps::Increment(&pending_invocations_);
73 thread->Post(posted_from, this, id, 73 thread->Post(posted_from, this, id,
74 new ScopedMessageData<AsyncClosure>(std::move(closure))); 74 new ScopedMessageData<AsyncClosure>(std::move(closure)));
75 } 75 }
76 76
77 void AsyncInvoker::DoInvokeDelayed(const Location& posted_from, 77 void AsyncInvoker::DoInvokeDelayed(const Location& posted_from,
78 Thread* thread, 78 Thread* thread,
79 std::unique_ptr<AsyncClosure> closure, 79 std::unique_ptr<AsyncClosure> closure,
80 uint32_t delay_ms, 80 uint32_t delay_ms,
81 uint32_t id) { 81 uint32_t id) {
82 if (AtomicOps::AcquireLoad(&destroying_)) { 82 if (destroying_) {
83 LOG(LS_WARNING) << "Tried to invoke while destroying the invoker."; 83 LOG(LS_WARNING) << "Tried to invoke while destroying the invoker.";
84 return; 84 return;
85 } 85 }
86 AtomicOps::Increment(&pending_invocations_);
86 thread->PostDelayed(posted_from, delay_ms, this, id, 87 thread->PostDelayed(posted_from, delay_ms, this, id,
87 new ScopedMessageData<AsyncClosure>(std::move(closure))); 88 new ScopedMessageData<AsyncClosure>(std::move(closure)));
88 } 89 }
89 90
90 GuardedAsyncInvoker::GuardedAsyncInvoker() : thread_(Thread::Current()) { 91 GuardedAsyncInvoker::GuardedAsyncInvoker() : thread_(Thread::Current()) {
91 thread_->SignalQueueDestroyed.connect(this, 92 thread_->SignalQueueDestroyed.connect(this,
92 &GuardedAsyncInvoker::ThreadDestroyed); 93 &GuardedAsyncInvoker::ThreadDestroyed);
93 } 94 }
94 95
95 GuardedAsyncInvoker::~GuardedAsyncInvoker() { 96 GuardedAsyncInvoker::~GuardedAsyncInvoker() {
96 } 97 }
97 98
98 bool GuardedAsyncInvoker::Flush(uint32_t id) { 99 bool GuardedAsyncInvoker::Flush(uint32_t id) {
99 rtc::CritScope cs(&crit_); 100 rtc::CritScope cs(&crit_);
100 if (thread_ == nullptr) 101 if (thread_ == nullptr)
101 return false; 102 return false;
102 invoker_.Flush(thread_, id); 103 invoker_.Flush(thread_, id);
103 return true; 104 return true;
104 } 105 }
105 106
106 void GuardedAsyncInvoker::ThreadDestroyed() { 107 void GuardedAsyncInvoker::ThreadDestroyed() {
107 rtc::CritScope cs(&crit_); 108 rtc::CritScope cs(&crit_);
108 // We should never get more than one notification about the thread dying. 109 // We should never get more than one notification about the thread dying.
109 RTC_DCHECK(thread_ != nullptr); 110 RTC_DCHECK(thread_ != nullptr);
110 thread_ = nullptr; 111 thread_ = nullptr;
111 } 112 }
112 113
113 AsyncClosure::AsyncClosure(AsyncInvoker* invoker) : invoker_(invoker) {
114 AtomicOps::Increment(&invoker_->pending_invocations_);
115 }
116
117 AsyncClosure::~AsyncClosure() { 114 AsyncClosure::~AsyncClosure() {
118 AtomicOps::Decrement(&invoker_->pending_invocations_); 115 AtomicOps::Decrement(&invoker_->pending_invocations_);
119 invoker_->invocation_complete_.Set(); 116 invoker_->invocation_complete_.Set();
120 } 117 }
121 118
122 } // namespace rtc 119 } // namespace rtc
OLDNEW
« no previous file with comments | « webrtc/base/asyncinvoker.h ('k') | webrtc/base/asyncinvoker-inl.h » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698