In rtc::Thread implement posting AnyInvocable
Lots of code call rtc::Thread directly instead of through TaskQueueBase
interface, thus to continue migration step by step rtc::Thread needs
to implement both old and new TaskQueueBase interfaces.
Bug: webrtc:14245
Change-Id: Ie7cac897a4c8a6227b8d467a39adb30aec6f1318
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/267984
Reviewed-by: Tomas Gunnarsson <tommi@webrtc.org>
Commit-Queue: Danil Chapovalov <danilchap@webrtc.org>
Cr-Commit-Position: refs/heads/main@{#37474}
diff --git a/rtc_base/thread.cc b/rtc_base/thread.cc
index e1db2bc..ef165ff 100644
--- a/rtc_base/thread.cc
+++ b/rtc_base/thread.cc
@@ -31,8 +31,8 @@
#include <utility>
#include "absl/algorithm/container.h"
+#include "absl/cleanup/cleanup.h"
#include "api/sequence_checker.h"
-#include "api/task_queue/to_queued_task.h"
#include "rtc_base/checks.h"
#include "rtc_base/deprecated/recursive_critical_section.h"
#include "rtc_base/event.h"
@@ -72,22 +72,25 @@
namespace rtc {
namespace {
-class MessageHandlerWithTask final : public MessageHandler {
+struct AnyInvocableMessage final : public MessageData {
+ explicit AnyInvocableMessage(absl::AnyInvocable<void() &&> task)
+ : task(std::move(task)) {}
+ absl::AnyInvocable<void() &&> task;
+};
+
+class AnyInvocableMessageHandler final : public MessageHandler {
public:
- MessageHandlerWithTask() {}
-
- MessageHandlerWithTask(const MessageHandlerWithTask&) = delete;
- MessageHandlerWithTask& operator=(const MessageHandlerWithTask&) = delete;
-
void OnMessage(Message* msg) override {
- static_cast<rtc_thread_internal::MessageLikeTask*>(msg->pdata)->Run();
+ std::move(static_cast<AnyInvocableMessage*>(msg->pdata)->task)();
delete msg->pdata;
}
-
- private:
- ~MessageHandlerWithTask() override {}
};
+MessageHandler* GetAnyInvocableMessageHandler() {
+ static MessageHandler* const handler = new AnyInvocableMessageHandler;
+ return handler;
+}
+
class RTC_SCOPED_LOCKABLE MarkProcessingCritScope {
public:
MarkProcessingCritScope(const RecursiveCriticalSection* cs,
@@ -761,8 +764,7 @@
void Thread::SetDispatchWarningMs(int deadline) {
if (!IsCurrent()) {
- PostTask(webrtc::ToQueuedTask(
- [this, deadline]() { SetDispatchWarningMs(deadline); }));
+ PostTask([this, deadline]() { SetDispatchWarningMs(deadline); });
return;
}
RTC_DCHECK_RUN_ON(this);
@@ -948,18 +950,19 @@
done_event.reset(new rtc::Event());
bool ready = false;
- PostTask(webrtc::ToQueuedTask(
- [&msg]() mutable { msg.phandler->OnMessage(&msg); },
- [this, &ready, current_thread, done = done_event.get()] {
- if (current_thread) {
- CritScope cs(&crit_);
- ready = true;
- current_thread->socketserver()->WakeUp();
- } else {
- done->Set();
- }
- }));
-
+ absl::Cleanup cleanup = [this, &ready, current_thread,
+ done = done_event.get()] {
+ if (current_thread) {
+ CritScope cs(&crit_);
+ ready = true;
+ current_thread->socketserver()->WakeUp();
+ } else {
+ done->Set();
+ }
+ };
+ PostTask([&msg, cleanup = std::move(cleanup)]() mutable {
+ msg.phandler->OnMessage(&msg);
+ });
if (current_thread) {
bool waited = false;
crit_.Enter();
@@ -1115,6 +1118,28 @@
delete this;
}
+void Thread::PostTask(absl::AnyInvocable<void() &&> task) {
+ // Though Post takes MessageData by raw pointer (last parameter), it still
+ // takes it with ownership.
+ Post(RTC_FROM_HERE, GetAnyInvocableMessageHandler(),
+ /*id=*/0, new AnyInvocableMessage(std::move(task)));
+}
+
+void Thread::PostDelayedTask(absl::AnyInvocable<void() &&> task,
+ webrtc::TimeDelta delay) {
+ // This implementation does not support low precision yet.
+ PostDelayedHighPrecisionTask(std::move(task), delay);
+}
+
+void Thread::PostDelayedHighPrecisionTask(absl::AnyInvocable<void() &&> task,
+ webrtc::TimeDelta delay) {
+ int delay_ms = delay.RoundUpTo(webrtc::TimeDelta::Millis(1)).ms<int>();
+ // Though PostDelayed takes MessageData by raw pointer (last parameter),
+ // it still takes it with ownership.
+ PostDelayed(RTC_FROM_HERE, delay_ms, GetAnyInvocableMessageHandler(),
+ /*id=*/0, new AnyInvocableMessage(std::move(task)));
+}
+
bool Thread::IsProcessingMessagesForTesting() {
return (owned_ || IsCurrent()) && !IsQuitting();
}
@@ -1183,13 +1208,6 @@
#endif
}
-// static
-MessageHandler* Thread::GetPostTaskMessageHandler() {
- // Allocate at first call, never deallocate.
- static MessageHandler* handler = new MessageHandlerWithTask;
- return handler;
-}
-
AutoThread::AutoThread()
: Thread(CreateDefaultSocketServer(), /*do_init=*/false) {
if (!ThreadManager::Instance()->CurrentThread()) {