1 /*
2 * Copyright 2004 The WebRTC Project Authors. All rights reserved.
3 *
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
6 * tree. An additional intellectual property rights grant can be found
7 * in the file PATENTS. All contributing project authors may
8 * be found in the AUTHORS file in the root of the source tree.
9 */
10
11 #include <memory>
12
13 #include "rtc_base/asyncinvoker.h"
14 #include "rtc_base/asyncudpsocket.h"
15 #include "rtc_base/event.h"
16 #include "rtc_base/gunit.h"
17 #include "rtc_base/nullsocketserver.h"
18 #include "rtc_base/physicalsocketserver.h"
19 #include "rtc_base/sigslot.h"
20 #include "rtc_base/socketaddress.h"
21 #include "rtc_base/thread.h"
22
23 #if defined(WEBRTC_WIN)
24 #include <comdef.h> // NOLINT
25 #endif
26
27 using namespace rtc;
28
29 // Generates a sequence of numbers (collaboratively).
30 class TestGenerator {
31 public:
TestGenerator()32 TestGenerator() : last(0), count(0) {}
33
Next(int prev)34 int Next(int prev) {
35 int result = prev + last;
36 last = result;
37 count += 1;
38 return result;
39 }
40
41 int last;
42 int count;
43 };
44
45 struct TestMessage : public MessageData {
TestMessageTestMessage46 explicit TestMessage(int v) : value(v) {}
47
48 int value;
49 };
50
51 // Receives on a socket and sends by posting messages.
52 class SocketClient : public TestGenerator, public sigslot::has_slots<> {
53 public:
SocketClient(AsyncSocket * socket,const SocketAddress & addr,Thread * post_thread,MessageHandler * phandler)54 SocketClient(AsyncSocket* socket, const SocketAddress& addr,
55 Thread* post_thread, MessageHandler* phandler)
56 : socket_(AsyncUDPSocket::Create(socket, addr)),
57 post_thread_(post_thread),
58 post_handler_(phandler) {
59 socket_->SignalReadPacket.connect(this, &SocketClient::OnPacket);
60 }
61
~SocketClient()62 ~SocketClient() override { delete socket_; }
63
address() const64 SocketAddress address() const { return socket_->GetLocalAddress(); }
65
OnPacket(AsyncPacketSocket * socket,const char * buf,size_t size,const SocketAddress & remote_addr,const PacketTime & packet_time)66 void OnPacket(AsyncPacketSocket* socket, const char* buf, size_t size,
67 const SocketAddress& remote_addr,
68 const PacketTime& packet_time) {
69 EXPECT_EQ(size, sizeof(uint32_t));
70 uint32_t prev = reinterpret_cast<const uint32_t*>(buf)[0];
71 uint32_t result = Next(prev);
72
73 post_thread_->PostDelayed(RTC_FROM_HERE, 200, post_handler_, 0,
74 new TestMessage(result));
75 }
76
77 private:
78 AsyncUDPSocket* socket_;
79 Thread* post_thread_;
80 MessageHandler* post_handler_;
81 };
82
83 // Receives messages and sends on a socket.
84 class MessageClient : public MessageHandler, public TestGenerator {
85 public:
MessageClient(Thread * pth,Socket * socket)86 MessageClient(Thread* pth, Socket* socket)
87 : socket_(socket) {
88 }
89
~MessageClient()90 ~MessageClient() override { delete socket_; }
91
OnMessage(Message * pmsg)92 void OnMessage(Message* pmsg) override {
93 TestMessage* msg = static_cast<TestMessage*>(pmsg->pdata);
94 int result = Next(msg->value);
95 EXPECT_GE(socket_->Send(&result, sizeof(result)), 0);
96 delete msg;
97 }
98
99 private:
100 Socket* socket_;
101 };
102
103 class CustomThread : public rtc::Thread {
104 public:
CustomThread()105 CustomThread()
106 : Thread(std::unique_ptr<SocketServer>(new rtc::NullSocketServer())) {}
~CustomThread()107 ~CustomThread() override { Stop(); }
Start()108 bool Start() { return false; }
109
WrapCurrent()110 bool WrapCurrent() {
111 return Thread::WrapCurrent();
112 }
UnwrapCurrent()113 void UnwrapCurrent() {
114 Thread::UnwrapCurrent();
115 }
116 };
117
118
119 // A thread that does nothing when it runs and signals an event
120 // when it is destroyed.
121 class SignalWhenDestroyedThread : public Thread {
122 public:
SignalWhenDestroyedThread(Event * event)123 SignalWhenDestroyedThread(Event* event)
124 : Thread(std::unique_ptr<SocketServer>(new NullSocketServer())),
125 event_(event) {}
126
~SignalWhenDestroyedThread()127 ~SignalWhenDestroyedThread() override {
128 Stop();
129 event_->Set();
130 }
131
Run()132 void Run() override {
133 // Do nothing.
134 }
135
136 private:
137 Event* event_;
138 };
139
140 // A bool wrapped in a mutex, to avoid data races. Using a volatile
141 // bool should be sufficient for correct code ("eventual consistency"
142 // between caches is sufficient), but we can't tell the compiler about
143 // that, and then tsan complains about a data race.
144
145 // See also discussion at
146 // http://stackoverflow.com/questions/7223164/is-mutex-needed-to-synchronize-a-simple-flag-between-pthreads
147
148 // Using std::atomic<bool> or std::atomic_flag in C++11 is probably
149 // the right thing to do, but those features are not yet allowed. Or
150 // rtc::AtomicInt, if/when that is added. Since the use isn't
151 // performance critical, use a plain critical section for the time
152 // being.
153
154 class AtomicBool {
155 public:
AtomicBool(bool value=false)156 explicit AtomicBool(bool value = false) : flag_(value) {}
operator =(bool value)157 AtomicBool& operator=(bool value) {
158 CritScope scoped_lock(&cs_);
159 flag_ = value;
160 return *this;
161 }
get() const162 bool get() const {
163 CritScope scoped_lock(&cs_);
164 return flag_;
165 }
166
167 private:
168 CriticalSection cs_;
169 bool flag_;
170 };
171
172 // Function objects to test Thread::Invoke.
173 struct FunctorA {
operator ()FunctorA174 int operator()() { return 42; }
175 };
176 class FunctorB {
177 public:
FunctorB(AtomicBool * flag)178 explicit FunctorB(AtomicBool* flag) : flag_(flag) {}
operator ()()179 void operator()() { if (flag_) *flag_ = true; }
180 private:
181 AtomicBool* flag_;
182 };
183 struct FunctorC {
operator ()FunctorC184 int operator()() {
185 Thread::Current()->ProcessMessages(50);
186 return 24;
187 }
188 };
189
190 // See: https://code.google.com/p/webrtc/issues/detail?id=2409
TEST(ThreadTest,DISABLED_Main)191 TEST(ThreadTest, DISABLED_Main) {
192 const SocketAddress addr("127.0.0.1", 0);
193
194 // Create the messaging client on its own thread.
195 auto th1 = Thread::CreateWithSocketServer();
196 Socket* socket =
197 th1->socketserver()->CreateAsyncSocket(addr.family(), SOCK_DGRAM);
198 MessageClient msg_client(th1.get(), socket);
199
200 // Create the socket client on its own thread.
201 auto th2 = Thread::CreateWithSocketServer();
202 AsyncSocket* asocket =
203 th2->socketserver()->CreateAsyncSocket(addr.family(), SOCK_DGRAM);
204 SocketClient sock_client(asocket, addr, th1.get(), &msg_client);
205
206 socket->Connect(sock_client.address());
207
208 th1->Start();
209 th2->Start();
210
211 // Get the messages started.
212 th1->PostDelayed(RTC_FROM_HERE, 100, &msg_client, 0, new TestMessage(1));
213
214 // Give the clients a little while to run.
215 // Messages will be processed at 100, 300, 500, 700, 900.
216 Thread* th_main = Thread::Current();
217 th_main->ProcessMessages(1000);
218
219 // Stop the sending client. Give the receiver a bit longer to run, in case
220 // it is running on a machine that is under load (e.g. the build machine).
221 th1->Stop();
222 th_main->ProcessMessages(200);
223 th2->Stop();
224
225 // Make sure the results were correct
226 EXPECT_EQ(5, msg_client.count);
227 EXPECT_EQ(34, msg_client.last);
228 EXPECT_EQ(5, sock_client.count);
229 EXPECT_EQ(55, sock_client.last);
230 }
231
232 // Test that setting thread names doesn't cause a malfunction.
233 // There's no easy way to verify the name was set properly at this time.
TEST(ThreadTest,Names)234 TEST(ThreadTest, Names) {
235 // Default name
236 auto thread = Thread::CreateWithSocketServer();
237 EXPECT_TRUE(thread->Start());
238 thread->Stop();
239 // Name with no object parameter
240 thread = Thread::CreateWithSocketServer();
241 EXPECT_TRUE(thread->SetName("No object", nullptr));
242 EXPECT_TRUE(thread->Start());
243 thread->Stop();
244 // Really long name
245 thread = Thread::CreateWithSocketServer();
246 EXPECT_TRUE(thread->SetName("Abcdefghijklmnopqrstuvwxyz1234567890", this));
247 EXPECT_TRUE(thread->Start());
248 thread->Stop();
249 }
250
TEST(ThreadTest,Wrap)251 TEST(ThreadTest, Wrap) {
252 Thread* current_thread = Thread::Current();
253 current_thread->UnwrapCurrent();
254 CustomThread* cthread = new CustomThread();
255 EXPECT_TRUE(cthread->WrapCurrent());
256 EXPECT_TRUE(cthread->RunningForTest());
257 EXPECT_FALSE(cthread->IsOwned());
258 cthread->UnwrapCurrent();
259 EXPECT_FALSE(cthread->RunningForTest());
260 delete cthread;
261 current_thread->WrapCurrent();
262 }
263
TEST(ThreadTest,Invoke)264 TEST(ThreadTest, Invoke) {
265 // Create and start the thread.
266 auto thread = Thread::CreateWithSocketServer();
267 thread->Start();
268 // Try calling functors.
269 EXPECT_EQ(42, thread->Invoke<int>(RTC_FROM_HERE, FunctorA()));
270 AtomicBool called;
271 FunctorB f2(&called);
272 thread->Invoke<void>(RTC_FROM_HERE, f2);
273 EXPECT_TRUE(called.get());
274 // Try calling bare functions.
275 struct LocalFuncs {
276 static int Func1() { return 999; }
277 static void Func2() {}
278 };
279 EXPECT_EQ(999, thread->Invoke<int>(RTC_FROM_HERE, &LocalFuncs::Func1));
280 thread->Invoke<void>(RTC_FROM_HERE, &LocalFuncs::Func2);
281 }
282
283 // Verifies that two threads calling Invoke on each other at the same time does
284 // not deadlock.
TEST(ThreadTest,TwoThreadsInvokeNoDeadlock)285 TEST(ThreadTest, TwoThreadsInvokeNoDeadlock) {
286 AutoThread thread;
287 Thread* current_thread = Thread::Current();
288 ASSERT_TRUE(current_thread != nullptr);
289
290 auto other_thread = Thread::CreateWithSocketServer();
291 other_thread->Start();
292
293 struct LocalFuncs {
294 static void Set(bool* out) { *out = true; }
295 static void InvokeSet(Thread* thread, bool* out) {
296 thread->Invoke<void>(RTC_FROM_HERE, Bind(&Set, out));
297 }
298 };
299
300 bool called = false;
301 other_thread->Invoke<void>(
302 RTC_FROM_HERE, Bind(&LocalFuncs::InvokeSet, current_thread, &called));
303
304 EXPECT_TRUE(called);
305 }
306
307 // Verifies that if thread A invokes a call on thread B and thread C is trying
308 // to invoke A at the same time, thread A does not handle C's invoke while
309 // invoking B.
TEST(ThreadTest,ThreeThreadsInvoke)310 TEST(ThreadTest, ThreeThreadsInvoke) {
311 AutoThread thread;
312 Thread* thread_a = Thread::Current();
313 auto thread_b = Thread::CreateWithSocketServer();
314 auto thread_c = Thread::CreateWithSocketServer();
315 thread_b->Start();
316 thread_c->Start();
317
318 class LockedBool {
319 public:
320 explicit LockedBool(bool value) : value_(value) {}
321
322 void Set(bool value) {
323 CritScope lock(&crit_);
324 value_ = value;
325 }
326
327 bool Get() {
328 CritScope lock(&crit_);
329 return value_;
330 }
331
332 private:
333 CriticalSection crit_;
334 bool value_ RTC_GUARDED_BY(crit_);
335 };
336
337 struct LocalFuncs {
338 static void Set(LockedBool* out) { out->Set(true); }
339 static void InvokeSet(Thread* thread, LockedBool* out) {
340 thread->Invoke<void>(RTC_FROM_HERE, Bind(&Set, out));
341 }
342
343 // Set |out| true and call InvokeSet on |thread|.
344 static void SetAndInvokeSet(LockedBool* out,
345 Thread* thread,
346 LockedBool* out_inner) {
347 out->Set(true);
348 InvokeSet(thread, out_inner);
349 }
350
351 // Asynchronously invoke SetAndInvokeSet on |thread1| and wait until
352 // |thread1| starts the call.
353 static void AsyncInvokeSetAndWait(AsyncInvoker* invoker,
354 Thread* thread1,
355 Thread* thread2,
356 LockedBool* out) {
357 CriticalSection crit;
358 LockedBool async_invoked(false);
359
360 invoker->AsyncInvoke<void>(
361 RTC_FROM_HERE, thread1,
362 Bind(&SetAndInvokeSet, &async_invoked, thread2, out));
363
364 EXPECT_TRUE_WAIT(async_invoked.Get(), 2000);
365 }
366 };
367
368 AsyncInvoker invoker;
369 LockedBool thread_a_called(false);
370
371 // Start the sequence A --(invoke)--> B --(async invoke)--> C --(invoke)--> A.
372 // Thread B returns when C receives the call and C should be blocked until A
373 // starts to process messages.
374 thread_b->Invoke<void>(RTC_FROM_HERE,
375 Bind(&LocalFuncs::AsyncInvokeSetAndWait, &invoker,
376 thread_c.get(), thread_a, &thread_a_called));
377 EXPECT_FALSE(thread_a_called.Get());
378
379 EXPECT_TRUE_WAIT(thread_a_called.Get(), 2000);
380 }
381
382 // Set the name on a thread when the underlying QueueDestroyed signal is
383 // triggered. This causes an error if the object is already partially
384 // destroyed.
385 class SetNameOnSignalQueueDestroyedTester : public sigslot::has_slots<> {
386 public:
SetNameOnSignalQueueDestroyedTester(Thread * thread)387 SetNameOnSignalQueueDestroyedTester(Thread* thread) : thread_(thread) {
388 thread->SignalQueueDestroyed.connect(
389 this, &SetNameOnSignalQueueDestroyedTester::OnQueueDestroyed);
390 }
391
OnQueueDestroyed()392 void OnQueueDestroyed() {
393 // Makes sure that if we access the Thread while it's being destroyed, that
394 // it doesn't cause a problem because the vtable has been modified.
395 thread_->SetName("foo", nullptr);
396 }
397
398 private:
399 Thread* thread_;
400 };
401
TEST(ThreadTest,SetNameOnSignalQueueDestroyed)402 TEST(ThreadTest, SetNameOnSignalQueueDestroyed) {
403 auto thread1 = Thread::CreateWithSocketServer();
404 SetNameOnSignalQueueDestroyedTester tester1(thread1.get());
405 thread1.reset();
406
407 Thread* thread2 = new AutoThread();
408 SetNameOnSignalQueueDestroyedTester tester2(thread2);
409 delete thread2;
410 }
411
412 class AsyncInvokeTest : public testing::Test {
413 public:
IntCallback(int value)414 void IntCallback(int value) {
415 EXPECT_EQ(expected_thread_, Thread::Current());
416 int_value_ = value;
417 }
SetExpectedThreadForIntCallback(Thread * thread)418 void SetExpectedThreadForIntCallback(Thread* thread) {
419 expected_thread_ = thread;
420 }
421
422 protected:
423 enum { kWaitTimeout = 1000 };
AsyncInvokeTest()424 AsyncInvokeTest()
425 : int_value_(0),
426 expected_thread_(nullptr) {}
427
428 int int_value_;
429 Thread* expected_thread_;
430 };
431
TEST_F(AsyncInvokeTest,FireAndForget)432 TEST_F(AsyncInvokeTest, FireAndForget) {
433 AsyncInvoker invoker;
434 // Create and start the thread.
435 auto thread = Thread::CreateWithSocketServer();
436 thread->Start();
437 // Try calling functor.
438 AtomicBool called;
439 invoker.AsyncInvoke<void>(RTC_FROM_HERE, thread.get(), FunctorB(&called));
440 EXPECT_TRUE_WAIT(called.get(), kWaitTimeout);
441 thread->Stop();
442 }
443
TEST_F(AsyncInvokeTest,KillInvokerDuringExecute)444 TEST_F(AsyncInvokeTest, KillInvokerDuringExecute) {
445 // Use these events to get in a state where the functor is in the middle of
446 // executing, and then to wait for it to finish, ensuring the "EXPECT_FALSE"
447 // is run.
448 Event functor_started(false, false);
449 Event functor_continue(false, false);
450 Event functor_finished(false, false);
451
452 auto thread = Thread::CreateWithSocketServer();
453 thread->Start();
454 volatile bool invoker_destroyed = false;
455 {
456 auto functor = [&functor_started, &functor_continue, &functor_finished,
457 &invoker_destroyed] {
458 functor_started.Set();
459 functor_continue.Wait(Event::kForever);
460 rtc::Thread::Current()->SleepMs(kWaitTimeout);
461 EXPECT_FALSE(invoker_destroyed);
462 functor_finished.Set();
463 };
464 AsyncInvoker invoker;
465 invoker.AsyncInvoke<void>(RTC_FROM_HERE, thread.get(), functor);
466 functor_started.Wait(Event::kForever);
467
468 // Destroy the invoker while the functor is still executing (doing
469 // SleepMs).
470 functor_continue.Set();
471 }
472
473 // If the destructor DIDN'T wait for the functor to finish executing, it will
474 // hit the EXPECT_FALSE(invoker_destroyed) after it finishes sleeping for a
475 // second.
476 invoker_destroyed = true;
477 functor_finished.Wait(Event::kForever);
478 }
479
480 // Variant of the above test where the async-invoked task calls AsyncInvoke
481 // *again*, for the thread on which the AsyncInvoker is currently being
482 // destroyed. This shouldn't deadlock or crash; this second invocation should
483 // just be ignored.
TEST_F(AsyncInvokeTest,KillInvokerDuringExecuteWithReentrantInvoke)484 TEST_F(AsyncInvokeTest, KillInvokerDuringExecuteWithReentrantInvoke) {
485 Event functor_started(false, false);
486 // Flag used to verify that the recursively invoked task never actually runs.
487 bool reentrant_functor_run = false;
488
489 Thread* main = Thread::Current();
490 Thread thread;
491 thread.Start();
492 {
493 AsyncInvoker invoker;
494 auto reentrant_functor = [&reentrant_functor_run] {
495 reentrant_functor_run = true;
496 };
497 auto functor = [&functor_started, &invoker, main, reentrant_functor] {
498 functor_started.Set();
499 Thread::Current()->SleepMs(kWaitTimeout);
500 invoker.AsyncInvoke<void>(RTC_FROM_HERE, main, reentrant_functor);
501 };
502 // This queues a task on |thread| to sleep for |kWaitTimeout| then queue a
503 // task on |main|. But this second queued task should never run, since the
504 // destructor will be entered before it's even invoked.
505 invoker.AsyncInvoke<void>(RTC_FROM_HERE, &thread, functor);
506 functor_started.Wait(Event::kForever);
507 }
508 EXPECT_FALSE(reentrant_functor_run);
509 }
510
TEST_F(AsyncInvokeTest,Flush)511 TEST_F(AsyncInvokeTest, Flush) {
512 AsyncInvoker invoker;
513 AtomicBool flag1;
514 AtomicBool flag2;
515 // Queue two async calls to the current thread.
516 invoker.AsyncInvoke<void>(RTC_FROM_HERE, Thread::Current(), FunctorB(&flag1));
517 invoker.AsyncInvoke<void>(RTC_FROM_HERE, Thread::Current(), FunctorB(&flag2));
518 // Because we haven't pumped messages, these should not have run yet.
519 EXPECT_FALSE(flag1.get());
520 EXPECT_FALSE(flag2.get());
521 // Force them to run now.
522 invoker.Flush(Thread::Current());
523 EXPECT_TRUE(flag1.get());
524 EXPECT_TRUE(flag2.get());
525 }
526
TEST_F(AsyncInvokeTest,FlushWithIds)527 TEST_F(AsyncInvokeTest, FlushWithIds) {
528 AsyncInvoker invoker;
529 AtomicBool flag1;
530 AtomicBool flag2;
531 // Queue two async calls to the current thread, one with a message id.
532 invoker.AsyncInvoke<void>(RTC_FROM_HERE, Thread::Current(), FunctorB(&flag1),
533 5);
534 invoker.AsyncInvoke<void>(RTC_FROM_HERE, Thread::Current(), FunctorB(&flag2));
535 // Because we haven't pumped messages, these should not have run yet.
536 EXPECT_FALSE(flag1.get());
537 EXPECT_FALSE(flag2.get());
538 // Execute pending calls with id == 5.
539 invoker.Flush(Thread::Current(), 5);
540 EXPECT_TRUE(flag1.get());
541 EXPECT_FALSE(flag2.get());
542 flag1 = false;
543 // Execute all pending calls. The id == 5 call should not execute again.
544 invoker.Flush(Thread::Current());
545 EXPECT_FALSE(flag1.get());
546 EXPECT_TRUE(flag2.get());
547 }
548
549 class GuardedAsyncInvokeTest : public testing::Test {
550 public:
IntCallback(int value)551 void IntCallback(int value) {
552 EXPECT_EQ(expected_thread_, Thread::Current());
553 int_value_ = value;
554 }
SetExpectedThreadForIntCallback(Thread * thread)555 void SetExpectedThreadForIntCallback(Thread* thread) {
556 expected_thread_ = thread;
557 }
558
559 protected:
560 const static int kWaitTimeout = 1000;
GuardedAsyncInvokeTest()561 GuardedAsyncInvokeTest()
562 : int_value_(0),
563 expected_thread_(nullptr) {}
564
565 int int_value_;
566 Thread* expected_thread_;
567 };
568
569 // Functor for creating an invoker.
570 struct CreateInvoker {
CreateInvokerCreateInvoker571 CreateInvoker(std::unique_ptr<GuardedAsyncInvoker>* invoker)
572 : invoker_(invoker) {}
operator ()CreateInvoker573 void operator()() { invoker_->reset(new GuardedAsyncInvoker()); }
574 std::unique_ptr<GuardedAsyncInvoker>* invoker_;
575 };
576
577 // Test that we can call AsyncInvoke<void>() after the thread died.
TEST_F(GuardedAsyncInvokeTest,KillThreadFireAndForget)578 TEST_F(GuardedAsyncInvokeTest, KillThreadFireAndForget) {
579 // Create and start the thread.
580 std::unique_ptr<Thread> thread(Thread::Create());
581 thread->Start();
582 std::unique_ptr<GuardedAsyncInvoker> invoker;
583 // Create the invoker on |thread|.
584 thread->Invoke<void>(RTC_FROM_HERE, CreateInvoker(&invoker));
585 // Kill |thread|.
586 thread = nullptr;
587 // Try calling functor.
588 AtomicBool called;
589 EXPECT_FALSE(invoker->AsyncInvoke<void>(RTC_FROM_HERE, FunctorB(&called)));
590 // With thread gone, nothing should happen.
591 WAIT(called.get(), kWaitTimeout);
592 EXPECT_FALSE(called.get());
593 }
594
595 // The remaining tests check that GuardedAsyncInvoker behaves as AsyncInvoker
596 // when Thread is still alive.
TEST_F(GuardedAsyncInvokeTest,FireAndForget)597 TEST_F(GuardedAsyncInvokeTest, FireAndForget) {
598 GuardedAsyncInvoker invoker;
599 // Try calling functor.
600 AtomicBool called;
601 EXPECT_TRUE(invoker.AsyncInvoke<void>(RTC_FROM_HERE, FunctorB(&called)));
602 EXPECT_TRUE_WAIT(called.get(), kWaitTimeout);
603 }
604
TEST_F(GuardedAsyncInvokeTest,Flush)605 TEST_F(GuardedAsyncInvokeTest, Flush) {
606 GuardedAsyncInvoker invoker;
607 AtomicBool flag1;
608 AtomicBool flag2;
609 // Queue two async calls to the current thread.
610 EXPECT_TRUE(invoker.AsyncInvoke<void>(RTC_FROM_HERE, FunctorB(&flag1)));
611 EXPECT_TRUE(invoker.AsyncInvoke<void>(RTC_FROM_HERE, FunctorB(&flag2)));
612 // Because we haven't pumped messages, these should not have run yet.
613 EXPECT_FALSE(flag1.get());
614 EXPECT_FALSE(flag2.get());
615 // Force them to run now.
616 EXPECT_TRUE(invoker.Flush());
617 EXPECT_TRUE(flag1.get());
618 EXPECT_TRUE(flag2.get());
619 }
620
TEST_F(GuardedAsyncInvokeTest,FlushWithIds)621 TEST_F(GuardedAsyncInvokeTest, FlushWithIds) {
622 GuardedAsyncInvoker invoker;
623 AtomicBool flag1;
624 AtomicBool flag2;
625 // Queue two async calls to the current thread, one with a message id.
626 EXPECT_TRUE(invoker.AsyncInvoke<void>(RTC_FROM_HERE, FunctorB(&flag1), 5));
627 EXPECT_TRUE(invoker.AsyncInvoke<void>(RTC_FROM_HERE, FunctorB(&flag2)));
628 // Because we haven't pumped messages, these should not have run yet.
629 EXPECT_FALSE(flag1.get());
630 EXPECT_FALSE(flag2.get());
631 // Execute pending calls with id == 5.
632 EXPECT_TRUE(invoker.Flush(5));
633 EXPECT_TRUE(flag1.get());
634 EXPECT_FALSE(flag2.get());
635 flag1 = false;
636 // Execute all pending calls. The id == 5 call should not execute again.
637 EXPECT_TRUE(invoker.Flush());
638 EXPECT_FALSE(flag1.get());
639 EXPECT_TRUE(flag2.get());
640 }
641