diff options
Diffstat (limited to 'third_party/libwebrtc/rtc_base/thread_unittest.cc')
| -rw-r--r-- | third_party/libwebrtc/rtc_base/thread_unittest.cc | 852 |
1 files changed, 852 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..cd733db2cd --- /dev/null +++ b/third_party/libwebrtc/rtc_base/thread_unittest.cc @@ -0,0 +1,852 @@ +/* + * 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/field_trials_view.h" +#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_udp_socket.h" +#include "rtc_base/checks.h" +#include "rtc_base/event.h" +#include "rtc_base/fake_clock.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/ref_counted_object.h" +#include "rtc_base/socket_address.h" +#include "rtc_base/synchronization/mutex.h" +#include "rtc_base/third_party/sigslot/sigslot.h" +#include "test/gmock.h" +#include "test/testsupport/rtc_expect_death.h" + +#if defined(WEBRTC_WIN) +#include <comdef.h> // NOLINT + +#endif + +namespace rtc { +namespace { + +using ::testing::ElementsAre; +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; +}; + +// Receives messages and sends on a socket. +class MessageClient : public TestGenerator { + public: + MessageClient(Thread* pth, Socket* socket) : socket_(socket) {} + + ~MessageClient() { delete socket_; } + + void OnValue(int value) { + int result = Next(value); + EXPECT_GE(socket_->Send(&result, sizeof(result)), 0); + } + + private: + Socket* socket_; +}; + +// 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, + MessageClient* 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_->PostDelayedTask([post_handler_ = post_handler_, + result] { post_handler_->OnValue(result); }, + TimeDelta::Millis(200)); + } + + private: + AsyncUDPSocket* socket_; + Thread* post_thread_; + MessageClient* post_handler_; +}; + +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_; +}; + +// See: https://code.google.com/p/webrtc/issues/detail?id=2409 +TEST(ThreadTest, DISABLED_Main) { + rtc::AutoThread main_thread; + 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->PostDelayedTask([&msg_client] { msg_client.OnValue(1); }, + TimeDelta::Millis(100)); + + // 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 `BlockingCall` serves a purpose in some configurations + // (and should not be used a general way to call methods on the same thread). + current.BlockingCall([]() {}); + 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->BlockingCall([]() { 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.BlockingCall([]() {}); + } + 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.BlockingCall([]() {}); + } + // 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, BlockingCall) { + // Create and start the thread. + auto thread = Thread::CreateWithSocketServer(); + thread->Start(); + // Try calling functors. + EXPECT_EQ(42, thread->BlockingCall([] { return 42; })); + bool called = false; + thread->BlockingCall([&] { called = true; }); + EXPECT_TRUE(called); + + // Try calling bare functions. + struct LocalFuncs { + static int Func1() { return 999; } + static void Func2() {} + }; + EXPECT_EQ(999, thread->BlockingCall(&LocalFuncs::Func1)); + thread->BlockingCall(&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->BlockingCall([main_thread] { + RTC_EXPECT_DEATH(main_thread->BlockingCall([] {}), "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->BlockingCall([&] { + third->BlockingCall( + [&] { RTC_EXPECT_DEATH(first->BlockingCall([] {}), "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, ThreeThreadsBlockingCall) { + 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->BlockingCall([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(Thread* thread1, + Thread* thread2, + LockedBool* out) { + LockedBool async_invoked(false); + + thread1->PostTask([&async_invoked, thread2, out] { + SetAndInvokeSet(&async_invoked, thread2, out); + }); + + EXPECT_TRUE_WAIT(async_invoked.Get(), 2000); + } + }; + + 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->BlockingCall([thread_c_ptr, thread_a, &thread_a_called] { + LocalFuncs::AsyncInvokeSetAndWait(thread_c_ptr, thread_a, &thread_a_called); + }); + EXPECT_FALSE(thread_a_called.Get()); + + EXPECT_TRUE_WAIT(thread_a_called.Get(), 2000); +} + +static void DelayedPostsWithIdenticalTimesAreProcessedInFifoOrder( + FakeClock& clock, + Thread& q) { + std::vector<int> run_order; + + Event done; + int64_t now = TimeMillis(); + q.PostDelayedTask([&] { run_order.push_back(3); }, TimeDelta::Millis(3)); + q.PostDelayedTask([&] { run_order.push_back(0); }, TimeDelta::Millis(1)); + q.PostDelayedTask([&] { run_order.push_back(1); }, TimeDelta::Millis(2)); + q.PostDelayedTask([&] { run_order.push_back(4); }, TimeDelta::Millis(3)); + q.PostDelayedTask([&] { run_order.push_back(2); }, TimeDelta::Millis(2)); + q.PostDelayedTask([&] { done.Set(); }, TimeDelta::Millis(4)); + // Validate time was frozen while tasks were posted. + RTC_DCHECK_EQ(TimeMillis(), now); + + // Change time to make all tasks ready to run and wait for them. + clock.AdvanceTime(TimeDelta::Millis(4)); + ASSERT_TRUE(done.Wait(TimeDelta::Seconds(1))); + + EXPECT_THAT(run_order, ElementsAre(0, 1, 2, 3, 4)); +} + +TEST(ThreadTest, DelayedPostsWithIdenticalTimesAreProcessedInFifoOrder) { + ScopedBaseFakeClock clock; + Thread q(CreateDefaultSocketServer(), true); + q.Start(); + DelayedPostsWithIdenticalTimesAreProcessedInFifoOrder(clock, q); + + NullSocketServer nullss; + Thread q_nullss(&nullss, true); + q_nullss.Start(); + DelayedPostsWithIdenticalTimesAreProcessedInFifoOrder(clock, q_nullss); +} + +// 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(); +} + +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(TimeDelta::Zero())); +} + +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()); + } +}; + +std::unique_ptr<webrtc::TaskQueueFactory> CreateDefaultThreadFactory( + const webrtc::FieldTrialsView*) { + return std::make_unique<ThreadFactory>(); +} + +using ::webrtc::TaskQueueTest; + +INSTANTIATE_TEST_SUITE_P(RtcThread, + TaskQueueTest, + ::testing::Values(CreateDefaultThreadFactory)); + +} // namespace +} // namespace rtc |