• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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/rtp_sender_egress.h"
12 
13 #include <algorithm>
14 #include <limits>
15 #include <memory>
16 #include <utility>
17 
18 #include "absl/strings/match.h"
19 #include "api/transport/field_trial_based_config.h"
20 #include "logging/rtc_event_log/events/rtc_event_rtp_packet_outgoing.h"
21 #include "rtc_base/logging.h"
22 #include "rtc_base/task_utils/to_queued_task.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 constexpr TimeDelta kUpdateInterval =
31     TimeDelta::Millis(kBitrateStatisticsWindowMs);
32 
IsEnabled(absl::string_view name,const WebRtcKeyValueConfig * field_trials)33 bool IsEnabled(absl::string_view name,
34                const WebRtcKeyValueConfig* field_trials) {
35   FieldTrialBasedConfig default_trials;
36   auto& trials = field_trials ? *field_trials : default_trials;
37   return absl::StartsWith(trials.Lookup(name), "Enabled");
38 }
39 }  // namespace
40 
NonPacedPacketSender(RtpSenderEgress * sender,SequenceNumberAssigner * sequence_number_assigner)41 RtpSenderEgress::NonPacedPacketSender::NonPacedPacketSender(
42     RtpSenderEgress* sender,
43     SequenceNumberAssigner* sequence_number_assigner)
44     : transport_sequence_number_(0),
45       sender_(sender),
46       sequence_number_assigner_(sequence_number_assigner) {
47   RTC_DCHECK(sequence_number_assigner_);
48 }
49 RtpSenderEgress::NonPacedPacketSender::~NonPacedPacketSender() = default;
50 
EnqueuePackets(std::vector<std::unique_ptr<RtpPacketToSend>> packets)51 void RtpSenderEgress::NonPacedPacketSender::EnqueuePackets(
52     std::vector<std::unique_ptr<RtpPacketToSend>> packets) {
53   for (auto& packet : packets) {
54     PrepareForSend(packet.get());
55     sender_->SendPacket(packet.get(), PacedPacketInfo());
56   }
57   auto fec_packets = sender_->FetchFecPackets();
58   if (!fec_packets.empty()) {
59     // Don't generate sequence numbers for flexfec, they are already running on
60     // an internally maintained sequence.
61     const bool generate_sequence_numbers = !sender_->FlexFecSsrc().has_value();
62 
63     for (auto& packet : fec_packets) {
64       if (generate_sequence_numbers) {
65         sequence_number_assigner_->AssignSequenceNumber(packet.get());
66       }
67       PrepareForSend(packet.get());
68     }
69     EnqueuePackets(std::move(fec_packets));
70   }
71 }
72 
PrepareForSend(RtpPacketToSend * packet)73 void RtpSenderEgress::NonPacedPacketSender::PrepareForSend(
74     RtpPacketToSend* packet) {
75   if (!packet->SetExtension<TransportSequenceNumber>(
76           ++transport_sequence_number_)) {
77     --transport_sequence_number_;
78   }
79   packet->ReserveExtension<TransmissionOffset>();
80   packet->ReserveExtension<AbsoluteSendTime>();
81 }
82 
RtpSenderEgress(const RtpRtcpInterface::Configuration & config,RtpPacketHistory * packet_history)83 RtpSenderEgress::RtpSenderEgress(const RtpRtcpInterface::Configuration& config,
84                                  RtpPacketHistory* packet_history)
85     : worker_queue_(TaskQueueBase::Current()),
86       ssrc_(config.local_media_ssrc),
87       rtx_ssrc_(config.rtx_send_ssrc),
88       flexfec_ssrc_(config.fec_generator ? config.fec_generator->FecSsrc()
89                                          : absl::nullopt),
90       populate_network2_timestamp_(config.populate_network2_timestamp),
91       send_side_bwe_with_overhead_(
92           IsEnabled("WebRTC-SendSideBwe-WithOverhead", config.field_trials)),
93       clock_(config.clock),
94       packet_history_(packet_history),
95       transport_(config.outgoing_transport),
96       event_log_(config.event_log),
97 #if BWE_TEST_LOGGING_COMPILE_TIME_ENABLE
98       is_audio_(config.audio),
99 #endif
100       need_rtp_packet_infos_(config.need_rtp_packet_infos),
101       fec_generator_(
102           IsEnabled("WebRTC-DeferredFecGeneration", config.field_trials)
103               ? config.fec_generator
104               : nullptr),
105       transport_feedback_observer_(config.transport_feedback_callback),
106       send_side_delay_observer_(config.send_side_delay_observer),
107       send_packet_observer_(config.send_packet_observer),
108       rtp_stats_callback_(config.rtp_stats_callback),
109       bitrate_callback_(config.send_bitrate_observer),
110       media_has_been_sent_(false),
111       force_part_of_allocation_(false),
112       timestamp_offset_(0),
113       max_delay_it_(send_delays_.end()),
114       sum_delays_ms_(0),
115       total_packet_send_delay_ms_(0),
116       send_rates_(kNumMediaTypes,
117                   {kBitrateStatisticsWindowMs, RateStatistics::kBpsScale}),
118       rtp_sequence_number_map_(need_rtp_packet_infos_
119                                    ? std::make_unique<RtpSequenceNumberMap>(
120                                          kRtpSequenceNumberMapMaxEntries)
121                                    : nullptr) {
122   RTC_DCHECK(worker_queue_);
123   pacer_checker_.Detach();
124   if (bitrate_callback_) {
125     update_task_ = RepeatingTaskHandle::DelayedStart(worker_queue_,
__anone124a4c50202() 126                                                      kUpdateInterval, [this]() {
127                                                        PeriodicUpdate();
128                                                        return kUpdateInterval;
129                                                      });
130   }
131 }
132 
~RtpSenderEgress()133 RtpSenderEgress::~RtpSenderEgress() {
134   RTC_DCHECK_RUN_ON(worker_queue_);
135   update_task_.Stop();
136 }
137 
SendPacket(RtpPacketToSend * packet,const PacedPacketInfo & pacing_info)138 void RtpSenderEgress::SendPacket(RtpPacketToSend* packet,
139                                  const PacedPacketInfo& pacing_info) {
140   RTC_DCHECK_RUN_ON(&pacer_checker_);
141   RTC_DCHECK(packet);
142 
143   RTC_DCHECK(packet->packet_type().has_value());
144   RTC_DCHECK(HasCorrectSsrc(*packet));
145 
146   const uint32_t packet_ssrc = packet->Ssrc();
147   const int64_t now_ms = clock_->TimeInMilliseconds();
148 
149 #if BWE_TEST_LOGGING_COMPILE_TIME_ENABLE
150   worker_queue_->PostTask(
151       ToQueuedTask(task_safety_, [this, now_ms, packet_ssrc]() {
152         BweTestLoggingPlot(now_ms, packet_ssrc);
153       }));
154 #endif
155 
156   if (need_rtp_packet_infos_ &&
157       packet->packet_type() == RtpPacketToSend::Type::kVideo) {
158     worker_queue_->PostTask(ToQueuedTask(
159         task_safety_,
160         [this, packet_timestamp = packet->Timestamp(),
161          is_first_packet_of_frame = packet->is_first_packet_of_frame(),
162          is_last_packet_of_frame = packet->Marker(),
163          sequence_number = packet->SequenceNumber()]() {
164           RTC_DCHECK_RUN_ON(worker_queue_);
165           // Last packet of a frame, add it to sequence number info map.
166           const uint32_t timestamp = packet_timestamp - timestamp_offset_;
167           rtp_sequence_number_map_->InsertPacket(
168               sequence_number,
169               RtpSequenceNumberMap::Info(timestamp, is_first_packet_of_frame,
170                                          is_last_packet_of_frame));
171         }));
172   }
173 
174   if (fec_generator_ && packet->fec_protect_packet()) {
175     // Deferred fec generation is used, add packet to generator.
176     RTC_DCHECK(fec_generator_);
177     RTC_DCHECK(packet->packet_type() == RtpPacketMediaType::kVideo);
178     absl::optional<std::pair<FecProtectionParams, FecProtectionParams>>
179         new_fec_params;
180     {
181       MutexLock lock(&lock_);
182       new_fec_params.swap(pending_fec_params_);
183     }
184     if (new_fec_params) {
185       fec_generator_->SetProtectionParameters(new_fec_params->first,
186                                               new_fec_params->second);
187     }
188     if (packet->is_red()) {
189       RtpPacketToSend unpacked_packet(*packet);
190 
191       const rtc::CopyOnWriteBuffer buffer = packet->Buffer();
192       // Grab media payload type from RED header.
193       const size_t headers_size = packet->headers_size();
194       unpacked_packet.SetPayloadType(buffer[headers_size]);
195 
196       // Copy the media payload into the unpacked buffer.
197       uint8_t* payload_buffer =
198           unpacked_packet.SetPayloadSize(packet->payload_size() - 1);
199       std::copy(&packet->payload()[0] + 1,
200                 &packet->payload()[0] + packet->payload_size(), payload_buffer);
201 
202       fec_generator_->AddPacketAndGenerateFec(unpacked_packet);
203     } else {
204       // If not RED encapsulated - we can just insert packet directly.
205       fec_generator_->AddPacketAndGenerateFec(*packet);
206     }
207   }
208 
209   // Bug webrtc:7859. While FEC is invoked from rtp_sender_video, and not after
210   // the pacer, these modifications of the header below are happening after the
211   // FEC protection packets are calculated. This will corrupt recovered packets
212   // at the same place. It's not an issue for extensions, which are present in
213   // all the packets (their content just may be incorrect on recovered packets).
214   // In case of VideoTimingExtension, since it's present not in every packet,
215   // data after rtp header may be corrupted if these packets are protected by
216   // the FEC.
217   int64_t diff_ms = now_ms - packet->capture_time_ms();
218   if (packet->HasExtension<TransmissionOffset>()) {
219     packet->SetExtension<TransmissionOffset>(kTimestampTicksPerMs * diff_ms);
220   }
221   if (packet->HasExtension<AbsoluteSendTime>()) {
222     packet->SetExtension<AbsoluteSendTime>(
223         AbsoluteSendTime::MsTo24Bits(now_ms));
224   }
225 
226   if (packet->HasExtension<VideoTimingExtension>()) {
227     if (populate_network2_timestamp_) {
228       packet->set_network2_time_ms(now_ms);
229     } else {
230       packet->set_pacer_exit_time_ms(now_ms);
231     }
232   }
233 
234   const bool is_media = packet->packet_type() == RtpPacketMediaType::kAudio ||
235                         packet->packet_type() == RtpPacketMediaType::kVideo;
236 
237   PacketOptions options;
238   {
239     MutexLock lock(&lock_);
240     options.included_in_allocation = force_part_of_allocation_;
241   }
242 
243   // Downstream code actually uses this flag to distinguish between media and
244   // everything else.
245   options.is_retransmit = !is_media;
246   if (auto packet_id = packet->GetExtension<TransportSequenceNumber>()) {
247     options.packet_id = *packet_id;
248     options.included_in_feedback = true;
249     options.included_in_allocation = true;
250     AddPacketToTransportFeedback(*packet_id, *packet, pacing_info);
251   }
252 
253   options.application_data.assign(packet->application_data().begin(),
254                                   packet->application_data().end());
255 
256   if (packet->packet_type() != RtpPacketMediaType::kPadding &&
257       packet->packet_type() != RtpPacketMediaType::kRetransmission) {
258     UpdateDelayStatistics(packet->capture_time_ms(), now_ms, packet_ssrc);
259     UpdateOnSendPacket(options.packet_id, packet->capture_time_ms(),
260                        packet_ssrc);
261   }
262 
263   const bool send_success = SendPacketToNetwork(*packet, options, pacing_info);
264 
265   // Put packet in retransmission history or update pending status even if
266   // actual sending fails.
267   if (is_media && packet->allow_retransmission()) {
268     packet_history_->PutRtpPacket(std::make_unique<RtpPacketToSend>(*packet),
269                                   now_ms);
270   } else if (packet->retransmitted_sequence_number()) {
271     packet_history_->MarkPacketAsSent(*packet->retransmitted_sequence_number());
272   }
273 
274   if (send_success) {
275     // |media_has_been_sent_| is used by RTPSender to figure out if it can send
276     // padding in the absence of transport-cc or abs-send-time.
277     // In those cases media must be sent first to set a reference timestamp.
278     media_has_been_sent_ = true;
279 
280     // TODO(sprang): Add support for FEC protecting all header extensions, add
281     // media packet to generator here instead.
282 
283     RTC_DCHECK(packet->packet_type().has_value());
284     RtpPacketMediaType packet_type = *packet->packet_type();
285     RtpPacketCounter counter(*packet);
286     size_t size = packet->size();
287     worker_queue_->PostTask(
288         ToQueuedTask(task_safety_, [this, now_ms, packet_ssrc, packet_type,
289                                     counter = std::move(counter), size]() {
290           RTC_DCHECK_RUN_ON(worker_queue_);
291           UpdateRtpStats(now_ms, packet_ssrc, packet_type, std::move(counter),
292                          size);
293         }));
294   }
295 }
296 
GetSendRates() const297 RtpSendRates RtpSenderEgress::GetSendRates() const {
298   MutexLock lock(&lock_);
299   const int64_t now_ms = clock_->TimeInMilliseconds();
300   return GetSendRatesLocked(now_ms);
301 }
302 
GetSendRatesLocked(int64_t now_ms) const303 RtpSendRates RtpSenderEgress::GetSendRatesLocked(int64_t now_ms) const {
304   RtpSendRates current_rates;
305   for (size_t i = 0; i < kNumMediaTypes; ++i) {
306     RtpPacketMediaType type = static_cast<RtpPacketMediaType>(i);
307     current_rates[type] =
308         DataRate::BitsPerSec(send_rates_[i].Rate(now_ms).value_or(0));
309   }
310   return current_rates;
311 }
312 
GetDataCounters(StreamDataCounters * rtp_stats,StreamDataCounters * rtx_stats) const313 void RtpSenderEgress::GetDataCounters(StreamDataCounters* rtp_stats,
314                                       StreamDataCounters* rtx_stats) const {
315   // TODO(bugs.webrtc.org/11581): make sure rtx_rtp_stats_ and rtp_stats_ are
316   // only touched on the worker thread.
317   MutexLock lock(&lock_);
318   *rtp_stats = rtp_stats_;
319   *rtx_stats = rtx_rtp_stats_;
320 }
321 
ForceIncludeSendPacketsInAllocation(bool part_of_allocation)322 void RtpSenderEgress::ForceIncludeSendPacketsInAllocation(
323     bool part_of_allocation) {
324   MutexLock lock(&lock_);
325   force_part_of_allocation_ = part_of_allocation;
326 }
327 
MediaHasBeenSent() const328 bool RtpSenderEgress::MediaHasBeenSent() const {
329   RTC_DCHECK_RUN_ON(&pacer_checker_);
330   return media_has_been_sent_;
331 }
332 
SetMediaHasBeenSent(bool media_sent)333 void RtpSenderEgress::SetMediaHasBeenSent(bool media_sent) {
334   RTC_DCHECK_RUN_ON(&pacer_checker_);
335   media_has_been_sent_ = media_sent;
336 }
337 
SetTimestampOffset(uint32_t timestamp)338 void RtpSenderEgress::SetTimestampOffset(uint32_t timestamp) {
339   RTC_DCHECK_RUN_ON(worker_queue_);
340   timestamp_offset_ = timestamp;
341 }
342 
GetSentRtpPacketInfos(rtc::ArrayView<const uint16_t> sequence_numbers) const343 std::vector<RtpSequenceNumberMap::Info> RtpSenderEgress::GetSentRtpPacketInfos(
344     rtc::ArrayView<const uint16_t> sequence_numbers) const {
345   RTC_DCHECK_RUN_ON(worker_queue_);
346   RTC_DCHECK(!sequence_numbers.empty());
347   if (!need_rtp_packet_infos_) {
348     return std::vector<RtpSequenceNumberMap::Info>();
349   }
350 
351   std::vector<RtpSequenceNumberMap::Info> results;
352   results.reserve(sequence_numbers.size());
353 
354   for (uint16_t sequence_number : sequence_numbers) {
355     const auto& info = rtp_sequence_number_map_->Get(sequence_number);
356     if (!info) {
357       // The empty vector will be returned. We can delay the clearing
358       // of the vector until after we exit the critical section.
359       return std::vector<RtpSequenceNumberMap::Info>();
360     }
361     results.push_back(*info);
362   }
363 
364   return results;
365 }
366 
SetFecProtectionParameters(const FecProtectionParams & delta_params,const FecProtectionParams & key_params)367 void RtpSenderEgress::SetFecProtectionParameters(
368     const FecProtectionParams& delta_params,
369     const FecProtectionParams& key_params) {
370   // TODO(sprang): Post task to pacer queue instead, one pacer is fully
371   // migrated to a task queue.
372   MutexLock lock(&lock_);
373   pending_fec_params_.emplace(delta_params, key_params);
374 }
375 
376 std::vector<std::unique_ptr<RtpPacketToSend>>
FetchFecPackets()377 RtpSenderEgress::FetchFecPackets() {
378   RTC_DCHECK_RUN_ON(&pacer_checker_);
379   if (fec_generator_) {
380     return fec_generator_->GetFecPackets();
381   }
382   return {};
383 }
384 
HasCorrectSsrc(const RtpPacketToSend & packet) const385 bool RtpSenderEgress::HasCorrectSsrc(const RtpPacketToSend& packet) const {
386   switch (*packet.packet_type()) {
387     case RtpPacketMediaType::kAudio:
388     case RtpPacketMediaType::kVideo:
389       return packet.Ssrc() == ssrc_;
390     case RtpPacketMediaType::kRetransmission:
391     case RtpPacketMediaType::kPadding:
392       // Both padding and retransmission must be on either the media or the
393       // RTX stream.
394       return packet.Ssrc() == rtx_ssrc_ || packet.Ssrc() == ssrc_;
395     case RtpPacketMediaType::kForwardErrorCorrection:
396       // FlexFEC is on separate SSRC, ULPFEC uses media SSRC.
397       return packet.Ssrc() == ssrc_ || packet.Ssrc() == flexfec_ssrc_;
398   }
399   return false;
400 }
401 
AddPacketToTransportFeedback(uint16_t packet_id,const RtpPacketToSend & packet,const PacedPacketInfo & pacing_info)402 void RtpSenderEgress::AddPacketToTransportFeedback(
403     uint16_t packet_id,
404     const RtpPacketToSend& packet,
405     const PacedPacketInfo& pacing_info) {
406   if (transport_feedback_observer_) {
407     size_t packet_size = packet.payload_size() + packet.padding_size();
408     if (send_side_bwe_with_overhead_) {
409       packet_size = packet.size();
410     }
411 
412     RtpPacketSendInfo packet_info;
413     packet_info.ssrc = ssrc_;
414     packet_info.transport_sequence_number = packet_id;
415     packet_info.rtp_sequence_number = packet.SequenceNumber();
416     packet_info.length = packet_size;
417     packet_info.pacing_info = pacing_info;
418     packet_info.packet_type = packet.packet_type();
419     transport_feedback_observer_->OnAddPacket(packet_info);
420   }
421 }
422 
UpdateDelayStatistics(int64_t capture_time_ms,int64_t now_ms,uint32_t ssrc)423 void RtpSenderEgress::UpdateDelayStatistics(int64_t capture_time_ms,
424                                             int64_t now_ms,
425                                             uint32_t ssrc) {
426   if (!send_side_delay_observer_ || capture_time_ms <= 0)
427     return;
428 
429   int avg_delay_ms = 0;
430   int max_delay_ms = 0;
431   uint64_t total_packet_send_delay_ms = 0;
432   {
433     MutexLock lock(&lock_);
434     // Compute the max and average of the recent capture-to-send delays.
435     // The time complexity of the current approach depends on the distribution
436     // of the delay values. This could be done more efficiently.
437 
438     // Remove elements older than kSendSideDelayWindowMs.
439     auto lower_bound =
440         send_delays_.lower_bound(now_ms - kSendSideDelayWindowMs);
441     for (auto it = send_delays_.begin(); it != lower_bound; ++it) {
442       if (max_delay_it_ == it) {
443         max_delay_it_ = send_delays_.end();
444       }
445       sum_delays_ms_ -= it->second;
446     }
447     send_delays_.erase(send_delays_.begin(), lower_bound);
448     if (max_delay_it_ == send_delays_.end()) {
449       // Removed the previous max. Need to recompute.
450       RecomputeMaxSendDelay();
451     }
452 
453     // Add the new element.
454     RTC_DCHECK_GE(now_ms, 0);
455     RTC_DCHECK_LE(now_ms, std::numeric_limits<int64_t>::max() / 2);
456     RTC_DCHECK_GE(capture_time_ms, 0);
457     RTC_DCHECK_LE(capture_time_ms, std::numeric_limits<int64_t>::max() / 2);
458     int64_t diff_ms = now_ms - capture_time_ms;
459     RTC_DCHECK_GE(diff_ms, static_cast<int64_t>(0));
460     RTC_DCHECK_LE(diff_ms, std::numeric_limits<int>::max());
461     int new_send_delay = rtc::dchecked_cast<int>(now_ms - capture_time_ms);
462     SendDelayMap::iterator it;
463     bool inserted;
464     std::tie(it, inserted) =
465         send_delays_.insert(std::make_pair(now_ms, new_send_delay));
466     if (!inserted) {
467       // TODO(terelius): If we have multiple delay measurements during the same
468       // millisecond then we keep the most recent one. It is not clear that this
469       // is the right decision, but it preserves an earlier behavior.
470       int previous_send_delay = it->second;
471       sum_delays_ms_ -= previous_send_delay;
472       it->second = new_send_delay;
473       if (max_delay_it_ == it && new_send_delay < previous_send_delay) {
474         RecomputeMaxSendDelay();
475       }
476     }
477     if (max_delay_it_ == send_delays_.end() ||
478         it->second >= max_delay_it_->second) {
479       max_delay_it_ = it;
480     }
481     sum_delays_ms_ += new_send_delay;
482     total_packet_send_delay_ms_ += new_send_delay;
483     total_packet_send_delay_ms = total_packet_send_delay_ms_;
484 
485     size_t num_delays = send_delays_.size();
486     RTC_DCHECK(max_delay_it_ != send_delays_.end());
487     max_delay_ms = rtc::dchecked_cast<int>(max_delay_it_->second);
488     int64_t avg_ms = (sum_delays_ms_ + num_delays / 2) / num_delays;
489     RTC_DCHECK_GE(avg_ms, static_cast<int64_t>(0));
490     RTC_DCHECK_LE(avg_ms,
491                   static_cast<int64_t>(std::numeric_limits<int>::max()));
492     avg_delay_ms =
493         rtc::dchecked_cast<int>((sum_delays_ms_ + num_delays / 2) / num_delays);
494   }
495   send_side_delay_observer_->SendSideDelayUpdated(
496       avg_delay_ms, max_delay_ms, total_packet_send_delay_ms, ssrc);
497 }
498 
RecomputeMaxSendDelay()499 void RtpSenderEgress::RecomputeMaxSendDelay() {
500   max_delay_it_ = send_delays_.begin();
501   for (auto it = send_delays_.begin(); it != send_delays_.end(); ++it) {
502     if (it->second >= max_delay_it_->second) {
503       max_delay_it_ = it;
504     }
505   }
506 }
507 
UpdateOnSendPacket(int packet_id,int64_t capture_time_ms,uint32_t ssrc)508 void RtpSenderEgress::UpdateOnSendPacket(int packet_id,
509                                          int64_t capture_time_ms,
510                                          uint32_t ssrc) {
511   if (!send_packet_observer_ || capture_time_ms <= 0 || packet_id == -1) {
512     return;
513   }
514 
515   send_packet_observer_->OnSendPacket(packet_id, capture_time_ms, ssrc);
516 }
517 
SendPacketToNetwork(const RtpPacketToSend & packet,const PacketOptions & options,const PacedPacketInfo & pacing_info)518 bool RtpSenderEgress::SendPacketToNetwork(const RtpPacketToSend& packet,
519                                           const PacketOptions& options,
520                                           const PacedPacketInfo& pacing_info) {
521   int bytes_sent = -1;
522   if (transport_) {
523     bytes_sent = transport_->SendRtp(packet.data(), packet.size(), options)
524                      ? static_cast<int>(packet.size())
525                      : -1;
526     if (event_log_ && bytes_sent > 0) {
527       event_log_->Log(std::make_unique<RtcEventRtpPacketOutgoing>(
528           packet, pacing_info.probe_cluster_id));
529     }
530   }
531 
532   if (bytes_sent <= 0) {
533     RTC_LOG(LS_WARNING) << "Transport failed to send packet.";
534     return false;
535   }
536   return true;
537 }
538 
UpdateRtpStats(int64_t now_ms,uint32_t packet_ssrc,RtpPacketMediaType packet_type,RtpPacketCounter counter,size_t packet_size)539 void RtpSenderEgress::UpdateRtpStats(int64_t now_ms,
540                                      uint32_t packet_ssrc,
541                                      RtpPacketMediaType packet_type,
542                                      RtpPacketCounter counter,
543                                      size_t packet_size) {
544   RTC_DCHECK_RUN_ON(worker_queue_);
545 
546   // TODO(bugs.webrtc.org/11581): send_rates_ should be touched only on the
547   // worker thread.
548   RtpSendRates send_rates;
549   {
550     MutexLock lock(&lock_);
551 
552     // TODO(bugs.webrtc.org/11581): make sure rtx_rtp_stats_ and rtp_stats_ are
553     // only touched on the worker thread.
554     StreamDataCounters* counters =
555         packet_ssrc == rtx_ssrc_ ? &rtx_rtp_stats_ : &rtp_stats_;
556 
557     if (counters->first_packet_time_ms == -1) {
558       counters->first_packet_time_ms = now_ms;
559     }
560 
561     if (packet_type == RtpPacketMediaType::kForwardErrorCorrection) {
562       counters->fec.Add(counter);
563     } else if (packet_type == RtpPacketMediaType::kRetransmission) {
564       counters->retransmitted.Add(counter);
565     }
566     counters->transmitted.Add(counter);
567 
568     send_rates_[static_cast<size_t>(packet_type)].Update(packet_size, now_ms);
569     if (bitrate_callback_) {
570       send_rates = GetSendRatesLocked(now_ms);
571     }
572 
573     if (rtp_stats_callback_) {
574       rtp_stats_callback_->DataCountersUpdated(*counters, packet_ssrc);
575     }
576   }
577 
578   // The bitrate_callback_ and rtp_stats_callback_ pointers in practice point
579   // to the same object, so these callbacks could be consolidated into one.
580   if (bitrate_callback_) {
581     bitrate_callback_->Notify(
582         send_rates.Sum().bps(),
583         send_rates[RtpPacketMediaType::kRetransmission].bps(), ssrc_);
584   }
585 }
586 
PeriodicUpdate()587 void RtpSenderEgress::PeriodicUpdate() {
588   RTC_DCHECK_RUN_ON(worker_queue_);
589   RTC_DCHECK(bitrate_callback_);
590   RtpSendRates send_rates = GetSendRates();
591   bitrate_callback_->Notify(
592       send_rates.Sum().bps(),
593       send_rates[RtpPacketMediaType::kRetransmission].bps(), ssrc_);
594 }
595 
596 #if BWE_TEST_LOGGING_COMPILE_TIME_ENABLE
BweTestLoggingPlot(int64_t now_ms,uint32_t packet_ssrc)597 void RtpSenderEgress::BweTestLoggingPlot(int64_t now_ms, uint32_t packet_ssrc) {
598   RTC_DCHECK_RUN_ON(worker_queue_);
599 
600   const auto rates = GetSendRates();
601   if (is_audio_) {
602     BWE_TEST_LOGGING_PLOT_WITH_SSRC(1, "AudioTotBitrate_kbps", now_ms,
603                                     rates.Sum().kbps(), packet_ssrc);
604     BWE_TEST_LOGGING_PLOT_WITH_SSRC(
605         1, "AudioNackBitrate_kbps", now_ms,
606         rates[RtpPacketMediaType::kRetransmission].kbps(), packet_ssrc);
607   } else {
608     BWE_TEST_LOGGING_PLOT_WITH_SSRC(1, "VideoTotBitrate_kbps", now_ms,
609                                     rates.Sum().kbps(), packet_ssrc);
610     BWE_TEST_LOGGING_PLOT_WITH_SSRC(
611         1, "VideoNackBitrate_kbps", now_ms,
612         rates[RtpPacketMediaType::kRetransmission].kbps(), packet_ssrc);
613   }
614 }
615 #endif  // BWE_TEST_LOGGING_COMPILE_TIME_ENABLE
616 
617 }  // namespace webrtc
618