Cleanup in NetworkEmulationManager

This prepares for an upcoming CL removing cross traffic processing
when it's not used.

Bug: webrtc:10365
Change-Id: I7f1f3998f7f38c2a627b888c3db6b0c459d8271d
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/133485
Reviewed-by: Artem Titov <titovartem@webrtc.org>
Commit-Queue: Sebastian Jansson <srte@webrtc.org>
Cr-Commit-Position: refs/heads/master@{#27682}

test/scenario/network/BUILD.gn

@@ -46,6 +46,7 @@ rtc_source_set("emulated_network") {
     "../../../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",

test/scenario/network/cross_traffic.cc

@@ -25,10 +25,13 @@ RandomWalkCrossTraffic::RandomWalkCrossTraffic(RandomWalkConfig config,
                                                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 @@ void RandomWalkCrossTraffic::Process(Timestamp at_time) {
 }
 
 DataRate RandomWalkCrossTraffic::TrafficRate() const {
+  RTC_DCHECK_RUN_ON(&sequence_checker_);
   return config_.peak_rate * intensity_;
 }
 
@@ -66,10 +70,13 @@ ColumnPrinter RandomWalkCrossTraffic::StatsPrinter() {
 
 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 @@ void PulsedPeaksCrossTraffic::Process(Timestamp at_time) {
 }
 
 DataRate PulsedPeaksCrossTraffic::TrafficRate() const {
+  RTC_DCHECK_RUN_ON(&sequence_checker_);
   return sending_ ? config_.peak_rate : DataRate::Zero();
 }
 

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 @@ class RandomWalkCrossTraffic {
   ColumnPrinter StatsPrinter();
 
  private:
-  RandomWalkConfig config_;
+  SequenceChecker sequence_checker_;
-  TrafficRoute* const traffic_route_;
+  const RandomWalkConfig config_;
-  webrtc::Random random_;
+  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_process_time_ RTC_GUARDED_BY(sequence_checker_) =
-  Timestamp last_update_time_ = Timestamp::MinusInfinity();
+      Timestamp::MinusInfinity();
-  Timestamp last_send_time_ = Timestamp::MinusInfinity();
+  Timestamp last_update_time_ RTC_GUARDED_BY(sequence_checker_) =
-  double intensity_ = 0;
+      Timestamp::MinusInfinity();
-  DataSize pending_size_ = DataSize::Zero();
+  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 @@ class PulsedPeaksCrossTraffic {
   ColumnPrinter StatsPrinter();
 
  private:
-  PulsedPeaksConfig config_;
+  SequenceChecker sequence_checker_;
-  TrafficRoute* const traffic_route_;
+  const PulsedPeaksConfig config_;
+  TrafficRoute* const traffic_route_ RTC_PT_GUARDED_BY(sequence_checker_);
 
-  Timestamp last_update_time_ = Timestamp::MinusInfinity();
+  Timestamp last_update_time_ RTC_GUARDED_BY(sequence_checker_) =
-  Timestamp last_send_time_ = Timestamp::MinusInfinity();
+      Timestamp::MinusInfinity();
-  bool sending_ = false;
+  Timestamp last_send_time_ RTC_GUARDED_BY(sequence_checker_) =
+      Timestamp::MinusInfinity();
+  bool sending_ RTC_GUARDED_BY(sequence_checker_) = false;
 };
 
 }  // namespace test

test/scenario/network/network_emulation_manager.cc

@@ -29,6 +29,28 @@ 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()
@@ -58,13 +80,10 @@ EmulatedNetworkNode* NetworkEmulationManagerImpl::CreateEmulatedNode(
   auto node = absl::make_unique<EmulatedNetworkNode>(
       clock_, &task_queue_, std::move(network_behavior));
   EmulatedNetworkNode* out = node.get();
-
+  task_queue_.PostTask(CreateResourceOwningTask(
-  struct Closure {
+      std::move(node), [this](std::unique_ptr<EmulatedNetworkNode> node) {
-    void operator()() { manager->network_nodes_.push_back(std::move(node)); }
+        network_nodes_.push_back(std::move(node));
-    NetworkEmulationManagerImpl* manager;
+      }));
-    std::unique_ptr<EmulatedNetworkNode> node;
-  };
-  task_queue_.PostTask(Closure{this, std::move(node)});
   return out;
 }
 
@@ -170,17 +189,15 @@ RandomWalkCrossTraffic*
 NetworkEmulationManagerImpl::CreateRandomWalkCrossTraffic(
     TrafficRoute* traffic_route,
     RandomWalkConfig config) {
-  auto traffic = absl::make_unique<RandomWalkCrossTraffic>(std::move(config),
+  auto traffic =
-                                                           traffic_route);
+      absl::make_unique<RandomWalkCrossTraffic>(config, traffic_route);
   RandomWalkCrossTraffic* out = traffic.get();
-  struct Closure {
+
-    void operator()() {
+  task_queue_.PostTask(CreateResourceOwningTask(
-      manager->random_cross_traffics_.push_back(std::move(traffic));
+      std::move(traffic),
-    }
+      [this](std::unique_ptr<RandomWalkCrossTraffic> traffic) {
-    NetworkEmulationManagerImpl* manager;
+        random_cross_traffics_.push_back(std::move(traffic));
-    std::unique_ptr<RandomWalkCrossTraffic> traffic;
+      }));
-  };
-  task_queue_.PostTask(Closure{this, std::move(traffic)});
   return out;
 }
 
@@ -188,17 +205,14 @@ PulsedPeaksCrossTraffic*
 NetworkEmulationManagerImpl::CreatePulsedPeaksCrossTraffic(
     TrafficRoute* traffic_route,
     PulsedPeaksConfig config) {
-  auto traffic = absl::make_unique<PulsedPeaksCrossTraffic>(std::move(config),
+  auto traffic =
-                                                            traffic_route);
+      absl::make_unique<PulsedPeaksCrossTraffic>(config, traffic_route);
   PulsedPeaksCrossTraffic* out = traffic.get();
-  struct Closure {
+  task_queue_.PostTask(CreateResourceOwningTask(
-    void operator()() {
+      std::move(traffic),
-      manager->pulsed_cross_traffics_.push_back(std::move(traffic));
+      [this](std::unique_ptr<PulsedPeaksCrossTraffic> traffic) {
-    }
+        pulsed_cross_traffics_.push_back(std::move(traffic));
-    NetworkEmulationManagerImpl* manager;
+      }));
-    std::unique_ptr<PulsedPeaksCrossTraffic> traffic;
-  };
-  task_queue_.PostTask(Closure{this, std::move(traffic)});
   return out;
 }
 

test/scenario/network/network_emulation_manager.h

@@ -66,8 +66,6 @@ class NetworkEmulationManagerImpl : public NetworkEmulationManager {
       const std::vector<EmulatedEndpoint*>& endpoints) override;
 
  private:
-  FakeNetworkSocketServer* CreateSocketServer(
-      const std::vector<EmulatedEndpoint*>& endpoints);
   absl::optional<rtc::IPAddress> GetNextIPv4Address();
   void ProcessNetworkPackets();
   Timestamp Now() const;