Move ScopedOperationsBatcher to its own source files Extract the ScopedOperationsBatcher class from sdp_offer_answer.cc into a new, dedicated library module. This change improves code organization by isolating the batching logic, which handles synchronous task execution on the worker thread and optional callback tasks on the calling thread. As part of this refactoring: * Created scoped_operations_batcher.h and .cc. * Defined a new rtc_library target in pc/BUILD.gn. * Added unit tests in scoped_operations_batcher_unittest.cc to verify task execution on both worker and signaling threads. * Updated sdp_offer_answer.cc to include the new header. Bug: webrtc:42222804 Change-Id: I090c1614917e83a3e3d9543991f1fa2dc4cf22f7 Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/454305 Commit-Queue: Tomas Gunnarsson <tommi@webrtc.org> Reviewed-by: Harald Alvestrand <hta@webrtc.org> Cr-Commit-Position: refs/heads/main@{#47199}
diff --git a/pc/BUILD.gn b/pc/BUILD.gn index 03fabe9..2755d73 100644 --- a/pc/BUILD.gn +++ b/pc/BUILD.gn
@@ -1069,6 +1069,19 @@ ] } +rtc_library("scoped_operations_batcher") { + visibility = [ ":*" ] + sources = [ + "scoped_operations_batcher.cc", + "scoped_operations_batcher.h", + ] + deps = [ + "../rtc_base:checks", + "../rtc_base:threading", + "//third_party/abseil-cpp/absl/functional:any_invocable", + ] +} + rtc_library("sdp_munging_detector") { visibility = [ ":*" ] sources = [ @@ -1121,6 +1134,7 @@ ":rtp_sender_proxy", ":rtp_transceiver", ":rtp_transmission_manager", + ":scoped_operations_batcher", ":sdp_munging_detector", ":sdp_payload_type_suggester", ":sdp_state_provider", @@ -2502,6 +2516,7 @@ "rtp_parameters_conversion_unittest.cc", "rtp_sender_receiver_unittest.cc", "rtp_transceiver_unittest.cc", + "scoped_operations_batcher_unittest.cc", "sctp_utils_unittest.cc", "sdp_munging_detector_unittest.cc", "sdp_offer_answer_unittest.cc", @@ -2552,6 +2567,7 @@ ":rtp_transceiver", ":rtp_transport", ":rtp_transport_internal", + ":scoped_operations_batcher", ":sctp_data_channel", ":sctp_transport", ":sctp_utils",
diff --git a/pc/scoped_operations_batcher.cc b/pc/scoped_operations_batcher.cc new file mode 100644 index 0000000..050d0c3 --- /dev/null +++ b/pc/scoped_operations_batcher.cc
@@ -0,0 +1,68 @@ +/* + * Copyright 2026 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 "pc/scoped_operations_batcher.h" + +#include <utility> +#include <vector> + +#include "absl/functional/any_invocable.h" +#include "rtc_base/checks.h" +#include "rtc_base/thread.h" + +namespace webrtc { + +ScopedOperationsBatcher::ScopedOperationsBatcher(Thread* worker_thread) + : worker_thread_(worker_thread) { + RTC_DCHECK(worker_thread_); +} + +ScopedOperationsBatcher::~ScopedOperationsBatcher() { + Run(); +} + +void ScopedOperationsBatcher::Run() { + std::vector<absl::AnyInvocable<void() &&>> signaling_tasks; + if (!tasks_.empty()) { + worker_thread_->BlockingCall([&] { + for (auto& task : tasks_) { + if (task.void_task) { + std::move(task.void_task)(); + } else { + RTC_DCHECK(task.returning_task); + auto ret = std::move(task.returning_task)(); + if (ret) { + signaling_tasks.push_back(std::move(ret)); + } + } + } + }); + tasks_.clear(); + } + + for (auto& task : signaling_tasks) { + std::move(task)(); + } +} + +void ScopedOperationsBatcher::push_back(absl::AnyInvocable<void() &&> task) { + if (task) { + tasks_.push_back({.void_task = std::move(task)}); + } +} + +void ScopedOperationsBatcher::push_back( + absl::AnyInvocable<absl::AnyInvocable<void() &&>() &&> task) { + if (task) { + tasks_.push_back({.returning_task = std::move(task)}); + } +} + +} // namespace webrtc
diff --git a/pc/scoped_operations_batcher.h b/pc/scoped_operations_batcher.h new file mode 100644 index 0000000..d9a4c23 --- /dev/null +++ b/pc/scoped_operations_batcher.h
@@ -0,0 +1,54 @@ +/* + * Copyright 2026 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 PC_SCOPED_OPERATIONS_BATCHER_H_ +#define PC_SCOPED_OPERATIONS_BATCHER_H_ + +#include <vector> + +#include "absl/functional/any_invocable.h" +#include "rtc_base/thread.h" + +namespace webrtc { + +// Batches operations to be executed synchronously on the worker thread. +// +// When the batcher goes out of scope (or `Run()` is explicitly called), it +// executes all queued tasks in a single `BlockingCall` to the worker thread. +// +// Tasks can either have a `void` return type, or return a new task. +// Any tasks returned by the executed worker thread tasks are collected +// and subsequently executed on the calling thread (typically the signaling +// thread) after the worker thread operations have completed. +class ScopedOperationsBatcher { + public: + explicit ScopedOperationsBatcher(Thread* worker_thread); + ~ScopedOperationsBatcher(); + + void Run(); + + // Queues non-nullptr tasks to be executed on the worker when the + // ScopedOperationsBatcher goes out of scope. + void push_back(absl::AnyInvocable<void() &&> task); + void push_back(absl::AnyInvocable<absl::AnyInvocable<void() &&>() &&> task); + + private: + struct BatchedTask { + absl::AnyInvocable<void() &&> void_task; + absl::AnyInvocable<absl::AnyInvocable<void() &&>() &&> returning_task; + }; + + Thread* const worker_thread_; + std::vector<BatchedTask> tasks_; +}; + +} // namespace webrtc + +#endif // PC_SCOPED_OPERATIONS_BATCHER_H_
diff --git a/pc/scoped_operations_batcher_unittest.cc b/pc/scoped_operations_batcher_unittest.cc new file mode 100644 index 0000000..7f15ad3 --- /dev/null +++ b/pc/scoped_operations_batcher_unittest.cc
@@ -0,0 +1,78 @@ +/* + * Copyright 2026 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 "pc/scoped_operations_batcher.h" + +#include <memory> +#include <utility> + +#include "absl/functional/any_invocable.h" +#include "rtc_base/thread.h" +#include "test/gtest.h" + +namespace webrtc { +namespace { + +TEST(ScopedOperationsBatcherTest, ExecutesTasksOnWorkerThread) { + auto worker_thread = Thread::Create(); + worker_thread->Start(); + + bool task_executed = false; + bool worker_checked = false; + + { + ScopedOperationsBatcher batcher(worker_thread.get()); + batcher.push_back([&] { + task_executed = true; + worker_checked = worker_thread->IsCurrent(); + }); + } + + EXPECT_TRUE(task_executed); + EXPECT_TRUE(worker_checked); +} + +TEST(ScopedOperationsBatcherTest, ExecutesReturnedTasksOnCallingThread) { + // Use current thread as the calling (signaling) thread. This test runs + // without an explicit RunLoop, because the returned tasks are executed + // synchronously within the caller's context of `Run()` or + // `~ScopedOperationsBatcher()`. + auto signaling_thread = Thread::Current(); + + auto worker_thread = Thread::Create(); + worker_thread->Start(); + + bool return_task_executed = false; + Thread* return_task_thread = nullptr; + bool task_executed = false; + Thread* task_thread = nullptr; + + { + ScopedOperationsBatcher batcher(worker_thread.get()); + absl::AnyInvocable<absl::AnyInvocable<void() &&>() &&> task = + [&]() -> absl::AnyInvocable<void() &&> { + task_executed = true; + task_thread = Thread::Current(); + return [&]() { + return_task_executed = true; + return_task_thread = Thread::Current(); + }; + }; + batcher.push_back(std::move(task)); + } + + EXPECT_TRUE(task_executed); + EXPECT_EQ(task_thread, worker_thread.get()); + EXPECT_TRUE(return_task_executed); + EXPECT_EQ(return_task_thread, signaling_thread); +} + +} // namespace +} // namespace webrtc
diff --git a/pc/sdp_offer_answer.cc b/pc/sdp_offer_answer.cc index 24abb0d..3f5c83f 100644 --- a/pc/sdp_offer_answer.cc +++ b/pc/sdp_offer_answer.cc
@@ -86,6 +86,7 @@ #include "pc/rtp_sender_proxy.h" #include "pc/rtp_transceiver.h" #include "pc/rtp_transmission_manager.h" +#include "pc/scoped_operations_batcher.h" #include "pc/sdp_munging_detector.h" #include "pc/session_description.h" #include "pc/simulcast_description.h" @@ -112,71 +113,6 @@ namespace webrtc { namespace { -// Batches operations to be executed synchronously on the worker thread. -// -// When the batcher goes out of scope (or `Run()` is explicitly called), it -// executes all queued tasks in a single `BlockingCall` to the worker thread. -// -// Tasks can either have a `void` return type, or return a new task. -// Any tasks returned by the executed worker thread tasks are collected -// and subsequently executed on the calling thread (typically the signaling -// thread) after the worker thread operations have completed. -class ScopedOperationsBatcher { - public: - explicit ScopedOperationsBatcher(Thread* worker_thread) - : worker_thread_(worker_thread) { - RTC_DCHECK(worker_thread_); - } - - ~ScopedOperationsBatcher() { Run(); } - - void Run() { - std::vector<absl::AnyInvocable<void() &&>> signaling_tasks; - if (!tasks_.empty()) { - worker_thread_->BlockingCall([&] { - for (auto& task : tasks_) { - if (task.void_task) { - std::move(task.void_task)(); - } else { - RTC_DCHECK(task.returning_task); - auto ret = std::move(task.returning_task)(); - if (ret) { - signaling_tasks.push_back(std::move(ret)); - } - } - } - }); - tasks_.clear(); - } - - for (auto& task : signaling_tasks) { - std::move(task)(); - } - } - - // Queues non-nullptr tasks to be executed on the worker when the - // ScopedOperationsBatcher goes out of scope. - void push_back(absl::AnyInvocable<void() &&> task) { - if (task) { - tasks_.push_back({.void_task = std::move(task)}); - } - } - - void push_back(absl::AnyInvocable<absl::AnyInvocable<void() &&>() &&> task) { - if (task) { - tasks_.push_back({.returning_task = std::move(task)}); - } - } - - private: - struct BatchedTask { - absl::AnyInvocable<void() &&> void_task; - absl::AnyInvocable<absl::AnyInvocable<void() &&>() &&> returning_task; - }; - - Thread* const worker_thread_; - std::vector<BatchedTask> tasks_; -}; struct DtlsTransportAndName { scoped_refptr<DtlsTransport> transport;