diff --git a/rtc_base/BUILD.gn b/rtc_base/BUILD.gn index 2ef79fa6c4..827559994b 100644 --- a/rtc_base/BUILD.gn +++ b/rtc_base/BUILD.gn @@ -526,6 +526,26 @@ rtc_source_set("rtc_task_queue_api") { ] } +rtc_source_set("rtc_cancelable_task") { + sources = [ + "cancelable_periodic_task.h", + "cancelable_task_handle.cc", + "cancelable_task_handle.h", + ] + deps = [ + ":checks", + ":logging", + ":macromagic", + ":ptr_util", + ":refcount", + ":rtc_task_queue", + ":safe_conversions", + ":sequenced_task_checker", + ":thread_checker", + "//third_party/abseil-cpp/absl/memory", + ] +} + if (rtc_enable_libevent) { rtc_source_set("rtc_task_queue_libevent") { visibility = [ ":rtc_task_queue_impl" ] @@ -1178,12 +1198,14 @@ if (rtc_include_tests) { testonly = true sources = [ + "cancelable_periodic_task_unittest.cc", "task_queue_unittest.cc", ] deps = [ ":rtc_base_approved", ":rtc_base_tests_main", ":rtc_base_tests_utils", + ":rtc_cancelable_task", ":rtc_task_queue", ":rtc_task_queue_for_test", "../test:test_support", diff --git a/rtc_base/cancelable_periodic_task.h b/rtc_base/cancelable_periodic_task.h new file mode 100644 index 0000000000..d11301597a --- /dev/null +++ b/rtc_base/cancelable_periodic_task.h @@ -0,0 +1,83 @@ +/* + * Copyright 2018 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. + */ + +#ifndef RTC_BASE_CANCELABLE_PERIODIC_TASK_H_ +#define RTC_BASE_CANCELABLE_PERIODIC_TASK_H_ + +#include +#include +#include + +#include "absl/memory/memory.h" +#include "rtc_base/cancelable_task_handle.h" +#include "rtc_base/checks.h" +#include "rtc_base/logging.h" +#include "rtc_base/numerics/safe_conversions.h" + +namespace rtc { +namespace cancelable_periodic_task_internal { +// CancelablePeriodicTask runs a closure multiple times with delay decided +// by the return value of the closure itself. +// The task can be canceled with the handle returned by GetCancelationHandle(). +// Note that the task can only be canceled on the task queue where it runs. +template +class CancelablePeriodicTask final : public BaseCancelableTask { + public: + // |closure| should return time in ms until next run. + explicit CancelablePeriodicTask(Closure&& closure) + : closure_(std::forward(closure)) {} + CancelablePeriodicTask(const CancelablePeriodicTask&) = delete; + CancelablePeriodicTask& operator=(const CancelablePeriodicTask&) = delete; + ~CancelablePeriodicTask() override = default; + + private: + bool Run() override { + // See QueuedTask::Run documentaion for return values meaning. + if (BaseCancelableTask::Canceled()) + return true; // Caller retains ownership of `this`, and will destroy it. + // Run the actual task. + auto delay = closure_(); + // Convert closure_() return type into uint32_t. + uint32_t delay_ms = 0; + if (rtc::IsValueInRangeForNumericType(delay)) { + delay_ms = static_cast(delay); + } else { + // Log and recover in production. + RTC_LOG(LS_ERROR) << "Invalid delay until next run: " << delay; + delay_ms = rtc::saturated_cast(delay); + // But crash in debug. + RTC_DCHECK(false); + } + // Reschedule. + auto owned_task = absl::WrapUnique(this); + if (delay_ms == 0) + TaskQueue::Current()->PostTask(std::move(owned_task)); + else + TaskQueue::Current()->PostDelayedTask(std::move(owned_task), delay_ms); + return false; // Caller will release ownership of `this`. + } + + Closure closure_; +}; +} // namespace cancelable_periodic_task_internal + +template +std::unique_ptr CreateCancelablePeriodicTask( + Closure&& closure) { + using CleanedClosure = typename std::remove_cv< + typename std::remove_reference::type>::type; + return absl::make_unique>( + std::forward(closure)); +} + +} // namespace rtc + +#endif // RTC_BASE_CANCELABLE_PERIODIC_TASK_H_ diff --git a/rtc_base/cancelable_periodic_task_unittest.cc b/rtc_base/cancelable_periodic_task_unittest.cc new file mode 100644 index 0000000000..fe27ea7d48 --- /dev/null +++ b/rtc_base/cancelable_periodic_task_unittest.cc @@ -0,0 +1,228 @@ +/* + * Copyright 2018 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/cancelable_periodic_task.h" + +#include "rtc_base/event.h" +#include "test/gmock.h" + +namespace { + +using ::testing::AtLeast; +using ::testing::Invoke; +using ::testing::NiceMock; +using ::testing::MockFunction; +using ::testing::Return; + +constexpr int kTimeoutMs = 1000; + +class MockClosure { + public: + MOCK_METHOD0(Call, int()); + MOCK_METHOD0(Delete, void()); +}; + +class MoveOnlyClosure { + public: + explicit MoveOnlyClosure(MockClosure* mock) : mock_(mock) {} + MoveOnlyClosure(const MoveOnlyClosure&) = delete; + MoveOnlyClosure(MoveOnlyClosure&& other) : mock_(other.mock_) { + other.mock_ = nullptr; + } + ~MoveOnlyClosure() { + if (mock_) + mock_->Delete(); + } + int operator()() { return mock_->Call(); } + + private: + MockClosure* mock_; +}; + +class CopyableClosure { + public: + explicit CopyableClosure(MockClosure* mock) : mock_(mock) {} + CopyableClosure(const CopyableClosure& other) : mock_(other.mock_) {} + ~CopyableClosure() { + if (mock_) { + mock_->Delete(); + mock_ = nullptr; + } + } + int operator()() { return mock_->Call(); } + + private: + MockClosure* mock_; +}; + +TEST(CancelablePeriodicTaskTest, CanCallCancelOnEmptyHandle) { + rtc::CancelableTaskHandle handle; + handle.Cancel(); +} + +TEST(CancelablePeriodicTaskTest, CancelTaskBeforeItRuns) { + rtc::Event done(false, false); + MockClosure mock; + EXPECT_CALL(mock, Call).Times(0); + EXPECT_CALL(mock, Delete).WillOnce(Invoke([&done] { done.Set(); })); + + rtc::TaskQueue task_queue("queue"); + + auto task = rtc::CreateCancelablePeriodicTask(MoveOnlyClosure(&mock)); + rtc::CancelableTaskHandle handle = task->GetCancellationHandle(); + task_queue.PostTask([handle] { handle.Cancel(); }); + task_queue.PostTask(std::move(task)); + + EXPECT_TRUE(done.Wait(kTimeoutMs)); +} + +TEST(CancelablePeriodicTaskTest, CancelDelayedTaskBeforeItRuns) { + rtc::Event done(false, false); + MockClosure mock; + EXPECT_CALL(mock, Call).Times(0); + EXPECT_CALL(mock, Delete).WillOnce(Invoke([&done] { done.Set(); })); + + rtc::TaskQueue task_queue("queue"); + + auto task = rtc::CreateCancelablePeriodicTask(MoveOnlyClosure(&mock)); + rtc::CancelableTaskHandle handle = task->GetCancellationHandle(); + task_queue.PostDelayedTask(std::move(task), 100); + task_queue.PostTask([handle] { handle.Cancel(); }); + + EXPECT_TRUE(done.Wait(kTimeoutMs)); +} + +TEST(CancelablePeriodicTaskTest, CancelTaskAfterItRuns) { + rtc::Event done(false, false); + MockClosure mock; + EXPECT_CALL(mock, Call).WillOnce(Return(100)); + EXPECT_CALL(mock, Delete).WillOnce(Invoke([&done] { done.Set(); })); + + rtc::TaskQueue task_queue("queue"); + + auto task = rtc::CreateCancelablePeriodicTask(MoveOnlyClosure(&mock)); + rtc::CancelableTaskHandle handle = task->GetCancellationHandle(); + task_queue.PostTask(std::move(task)); + task_queue.PostTask([handle] { handle.Cancel(); }); + + EXPECT_TRUE(done.Wait(kTimeoutMs)); +} + +TEST(CancelablePeriodicTaskTest, ZeroReturnValueRepostsTheTask) { + NiceMock closure; + rtc::Event done(false, false); + EXPECT_CALL(closure, Call()).WillOnce(Return(0)).WillOnce(Invoke([&done] { + done.Set(); + return kTimeoutMs; + })); + rtc::TaskQueue task_queue("queue"); + task_queue.PostTask( + rtc::CreateCancelablePeriodicTask(MoveOnlyClosure(&closure))); + EXPECT_TRUE(done.Wait(kTimeoutMs)); +} + +TEST(CancelablePeriodicTaskTest, StartPeriodicTask) { + MockFunction closure; + rtc::Event done(false, false); + EXPECT_CALL(closure, Call()) + .WillOnce(Return(20)) + .WillOnce(Return(20)) + .WillOnce(Invoke([&done] { + done.Set(); + return kTimeoutMs; + })); + rtc::TaskQueue task_queue("queue"); + task_queue.PostTask( + rtc::CreateCancelablePeriodicTask(closure.AsStdFunction())); + EXPECT_TRUE(done.Wait(kTimeoutMs)); +} + +// Validates perfect forwarding doesn't keep reference to deleted copy. +TEST(CancelablePeriodicTaskTest, CreateWithCopyOfAClosure) { + rtc::Event done(false, false); + MockClosure mock; + EXPECT_CALL(mock, Call).WillOnce(Invoke([&done] { + done.Set(); + return kTimeoutMs; + })); + EXPECT_CALL(mock, Delete).Times(AtLeast(2)); + CopyableClosure closure(&mock); + std::unique_ptr task; + { + CopyableClosure copy = closure; + task = rtc::CreateCancelablePeriodicTask(copy); + } + + rtc::TaskQueue task_queue("queue"); + task_queue.PostTask(std::move(task)); + EXPECT_TRUE(done.Wait(kTimeoutMs)); +} + +TEST(CancelablePeriodicTaskTest, DeletingHandleDoesntStopTheTask) { + rtc::Event run(false, false); + rtc::TaskQueue task_queue("queue"); + auto task = rtc::CreateCancelablePeriodicTask(([&] { + run.Set(); + return kTimeoutMs; + })); + rtc::CancelableTaskHandle handle = task->GetCancellationHandle(); + handle = {}; // delete the handle. + task_queue.PostTask(std::move(task)); + EXPECT_TRUE(run.Wait(kTimeoutMs)); +} + +// Example to test there are no thread races and use after free for suggested +// typical usage of the CancelablePeriodicTask +TEST(CancelablePeriodicTaskTest, Example) { + class ObjectOnTaskQueue { + public: + void DoPeriodicTask() {} + int TimeUntilNextRunMs() { return 100; } + + rtc::CancelableTaskHandle StartPeriodicTask(rtc::TaskQueue* task_queue) { + auto periodic_task = rtc::CreateCancelablePeriodicTask([this] { + DoPeriodicTask(); + return TimeUntilNextRunMs(); + }); + rtc::CancelableTaskHandle handle = periodic_task->GetCancellationHandle(); + task_queue->PostTask(std::move(periodic_task)); + return handle; + } + }; + + rtc::TaskQueue task_queue("queue"); + + auto object = absl::make_unique(); + // Create and start the periodic task. + rtc::CancelableTaskHandle handle = object->StartPeriodicTask(&task_queue); + + // Restart the task + task_queue.PostTask([handle] { handle.Cancel(); }); + handle = object->StartPeriodicTask(&task_queue); + + // Stop the task and destroy the object. + struct Destructor { + void operator()() { + // Cancel must be run on the task_queue, but if task failed to start + // because of task queue destruction, there is no need to run Cancel. + handle.Cancel(); + } + // Destruction will happen either on the task queue or because task + // queue is destroyed. + + std::unique_ptr object; + rtc::CancelableTaskHandle handle; + }; + task_queue.PostTask(Destructor{std::move(object), std::move(handle)}); + // Do not wait for the Destructor closure in order to create a race between + // task queue destruction and running the Desctructor closure. +} + +} // namespace diff --git a/rtc_base/cancelable_task_handle.cc b/rtc_base/cancelable_task_handle.cc new file mode 100644 index 0000000000..372e766e9d --- /dev/null +++ b/rtc_base/cancelable_task_handle.cc @@ -0,0 +1,85 @@ +/* + * Copyright 2018 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/cancelable_task_handle.h" + +#include + +#include "rtc_base/refcounter.h" +#include "rtc_base/sequenced_task_checker.h" +#include "rtc_base/thread_annotations.h" +#include "rtc_base/thread_checker.h" + +namespace rtc { + +class CancelableTaskHandle::CancellationToken { + public: + CancellationToken() : canceled_(false), ref_count_(0) { checker_.Detach(); } + CancellationToken(const CancellationToken&) = delete; + CancellationToken& operator=(const CancellationToken&) = delete; + + void Cancel() { + RTC_DCHECK_RUN_ON(&checker_); + canceled_ = true; + } + + bool Canceled() { + RTC_DCHECK_RUN_ON(&checker_); + return canceled_; + } + + void AddRef() { ref_count_.IncRef(); } + + void Release() { + if (ref_count_.DecRef() == rtc::RefCountReleaseStatus::kDroppedLastRef) + delete this; + } + + private: + ~CancellationToken() = default; + + rtc::SequencedTaskChecker checker_; + bool canceled_ RTC_GUARDED_BY(checker_); + webrtc::webrtc_impl::RefCounter ref_count_; +}; + +CancelableTaskHandle::CancelableTaskHandle() = default; +CancelableTaskHandle::CancelableTaskHandle(const CancelableTaskHandle&) = + default; +CancelableTaskHandle::CancelableTaskHandle(CancelableTaskHandle&&) = default; +CancelableTaskHandle& CancelableTaskHandle::operator=( + const CancelableTaskHandle&) = default; +CancelableTaskHandle& CancelableTaskHandle::operator=(CancelableTaskHandle&&) = + default; +CancelableTaskHandle::~CancelableTaskHandle() = default; + +void CancelableTaskHandle::Cancel() const { + if (cancellation_token_.get() != nullptr) + cancellation_token_->Cancel(); +} + +CancelableTaskHandle::CancelableTaskHandle( + rtc::scoped_refptr cancellation_token) + : cancellation_token_(std::move(cancellation_token)) {} + +BaseCancelableTask::~BaseCancelableTask() = default; + +CancelableTaskHandle BaseCancelableTask::GetCancellationHandle() const { + return CancelableTaskHandle(cancellation_token_); +} + +BaseCancelableTask::BaseCancelableTask() + : cancellation_token_(new CancelableTaskHandle::CancellationToken) {} + +bool BaseCancelableTask::Canceled() const { + return cancellation_token_->Canceled(); +} + +} // namespace rtc diff --git a/rtc_base/cancelable_task_handle.h b/rtc_base/cancelable_task_handle.h new file mode 100644 index 0000000000..3b1f0d5a6b --- /dev/null +++ b/rtc_base/cancelable_task_handle.h @@ -0,0 +1,65 @@ +/* + * Copyright 2018 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. + */ + +#ifndef RTC_BASE_CANCELABLE_TASK_HANDLE_H_ +#define RTC_BASE_CANCELABLE_TASK_HANDLE_H_ + +#include "rtc_base/scoped_ref_ptr.h" +#include "rtc_base/task_queue.h" + +namespace rtc { + +class BaseCancelableTask; + +// Allows to cancel a cancelable task. Non-empty handle can be acquired by +// calling GetCancellationHandle() on a cancelable task. +class CancelableTaskHandle { + public: + // This class is copyable and cheaply movable. + CancelableTaskHandle(); + CancelableTaskHandle(const CancelableTaskHandle&); + CancelableTaskHandle(CancelableTaskHandle&&); + CancelableTaskHandle& operator=(const CancelableTaskHandle&); + CancelableTaskHandle& operator=(CancelableTaskHandle&&); + // Deleting the handler doesn't Cancel the task. + ~CancelableTaskHandle(); + + // Prevents the cancelable task to run. + // Must be executed on the same task queue as the task itself. + void Cancel() const; + + private: + friend class BaseCancelableTask; + class CancellationToken; + explicit CancelableTaskHandle( + rtc::scoped_refptr cancelation_token); + + rtc::scoped_refptr cancellation_token_; +}; + +class BaseCancelableTask : public QueuedTask { + public: + ~BaseCancelableTask() override; + + CancelableTaskHandle GetCancellationHandle() const; + + protected: + BaseCancelableTask(); + + bool Canceled() const; + + private: + rtc::scoped_refptr + cancellation_token_; +}; + +} // namespace rtc + +#endif // RTC_BASE_CANCELABLE_TASK_HANDLE_H_