diff options
Diffstat (limited to 'third_party/libwebrtc/rtc_base/thread_unittest.cc')
-rw-r--r-- | third_party/libwebrtc/rtc_base/thread_unittest.cc | 1148 |
1 files changed, 1148 insertions, 0 deletions
diff --git a/third_party/libwebrtc/rtc_base/thread_unittest.cc b/third_party/libwebrtc/rtc_base/thread_unittest.cc new file mode 100644 index 0000000000..7fcf7ca833 --- /dev/null +++ b/third_party/libwebrtc/rtc_base/thread_unittest.cc @@ -0,0 +1,1148 @@ +/* + * Copyright 2004 The WebRTC Project Authors. All rights reserved. + * + * Use of this source code is governed by a BSD-style license + * that can be found in the LICENSE file in the root of the source + * tree. An additional intellectual property rights grant can be found + * in the file PATENTS. All contributing project authors may + * be found in the AUTHORS file in the root of the source tree. + */ + +#include "rtc_base/thread.h" + +#include <memory> + +#include "api/task_queue/task_queue_factory.h" +#include "api/task_queue/task_queue_test.h" +#include "api/units/time_delta.h" +#include "rtc_base/async_invoker.h" +#include "rtc_base/async_udp_socket.h" +#include "rtc_base/checks.h" +#include "rtc_base/event.h" +#include "rtc_base/gunit.h" +#include "rtc_base/internal/default_socket_server.h" +#include "rtc_base/null_socket_server.h" +#include "rtc_base/physical_socket_server.h" +#include "rtc_base/socket_address.h" +#include "rtc_base/synchronization/mutex.h" +#include "rtc_base/third_party/sigslot/sigslot.h" +#include "test/testsupport/rtc_expect_death.h" + +#if defined(WEBRTC_WIN) +#include <comdef.h> // NOLINT + +#endif + +namespace rtc { +namespace { + +using ::webrtc::TimeDelta; + +// Generates a sequence of numbers (collaboratively). +class TestGenerator { + public: + TestGenerator() : last(0), count(0) {} + + int Next(int prev) { + int result = prev + last; + last = result; + count += 1; + return result; + } + + int last; + int count; +}; + +struct TestMessage : public MessageData { + explicit TestMessage(int v) : value(v) {} + + int value; +}; + +// Receives on a socket and sends by posting messages. +class SocketClient : public TestGenerator, public sigslot::has_slots<> { + public: + SocketClient(Socket* socket, + const SocketAddress& addr, + Thread* post_thread, + MessageHandler* phandler) + : socket_(AsyncUDPSocket::Create(socket, addr)), + post_thread_(post_thread), + post_handler_(phandler) { + socket_->SignalReadPacket.connect(this, &SocketClient::OnPacket); + } + + ~SocketClient() override { delete socket_; } + + SocketAddress address() const { return socket_->GetLocalAddress(); } + + void OnPacket(AsyncPacketSocket* socket, + const char* buf, + size_t size, + const SocketAddress& remote_addr, + const int64_t& packet_time_us) { + EXPECT_EQ(size, sizeof(uint32_t)); + uint32_t prev = reinterpret_cast<const uint32_t*>(buf)[0]; + uint32_t result = Next(prev); + + post_thread_->PostDelayed(RTC_FROM_HERE, 200, post_handler_, 0, + new TestMessage(result)); + } + + private: + AsyncUDPSocket* socket_; + Thread* post_thread_; + MessageHandler* post_handler_; +}; + +// Receives messages and sends on a socket. +class MessageClient : public MessageHandlerAutoCleanup, public TestGenerator { + public: + MessageClient(Thread* pth, Socket* socket) : socket_(socket) {} + + ~MessageClient() override { delete socket_; } + + void OnMessage(Message* pmsg) override { + TestMessage* msg = static_cast<TestMessage*>(pmsg->pdata); + int result = Next(msg->value); + EXPECT_GE(socket_->Send(&result, sizeof(result)), 0); + delete msg; + } + + private: + Socket* socket_; +}; + +class CustomThread : public rtc::Thread { + public: + CustomThread() + : Thread(std::unique_ptr<SocketServer>(new rtc::NullSocketServer())) {} + ~CustomThread() override { Stop(); } + bool Start() { return false; } + + bool WrapCurrent() { return Thread::WrapCurrent(); } + void UnwrapCurrent() { Thread::UnwrapCurrent(); } +}; + +// A thread that does nothing when it runs and signals an event +// when it is destroyed. +class SignalWhenDestroyedThread : public Thread { + public: + SignalWhenDestroyedThread(Event* event) + : Thread(std::unique_ptr<SocketServer>(new NullSocketServer())), + event_(event) {} + + ~SignalWhenDestroyedThread() override { + Stop(); + event_->Set(); + } + + void Run() override { + // Do nothing. + } + + private: + Event* event_; +}; + +// A bool wrapped in a mutex, to avoid data races. Using a volatile +// bool should be sufficient for correct code ("eventual consistency" +// between caches is sufficient), but we can't tell the compiler about +// that, and then tsan complains about a data race. + +// See also discussion at +// http://stackoverflow.com/questions/7223164/is-mutex-needed-to-synchronize-a-simple-flag-between-pthreads + +// Using std::atomic<bool> or std::atomic_flag in C++11 is probably +// the right thing to do, but those features are not yet allowed. Or +// rtc::AtomicInt, if/when that is added. Since the use isn't +// performance critical, use a plain critical section for the time +// being. + +class AtomicBool { + public: + explicit AtomicBool(bool value = false) : flag_(value) {} + AtomicBool& operator=(bool value) { + webrtc::MutexLock scoped_lock(&mutex_); + flag_ = value; + return *this; + } + bool get() const { + webrtc::MutexLock scoped_lock(&mutex_); + return flag_; + } + + private: + mutable webrtc::Mutex mutex_; + bool flag_; +}; + +// Function objects to test Thread::Invoke. +struct FunctorA { + int operator()() { return 42; } +}; +class FunctorB { + public: + explicit FunctorB(AtomicBool* flag) : flag_(flag) {} + void operator()() { + if (flag_) + *flag_ = true; + } + + private: + AtomicBool* flag_; +}; +struct FunctorC { + int operator()() { + Thread::Current()->ProcessMessages(50); + return 24; + } +}; +struct FunctorD { + public: + explicit FunctorD(AtomicBool* flag) : flag_(flag) {} + FunctorD(FunctorD&&) = default; + + FunctorD(const FunctorD&) = delete; + FunctorD& operator=(const FunctorD&) = delete; + + FunctorD& operator=(FunctorD&&) = default; + void operator()() { + if (flag_) + *flag_ = true; + } + + private: + AtomicBool* flag_; +}; + +// See: https://code.google.com/p/webrtc/issues/detail?id=2409 +TEST(ThreadTest, DISABLED_Main) { + const SocketAddress addr("127.0.0.1", 0); + + // Create the messaging client on its own thread. + auto th1 = Thread::CreateWithSocketServer(); + Socket* socket = th1->socketserver()->CreateSocket(addr.family(), SOCK_DGRAM); + MessageClient msg_client(th1.get(), socket); + + // Create the socket client on its own thread. + auto th2 = Thread::CreateWithSocketServer(); + Socket* asocket = + th2->socketserver()->CreateSocket(addr.family(), SOCK_DGRAM); + SocketClient sock_client(asocket, addr, th1.get(), &msg_client); + + socket->Connect(sock_client.address()); + + th1->Start(); + th2->Start(); + + // Get the messages started. + th1->PostDelayed(RTC_FROM_HERE, 100, &msg_client, 0, new TestMessage(1)); + + // Give the clients a little while to run. + // Messages will be processed at 100, 300, 500, 700, 900. + Thread* th_main = Thread::Current(); + th_main->ProcessMessages(1000); + + // Stop the sending client. Give the receiver a bit longer to run, in case + // it is running on a machine that is under load (e.g. the build machine). + th1->Stop(); + th_main->ProcessMessages(200); + th2->Stop(); + + // Make sure the results were correct + EXPECT_EQ(5, msg_client.count); + EXPECT_EQ(34, msg_client.last); + EXPECT_EQ(5, sock_client.count); + EXPECT_EQ(55, sock_client.last); +} + +TEST(ThreadTest, CountBlockingCalls) { + rtc::AutoThread current; + + // When the test runs, this will print out: + // (thread_unittest.cc:262): Blocking TestBody: total=2 (actual=1, could=1) + RTC_LOG_THREAD_BLOCK_COUNT(); +#if RTC_DCHECK_IS_ON + rtc::Thread::ScopedCountBlockingCalls blocked_calls( + [&](uint32_t actual_block, uint32_t could_block) { + EXPECT_EQ(1u, actual_block); + EXPECT_EQ(1u, could_block); + }); + + EXPECT_EQ(0u, blocked_calls.GetBlockingCallCount()); + EXPECT_EQ(0u, blocked_calls.GetCouldBeBlockingCallCount()); + EXPECT_EQ(0u, blocked_calls.GetTotalBlockedCallCount()); + + // Test invoking on the current thread. This should not count as an 'actual' + // invoke, but should still count as an invoke that could block since we + // that the call to Invoke serves a purpose in some configurations (and should + // not be used a general way to call methods on the same thread). + current.Invoke<void>(RTC_FROM_HERE, []() {}); + EXPECT_EQ(0u, blocked_calls.GetBlockingCallCount()); + EXPECT_EQ(1u, blocked_calls.GetCouldBeBlockingCallCount()); + EXPECT_EQ(1u, blocked_calls.GetTotalBlockedCallCount()); + + // Create a new thread to invoke on. + auto thread = Thread::CreateWithSocketServer(); + thread->Start(); + EXPECT_EQ(42, thread->Invoke<int>(RTC_FROM_HERE, []() { return 42; })); + EXPECT_EQ(1u, blocked_calls.GetBlockingCallCount()); + EXPECT_EQ(1u, blocked_calls.GetCouldBeBlockingCallCount()); + EXPECT_EQ(2u, blocked_calls.GetTotalBlockedCallCount()); + thread->Stop(); + RTC_DCHECK_BLOCK_COUNT_NO_MORE_THAN(2); +#else + RTC_DCHECK_BLOCK_COUNT_NO_MORE_THAN(0); + RTC_LOG(LS_INFO) << "Test not active in this config"; +#endif +} + +#if RTC_DCHECK_IS_ON +TEST(ThreadTest, CountBlockingCallsOneCallback) { + rtc::AutoThread current; + bool was_called_back = false; + { + rtc::Thread::ScopedCountBlockingCalls blocked_calls( + [&](uint32_t actual_block, uint32_t could_block) { + was_called_back = true; + }); + current.Invoke<void>(RTC_FROM_HERE, []() {}); + } + EXPECT_TRUE(was_called_back); +} + +TEST(ThreadTest, CountBlockingCallsSkipCallback) { + rtc::AutoThread current; + bool was_called_back = false; + { + rtc::Thread::ScopedCountBlockingCalls blocked_calls( + [&](uint32_t actual_block, uint32_t could_block) { + was_called_back = true; + }); + // Changed `blocked_calls` to not issue the callback if there are 1 or + // fewer blocking calls (i.e. we set the minimum required number to 2). + blocked_calls.set_minimum_call_count_for_callback(2); + current.Invoke<void>(RTC_FROM_HERE, []() {}); + } + // We should not have gotten a call back. + EXPECT_FALSE(was_called_back); +} +#endif + +// Test that setting thread names doesn't cause a malfunction. +// There's no easy way to verify the name was set properly at this time. +TEST(ThreadTest, Names) { + // Default name + auto thread = Thread::CreateWithSocketServer(); + EXPECT_TRUE(thread->Start()); + thread->Stop(); + // Name with no object parameter + thread = Thread::CreateWithSocketServer(); + EXPECT_TRUE(thread->SetName("No object", nullptr)); + EXPECT_TRUE(thread->Start()); + thread->Stop(); + // Really long name + thread = Thread::CreateWithSocketServer(); + EXPECT_TRUE(thread->SetName("Abcdefghijklmnopqrstuvwxyz1234567890", this)); + EXPECT_TRUE(thread->Start()); + thread->Stop(); +} + +TEST(ThreadTest, Wrap) { + Thread* current_thread = Thread::Current(); + ThreadManager::Instance()->SetCurrentThread(nullptr); + + { + CustomThread cthread; + EXPECT_TRUE(cthread.WrapCurrent()); + EXPECT_EQ(&cthread, Thread::Current()); + EXPECT_TRUE(cthread.RunningForTest()); + EXPECT_FALSE(cthread.IsOwned()); + cthread.UnwrapCurrent(); + EXPECT_FALSE(cthread.RunningForTest()); + } + ThreadManager::Instance()->SetCurrentThread(current_thread); +} + +#if (!defined(NDEBUG) || RTC_DCHECK_IS_ON) +TEST(ThreadTest, InvokeToThreadAllowedReturnsTrueWithoutPolicies) { + rtc::AutoThread main_thread; + // Create and start the thread. + auto thread1 = Thread::CreateWithSocketServer(); + auto thread2 = Thread::CreateWithSocketServer(); + + thread1->PostTask( + [&]() { EXPECT_TRUE(thread1->IsInvokeToThreadAllowed(thread2.get())); }); + main_thread.ProcessMessages(100); +} + +TEST(ThreadTest, InvokeAllowedWhenThreadsAdded) { + rtc::AutoThread main_thread; + // Create and start the thread. + auto thread1 = Thread::CreateWithSocketServer(); + auto thread2 = Thread::CreateWithSocketServer(); + auto thread3 = Thread::CreateWithSocketServer(); + auto thread4 = Thread::CreateWithSocketServer(); + + thread1->AllowInvokesToThread(thread2.get()); + thread1->AllowInvokesToThread(thread3.get()); + + thread1->PostTask([&]() { + EXPECT_TRUE(thread1->IsInvokeToThreadAllowed(thread2.get())); + EXPECT_TRUE(thread1->IsInvokeToThreadAllowed(thread3.get())); + EXPECT_FALSE(thread1->IsInvokeToThreadAllowed(thread4.get())); + }); + main_thread.ProcessMessages(100); +} + +TEST(ThreadTest, InvokesDisallowedWhenDisallowAllInvokes) { + rtc::AutoThread main_thread; + // Create and start the thread. + auto thread1 = Thread::CreateWithSocketServer(); + auto thread2 = Thread::CreateWithSocketServer(); + + thread1->DisallowAllInvokes(); + + thread1->PostTask( + [&]() { EXPECT_FALSE(thread1->IsInvokeToThreadAllowed(thread2.get())); }); + main_thread.ProcessMessages(100); +} +#endif // (!defined(NDEBUG) || RTC_DCHECK_IS_ON) + +TEST(ThreadTest, InvokesAllowedByDefault) { + rtc::AutoThread main_thread; + // Create and start the thread. + auto thread1 = Thread::CreateWithSocketServer(); + auto thread2 = Thread::CreateWithSocketServer(); + + thread1->PostTask( + [&]() { EXPECT_TRUE(thread1->IsInvokeToThreadAllowed(thread2.get())); }); + main_thread.ProcessMessages(100); +} + +TEST(ThreadTest, Invoke) { + // Create and start the thread. + auto thread = Thread::CreateWithSocketServer(); + thread->Start(); + // Try calling functors. + EXPECT_EQ(42, thread->Invoke<int>(RTC_FROM_HERE, FunctorA())); + AtomicBool called; + FunctorB f2(&called); + thread->Invoke<void>(RTC_FROM_HERE, f2); + EXPECT_TRUE(called.get()); + // Try calling bare functions. + struct LocalFuncs { + static int Func1() { return 999; } + static void Func2() {} + }; + EXPECT_EQ(999, thread->Invoke<int>(RTC_FROM_HERE, &LocalFuncs::Func1)); + thread->Invoke<void>(RTC_FROM_HERE, &LocalFuncs::Func2); +} + +// Verifies that two threads calling Invoke on each other at the same time does +// not deadlock but crash. +#if RTC_DCHECK_IS_ON && GTEST_HAS_DEATH_TEST && !defined(WEBRTC_ANDROID) +TEST(ThreadTest, TwoThreadsInvokeDeathTest) { + GTEST_FLAG_SET(death_test_style, "threadsafe"); + AutoThread thread; + Thread* main_thread = Thread::Current(); + auto other_thread = Thread::CreateWithSocketServer(); + other_thread->Start(); + other_thread->Invoke<void>(RTC_FROM_HERE, [main_thread] { + RTC_EXPECT_DEATH(main_thread->Invoke<void>(RTC_FROM_HERE, [] {}), "loop"); + }); +} + +TEST(ThreadTest, ThreeThreadsInvokeDeathTest) { + GTEST_FLAG_SET(death_test_style, "threadsafe"); + AutoThread thread; + Thread* first = Thread::Current(); + + auto second = Thread::Create(); + second->Start(); + auto third = Thread::Create(); + third->Start(); + + second->Invoke<void>(RTC_FROM_HERE, [&] { + third->Invoke<void>(RTC_FROM_HERE, [&] { + RTC_EXPECT_DEATH(first->Invoke<void>(RTC_FROM_HERE, [] {}), "loop"); + }); + }); +} + +#endif + +// Verifies that if thread A invokes a call on thread B and thread C is trying +// to invoke A at the same time, thread A does not handle C's invoke while +// invoking B. +TEST(ThreadTest, ThreeThreadsInvoke) { + AutoThread thread; + Thread* thread_a = Thread::Current(); + auto thread_b = Thread::CreateWithSocketServer(); + auto thread_c = Thread::CreateWithSocketServer(); + thread_b->Start(); + thread_c->Start(); + + class LockedBool { + public: + explicit LockedBool(bool value) : value_(value) {} + + void Set(bool value) { + webrtc::MutexLock lock(&mutex_); + value_ = value; + } + + bool Get() { + webrtc::MutexLock lock(&mutex_); + return value_; + } + + private: + webrtc::Mutex mutex_; + bool value_ RTC_GUARDED_BY(mutex_); + }; + + struct LocalFuncs { + static void Set(LockedBool* out) { out->Set(true); } + static void InvokeSet(Thread* thread, LockedBool* out) { + thread->Invoke<void>(RTC_FROM_HERE, [out] { Set(out); }); + } + + // Set `out` true and call InvokeSet on `thread`. + static void SetAndInvokeSet(LockedBool* out, + Thread* thread, + LockedBool* out_inner) { + out->Set(true); + InvokeSet(thread, out_inner); + } + + // Asynchronously invoke SetAndInvokeSet on `thread1` and wait until + // `thread1` starts the call. + static void AsyncInvokeSetAndWait(DEPRECATED_AsyncInvoker* invoker, + Thread* thread1, + Thread* thread2, + LockedBool* out) { + LockedBool async_invoked(false); + + invoker->AsyncInvoke<void>( + RTC_FROM_HERE, thread1, [&async_invoked, thread2, out] { + SetAndInvokeSet(&async_invoked, thread2, out); + }); + + EXPECT_TRUE_WAIT(async_invoked.Get(), 2000); + } + }; + + DEPRECATED_AsyncInvoker invoker; + LockedBool thread_a_called(false); + + // Start the sequence A --(invoke)--> B --(async invoke)--> C --(invoke)--> A. + // Thread B returns when C receives the call and C should be blocked until A + // starts to process messages. + Thread* thread_c_ptr = thread_c.get(); + thread_b->Invoke<void>( + RTC_FROM_HERE, [&invoker, thread_c_ptr, thread_a, &thread_a_called] { + LocalFuncs::AsyncInvokeSetAndWait(&invoker, thread_c_ptr, thread_a, + &thread_a_called); + }); + EXPECT_FALSE(thread_a_called.Get()); + + EXPECT_TRUE_WAIT(thread_a_called.Get(), 2000); +} + +class ThreadQueueTest : public ::testing::Test, public Thread { + public: + ThreadQueueTest() : Thread(CreateDefaultSocketServer(), true) {} + bool IsLocked_Worker() { + if (!CritForTest()->TryEnter()) { + return true; + } + CritForTest()->Leave(); + return false; + } + bool IsLocked() { + // We have to do this on a worker thread, or else the TryEnter will + // succeed, since our critical sections are reentrant. + std::unique_ptr<Thread> worker(Thread::CreateWithSocketServer()); + worker->Start(); + return worker->Invoke<bool>(RTC_FROM_HERE, + [this] { return IsLocked_Worker(); }); + } +}; + +struct DeletedLockChecker { + DeletedLockChecker(ThreadQueueTest* test, bool* was_locked, bool* deleted) + : test(test), was_locked(was_locked), deleted(deleted) {} + ~DeletedLockChecker() { + *deleted = true; + *was_locked = test->IsLocked(); + } + ThreadQueueTest* test; + bool* was_locked; + bool* deleted; +}; + +static void DelayedPostsWithIdenticalTimesAreProcessedInFifoOrder(Thread* q) { + EXPECT_TRUE(q != nullptr); + int64_t now = TimeMillis(); + q->PostAt(RTC_FROM_HERE, now, nullptr, 3); + q->PostAt(RTC_FROM_HERE, now - 2, nullptr, 0); + q->PostAt(RTC_FROM_HERE, now - 1, nullptr, 1); + q->PostAt(RTC_FROM_HERE, now, nullptr, 4); + q->PostAt(RTC_FROM_HERE, now - 1, nullptr, 2); + + Message msg; + for (size_t i = 0; i < 5; ++i) { + memset(&msg, 0, sizeof(msg)); + EXPECT_TRUE(q->Get(&msg, 0)); + EXPECT_EQ(i, msg.message_id); + } + + EXPECT_FALSE(q->Get(&msg, 0)); // No more messages +} + +TEST_F(ThreadQueueTest, DelayedPostsWithIdenticalTimesAreProcessedInFifoOrder) { + Thread q(CreateDefaultSocketServer(), true); + DelayedPostsWithIdenticalTimesAreProcessedInFifoOrder(&q); + + NullSocketServer nullss; + Thread q_nullss(&nullss, true); + DelayedPostsWithIdenticalTimesAreProcessedInFifoOrder(&q_nullss); +} + +TEST_F(ThreadQueueTest, DisposeNotLocked) { + bool was_locked = true; + bool deleted = false; + DeletedLockChecker* d = new DeletedLockChecker(this, &was_locked, &deleted); + Dispose(d); + Message msg; + EXPECT_FALSE(Get(&msg, 0)); + EXPECT_TRUE(deleted); + EXPECT_FALSE(was_locked); +} + +class DeletedMessageHandler : public MessageHandlerAutoCleanup { + public: + explicit DeletedMessageHandler(bool* deleted) : deleted_(deleted) {} + ~DeletedMessageHandler() override { *deleted_ = true; } + void OnMessage(Message* msg) override {} + + private: + bool* deleted_; +}; + +TEST_F(ThreadQueueTest, DiposeHandlerWithPostedMessagePending) { + bool deleted = false; + DeletedMessageHandler* handler = new DeletedMessageHandler(&deleted); + // First, post a dispose. + Dispose(handler); + // Now, post a message, which should *not* be returned by Get(). + Post(RTC_FROM_HERE, handler, 1); + Message msg; + EXPECT_FALSE(Get(&msg, 0)); + EXPECT_TRUE(deleted); +} + +// Ensure that ProcessAllMessageQueues does its essential function; process +// all messages (both delayed and non delayed) up until the current time, on +// all registered message queues. +TEST(ThreadManager, ProcessAllMessageQueues) { + rtc::AutoThread main_thread; + Event entered_process_all_message_queues(true, false); + auto a = Thread::CreateWithSocketServer(); + auto b = Thread::CreateWithSocketServer(); + a->Start(); + b->Start(); + + std::atomic<int> messages_processed(0); + auto incrementer = [&messages_processed, + &entered_process_all_message_queues] { + // Wait for event as a means to ensure Increment doesn't occur outside + // of ProcessAllMessageQueues. The event is set by a message posted to + // the main thread, which is guaranteed to be handled inside + // ProcessAllMessageQueues. + entered_process_all_message_queues.Wait(Event::kForever); + messages_processed.fetch_add(1); + }; + auto event_signaler = [&entered_process_all_message_queues] { + entered_process_all_message_queues.Set(); + }; + + // Post messages (both delayed and non delayed) to both threads. + a->PostTask(incrementer); + b->PostTask(incrementer); + a->PostDelayedTask(incrementer, TimeDelta::Zero()); + b->PostDelayedTask(incrementer, TimeDelta::Zero()); + main_thread.PostTask(event_signaler); + + ThreadManager::ProcessAllMessageQueuesForTesting(); + EXPECT_EQ(4, messages_processed.load(std::memory_order_acquire)); +} + +// Test that ProcessAllMessageQueues doesn't hang if a thread is quitting. +TEST(ThreadManager, ProcessAllMessageQueuesWithQuittingThread) { + auto t = Thread::CreateWithSocketServer(); + t->Start(); + t->Quit(); + ThreadManager::ProcessAllMessageQueuesForTesting(); +} + +// Test that ProcessAllMessageQueues doesn't hang if a queue clears its +// messages. +TEST(ThreadManager, ProcessAllMessageQueuesWithClearedQueue) { + rtc::AutoThread main_thread; + Event entered_process_all_message_queues(true, false); + auto t = Thread::CreateWithSocketServer(); + t->Start(); + + auto clearer = [&entered_process_all_message_queues] { + // Wait for event as a means to ensure Clear doesn't occur outside of + // ProcessAllMessageQueues. The event is set by a message posted to the + // main thread, which is guaranteed to be handled inside + // ProcessAllMessageQueues. + entered_process_all_message_queues.Wait(Event::kForever); + rtc::Thread::Current()->Clear(nullptr); + }; + auto event_signaler = [&entered_process_all_message_queues] { + entered_process_all_message_queues.Set(); + }; + + // Post messages (both delayed and non delayed) to both threads. + t->PostTask(clearer); + main_thread.PostTask(event_signaler); + ThreadManager::ProcessAllMessageQueuesForTesting(); +} + +class RefCountedHandler : public MessageHandlerAutoCleanup, + public rtc::RefCountInterface { + public: + void OnMessage(Message* msg) override {} +}; + +class EmptyHandler : public MessageHandlerAutoCleanup { + public: + void OnMessage(Message* msg) override {} +}; + +TEST(ThreadManager, ClearReentrant) { + std::unique_ptr<Thread> t(Thread::Create()); + EmptyHandler handler; + RefCountedHandler* inner_handler( + new rtc::RefCountedObject<RefCountedHandler>()); + // When the empty handler is destroyed, it will clear messages queued for + // itself. The message to be cleared itself wraps a MessageHandler object + // (RefCountedHandler) so this will cause the message queue to be cleared + // again in a re-entrant fashion, which previously triggered a DCHECK. + // The inner handler will be removed in a re-entrant fashion from the + // message queue of the thread while the outer handler is removed, verifying + // that the iterator is not invalidated in "MessageQueue::Clear". + t->Post(RTC_FROM_HERE, inner_handler, 0); + t->Post(RTC_FROM_HERE, &handler, 0, + new ScopedRefMessageData<RefCountedHandler>(inner_handler)); +} + +class DEPRECATED_AsyncInvokeTest : public ::testing::Test { + public: + void IntCallback(int value) { + EXPECT_EQ(expected_thread_, Thread::Current()); + int_value_ = value; + } + void SetExpectedThreadForIntCallback(Thread* thread) { + expected_thread_ = thread; + } + + protected: + enum { kWaitTimeout = 1000 }; + DEPRECATED_AsyncInvokeTest() : int_value_(0), expected_thread_(nullptr) {} + + rtc::AutoThread main_thread_; + int int_value_; + Thread* expected_thread_; +}; + +TEST_F(DEPRECATED_AsyncInvokeTest, FireAndForget) { + DEPRECATED_AsyncInvoker invoker; + // Create and start the thread. + auto thread = Thread::CreateWithSocketServer(); + thread->Start(); + // Try calling functor. + AtomicBool called; + invoker.AsyncInvoke<void>(RTC_FROM_HERE, thread.get(), FunctorB(&called)); + EXPECT_TRUE_WAIT(called.get(), kWaitTimeout); + thread->Stop(); +} + +TEST_F(DEPRECATED_AsyncInvokeTest, NonCopyableFunctor) { + DEPRECATED_AsyncInvoker invoker; + // Create and start the thread. + auto thread = Thread::CreateWithSocketServer(); + thread->Start(); + // Try calling functor. + AtomicBool called; + invoker.AsyncInvoke<void>(RTC_FROM_HERE, thread.get(), FunctorD(&called)); + EXPECT_TRUE_WAIT(called.get(), kWaitTimeout); + thread->Stop(); +} + +TEST_F(DEPRECATED_AsyncInvokeTest, KillInvokerDuringExecute) { + // Use these events to get in a state where the functor is in the middle of + // executing, and then to wait for it to finish, ensuring the "EXPECT_FALSE" + // is run. + Event functor_started; + Event functor_continue; + Event functor_finished; + + auto thread = Thread::CreateWithSocketServer(); + thread->Start(); + volatile bool invoker_destroyed = false; + { + auto functor = [&functor_started, &functor_continue, &functor_finished, + &invoker_destroyed] { + functor_started.Set(); + functor_continue.Wait(Event::kForever); + rtc::Thread::Current()->SleepMs(kWaitTimeout); + EXPECT_FALSE(invoker_destroyed); + functor_finished.Set(); + }; + DEPRECATED_AsyncInvoker invoker; + invoker.AsyncInvoke<void>(RTC_FROM_HERE, thread.get(), functor); + functor_started.Wait(Event::kForever); + + // Destroy the invoker while the functor is still executing (doing + // SleepMs). + functor_continue.Set(); + } + + // If the destructor DIDN'T wait for the functor to finish executing, it will + // hit the EXPECT_FALSE(invoker_destroyed) after it finishes sleeping for a + // second. + invoker_destroyed = true; + functor_finished.Wait(Event::kForever); +} + +// Variant of the above test where the async-invoked task calls AsyncInvoke +// *again*, for the thread on which the invoker is currently being destroyed. +// This shouldn't deadlock or crash. The second invocation should be ignored. +TEST_F(DEPRECATED_AsyncInvokeTest, + KillInvokerDuringExecuteWithReentrantInvoke) { + Event functor_started; + // Flag used to verify that the recursively invoked task never actually runs. + bool reentrant_functor_run = false; + + Thread* main = Thread::Current(); + Thread thread(std::make_unique<NullSocketServer>()); + thread.Start(); + { + DEPRECATED_AsyncInvoker invoker; + auto reentrant_functor = [&reentrant_functor_run] { + reentrant_functor_run = true; + }; + auto functor = [&functor_started, &invoker, main, reentrant_functor] { + functor_started.Set(); + Thread::Current()->SleepMs(kWaitTimeout); + invoker.AsyncInvoke<void>(RTC_FROM_HERE, main, reentrant_functor); + }; + // This queues a task on `thread` to sleep for `kWaitTimeout` then queue a + // task on `main`. But this second queued task should never run, since the + // destructor will be entered before it's even invoked. + invoker.AsyncInvoke<void>(RTC_FROM_HERE, &thread, functor); + functor_started.Wait(Event::kForever); + } + EXPECT_FALSE(reentrant_functor_run); +} + +void WaitAndSetEvent(Event* wait_event, Event* set_event) { + wait_event->Wait(Event::kForever); + set_event->Set(); +} + +// A functor that keeps track of the number of copies and moves. +class LifeCycleFunctor { + public: + struct Stats { + size_t copy_count = 0; + size_t move_count = 0; + }; + + LifeCycleFunctor(Stats* stats, Event* event) : stats_(stats), event_(event) {} + LifeCycleFunctor(const LifeCycleFunctor& other) { *this = other; } + LifeCycleFunctor(LifeCycleFunctor&& other) { *this = std::move(other); } + + LifeCycleFunctor& operator=(const LifeCycleFunctor& other) { + stats_ = other.stats_; + event_ = other.event_; + ++stats_->copy_count; + return *this; + } + + LifeCycleFunctor& operator=(LifeCycleFunctor&& other) { + stats_ = other.stats_; + event_ = other.event_; + ++stats_->move_count; + return *this; + } + + void operator()() { event_->Set(); } + + private: + Stats* stats_; + Event* event_; +}; + +// A functor that verifies the thread it was destroyed on. +class DestructionFunctor { + public: + DestructionFunctor(Thread* thread, bool* thread_was_current, Event* event) + : thread_(thread), + thread_was_current_(thread_was_current), + event_(event) {} + ~DestructionFunctor() { + // Only signal the event if this was the functor that was invoked to avoid + // the event being signaled due to the destruction of temporary/moved + // versions of this object. + if (was_invoked_) { + *thread_was_current_ = thread_->IsCurrent(); + event_->Set(); + } + } + + void operator()() { was_invoked_ = true; } + + private: + Thread* thread_; + bool* thread_was_current_; + Event* event_; + bool was_invoked_ = false; +}; + +TEST(ThreadPostTaskTest, InvokesWithLambda) { + std::unique_ptr<rtc::Thread> background_thread(rtc::Thread::Create()); + background_thread->Start(); + + Event event; + background_thread->PostTask([&event] { event.Set(); }); + event.Wait(Event::kForever); +} + +TEST(ThreadPostTaskTest, InvokesWithCopiedFunctor) { + std::unique_ptr<rtc::Thread> background_thread(rtc::Thread::Create()); + background_thread->Start(); + + LifeCycleFunctor::Stats stats; + Event event; + LifeCycleFunctor functor(&stats, &event); + background_thread->PostTask(functor); + event.Wait(Event::kForever); + + EXPECT_EQ(1u, stats.copy_count); + EXPECT_EQ(0u, stats.move_count); +} + +TEST(ThreadPostTaskTest, InvokesWithMovedFunctor) { + std::unique_ptr<rtc::Thread> background_thread(rtc::Thread::Create()); + background_thread->Start(); + + LifeCycleFunctor::Stats stats; + Event event; + LifeCycleFunctor functor(&stats, &event); + background_thread->PostTask(std::move(functor)); + event.Wait(Event::kForever); + + EXPECT_EQ(0u, stats.copy_count); + EXPECT_EQ(1u, stats.move_count); +} + +TEST(ThreadPostTaskTest, InvokesWithReferencedFunctorShouldCopy) { + std::unique_ptr<rtc::Thread> background_thread(rtc::Thread::Create()); + background_thread->Start(); + + LifeCycleFunctor::Stats stats; + Event event; + LifeCycleFunctor functor(&stats, &event); + LifeCycleFunctor& functor_ref = functor; + background_thread->PostTask(functor_ref); + event.Wait(Event::kForever); + + EXPECT_EQ(1u, stats.copy_count); + EXPECT_EQ(0u, stats.move_count); +} + +TEST(ThreadPostTaskTest, InvokesWithCopiedFunctorDestroyedOnTargetThread) { + std::unique_ptr<rtc::Thread> background_thread(rtc::Thread::Create()); + background_thread->Start(); + + Event event; + bool was_invoked_on_background_thread = false; + DestructionFunctor functor(background_thread.get(), + &was_invoked_on_background_thread, &event); + background_thread->PostTask(functor); + event.Wait(Event::kForever); + + EXPECT_TRUE(was_invoked_on_background_thread); +} + +TEST(ThreadPostTaskTest, InvokesWithMovedFunctorDestroyedOnTargetThread) { + std::unique_ptr<rtc::Thread> background_thread(rtc::Thread::Create()); + background_thread->Start(); + + Event event; + bool was_invoked_on_background_thread = false; + DestructionFunctor functor(background_thread.get(), + &was_invoked_on_background_thread, &event); + background_thread->PostTask(std::move(functor)); + event.Wait(Event::kForever); + + EXPECT_TRUE(was_invoked_on_background_thread); +} + +TEST(ThreadPostTaskTest, + InvokesWithReferencedFunctorShouldCopyAndDestroyedOnTargetThread) { + std::unique_ptr<rtc::Thread> background_thread(rtc::Thread::Create()); + background_thread->Start(); + + Event event; + bool was_invoked_on_background_thread = false; + DestructionFunctor functor(background_thread.get(), + &was_invoked_on_background_thread, &event); + DestructionFunctor& functor_ref = functor; + background_thread->PostTask(functor_ref); + event.Wait(Event::kForever); + + EXPECT_TRUE(was_invoked_on_background_thread); +} + +TEST(ThreadPostTaskTest, InvokesOnBackgroundThread) { + std::unique_ptr<rtc::Thread> background_thread(rtc::Thread::Create()); + background_thread->Start(); + + Event event; + bool was_invoked_on_background_thread = false; + Thread* background_thread_ptr = background_thread.get(); + background_thread->PostTask( + [background_thread_ptr, &was_invoked_on_background_thread, &event] { + was_invoked_on_background_thread = background_thread_ptr->IsCurrent(); + event.Set(); + }); + event.Wait(Event::kForever); + + EXPECT_TRUE(was_invoked_on_background_thread); +} + +TEST(ThreadPostTaskTest, InvokesAsynchronously) { + std::unique_ptr<rtc::Thread> background_thread(rtc::Thread::Create()); + background_thread->Start(); + + // The first event ensures that SendSingleMessage() is not blocking this + // thread. The second event ensures that the message is processed. + Event event_set_by_test_thread; + Event event_set_by_background_thread; + background_thread->PostTask( + [&event_set_by_test_thread, &event_set_by_background_thread] { + WaitAndSetEvent(&event_set_by_test_thread, + &event_set_by_background_thread); + }); + event_set_by_test_thread.Set(); + event_set_by_background_thread.Wait(Event::kForever); +} + +TEST(ThreadPostTaskTest, InvokesInPostedOrder) { + std::unique_ptr<rtc::Thread> background_thread(rtc::Thread::Create()); + background_thread->Start(); + + Event first; + Event second; + Event third; + Event fourth; + + background_thread->PostTask( + [&first, &second] { WaitAndSetEvent(&first, &second); }); + background_thread->PostTask( + [&second, &third] { WaitAndSetEvent(&second, &third); }); + background_thread->PostTask( + [&third, &fourth] { WaitAndSetEvent(&third, &fourth); }); + + // All tasks have been posted before the first one is unblocked. + first.Set(); + // Only if the chain is invoked in posted order will the last event be set. + fourth.Wait(Event::kForever); +} + +TEST(ThreadPostDelayedTaskTest, InvokesAsynchronously) { + std::unique_ptr<rtc::Thread> background_thread(rtc::Thread::Create()); + background_thread->Start(); + + // The first event ensures that SendSingleMessage() is not blocking this + // thread. The second event ensures that the message is processed. + Event event_set_by_test_thread; + Event event_set_by_background_thread; + background_thread->PostDelayedTask( + [&event_set_by_test_thread, &event_set_by_background_thread] { + WaitAndSetEvent(&event_set_by_test_thread, + &event_set_by_background_thread); + }, + TimeDelta::Millis(10)); + event_set_by_test_thread.Set(); + event_set_by_background_thread.Wait(Event::kForever); +} + +TEST(ThreadPostDelayedTaskTest, InvokesInDelayOrder) { + ScopedFakeClock clock; + std::unique_ptr<rtc::Thread> background_thread(rtc::Thread::Create()); + background_thread->Start(); + + Event first; + Event second; + Event third; + Event fourth; + + background_thread->PostDelayedTask( + [&third, &fourth] { WaitAndSetEvent(&third, &fourth); }, + TimeDelta::Millis(11)); + background_thread->PostDelayedTask( + [&first, &second] { WaitAndSetEvent(&first, &second); }, + TimeDelta::Millis(9)); + background_thread->PostDelayedTask( + [&second, &third] { WaitAndSetEvent(&second, &third); }, + TimeDelta::Millis(10)); + + // All tasks have been posted before the first one is unblocked. + first.Set(); + // Only if the chain is invoked in delay order will the last event be set. + clock.AdvanceTime(TimeDelta::Millis(11)); + EXPECT_TRUE(fourth.Wait(0)); +} + +TEST(ThreadPostDelayedTaskTest, IsCurrentTaskQueue) { + auto current_tq = webrtc::TaskQueueBase::Current(); + { + std::unique_ptr<rtc::Thread> thread(rtc::Thread::Create()); + thread->WrapCurrent(); + EXPECT_EQ(webrtc::TaskQueueBase::Current(), + static_cast<webrtc::TaskQueueBase*>(thread.get())); + thread->UnwrapCurrent(); + } + EXPECT_EQ(webrtc::TaskQueueBase::Current(), current_tq); +} + +class ThreadFactory : public webrtc::TaskQueueFactory { + public: + std::unique_ptr<webrtc::TaskQueueBase, webrtc::TaskQueueDeleter> + CreateTaskQueue(absl::string_view /* name */, + Priority /*priority*/) const override { + std::unique_ptr<Thread> thread = Thread::Create(); + thread->Start(); + return std::unique_ptr<webrtc::TaskQueueBase, webrtc::TaskQueueDeleter>( + thread.release()); + } +}; + +using ::webrtc::TaskQueueTest; + +INSTANTIATE_TEST_SUITE_P(RtcThread, + TaskQueueTest, + ::testing::Values(std::make_unique<ThreadFactory>)); + +} // namespace +} // namespace rtc |