1 /*
2 * Copyright 2017 The WebRTC project authors. All Rights Reserved.
3 *
4 * Use of this source code is governed by a BSD-style license
5 * that can be found in the LICENSE file in the root of the source
6 * tree. An additional intellectual property rights grant can be found
7 * in the file PATENTS. All contributing project authors may
8 * be found in the AUTHORS file in the root of the source tree.
9 */
10
11 #include "rtc_tools/network_tester/packet_sender.h"
12
13 #include <algorithm>
14 #include <memory>
15 #include <string>
16 #include <utility>
17
18 #include "absl/functional/any_invocable.h"
19 #include "api/task_queue/pending_task_safety_flag.h"
20 #include "api/task_queue/task_queue_base.h"
21 #include "rtc_base/time_utils.h"
22 #include "rtc_tools/network_tester/config_reader.h"
23 #include "rtc_tools/network_tester/test_controller.h"
24
25 namespace webrtc {
26
27 namespace {
28
SendPacketTask(PacketSender * packet_sender,rtc::scoped_refptr<webrtc::PendingTaskSafetyFlag> task_safety_flag,int64_t target_time_ms=rtc::TimeMillis ())29 absl::AnyInvocable<void() &&> SendPacketTask(
30 PacketSender* packet_sender,
31 rtc::scoped_refptr<webrtc::PendingTaskSafetyFlag> task_safety_flag,
32 int64_t target_time_ms = rtc::TimeMillis()) {
33 return [target_time_ms, packet_sender,
34 task_safety_flag = std::move(task_safety_flag)]() mutable {
35 if (task_safety_flag->alive() && packet_sender->IsSending()) {
36 packet_sender->SendPacket();
37 target_time_ms += packet_sender->GetSendIntervalMs();
38 int64_t delay_ms =
39 std::max(static_cast<int64_t>(0), target_time_ms - rtc::TimeMillis());
40 TaskQueueBase::Current()->PostDelayedTask(
41 SendPacketTask(packet_sender, std::move(task_safety_flag),
42 target_time_ms),
43 TimeDelta::Millis(delay_ms));
44 }
45 };
46 }
47
UpdateTestSettingTask(PacketSender * packet_sender,std::unique_ptr<ConfigReader> config_reader,rtc::scoped_refptr<webrtc::PendingTaskSafetyFlag> task_safety_flag)48 absl::AnyInvocable<void() &&> UpdateTestSettingTask(
49 PacketSender* packet_sender,
50 std::unique_ptr<ConfigReader> config_reader,
51 rtc::scoped_refptr<webrtc::PendingTaskSafetyFlag> task_safety_flag) {
52 return [packet_sender, config_reader = std::move(config_reader),
53 task_safety_flag = std::move(task_safety_flag)]() mutable {
54 if (!task_safety_flag->alive()) {
55 return;
56 }
57 if (absl::optional<ConfigReader::Config> config =
58 config_reader->GetNextConfig()) {
59 packet_sender->UpdateTestSetting(config->packet_size,
60 config->packet_send_interval_ms);
61 TaskQueueBase::Current()->PostDelayedTask(
62 UpdateTestSettingTask(packet_sender, std::move(config_reader),
63 std::move(task_safety_flag)),
64 TimeDelta::Millis(config->execution_time_ms));
65 } else {
66 packet_sender->StopSending();
67 }
68 };
69 }
70
71 } // namespace
72
PacketSender(TestController * test_controller,webrtc::TaskQueueBase * worker_queue,rtc::scoped_refptr<webrtc::PendingTaskSafetyFlag> task_safety_flag,const std::string & config_file_path)73 PacketSender::PacketSender(
74 TestController* test_controller,
75 webrtc::TaskQueueBase* worker_queue,
76 rtc::scoped_refptr<webrtc::PendingTaskSafetyFlag> task_safety_flag,
77 const std::string& config_file_path)
78 : packet_size_(0),
79 send_interval_ms_(0),
80 sequence_number_(0),
81 sending_(false),
82 config_file_path_(config_file_path),
83 test_controller_(test_controller),
84 worker_queue_(worker_queue),
85 task_safety_flag_(task_safety_flag) {}
86
87 PacketSender::~PacketSender() = default;
88
StartSending()89 void PacketSender::StartSending() {
90 worker_queue_checker_.Detach();
91 worker_queue_->PostTask(SafeTask(task_safety_flag_, [this]() {
92 RTC_DCHECK_RUN_ON(&worker_queue_checker_);
93 sending_ = true;
94 }));
95 worker_queue_->PostTask(UpdateTestSettingTask(
96 this, std::make_unique<ConfigReader>(config_file_path_),
97 task_safety_flag_));
98 worker_queue_->PostTask(SendPacketTask(this, task_safety_flag_));
99 }
100
StopSending()101 void PacketSender::StopSending() {
102 RTC_DCHECK_RUN_ON(&worker_queue_checker_);
103 sending_ = false;
104 test_controller_->OnTestDone();
105 }
106
IsSending() const107 bool PacketSender::IsSending() const {
108 RTC_DCHECK_RUN_ON(&worker_queue_checker_);
109 return sending_;
110 }
111
SendPacket()112 void PacketSender::SendPacket() {
113 RTC_DCHECK_RUN_ON(&worker_queue_checker_);
114 NetworkTesterPacket packet;
115 packet.set_type(NetworkTesterPacket::TEST_DATA);
116 packet.set_sequence_number(sequence_number_++);
117 packet.set_send_timestamp(rtc::TimeMicros());
118 test_controller_->SendData(packet, packet_size_);
119 }
120
GetSendIntervalMs() const121 int64_t PacketSender::GetSendIntervalMs() const {
122 RTC_DCHECK_RUN_ON(&worker_queue_checker_);
123 return send_interval_ms_;
124 }
125
UpdateTestSetting(size_t packet_size,int64_t send_interval_ms)126 void PacketSender::UpdateTestSetting(size_t packet_size,
127 int64_t send_interval_ms) {
128 RTC_DCHECK_RUN_ON(&worker_queue_checker_);
129 send_interval_ms_ = send_interval_ms;
130 packet_size_ = packet_size;
131 }
132
133 } // namespace webrtc
134