Refactor RepeatingTask to use absl::AnyInvocable functions of TaskQueue
Bug: webrtc:14245
Change-Id: Ie02755a4bb732cc25b3a22511e6d8920fc434c65
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/267847
Reviewed-by: Tomas Gunnarsson <tommi@webrtc.org>
Commit-Queue: Danil Chapovalov <danilchap@webrtc.org>
Cr-Commit-Position: refs/heads/main@{#37461}
diff --git a/rtc_base/task_utils/BUILD.gn b/rtc_base/task_utils/BUILD.gn
index c01e82e..732f6d7 100644
--- a/rtc_base/task_utils/BUILD.gn
+++ b/rtc_base/task_utils/BUILD.gn
@@ -19,15 +19,11 @@
"../../api:sequence_checker",
"../../api/task_queue",
"../../api/task_queue:pending_task_safety_flag",
- "../../api/task_queue:to_queued_task",
"../../api/units:time_delta",
"../../api/units:timestamp",
"../../system_wrappers:system_wrappers",
]
- absl_deps = [
- "//third_party/abseil-cpp/absl/functional:any_invocable",
- "//third_party/abseil-cpp/absl/memory",
- ]
+ absl_deps = [ "//third_party/abseil-cpp/absl/functional:any_invocable" ]
}
rtc_library("pending_task_safety_flag") {
@@ -51,9 +47,11 @@
"..:task_queue_for_test",
"../../api/task_queue",
"../../api/task_queue:to_queued_task",
+ "../../api/units:time_delta",
"../../api/units:timestamp",
"../../system_wrappers:system_wrappers",
"../../test:test_support",
]
+ absl_deps = [ "//third_party/abseil-cpp/absl/functional:any_invocable" ]
}
}
diff --git a/rtc_base/task_utils/repeating_task.cc b/rtc_base/task_utils/repeating_task.cc
index fa6ae79..222ab1a 100644
--- a/rtc_base/task_utils/repeating_task.cc
+++ b/rtc_base/task_utils/repeating_task.cc
@@ -11,16 +11,13 @@
#include "rtc_base/task_utils/repeating_task.h"
#include "absl/functional/any_invocable.h"
-#include "absl/memory/memory.h"
#include "api/task_queue/pending_task_safety_flag.h"
-#include "api/task_queue/to_queued_task.h"
#include "rtc_base/logging.h"
-#include "rtc_base/time_utils.h"
namespace webrtc {
namespace {
-class RepeatingTask : public QueuedTask {
+class RepeatingTask {
public:
RepeatingTask(TaskQueueBase* task_queue,
TaskQueueBase::DelayPrecision precision,
@@ -28,11 +25,13 @@
absl::AnyInvocable<TimeDelta()> task,
Clock* clock,
rtc::scoped_refptr<PendingTaskSafetyFlag> alive_flag);
- ~RepeatingTask() override = default;
+ RepeatingTask(RepeatingTask&&) = default;
+ RepeatingTask& operator=(RepeatingTask&&) = delete;
+ ~RepeatingTask() = default;
+
+ void operator()() &&;
private:
- bool Run() final;
-
TaskQueueBase* const task_queue_;
const TaskQueueBase::DelayPrecision precision_;
Clock* const clock_;
@@ -57,33 +56,27 @@
next_run_time_(clock_->CurrentTime() + first_delay),
alive_flag_(std::move(alive_flag)) {}
-bool RepeatingTask::Run() {
+void RepeatingTask::operator()() && {
RTC_DCHECK_RUN_ON(task_queue_);
- // Return true to tell the TaskQueue to destruct this object.
if (!alive_flag_->alive())
- return true;
+ return;
webrtc_repeating_task_impl::RepeatingTaskImplDTraceProbeRun();
TimeDelta delay = task_();
RTC_DCHECK_GE(delay, TimeDelta::Zero());
// A delay of +infinity means that the task should not be run again.
- // Alternatively, the closure might have stopped this task. In either which
- // case we return true to destruct this object.
+ // Alternatively, the closure might have stopped this task.
if (delay.IsPlusInfinity() || !alive_flag_->alive())
- return true;
+ return;
TimeDelta lost_time = clock_->CurrentTime() - next_run_time_;
next_run_time_ += delay;
delay -= lost_time;
delay = std::max(delay, TimeDelta::Zero());
- task_queue_->PostDelayedTaskWithPrecision(precision_, absl::WrapUnique(this),
- delay.ms());
-
- // Return false to tell the TaskQueue to not destruct this object since we
- // have taken ownership with absl::WrapUnique.
- return false;
+ task_queue_->PostDelayedTaskWithPrecision(precision_, std::move(*this),
+ delay);
}
} // namespace
@@ -95,9 +88,8 @@
Clock* clock) {
auto alive_flag = PendingTaskSafetyFlag::CreateDetached();
webrtc_repeating_task_impl::RepeatingTaskHandleDTraceProbeStart();
- task_queue->PostTask(
- std::make_unique<RepeatingTask>(task_queue, precision, TimeDelta::Zero(),
- std::move(closure), clock, alive_flag));
+ task_queue->PostTask(RepeatingTask(task_queue, precision, TimeDelta::Zero(),
+ std::move(closure), clock, alive_flag));
return RepeatingTaskHandle(std::move(alive_flag));
}
@@ -113,9 +105,9 @@
webrtc_repeating_task_impl::RepeatingTaskHandleDTraceProbeDelayedStart();
task_queue->PostDelayedTaskWithPrecision(
precision,
- std::make_unique<RepeatingTask>(task_queue, precision, first_delay,
- std::move(closure), clock, alive_flag),
- first_delay.ms());
+ RepeatingTask(task_queue, precision, first_delay, std::move(closure),
+ clock, alive_flag),
+ first_delay);
return RepeatingTaskHandle(std::move(alive_flag));
}
diff --git a/rtc_base/task_utils/repeating_task_unittest.cc b/rtc_base/task_utils/repeating_task_unittest.cc
index ca33446..4cb6816 100644
--- a/rtc_base/task_utils/repeating_task_unittest.cc
+++ b/rtc_base/task_utils/repeating_task_unittest.cc
@@ -13,9 +13,9 @@
#include <atomic>
#include <memory>
-#include "api/task_queue/queued_task.h"
+#include "absl/functional/any_invocable.h"
#include "api/task_queue/task_queue_base.h"
-#include "api/task_queue/to_queued_task.h"
+#include "api/units/time_delta.h"
#include "api/units/timestamp.h"
#include "rtc_base/event.h"
#include "rtc_base/task_queue_for_test.h"
@@ -46,10 +46,14 @@
MockTaskQueue() : task_queue_setter_(this) {}
MOCK_METHOD(void, Delete, (), (override));
- MOCK_METHOD(void, PostTask, (std::unique_ptr<QueuedTask> task), (override));
+ MOCK_METHOD(void, PostTask, (absl::AnyInvocable<void() &&>), (override));
MOCK_METHOD(void,
PostDelayedTask,
- (std::unique_ptr<QueuedTask> task, uint32_t milliseconds),
+ (absl::AnyInvocable<void() &&>, TimeDelta),
+ (override));
+ MOCK_METHOD(void,
+ PostDelayedHighPrecisionTask,
+ (absl::AnyInvocable<void() &&>, TimeDelta),
(override));
private:
@@ -63,45 +67,41 @@
void Delete() override {}
- void PostTask(std::unique_ptr<QueuedTask> task) override {
+ void PostTask(absl::AnyInvocable<void() &&> task) override {
last_task_ = std::move(task);
last_precision_ = absl::nullopt;
- last_delay_ = 0;
+ last_delay_ = TimeDelta::Zero();
}
- void PostDelayedTask(std::unique_ptr<QueuedTask> task,
- uint32_t milliseconds) override {
+ void PostDelayedTask(absl::AnyInvocable<void() &&> task,
+ TimeDelta delay) override {
last_task_ = std::move(task);
last_precision_ = TaskQueueBase::DelayPrecision::kLow;
- last_delay_ = milliseconds;
+ last_delay_ = delay;
}
- void PostDelayedHighPrecisionTask(std::unique_ptr<QueuedTask> task,
- uint32_t milliseconds) override {
+ void PostDelayedHighPrecisionTask(absl::AnyInvocable<void() &&> task,
+ TimeDelta delay) override {
last_task_ = std::move(task);
last_precision_ = TaskQueueBase::DelayPrecision::kHigh;
- last_delay_ = milliseconds;
+ last_delay_ = delay;
}
bool AdvanceTimeAndRunLastTask() {
EXPECT_TRUE(last_task_);
- EXPECT_TRUE(last_delay_);
- clock_->AdvanceTimeMilliseconds(last_delay_.value_or(0));
- last_delay_.reset();
+ EXPECT_TRUE(last_delay_.IsFinite());
+ clock_->AdvanceTime(last_delay_);
+ last_delay_ = TimeDelta::MinusInfinity();
auto task = std::move(last_task_);
- bool delete_task = task->Run();
- if (!delete_task) {
- // If the task should not be deleted then just release it.
- task.release();
- }
- return delete_task;
+ std::move(task)();
+ return last_task_ == nullptr;
}
bool IsTaskQueued() { return !!last_task_; }
- uint32_t last_delay() const {
- EXPECT_TRUE(last_delay_.has_value());
- return last_delay_.value_or(-1);
+ TimeDelta last_delay() const {
+ EXPECT_TRUE(last_delay_.IsFinite());
+ return last_delay_;
}
absl::optional<TaskQueueBase::DelayPrecision> last_precision() const {
@@ -111,8 +111,8 @@
private:
CurrentTaskQueueSetter task_queue_setter_;
SimulatedClock* clock_;
- std::unique_ptr<QueuedTask> last_task_;
- absl::optional<uint32_t> last_delay_;
+ absl::AnyInvocable<void() &&> last_task_;
+ TimeDelta last_delay_ = TimeDelta::MinusInfinity();
absl::optional<TaskQueueBase::DelayPrecision> last_precision_;
};
@@ -146,16 +146,19 @@
SimulatedClock clock(Timestamp::Zero());
FakeTaskQueue task_queue(&clock);
std::atomic_int counter(0);
- auto handle = RepeatingTaskHandle::Start(&task_queue, [&] {
- counter++;
- return kShortInterval;
- });
- EXPECT_EQ(task_queue.last_delay(), 0u);
+ auto handle = RepeatingTaskHandle::Start(
+ &task_queue,
+ [&] {
+ counter++;
+ return kShortInterval;
+ },
+ TaskQueueBase::DelayPrecision::kLow, &clock);
+ EXPECT_EQ(task_queue.last_delay(), TimeDelta::Zero());
EXPECT_FALSE(task_queue.AdvanceTimeAndRunLastTask());
EXPECT_EQ(counter.load(), 1);
// The handle reposted at the short interval.
- EXPECT_EQ(task_queue.last_delay(), kShortInterval.ms());
+ EXPECT_EQ(task_queue.last_delay(), kShortInterval);
// Stop the handle. This prevernts the counter from incrementing.
handle.Stop();
@@ -182,12 +185,12 @@
},
TaskQueueBase::DelayPrecision::kLow, &clock);
- EXPECT_EQ(task_queue.last_delay(), 0u);
+ EXPECT_EQ(task_queue.last_delay(), TimeDelta::Zero());
EXPECT_FALSE(task_queue.AdvanceTimeAndRunLastTask());
// Task is posted right away since it took longer to run then the repeat
// interval.
- EXPECT_EQ(task_queue.last_delay(), 0u);
+ EXPECT_EQ(task_queue.last_delay(), TimeDelta::Zero());
EXPECT_EQ(counter.load(), 1);
}
@@ -206,11 +209,11 @@
TaskQueueBase::DelayPrecision::kLow, &clock);
// Expect instant post task.
- EXPECT_EQ(task_queue.last_delay(), 0u);
+ EXPECT_EQ(task_queue.last_delay(), TimeDelta::Zero());
// Task should be retained by the handler since it is not cancelled.
EXPECT_FALSE(task_queue.AdvanceTimeAndRunLastTask());
// New delay should be 200ms since repeat delay was 300ms but task took 100ms.
- EXPECT_EQ(task_queue.last_delay(), 200u);
+ EXPECT_EQ(task_queue.last_delay(), TimeDelta::Millis(200));
}
TEST(RepeatingTaskTest, CancelDelayedTaskBeforeItRuns) {
@@ -248,7 +251,7 @@
handle.Stop();
return TimeDelta::Millis(2);
});
- EXPECT_EQ(task_queue.last_delay(), 0u);
+ EXPECT_EQ(task_queue.last_delay(), TimeDelta::Zero());
// Task cancelled itself so wants to be released.
EXPECT_TRUE(task_queue.AdvanceTimeAndRunLastTask());
EXPECT_EQ(counter.load(), 1);
@@ -262,7 +265,7 @@
++counter;
return TimeDelta::PlusInfinity();
});
- EXPECT_EQ(task_queue.last_delay(), 0u);
+ EXPECT_EQ(task_queue.last_delay(), TimeDelta::Zero());
// Task cancelled itself so wants to be released.
EXPECT_TRUE(task_queue.AdvanceTimeAndRunLastTask());
EXPECT_EQ(counter.load(), 1);
@@ -331,20 +334,18 @@
}
TEST(RepeatingTaskTest, ClockIntegration) {
- std::unique_ptr<QueuedTask> delayed_task;
- uint32_t expected_ms = 0;
+ absl::AnyInvocable<void() &&> delayed_task;
+ TimeDelta expected_delay = TimeDelta::Zero();
SimulatedClock clock(Timestamp::Millis(0));
NiceMock<MockTaskQueue> task_queue;
ON_CALL(task_queue, PostDelayedTask)
- .WillByDefault(
- Invoke([&delayed_task, &expected_ms](std::unique_ptr<QueuedTask> task,
- uint32_t milliseconds) {
- EXPECT_EQ(milliseconds, expected_ms);
- delayed_task = std::move(task);
- }));
+ .WillByDefault([&](absl::AnyInvocable<void() &&> task, TimeDelta delay) {
+ EXPECT_EQ(delay, expected_delay);
+ delayed_task = std::move(task);
+ });
- expected_ms = 100;
+ expected_delay = TimeDelta::Millis(100);
RepeatingTaskHandle handle = RepeatingTaskHandle::DelayedStart(
&task_queue, TimeDelta::Millis(100),
[&clock]() {
@@ -356,19 +357,19 @@
TaskQueueBase::DelayPrecision::kLow, &clock);
clock.AdvanceTimeMilliseconds(100);
- QueuedTask* task_to_run = delayed_task.release();
- expected_ms = 90;
- EXPECT_FALSE(task_to_run->Run());
- EXPECT_NE(nullptr, delayed_task.get());
+ absl::AnyInvocable<void()&&> task_to_run = std::move(delayed_task);
+ expected_delay = TimeDelta::Millis(90);
+ std::move(task_to_run)();
+ EXPECT_NE(delayed_task, nullptr);
handle.Stop();
}
TEST(RepeatingTaskTest, CanBeStoppedAfterTaskQueueDeletedTheRepeatingTask) {
- std::unique_ptr<QueuedTask> repeating_task;
+ absl::AnyInvocable<void() &&> repeating_task;
MockTaskQueue task_queue;
EXPECT_CALL(task_queue, PostDelayedTask)
- .WillOnce([&](std::unique_ptr<QueuedTask> task, uint32_t milliseconds) {
+ .WillOnce([&](absl::AnyInvocable<void() &&> task, TimeDelta delay) {
repeating_task = std::move(task);
});