1 /*
2 * Copyright (c) 2019 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 "modules/rtp_rtcp/source/deprecated/deprecated_rtp_sender_egress.h"
12
13 #include <limits>
14 #include <memory>
15 #include <utility>
16
17 #include "absl/strings/match.h"
18 #include "api/transport/field_trial_based_config.h"
19 #include "api/units/timestamp.h"
20 #include "logging/rtc_event_log/events/rtc_event_rtp_packet_outgoing.h"
21 #include "modules/remote_bitrate_estimator/test/bwe_test_logging.h"
22 #include "rtc_base/logging.h"
23
24 namespace webrtc {
25 namespace {
26 constexpr uint32_t kTimestampTicksPerMs = 90;
27 constexpr int kSendSideDelayWindowMs = 1000;
28 constexpr int kBitrateStatisticsWindowMs = 1000;
29 constexpr size_t kRtpSequenceNumberMapMaxEntries = 1 << 13;
30
31 } // namespace
32
NonPacedPacketSender(DEPRECATED_RtpSenderEgress * sender,PacketSequencer * sequence_number_assigner)33 DEPRECATED_RtpSenderEgress::NonPacedPacketSender::NonPacedPacketSender(
34 DEPRECATED_RtpSenderEgress* sender,
35 PacketSequencer* sequence_number_assigner)
36 : transport_sequence_number_(0),
37 sender_(sender),
38 sequence_number_assigner_(sequence_number_assigner) {
39 RTC_DCHECK(sequence_number_assigner_);
40 }
41 DEPRECATED_RtpSenderEgress::NonPacedPacketSender::~NonPacedPacketSender() =
42 default;
43
EnqueuePackets(std::vector<std::unique_ptr<RtpPacketToSend>> packets)44 void DEPRECATED_RtpSenderEgress::NonPacedPacketSender::EnqueuePackets(
45 std::vector<std::unique_ptr<RtpPacketToSend>> packets) {
46 for (auto& packet : packets) {
47 // Assign sequence numbers, but not for flexfec which is already running on
48 // an internally maintained sequence number series.
49 if (packet->Ssrc() != sender_->FlexFecSsrc()) {
50 sequence_number_assigner_->Sequence(*packet);
51 }
52 if (!packet->SetExtension<TransportSequenceNumber>(
53 ++transport_sequence_number_)) {
54 --transport_sequence_number_;
55 }
56 packet->ReserveExtension<TransmissionOffset>();
57 packet->ReserveExtension<AbsoluteSendTime>();
58 sender_->SendPacket(packet.get(), PacedPacketInfo());
59 }
60 }
61
DEPRECATED_RtpSenderEgress(const RtpRtcpInterface::Configuration & config,RtpPacketHistory * packet_history)62 DEPRECATED_RtpSenderEgress::DEPRECATED_RtpSenderEgress(
63 const RtpRtcpInterface::Configuration& config,
64 RtpPacketHistory* packet_history)
65 : ssrc_(config.local_media_ssrc),
66 rtx_ssrc_(config.rtx_send_ssrc),
67 flexfec_ssrc_(config.fec_generator ? config.fec_generator->FecSsrc()
68 : absl::nullopt),
69 populate_network2_timestamp_(config.populate_network2_timestamp),
70 clock_(config.clock),
71 packet_history_(packet_history),
72 transport_(config.outgoing_transport),
73 event_log_(config.event_log),
74 is_audio_(config.audio),
75 need_rtp_packet_infos_(config.need_rtp_packet_infos),
76 transport_feedback_observer_(config.transport_feedback_callback),
77 send_side_delay_observer_(config.send_side_delay_observer),
78 send_packet_observer_(config.send_packet_observer),
79 rtp_stats_callback_(config.rtp_stats_callback),
80 bitrate_callback_(config.send_bitrate_observer),
81 media_has_been_sent_(false),
82 force_part_of_allocation_(false),
83 timestamp_offset_(0),
84 max_delay_it_(send_delays_.end()),
85 sum_delays_ms_(0),
86 send_rates_(kNumMediaTypes,
87 {kBitrateStatisticsWindowMs, RateStatistics::kBpsScale}),
88 rtp_sequence_number_map_(need_rtp_packet_infos_
89 ? std::make_unique<RtpSequenceNumberMap>(
90 kRtpSequenceNumberMapMaxEntries)
91 : nullptr) {}
92
SendPacket(RtpPacketToSend * packet,const PacedPacketInfo & pacing_info)93 void DEPRECATED_RtpSenderEgress::SendPacket(
94 RtpPacketToSend* packet,
95 const PacedPacketInfo& pacing_info) {
96 RTC_DCHECK(packet);
97
98 const uint32_t packet_ssrc = packet->Ssrc();
99 RTC_DCHECK(packet->packet_type().has_value());
100 RTC_DCHECK(HasCorrectSsrc(*packet));
101 Timestamp now = clock_->CurrentTime();
102 int64_t now_ms = now.ms();
103
104 if (is_audio_) {
105 #if BWE_TEST_LOGGING_COMPILE_TIME_ENABLE
106 BWE_TEST_LOGGING_PLOT_WITH_SSRC(1, "AudioTotBitrate_kbps", now_ms,
107 GetSendRates().Sum().kbps(), packet_ssrc);
108 BWE_TEST_LOGGING_PLOT_WITH_SSRC(
109 1, "AudioNackBitrate_kbps", now_ms,
110 GetSendRates()[RtpPacketMediaType::kRetransmission].kbps(),
111 packet_ssrc);
112 #endif
113 } else {
114 #if BWE_TEST_LOGGING_COMPILE_TIME_ENABLE
115 BWE_TEST_LOGGING_PLOT_WITH_SSRC(1, "VideoTotBitrate_kbps", now_ms,
116 GetSendRates().Sum().kbps(), packet_ssrc);
117 BWE_TEST_LOGGING_PLOT_WITH_SSRC(
118 1, "VideoNackBitrate_kbps", now_ms,
119 GetSendRates()[RtpPacketMediaType::kRetransmission].kbps(),
120 packet_ssrc);
121 #endif
122 }
123
124 PacketOptions options;
125 {
126 MutexLock lock(&lock_);
127 options.included_in_allocation = force_part_of_allocation_;
128
129 if (need_rtp_packet_infos_ &&
130 packet->packet_type() == RtpPacketToSend::Type::kVideo) {
131 RTC_DCHECK(rtp_sequence_number_map_);
132 // Last packet of a frame, add it to sequence number info map.
133 const uint32_t timestamp = packet->Timestamp() - timestamp_offset_;
134 bool is_first_packet_of_frame = packet->is_first_packet_of_frame();
135 bool is_last_packet_of_frame = packet->Marker();
136
137 rtp_sequence_number_map_->InsertPacket(
138 packet->SequenceNumber(),
139 RtpSequenceNumberMap::Info(timestamp, is_first_packet_of_frame,
140 is_last_packet_of_frame));
141 }
142 }
143
144 // Bug webrtc:7859. While FEC is invoked from rtp_sender_video, and not after
145 // the pacer, these modifications of the header below are happening after the
146 // FEC protection packets are calculated. This will corrupt recovered packets
147 // at the same place. It's not an issue for extensions, which are present in
148 // all the packets (their content just may be incorrect on recovered packets).
149 // In case of VideoTimingExtension, since it's present not in every packet,
150 // data after rtp header may be corrupted if these packets are protected by
151 // the FEC.
152 int64_t diff_ms = now_ms - packet->capture_time().ms();
153 if (packet->HasExtension<TransmissionOffset>()) {
154 packet->SetExtension<TransmissionOffset>(kTimestampTicksPerMs * diff_ms);
155 }
156 if (packet->HasExtension<AbsoluteSendTime>()) {
157 packet->SetExtension<AbsoluteSendTime>(AbsoluteSendTime::To24Bits(now));
158 }
159
160 if (packet->HasExtension<VideoTimingExtension>()) {
161 if (populate_network2_timestamp_) {
162 packet->set_network2_time(now);
163 } else {
164 packet->set_pacer_exit_time(now);
165 }
166 }
167
168 const bool is_media = packet->packet_type() == RtpPacketMediaType::kAudio ||
169 packet->packet_type() == RtpPacketMediaType::kVideo;
170
171 // Downstream code actually uses this flag to distinguish between media and
172 // everything else.
173 options.is_retransmit = !is_media;
174 if (auto packet_id = packet->GetExtension<TransportSequenceNumber>()) {
175 options.packet_id = *packet_id;
176 options.included_in_feedback = true;
177 options.included_in_allocation = true;
178 AddPacketToTransportFeedback(*packet_id, *packet, pacing_info);
179 }
180
181 options.additional_data = packet->additional_data();
182
183 if (packet->packet_type() != RtpPacketMediaType::kPadding &&
184 packet->packet_type() != RtpPacketMediaType::kRetransmission) {
185 UpdateDelayStatistics(packet->capture_time().ms(), now_ms, packet_ssrc);
186 UpdateOnSendPacket(options.packet_id, packet->capture_time().ms(),
187 packet_ssrc);
188 }
189
190 const bool send_success = SendPacketToNetwork(*packet, options, pacing_info);
191
192 // Put packet in retransmission history or update pending status even if
193 // actual sending fails.
194 if (is_media && packet->allow_retransmission()) {
195 packet_history_->PutRtpPacket(std::make_unique<RtpPacketToSend>(*packet),
196 now);
197 } else if (packet->retransmitted_sequence_number()) {
198 packet_history_->MarkPacketAsSent(*packet->retransmitted_sequence_number());
199 }
200
201 if (send_success) {
202 MutexLock lock(&lock_);
203 UpdateRtpStats(*packet);
204 media_has_been_sent_ = true;
205 }
206 }
207
ProcessBitrateAndNotifyObservers()208 void DEPRECATED_RtpSenderEgress::ProcessBitrateAndNotifyObservers() {
209 if (!bitrate_callback_)
210 return;
211
212 MutexLock lock(&lock_);
213 RtpSendRates send_rates = GetSendRatesLocked();
214 bitrate_callback_->Notify(
215 send_rates.Sum().bps(),
216 send_rates[RtpPacketMediaType::kRetransmission].bps(), ssrc_);
217 }
218
GetSendRates() const219 RtpSendRates DEPRECATED_RtpSenderEgress::GetSendRates() const {
220 MutexLock lock(&lock_);
221 return GetSendRatesLocked();
222 }
223
GetSendRatesLocked() const224 RtpSendRates DEPRECATED_RtpSenderEgress::GetSendRatesLocked() const {
225 const int64_t now_ms = clock_->TimeInMilliseconds();
226 RtpSendRates current_rates;
227 for (size_t i = 0; i < kNumMediaTypes; ++i) {
228 RtpPacketMediaType type = static_cast<RtpPacketMediaType>(i);
229 current_rates[type] =
230 DataRate::BitsPerSec(send_rates_[i].Rate(now_ms).value_or(0));
231 }
232 return current_rates;
233 }
234
GetDataCounters(StreamDataCounters * rtp_stats,StreamDataCounters * rtx_stats) const235 void DEPRECATED_RtpSenderEgress::GetDataCounters(
236 StreamDataCounters* rtp_stats,
237 StreamDataCounters* rtx_stats) const {
238 MutexLock lock(&lock_);
239 *rtp_stats = rtp_stats_;
240 *rtx_stats = rtx_rtp_stats_;
241 }
242
ForceIncludeSendPacketsInAllocation(bool part_of_allocation)243 void DEPRECATED_RtpSenderEgress::ForceIncludeSendPacketsInAllocation(
244 bool part_of_allocation) {
245 MutexLock lock(&lock_);
246 force_part_of_allocation_ = part_of_allocation;
247 }
248
MediaHasBeenSent() const249 bool DEPRECATED_RtpSenderEgress::MediaHasBeenSent() const {
250 MutexLock lock(&lock_);
251 return media_has_been_sent_;
252 }
253
SetMediaHasBeenSent(bool media_sent)254 void DEPRECATED_RtpSenderEgress::SetMediaHasBeenSent(bool media_sent) {
255 MutexLock lock(&lock_);
256 media_has_been_sent_ = media_sent;
257 }
258
SetTimestampOffset(uint32_t timestamp)259 void DEPRECATED_RtpSenderEgress::SetTimestampOffset(uint32_t timestamp) {
260 MutexLock lock(&lock_);
261 timestamp_offset_ = timestamp;
262 }
263
264 std::vector<RtpSequenceNumberMap::Info>
GetSentRtpPacketInfos(rtc::ArrayView<const uint16_t> sequence_numbers) const265 DEPRECATED_RtpSenderEgress::GetSentRtpPacketInfos(
266 rtc::ArrayView<const uint16_t> sequence_numbers) const {
267 RTC_DCHECK(!sequence_numbers.empty());
268 if (!need_rtp_packet_infos_) {
269 return std::vector<RtpSequenceNumberMap::Info>();
270 }
271
272 std::vector<RtpSequenceNumberMap::Info> results;
273 results.reserve(sequence_numbers.size());
274
275 MutexLock lock(&lock_);
276 for (uint16_t sequence_number : sequence_numbers) {
277 const auto& info = rtp_sequence_number_map_->Get(sequence_number);
278 if (!info) {
279 // The empty vector will be returned. We can delay the clearing
280 // of the vector until after we exit the critical section.
281 return std::vector<RtpSequenceNumberMap::Info>();
282 }
283 results.push_back(*info);
284 }
285
286 return results;
287 }
288
HasCorrectSsrc(const RtpPacketToSend & packet) const289 bool DEPRECATED_RtpSenderEgress::HasCorrectSsrc(
290 const RtpPacketToSend& packet) const {
291 switch (*packet.packet_type()) {
292 case RtpPacketMediaType::kAudio:
293 case RtpPacketMediaType::kVideo:
294 return packet.Ssrc() == ssrc_;
295 case RtpPacketMediaType::kRetransmission:
296 case RtpPacketMediaType::kPadding:
297 // Both padding and retransmission must be on either the media or the
298 // RTX stream.
299 return packet.Ssrc() == rtx_ssrc_ || packet.Ssrc() == ssrc_;
300 case RtpPacketMediaType::kForwardErrorCorrection:
301 // FlexFEC is on separate SSRC, ULPFEC uses media SSRC.
302 return packet.Ssrc() == ssrc_ || packet.Ssrc() == flexfec_ssrc_;
303 }
304 return false;
305 }
306
AddPacketToTransportFeedback(uint16_t packet_id,const RtpPacketToSend & packet,const PacedPacketInfo & pacing_info)307 void DEPRECATED_RtpSenderEgress::AddPacketToTransportFeedback(
308 uint16_t packet_id,
309 const RtpPacketToSend& packet,
310 const PacedPacketInfo& pacing_info) {
311 if (transport_feedback_observer_) {
312 RtpPacketSendInfo packet_info;
313 packet_info.media_ssrc = ssrc_;
314 packet_info.transport_sequence_number = packet_id;
315 packet_info.rtp_sequence_number = packet.SequenceNumber();
316 packet_info.length = packet.size();
317 packet_info.pacing_info = pacing_info;
318 packet_info.packet_type = packet.packet_type();
319 transport_feedback_observer_->OnAddPacket(packet_info);
320 }
321 }
322
UpdateDelayStatistics(int64_t capture_time_ms,int64_t now_ms,uint32_t ssrc)323 void DEPRECATED_RtpSenderEgress::UpdateDelayStatistics(int64_t capture_time_ms,
324 int64_t now_ms,
325 uint32_t ssrc) {
326 if (!send_side_delay_observer_ || capture_time_ms <= 0)
327 return;
328
329 int avg_delay_ms = 0;
330 int max_delay_ms = 0;
331 {
332 MutexLock lock(&lock_);
333 // Compute the max and average of the recent capture-to-send delays.
334 // The time complexity of the current approach depends on the distribution
335 // of the delay values. This could be done more efficiently.
336
337 // Remove elements older than kSendSideDelayWindowMs.
338 auto lower_bound =
339 send_delays_.lower_bound(now_ms - kSendSideDelayWindowMs);
340 for (auto it = send_delays_.begin(); it != lower_bound; ++it) {
341 if (max_delay_it_ == it) {
342 max_delay_it_ = send_delays_.end();
343 }
344 sum_delays_ms_ -= it->second;
345 }
346 send_delays_.erase(send_delays_.begin(), lower_bound);
347 if (max_delay_it_ == send_delays_.end()) {
348 // Removed the previous max. Need to recompute.
349 RecomputeMaxSendDelay();
350 }
351
352 // Add the new element.
353 RTC_DCHECK_GE(now_ms, 0);
354 RTC_DCHECK_LE(now_ms, std::numeric_limits<int64_t>::max() / 2);
355 RTC_DCHECK_GE(capture_time_ms, 0);
356 RTC_DCHECK_LE(capture_time_ms, std::numeric_limits<int64_t>::max() / 2);
357 int64_t diff_ms = now_ms - capture_time_ms;
358 RTC_DCHECK_GE(diff_ms, static_cast<int64_t>(0));
359 RTC_DCHECK_LE(diff_ms, std::numeric_limits<int>::max());
360 int new_send_delay = rtc::dchecked_cast<int>(now_ms - capture_time_ms);
361 SendDelayMap::iterator it;
362 bool inserted;
363 std::tie(it, inserted) =
364 send_delays_.insert(std::make_pair(now_ms, new_send_delay));
365 if (!inserted) {
366 // TODO(terelius): If we have multiple delay measurements during the same
367 // millisecond then we keep the most recent one. It is not clear that this
368 // is the right decision, but it preserves an earlier behavior.
369 int previous_send_delay = it->second;
370 sum_delays_ms_ -= previous_send_delay;
371 it->second = new_send_delay;
372 if (max_delay_it_ == it && new_send_delay < previous_send_delay) {
373 RecomputeMaxSendDelay();
374 }
375 }
376 if (max_delay_it_ == send_delays_.end() ||
377 it->second >= max_delay_it_->second) {
378 max_delay_it_ = it;
379 }
380 sum_delays_ms_ += new_send_delay;
381
382 size_t num_delays = send_delays_.size();
383 RTC_DCHECK(max_delay_it_ != send_delays_.end());
384 max_delay_ms = rtc::dchecked_cast<int>(max_delay_it_->second);
385 int64_t avg_ms = (sum_delays_ms_ + num_delays / 2) / num_delays;
386 RTC_DCHECK_GE(avg_ms, static_cast<int64_t>(0));
387 RTC_DCHECK_LE(avg_ms,
388 static_cast<int64_t>(std::numeric_limits<int>::max()));
389 avg_delay_ms =
390 rtc::dchecked_cast<int>((sum_delays_ms_ + num_delays / 2) / num_delays);
391 }
392 send_side_delay_observer_->SendSideDelayUpdated(avg_delay_ms, max_delay_ms,
393 ssrc);
394 }
395
RecomputeMaxSendDelay()396 void DEPRECATED_RtpSenderEgress::RecomputeMaxSendDelay() {
397 max_delay_it_ = send_delays_.begin();
398 for (auto it = send_delays_.begin(); it != send_delays_.end(); ++it) {
399 if (it->second >= max_delay_it_->second) {
400 max_delay_it_ = it;
401 }
402 }
403 }
404
UpdateOnSendPacket(int packet_id,int64_t capture_time_ms,uint32_t ssrc)405 void DEPRECATED_RtpSenderEgress::UpdateOnSendPacket(int packet_id,
406 int64_t capture_time_ms,
407 uint32_t ssrc) {
408 if (!send_packet_observer_ || capture_time_ms <= 0 || packet_id == -1) {
409 return;
410 }
411
412 send_packet_observer_->OnSendPacket(packet_id, capture_time_ms, ssrc);
413 }
414
SendPacketToNetwork(const RtpPacketToSend & packet,const PacketOptions & options,const PacedPacketInfo & pacing_info)415 bool DEPRECATED_RtpSenderEgress::SendPacketToNetwork(
416 const RtpPacketToSend& packet,
417 const PacketOptions& options,
418 const PacedPacketInfo& pacing_info) {
419 int bytes_sent = -1;
420 if (transport_) {
421 bytes_sent = transport_->SendRtp(packet.data(), packet.size(), options)
422 ? static_cast<int>(packet.size())
423 : -1;
424 if (event_log_ && bytes_sent > 0) {
425 event_log_->Log(std::make_unique<RtcEventRtpPacketOutgoing>(
426 packet, pacing_info.probe_cluster_id));
427 }
428 }
429
430 if (bytes_sent <= 0) {
431 RTC_LOG(LS_WARNING) << "Transport failed to send packet.";
432 return false;
433 }
434 return true;
435 }
436
UpdateRtpStats(const RtpPacketToSend & packet)437 void DEPRECATED_RtpSenderEgress::UpdateRtpStats(const RtpPacketToSend& packet) {
438 int64_t now_ms = clock_->TimeInMilliseconds();
439
440 StreamDataCounters* counters =
441 packet.Ssrc() == rtx_ssrc_ ? &rtx_rtp_stats_ : &rtp_stats_;
442
443 if (counters->first_packet_time_ms == -1) {
444 counters->first_packet_time_ms = now_ms;
445 }
446
447 if (packet.packet_type() == RtpPacketMediaType::kForwardErrorCorrection) {
448 counters->fec.AddPacket(packet);
449 }
450
451 if (packet.packet_type() == RtpPacketMediaType::kRetransmission) {
452 counters->retransmitted.AddPacket(packet);
453 }
454 counters->transmitted.AddPacket(packet);
455
456 RTC_DCHECK(packet.packet_type().has_value());
457 send_rates_[static_cast<size_t>(*packet.packet_type())].Update(packet.size(),
458 now_ms);
459
460 if (rtp_stats_callback_) {
461 rtp_stats_callback_->DataCountersUpdated(*counters, packet.Ssrc());
462 }
463 }
464
465 } // namespace webrtc
466