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}

test/scenario/network/network_emulation_manager.cc

@@ -23,12 +23,33 @@ namespace webrtc {
 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 @@ NetworkEmulationManagerImpl::NetworkEmulationManagerImpl(
       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 @@ RandomWalkCrossTraffic*
 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 @@ PulsedPeaksCrossTraffic*
 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 @@ NetworkEmulationManagerImpl::GetNextIPv4Address() {
   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());
 }