Only process cross traffic simulation if added.
This avoids extra processing overhead when there's no cross traffic
simulation active. To allow the use of SendTask, GlobalTimeController
is adjusted so it always overrides yield behavior.
Also adding sequence checkers to protect against races on
read access.
Bug: webrtc:10365
Change-Id: I55c6ceb22f36ec19a4fca48cff500905192d1a16
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/133167
Reviewed-by: Niels Moller <nisse@webrtc.org>
Reviewed-by: Artem Titov <titovartem@webrtc.org>
Commit-Queue: Sebastian Jansson <srte@webrtc.org>
Cr-Commit-Position: refs/heads/master@{#27658}diff --git a/test/scenario/network/BUILD.gn b/test/scenario/network/BUILD.gn
index 0bcd65d..1f690c5 100644
--- a/test/scenario/network/BUILD.gn
+++ b/test/scenario/network/BUILD.gn
@@ -46,6 +46,7 @@
"../../../rtc_base:rtc_task_queue",
"../../../rtc_base:safe_minmax",
"../../../rtc_base:task_queue_for_test",
+ "../../../rtc_base/synchronization:sequence_checker",
"../../../rtc_base/task_utils:repeating_task",
"../../../rtc_base/third_party/sigslot",
"../../../system_wrappers",
diff --git a/test/scenario/network/cross_traffic.cc b/test/scenario/network/cross_traffic.cc
index bd63a08..1d4efef 100644
--- a/test/scenario/network/cross_traffic.cc
+++ b/test/scenario/network/cross_traffic.cc
@@ -25,10 +25,13 @@
TrafficRoute* traffic_route)
: config_(config),
traffic_route_(traffic_route),
- random_(config_.random_seed) {}
+ random_(config_.random_seed) {
+ sequence_checker_.Detach();
+}
RandomWalkCrossTraffic::~RandomWalkCrossTraffic() = default;
void RandomWalkCrossTraffic::Process(Timestamp at_time) {
+ RTC_DCHECK_RUN_ON(&sequence_checker_);
if (last_process_time_.IsMinusInfinity()) {
last_process_time_ = at_time;
}
@@ -52,6 +55,7 @@
}
DataRate RandomWalkCrossTraffic::TrafficRate() const {
+ RTC_DCHECK_RUN_ON(&sequence_checker_);
return config_.peak_rate * intensity_;
}
@@ -66,10 +70,13 @@
PulsedPeaksCrossTraffic::PulsedPeaksCrossTraffic(PulsedPeaksConfig config,
TrafficRoute* traffic_route)
- : config_(config), traffic_route_(traffic_route) {}
+ : config_(config), traffic_route_(traffic_route) {
+ sequence_checker_.Detach();
+}
PulsedPeaksCrossTraffic::~PulsedPeaksCrossTraffic() = default;
void PulsedPeaksCrossTraffic::Process(Timestamp at_time) {
+ RTC_DCHECK_RUN_ON(&sequence_checker_);
TimeDelta time_since_toggle = at_time - last_update_time_;
if (time_since_toggle.IsInfinite() ||
(sending_ && time_since_toggle >= config_.send_duration)) {
@@ -94,6 +101,7 @@
}
DataRate PulsedPeaksCrossTraffic::TrafficRate() const {
+ RTC_DCHECK_RUN_ON(&sequence_checker_);
return sending_ ? config_.peak_rate : DataRate::Zero();
}
diff --git a/test/scenario/network/cross_traffic.h b/test/scenario/network/cross_traffic.h
index b5a65fb..e888274 100644
--- a/test/scenario/network/cross_traffic.h
+++ b/test/scenario/network/cross_traffic.h
@@ -18,6 +18,7 @@
#include "api/units/time_delta.h"
#include "api/units/timestamp.h"
#include "rtc_base/random.h"
+#include "rtc_base/synchronization/sequence_checker.h"
#include "test/scenario/column_printer.h"
#include "test/scenario/network/traffic_route.h"
@@ -44,15 +45,19 @@
ColumnPrinter StatsPrinter();
private:
- RandomWalkConfig config_;
- TrafficRoute* const traffic_route_;
- webrtc::Random random_;
+ SequenceChecker sequence_checker_;
+ const RandomWalkConfig config_;
+ TrafficRoute* const traffic_route_ RTC_PT_GUARDED_BY(sequence_checker_);
+ webrtc::Random random_ RTC_GUARDED_BY(sequence_checker_);
- Timestamp last_process_time_ = Timestamp::MinusInfinity();
- Timestamp last_update_time_ = Timestamp::MinusInfinity();
- Timestamp last_send_time_ = Timestamp::MinusInfinity();
- double intensity_ = 0;
- DataSize pending_size_ = DataSize::Zero();
+ Timestamp last_process_time_ RTC_GUARDED_BY(sequence_checker_) =
+ Timestamp::MinusInfinity();
+ Timestamp last_update_time_ RTC_GUARDED_BY(sequence_checker_) =
+ Timestamp::MinusInfinity();
+ Timestamp last_send_time_ RTC_GUARDED_BY(sequence_checker_) =
+ Timestamp::MinusInfinity();
+ double intensity_ RTC_GUARDED_BY(sequence_checker_) = 0;
+ DataSize pending_size_ RTC_GUARDED_BY(sequence_checker_) = DataSize::Zero();
};
struct PulsedPeaksConfig {
@@ -74,12 +79,15 @@
ColumnPrinter StatsPrinter();
private:
- PulsedPeaksConfig config_;
- TrafficRoute* const traffic_route_;
+ SequenceChecker sequence_checker_;
+ const PulsedPeaksConfig config_;
+ TrafficRoute* const traffic_route_ RTC_PT_GUARDED_BY(sequence_checker_);
- Timestamp last_update_time_ = Timestamp::MinusInfinity();
- Timestamp last_send_time_ = Timestamp::MinusInfinity();
- bool sending_ = false;
+ Timestamp last_update_time_ RTC_GUARDED_BY(sequence_checker_) =
+ Timestamp::MinusInfinity();
+ Timestamp last_send_time_ RTC_GUARDED_BY(sequence_checker_) =
+ Timestamp::MinusInfinity();
+ bool sending_ RTC_GUARDED_BY(sequence_checker_) = false;
};
} // namespace test
diff --git a/test/scenario/network/network_emulation_manager.cc b/test/scenario/network/network_emulation_manager.cc
index ae67f6f..b132e80 100644
--- a/test/scenario/network/network_emulation_manager.cc
+++ b/test/scenario/network/network_emulation_manager.cc
@@ -23,12 +23,33 @@
namespace test {
namespace {
-constexpr int64_t kPacketProcessingIntervalMs = 1;
// uint32_t representation of 192.168.0.0 address
constexpr uint32_t kMinIPv4Address = 0xC0A80000;
// uint32_t representation of 192.168.255.255 address
constexpr uint32_t kMaxIPv4Address = 0xC0A8FFFF;
+template <typename T, typename Closure>
+class ResourceOwningTask final : public QueuedTask {
+ public:
+ ResourceOwningTask(T&& resource, Closure&& handler)
+ : resource_(std::move(resource)),
+ handler_(std::forward<Closure>(handler)) {}
+
+ bool Run() override {
+ handler_(std::move(resource_));
+ return true;
+ }
+
+ private:
+ T resource_;
+ Closure handler_;
+};
+template <typename T, typename Closure>
+std::unique_ptr<QueuedTask> CreateResourceOwningTask(T resource,
+ Closure&& closure) {
+ return absl::make_unique<ResourceOwningTask<T, Closure>>(
+ std::forward<T>(resource), std::forward<Closure>(closure));
+}
} // namespace
NetworkEmulationManagerImpl::NetworkEmulationManagerImpl()
@@ -42,10 +63,6 @@
task_queue_(time_controller->GetTaskQueueFactory()->CreateTaskQueue(
"NetworkEmulation",
TaskQueueFactory::Priority::NORMAL)) {
- process_task_handle_ = RepeatingTaskHandle::Start(task_queue_.Get(), [this] {
- ProcessNetworkPackets();
- return TimeDelta::ms(kPacketProcessingIntervalMs);
- });
}
// TODO(srte): Ensure that any pending task that must be run for consistency
@@ -170,17 +187,21 @@
NetworkEmulationManagerImpl::CreateRandomWalkCrossTraffic(
TrafficRoute* traffic_route,
RandomWalkConfig config) {
- auto traffic = absl::make_unique<RandomWalkCrossTraffic>(std::move(config),
- traffic_route);
+ auto traffic =
+ absl::make_unique<RandomWalkCrossTraffic>(config, traffic_route);
RandomWalkCrossTraffic* out = traffic.get();
- struct Closure {
- void operator()() {
- manager->random_cross_traffics_.push_back(std::move(traffic));
- }
- NetworkEmulationManagerImpl* manager;
- std::unique_ptr<RandomWalkCrossTraffic> traffic;
- };
- task_queue_.PostTask(Closure{this, std::move(traffic)});
+
+ task_queue_.PostTask(CreateResourceOwningTask(
+ std::move(traffic),
+ [this, config](std::unique_ptr<RandomWalkCrossTraffic> traffic) {
+ auto* traffic_ptr = traffic.get();
+ random_cross_traffics_.push_back(std::move(traffic));
+ RepeatingTaskHandle::Start(task_queue_.Get(),
+ [this, config, traffic_ptr] {
+ traffic_ptr->Process(Now());
+ return config.min_packet_interval;
+ });
+ }));
return out;
}
@@ -188,17 +209,20 @@
NetworkEmulationManagerImpl::CreatePulsedPeaksCrossTraffic(
TrafficRoute* traffic_route,
PulsedPeaksConfig config) {
- auto traffic = absl::make_unique<PulsedPeaksCrossTraffic>(std::move(config),
- traffic_route);
+ auto traffic =
+ absl::make_unique<PulsedPeaksCrossTraffic>(config, traffic_route);
PulsedPeaksCrossTraffic* out = traffic.get();
- struct Closure {
- void operator()() {
- manager->pulsed_cross_traffics_.push_back(std::move(traffic));
- }
- NetworkEmulationManagerImpl* manager;
- std::unique_ptr<PulsedPeaksCrossTraffic> traffic;
- };
- task_queue_.PostTask(Closure{this, std::move(traffic)});
+ task_queue_.PostTask(CreateResourceOwningTask(
+ std::move(traffic),
+ [this, config](std::unique_ptr<PulsedPeaksCrossTraffic> traffic) {
+ auto* traffic_ptr = traffic.get();
+ pulsed_cross_traffics_.push_back(std::move(traffic));
+ RepeatingTaskHandle::Start(task_queue_.Get(),
+ [this, config, traffic_ptr] {
+ traffic_ptr->Process(Now());
+ return config.min_packet_interval;
+ });
+ }));
return out;
}
@@ -242,16 +266,6 @@
return absl::nullopt;
}
-void NetworkEmulationManagerImpl::ProcessNetworkPackets() {
- Timestamp current_time = Now();
- for (auto& traffic : random_cross_traffics_) {
- traffic->Process(current_time);
- }
- for (auto& traffic : pulsed_cross_traffics_) {
- traffic->Process(current_time);
- }
-}
-
Timestamp NetworkEmulationManagerImpl::Now() const {
return Timestamp::us(clock_->TimeInMicroseconds());
}
diff --git a/test/scenario/network/network_emulation_manager.h b/test/scenario/network/network_emulation_manager.h
index b3fd2a5..72f15fdb 100644
--- a/test/scenario/network/network_emulation_manager.h
+++ b/test/scenario/network/network_emulation_manager.h
@@ -69,7 +69,6 @@
FakeNetworkSocketServer* CreateSocketServer(
const std::vector<EmulatedEndpoint*>& endpoints);
absl::optional<rtc::IPAddress> GetNextIPv4Address();
- void ProcessNetworkPackets();
Timestamp Now() const;
Clock* const clock_;
