third_party/chromium/media/cast/net/pacing/paced_sender.cc - cobalt - Git at Google

 // Copyright 2014 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.

 #include "media/cast/net/pacing/paced_sender.h"

 #include "base/big_endian.h"
 #include "base/bind.h"
 #include "base/logging.h"
 #include "base/numerics/safe_conversions.h"

 namespace media {
 namespace cast {

 namespace {

 static const int64_t kPacingIntervalMs = 10;
 // Each frame will be split into no more than kPacingMaxBurstsPerFrame
 // bursts of packets.
 static const size_t kPacingMaxBurstsPerFrame = 3;
 static const size_t kMaxDedupeWindowMs = 500;

 }  // namespace

 DedupInfo::DedupInfo() : last_byte_acked_for_audio(0) {}

 PacketKey::PacketKey() : ssrc(0), packet_id(0) {}

 PacketKey::PacketKey(base::TimeTicks capture_time,
                      uint32_t ssrc,
                      FrameId frame_id,
                      uint16_t packet_id)
     : capture_time(capture_time),
       ssrc(ssrc),
       frame_id(frame_id),
       packet_id(packet_id) {}

 PacketKey::PacketKey(const PacketKey& other) = default;

 PacketKey::~PacketKey() = default;

 struct PacedSender::PacketSendRecord {
   PacketSendRecord()
       : last_byte_sent(0), last_byte_sent_for_audio(0), cancel_count(0) {}

   base::TimeTicks time;    // Time when the packet was sent.
   int64_t last_byte_sent;  // Number of bytes sent to network just after this
                            // packet was sent.
   int64_t last_byte_sent_for_audio;  // Number of bytes sent to network from
                                      // audio stream just before this packet.
   int cancel_count;  // Number of times the packet was canceled (debugging).
 };

 struct PacedSender::RtpSession {
   explicit RtpSession(bool is_audio_stream)
       : last_byte_sent(0), is_audio(is_audio_stream) {}
   RtpSession() = default;

   // Tracks recently-logged RTP timestamps so that it can expand the truncated
   // values found in packets.
   RtpTimeTicks last_logged_rtp_timestamp_;
   int64_t last_byte_sent;
   bool is_audio;
 };

 PacedSender::PacedSender(
     size_t target_burst_size,
     size_t max_burst_size,
     const base::TickClock* clock,
     std::vector<PacketEvent>* recent_packet_events,
     PacketTransport* transport,
     const scoped_refptr<base::SingleThreadTaskRunner>& transport_task_runner)
     : clock_(clock),
       recent_packet_events_(recent_packet_events),
       transport_(transport),
       transport_task_runner_(transport_task_runner),
       last_byte_sent_for_audio_(0),
       target_burst_size_(target_burst_size),
       max_burst_size_(max_burst_size),
       current_max_burst_size_(target_burst_size_),
       next_max_burst_size_(target_burst_size_),
       next_next_max_burst_size_(target_burst_size_),
       current_burst_size_(0),
       state_(State_Unblocked) {}

 PacedSender::~PacedSender() = default;

 void PacedSender::RegisterSsrc(uint32_t ssrc, bool is_audio) {
   if (sessions_.find(ssrc) != sessions_.end())
     DVLOG(1) << "Re-register ssrc: " << ssrc;

   sessions_[ssrc] = RtpSession(is_audio);
 }

 void PacedSender::RegisterPrioritySsrc(uint32_t ssrc) {
   priority_ssrcs_.push_back(ssrc);
 }

 int64_t PacedSender::GetLastByteSentForPacket(const PacketKey& packet_key) {
   PacketSendHistory::const_iterator it = send_history_.find(packet_key);
   if (it == send_history_.end())
     return 0;
   return it->second.last_byte_sent;
 }

 int64_t PacedSender::GetLastByteSentForSsrc(uint32_t ssrc) {
   auto it = sessions_.find(ssrc);
   // Return 0 for unknown session.
   if (it == sessions_.end())
     return 0;
   return it->second.last_byte_sent;
 }

 bool PacedSender::SendPackets(const SendPacketVector& packets) {
   if (packets.empty()) {
     return true;
   }
   const bool high_priority = IsHighPriority(packets.begin()->first);
   for (size_t i = 0; i < packets.size(); i++) {
     if (VLOG_IS_ON(2)) {
       PacketSendHistory::const_iterator history_it =
           send_history_.find(packets[i].first);
       if (history_it != send_history_.end() &&
           history_it->second.cancel_count > 0) {
         VLOG(2) << "PacedSender::SendPackets() called for packet CANCELED "
                 << history_it->second.cancel_count << " times: "
                 << "ssrc=" << packets[i].first.ssrc
                 << ", frame_id=" << packets[i].first.frame_id
                 << ", packet_id=" << packets[i].first.packet_id;
       }
     }

     DCHECK(IsHighPriority(packets[i].first) == high_priority);
     if (high_priority) {
       priority_packet_list_[packets[i].first] =
           make_pair(PacketType_Normal, packets[i].second);
     } else {
       packet_list_[packets[i].first] =
           make_pair(PacketType_Normal, packets[i].second);
     }
   }
   if (state_ == State_Unblocked) {
     SendStoredPackets();
   }
   return true;
 }

 bool PacedSender::ShouldResend(const PacketKey& packet_key,
                                const DedupInfo& dedup_info,
                                const base::TimeTicks& now) {
   PacketSendHistory::const_iterator it = send_history_.find(packet_key);

   // No history of previous transmission. It might be sent too long ago.
   if (it == send_history_.end())
     return true;

   // Suppose there is request to retransmit X and there is an audio
   // packet Y sent just before X. Reject retransmission of X if ACK for
   // Y has not been received.
   // Only do this for video packets.
   //
   // TODO(miu): This sounds wrong.  Audio packets are always transmitted first
   // (because they are put in |priority_packet_list_|, see PopNextPacket()).
   auto session_it = sessions_.find(packet_key.ssrc);
   // The session should always have been registered in |sessions_|.
   DCHECK(session_it != sessions_.end());
   if (!session_it->second.is_audio) {
     if (dedup_info.last_byte_acked_for_audio &&
         it->second.last_byte_sent_for_audio &&
         dedup_info.last_byte_acked_for_audio <
         it->second.last_byte_sent_for_audio) {
       return false;
     }
   }
   // Retransmission interval has to be greater than |resend_interval|.
   if (now - it->second.time < dedup_info.resend_interval)
     return false;
   return true;
 }

 bool PacedSender::ResendPackets(const SendPacketVector& packets,
                                 const DedupInfo& dedup_info) {
   if (packets.empty()) {
     return true;
   }
   const bool high_priority = IsHighPriority(packets.begin()->first);
   const base::TimeTicks now = clock_->NowTicks();
   for (size_t i = 0; i < packets.size(); i++) {
     if (VLOG_IS_ON(2)) {
       PacketSendHistory::const_iterator history_it =
           send_history_.find(packets[i].first);
       if (history_it != send_history_.end() &&
           history_it->second.cancel_count > 0) {
         VLOG(2) << "PacedSender::ReendPackets() called for packet CANCELED "
                 << history_it->second.cancel_count << " times: "
                 << "ssrc=" << packets[i].first.ssrc
                 << ", frame_id=" << packets[i].first.frame_id
                 << ", packet_id=" << packets[i].first.packet_id;
       }
     }

     if (!ShouldResend(packets[i].first, dedup_info, now)) {
       LogPacketEvent(packets[i].second->data, PACKET_RTX_REJECTED);
       continue;
     }

     DCHECK(IsHighPriority(packets[i].first) == high_priority);
     if (high_priority) {
       priority_packet_list_[packets[i].first] =
           make_pair(PacketType_Resend, packets[i].second);
     } else {
       packet_list_[packets[i].first] =
           make_pair(PacketType_Resend, packets[i].second);
     }
   }
   if (state_ == State_Unblocked) {
     SendStoredPackets();
   }
   return true;
 }

 bool PacedSender::SendRtcpPacket(uint32_t ssrc, PacketRef packet) {
   if (state_ == State_TransportBlocked) {
     const PacketKey key(base::TimeTicks(), ssrc, FrameId::first(), 0);
     priority_packet_list_[key] = make_pair(PacketType_RTCP, packet);
   } else {
     // We pass the RTCP packets straight through.
     if (!transport_->SendPacket(packet,
                                 base::BindOnce(&PacedSender::SendStoredPackets,
                                                weak_factory_.GetWeakPtr()))) {
       state_ = State_TransportBlocked;
     }
   }
   return true;
 }

 void PacedSender::CancelSendingPacket(const PacketKey& packet_key) {
   packet_list_.erase(packet_key);
   priority_packet_list_.erase(packet_key);

   if (VLOG_IS_ON(2)) {
     auto history_it = send_history_.find(packet_key);
     if (history_it != send_history_.end())
       ++history_it->second.cancel_count;
   }
 }

 PacketRef PacedSender::PopNextPacket(PacketType* packet_type,
                                      PacketKey* packet_key) {
   // Always pop from the priority list first.
   PacketList* list =
       !priority_packet_list_.empty() ? &priority_packet_list_ : &packet_list_;
   DCHECK(!list->empty());

   // Determine which packet in the frame should be popped by examining the
   // |send_history_| for prior transmission attempts.  Packets that have never
   // been transmitted will be popped first.  If all packets have transmitted
   // before, pop the one that has not been re-attempted for the longest time.
   auto it = list->begin();
   PacketKey last_key = it->first;
   last_key.packet_id = UINT16_C(0xffff);
   PacketSendHistory::const_iterator history_it =
       send_history_.lower_bound(it->first);
   base::TimeTicks earliest_send_time =
       base::TimeTicks() + base::TimeDelta::Max();
   auto found_it = it;
   while (true) {
     if (history_it == send_history_.end() || it->first < history_it->first) {
       // There is no send history for this packet, which means it has not been
       // transmitted yet.
       found_it = it;
       break;
     }

     DCHECK(it->first == history_it->first);
     if (history_it->second.time < earliest_send_time) {
       earliest_send_time = history_it->second.time;
       found_it = it;
     }

     // Advance to next packet for the current frame, or break if there are no
     // more.
     ++it;
     if (it == list->end() || last_key < it->first)
       break;

     // Advance to next history entry.  Since there may be "holes" in the packet
     // list (e.g., due to packets canceled for retransmission), it's possible
     // |history_it| will have to be advanced more than once even though |it| was
     // only advanced once.
     do {
       ++history_it;
     } while (history_it != send_history_.end() &&
              history_it->first < it->first);
   }

   *packet_type = found_it->second.first;
   *packet_key = found_it->first;
   PacketRef ret = found_it->second.second;
   list->erase(found_it);
   return ret;
 }

 bool PacedSender::IsHighPriority(const PacketKey& packet_key) const {
   return std::find(priority_ssrcs_.begin(), priority_ssrcs_.end(),
                    packet_key.ssrc) != priority_ssrcs_.end();
 }

 bool PacedSender::empty() const {
   return packet_list_.empty() && priority_packet_list_.empty();
 }

 size_t PacedSender::size() const {
   return packet_list_.size() + priority_packet_list_.size();
 }

 // This function can be called from three places:
 // 1. User called one of the Send* functions and we were in an unblocked state.
 // 2. state_ == State_TransportBlocked and the transport is calling us to
 //    let us know that it's ok to send again.
 // 3. state_ == State_BurstFull and there are still packets to send. In this
 //    case we called PostDelayedTask on this function to start a new burst.
 void PacedSender::SendStoredPackets() {
   State previous_state = state_;
   state_ = State_Unblocked;
   if (empty()) {
     return;
   }

   base::TimeTicks now = clock_->NowTicks();
   // I don't actually trust that PostDelayTask(x - now) will mean that
   // now >= x when the call happens, so check if the previous state was
   // State_BurstFull too.
   if (now >= burst_end_ || previous_state == State_BurstFull) {
     // Start a new burst.
     current_burst_size_ = 0;
     burst_end_ = now + base::Milliseconds(kPacingIntervalMs);

     // The goal here is to try to send out the queued packets over the next
     // three bursts, while trying to keep the burst size below 10 if possible.
     // We have some evidence that sending more than 12 packets in a row doesn't
     // work very well, but we don't actually know why yet. Sending out packets
     // sooner is better than sending out packets later as that gives us more
     // time to re-send them if needed. So if we have less than 30 packets, just
     // send 10 at a time. If we have less than 60 packets, send n / 3 at a time.
     // if we have more than 60, we send 20 at a time. 20 packets is ~24Mbit/s
     // which is more bandwidth than the cast library should need, and sending
     // out more data per second is unlikely to be helpful.
     size_t max_burst_size = std::min(
         max_burst_size_,
         std::max(target_burst_size_, size() / kPacingMaxBurstsPerFrame));
     current_max_burst_size_ = std::max(next_max_burst_size_, max_burst_size);
     next_max_burst_size_ = std::max(next_next_max_burst_size_, max_burst_size);
     next_next_max_burst_size_ = max_burst_size;
   }

   base::RepeatingClosure cb = base::BindRepeating(
       &PacedSender::SendStoredPackets, weak_factory_.GetWeakPtr());
   while (!empty()) {
     if (current_burst_size_ >= current_max_burst_size_) {
       transport_task_runner_->PostDelayedTask(FROM_HERE,
                                               cb,
                                               burst_end_ - now);
       state_ = State_BurstFull;
       return;
     }
     PacketType packet_type;
     PacketKey packet_key;
     PacketRef packet = PopNextPacket(&packet_type, &packet_key);
     PacketSendRecord* const send_record = &(send_history_[packet_key]);
     send_record->time = now;

     if (send_record->cancel_count > 0 && packet_type != PacketType_RTCP) {
       VLOG(2) << "PacedSender is sending a packet known to have been CANCELED "
               << send_record->cancel_count << " times: "
               << "ssrc=" << packet_key.ssrc
               << ", frame_id=" << packet_key.frame_id
               << ", packet_id=" << packet_key.packet_id;
     }

     switch (packet_type) {
       case PacketType_Resend:
         LogPacketEvent(packet->data, PACKET_RETRANSMITTED);
         break;
       case PacketType_Normal:
         LogPacketEvent(packet->data, PACKET_SENT_TO_NETWORK);
         break;
       case PacketType_RTCP:
         break;
     }

     const bool socket_blocked = !transport_->SendPacket(packet, cb);

     // Save the send record.
     send_record->last_byte_sent = transport_->GetBytesSent();
     send_record->last_byte_sent_for_audio = last_byte_sent_for_audio_;
     send_history_buffer_[packet_key] = *send_record;

     auto it = sessions_.find(packet_key.ssrc);
     // The session should always have been registered in |sessions_|.
     DCHECK(it != sessions_.end());
     it->second.last_byte_sent = send_record->last_byte_sent;
     if (it->second.is_audio)
       last_byte_sent_for_audio_ = send_record->last_byte_sent;

     if (socket_blocked) {
       state_ = State_TransportBlocked;
       return;
     }
     current_burst_size_++;
   }

   // Keep ~0.5 seconds of data (1000 packets).
   //
   // TODO(miu): This has no relation to the actual size of the frames, and so
   // there's no way to reason whether 1000 is enough or too much, or whatever.
   if (send_history_buffer_.size() >=
       max_burst_size_ * kMaxDedupeWindowMs / kPacingIntervalMs) {
     send_history_.swap(send_history_buffer_);
     send_history_buffer_.clear();
   }
   DCHECK_LE(send_history_buffer_.size(),
             max_burst_size_ * kMaxDedupeWindowMs / kPacingIntervalMs);
   state_ = State_Unblocked;
 }

 void PacedSender::LogPacketEvent(const Packet& packet, CastLoggingEvent type) {
   if (!recent_packet_events_)
     return;

   recent_packet_events_->push_back(PacketEvent());
   PacketEvent& event = recent_packet_events_->back();

   // Populate the new PacketEvent by parsing the wire-format |packet|.
   //
   // TODO(miu): This parsing logic belongs in RtpParser.
   event.timestamp = clock_->NowTicks();
   event.type = type;
   base::BigEndianReader reader(reinterpret_cast<const char*>(&packet[0]),
                                packet.size());
   bool success = reader.Skip(4);
   uint32_t truncated_rtp_timestamp;
   success &= reader.ReadU32(&truncated_rtp_timestamp);
   uint32_t ssrc;
   success &= reader.ReadU32(&ssrc);

   auto it = sessions_.find(ssrc);
   // The session should always have been registered in |sessions_|.
   DCHECK(it != sessions_.end());
   event.rtp_timestamp = it->second.last_logged_rtp_timestamp_ =
       it->second.last_logged_rtp_timestamp_.Expand(truncated_rtp_timestamp);
   event.media_type = it->second.is_audio ? AUDIO_EVENT : VIDEO_EVENT;
   success &= reader.Skip(2);
   success &= reader.ReadU16(&event.packet_id);
   success &= reader.ReadU16(&event.max_packet_id);
   event.size = base::checked_cast<uint32_t>(packet.size());
   DCHECK(success);
 }

 }  // namespace cast
 }  // namespace media
	// Copyright 2014 The Chromium Authors. All rights reserved.
	// Use of this source code is governed by a BSD-style license that can be
	// found in the LICENSE file.

	#include "media/cast/net/pacing/paced_sender.h"

	#include "base/big_endian.h"
	#include "base/bind.h"
	#include "base/logging.h"
	#include "base/numerics/safe_conversions.h"

	namespace media {
	namespace cast {

	namespace {

	static const int64_t kPacingIntervalMs = 10;
	// Each frame will be split into no more than kPacingMaxBurstsPerFrame
	// bursts of packets.
	static const size_t kPacingMaxBurstsPerFrame = 3;
	static const size_t kMaxDedupeWindowMs = 500;

	} // namespace

	DedupInfo::DedupInfo() : last_byte_acked_for_audio(0) {}

	PacketKey::PacketKey() : ssrc(0), packet_id(0) {}

	PacketKey::PacketKey(base::TimeTicks capture_time,
	uint32_t ssrc,
	FrameId frame_id,
	uint16_t packet_id)
	: capture_time(capture_time),
	ssrc(ssrc),
	frame_id(frame_id),
	packet_id(packet_id) {}

	PacketKey::PacketKey(const PacketKey& other) = default;

	PacketKey::~PacketKey() = default;

	struct PacedSender::PacketSendRecord {
	PacketSendRecord()
	: last_byte_sent(0), last_byte_sent_for_audio(0), cancel_count(0) {}

	base::TimeTicks time; // Time when the packet was sent.
	int64_t last_byte_sent; // Number of bytes sent to network just after this
	// packet was sent.
	int64_t last_byte_sent_for_audio; // Number of bytes sent to network from
	// audio stream just before this packet.
	int cancel_count; // Number of times the packet was canceled (debugging).
	};

	struct PacedSender::RtpSession {
	explicit RtpSession(bool is_audio_stream)
	: last_byte_sent(0), is_audio(is_audio_stream) {}
	RtpSession() = default;

	// Tracks recently-logged RTP timestamps so that it can expand the truncated
	// values found in packets.
	RtpTimeTicks last_logged_rtp_timestamp_;
	int64_t last_byte_sent;
	bool is_audio;
	};

	PacedSender::PacedSender(
	size_t target_burst_size,
	size_t max_burst_size,
	const base::TickClock* clock,
	std::vector<PacketEvent>* recent_packet_events,
	PacketTransport* transport,
	const scoped_refptr<base::SingleThreadTaskRunner>& transport_task_runner)
	: clock_(clock),
	recent_packet_events_(recent_packet_events),
	transport_(transport),
	transport_task_runner_(transport_task_runner),
	last_byte_sent_for_audio_(0),
	target_burst_size_(target_burst_size),
	max_burst_size_(max_burst_size),
	current_max_burst_size_(target_burst_size_),
	next_max_burst_size_(target_burst_size_),
	next_next_max_burst_size_(target_burst_size_),
	current_burst_size_(0),
	state_(State_Unblocked) {}

	PacedSender::~PacedSender() = default;

	void PacedSender::RegisterSsrc(uint32_t ssrc, bool is_audio) {
	if (sessions_.find(ssrc) != sessions_.end())
	DVLOG(1) << "Re-register ssrc: " << ssrc;

	sessions_[ssrc] = RtpSession(is_audio);
	}

	void PacedSender::RegisterPrioritySsrc(uint32_t ssrc) {
	priority_ssrcs_.push_back(ssrc);
	}

	int64_t PacedSender::GetLastByteSentForPacket(const PacketKey& packet_key) {
	PacketSendHistory::const_iterator it = send_history_.find(packet_key);
	if (it == send_history_.end())
	return 0;
	return it->second.last_byte_sent;
	}

	int64_t PacedSender::GetLastByteSentForSsrc(uint32_t ssrc) {
	auto it = sessions_.find(ssrc);
	// Return 0 for unknown session.
	if (it == sessions_.end())
	return 0;
	return it->second.last_byte_sent;
	}

	bool PacedSender::SendPackets(const SendPacketVector& packets) {
	if (packets.empty()) {
	return true;
	}
	const bool high_priority = IsHighPriority(packets.begin()->first);
	for (size_t i = 0; i < packets.size(); i++) {
	if (VLOG_IS_ON(2)) {
	PacketSendHistory::const_iterator history_it =
	send_history_.find(packets[i].first);
	if (history_it != send_history_.end() &&
	history_it->second.cancel_count > 0) {
	VLOG(2) << "PacedSender::SendPackets() called for packet CANCELED "
	<< history_it->second.cancel_count << " times: "
	<< "ssrc=" << packets[i].first.ssrc
	<< ", frame_id=" << packets[i].first.frame_id
	<< ", packet_id=" << packets[i].first.packet_id;
	}
	}

	DCHECK(IsHighPriority(packets[i].first) == high_priority);
	if (high_priority) {
	priority_packet_list_[packets[i].first] =
	make_pair(PacketType_Normal, packets[i].second);
	} else {
	packet_list_[packets[i].first] =
	make_pair(PacketType_Normal, packets[i].second);
	}
	}
	if (state_ == State_Unblocked) {
	SendStoredPackets();
	}
	return true;
	}

	bool PacedSender::ShouldResend(const PacketKey& packet_key,
	const DedupInfo& dedup_info,
	const base::TimeTicks& now) {
	PacketSendHistory::const_iterator it = send_history_.find(packet_key);

	// No history of previous transmission. It might be sent too long ago.
	if (it == send_history_.end())
	return true;

	// Suppose there is request to retransmit X and there is an audio
	// packet Y sent just before X. Reject retransmission of X if ACK for
	// Y has not been received.
	// Only do this for video packets.
	//
	// TODO(miu): This sounds wrong. Audio packets are always transmitted first
	// (because they are put in \|priority_packet_list_\|, see PopNextPacket()).
	auto session_it = sessions_.find(packet_key.ssrc);
	// The session should always have been registered in \|sessions_\|.
	DCHECK(session_it != sessions_.end());
	if (!session_it->second.is_audio) {
	if (dedup_info.last_byte_acked_for_audio &&
	it->second.last_byte_sent_for_audio &&
	dedup_info.last_byte_acked_for_audio <
	it->second.last_byte_sent_for_audio) {
	return false;
	}
	}
	// Retransmission interval has to be greater than \|resend_interval\|.
	if (now - it->second.time < dedup_info.resend_interval)
	return false;
	return true;
	}

	bool PacedSender::ResendPackets(const SendPacketVector& packets,
	const DedupInfo& dedup_info) {
	if (packets.empty()) {
	return true;
	}
	const bool high_priority = IsHighPriority(packets.begin()->first);
	const base::TimeTicks now = clock_->NowTicks();
	for (size_t i = 0; i < packets.size(); i++) {
	if (VLOG_IS_ON(2)) {
	PacketSendHistory::const_iterator history_it =
	send_history_.find(packets[i].first);
	if (history_it != send_history_.end() &&
	history_it->second.cancel_count > 0) {
	VLOG(2) << "PacedSender::ReendPackets() called for packet CANCELED "
	<< history_it->second.cancel_count << " times: "
	<< "ssrc=" << packets[i].first.ssrc
	<< ", frame_id=" << packets[i].first.frame_id
	<< ", packet_id=" << packets[i].first.packet_id;
	}
	}

	if (!ShouldResend(packets[i].first, dedup_info, now)) {
	LogPacketEvent(packets[i].second->data, PACKET_RTX_REJECTED);
	continue;
	}

	DCHECK(IsHighPriority(packets[i].first) == high_priority);
	if (high_priority) {
	priority_packet_list_[packets[i].first] =
	make_pair(PacketType_Resend, packets[i].second);
	} else {
	packet_list_[packets[i].first] =
	make_pair(PacketType_Resend, packets[i].second);
	}
	}
	if (state_ == State_Unblocked) {
	SendStoredPackets();
	}
	return true;
	}

	bool PacedSender::SendRtcpPacket(uint32_t ssrc, PacketRef packet) {
	if (state_ == State_TransportBlocked) {
	const PacketKey key(base::TimeTicks(), ssrc, FrameId::first(), 0);
	priority_packet_list_[key] = make_pair(PacketType_RTCP, packet);
	} else {
	// We pass the RTCP packets straight through.
	if (!transport_->SendPacket(packet,
	base::BindOnce(&PacedSender::SendStoredPackets,
	weak_factory_.GetWeakPtr()))) {
	state_ = State_TransportBlocked;
	}
	}
	return true;
	}

	void PacedSender::CancelSendingPacket(const PacketKey& packet_key) {
	packet_list_.erase(packet_key);
	priority_packet_list_.erase(packet_key);

	if (VLOG_IS_ON(2)) {
	auto history_it = send_history_.find(packet_key);
	if (history_it != send_history_.end())
	++history_it->second.cancel_count;
	}
	}

	PacketRef PacedSender::PopNextPacket(PacketType* packet_type,
	PacketKey* packet_key) {
	// Always pop from the priority list first.
	PacketList* list =
	!priority_packet_list_.empty() ? &priority_packet_list_ : &packet_list_;
	DCHECK(!list->empty());

	// Determine which packet in the frame should be popped by examining the
	// \|send_history_\| for prior transmission attempts. Packets that have never
	// been transmitted will be popped first. If all packets have transmitted
	// before, pop the one that has not been re-attempted for the longest time.
	auto it = list->begin();
	PacketKey last_key = it->first;
	last_key.packet_id = UINT16_C(0xffff);
	PacketSendHistory::const_iterator history_it =
	send_history_.lower_bound(it->first);
	base::TimeTicks earliest_send_time =
	base::TimeTicks() + base::TimeDelta::Max();
	auto found_it = it;
	while (true) {
	if (history_it == send_history_.end() \|\| it->first < history_it->first) {
	// There is no send history for this packet, which means it has not been
	// transmitted yet.
	found_it = it;
	break;
	}

	DCHECK(it->first == history_it->first);
	if (history_it->second.time < earliest_send_time) {
	earliest_send_time = history_it->second.time;
	found_it = it;
	}

	// Advance to next packet for the current frame, or break if there are no
	// more.
	++it;
	if (it == list->end() \|\| last_key < it->first)
	break;

	// Advance to next history entry. Since there may be "holes" in the packet
	// list (e.g., due to packets canceled for retransmission), it's possible
	// \|history_it\| will have to be advanced more than once even though \|it\| was
	// only advanced once.
	do {
	++history_it;
	} while (history_it != send_history_.end() &&
	history_it->first < it->first);
	}

	*packet_type = found_it->second.first;
	*packet_key = found_it->first;
	PacketRef ret = found_it->second.second;
	list->erase(found_it);
	return ret;
	}

	bool PacedSender::IsHighPriority(const PacketKey& packet_key) const {
	return std::find(priority_ssrcs_.begin(), priority_ssrcs_.end(),
	packet_key.ssrc) != priority_ssrcs_.end();
	}

	bool PacedSender::empty() const {
	return packet_list_.empty() && priority_packet_list_.empty();
	}

	size_t PacedSender::size() const {
	return packet_list_.size() + priority_packet_list_.size();
	}

	// This function can be called from three places:
	// 1. User called one of the Send* functions and we were in an unblocked state.
	// 2. state_ == State_TransportBlocked and the transport is calling us to
	// let us know that it's ok to send again.
	// 3. state_ == State_BurstFull and there are still packets to send. In this
	// case we called PostDelayedTask on this function to start a new burst.
	void PacedSender::SendStoredPackets() {
	State previous_state = state_;
	state_ = State_Unblocked;
	if (empty()) {
	return;
	}

	base::TimeTicks now = clock_->NowTicks();
	// I don't actually trust that PostDelayTask(x - now) will mean that
	// now >= x when the call happens, so check if the previous state was
	// State_BurstFull too.
	if (now >= burst_end_ \|\| previous_state == State_BurstFull) {
	// Start a new burst.
	current_burst_size_ = 0;
	burst_end_ = now + base::Milliseconds(kPacingIntervalMs);

	// The goal here is to try to send out the queued packets over the next
	// three bursts, while trying to keep the burst size below 10 if possible.
	// We have some evidence that sending more than 12 packets in a row doesn't
	// work very well, but we don't actually know why yet. Sending out packets
	// sooner is better than sending out packets later as that gives us more
	// time to re-send them if needed. So if we have less than 30 packets, just
	// send 10 at a time. If we have less than 60 packets, send n / 3 at a time.
	// if we have more than 60, we send 20 at a time. 20 packets is ~24Mbit/s
	// which is more bandwidth than the cast library should need, and sending
	// out more data per second is unlikely to be helpful.
	size_t max_burst_size = std::min(
	max_burst_size_,
	std::max(target_burst_size_, size() / kPacingMaxBurstsPerFrame));
	current_max_burst_size_ = std::max(next_max_burst_size_, max_burst_size);
	next_max_burst_size_ = std::max(next_next_max_burst_size_, max_burst_size);
	next_next_max_burst_size_ = max_burst_size;
	}

	base::RepeatingClosure cb = base::BindRepeating(
	&PacedSender::SendStoredPackets, weak_factory_.GetWeakPtr());
	while (!empty()) {
	if (current_burst_size_ >= current_max_burst_size_) {
	transport_task_runner_->PostDelayedTask(FROM_HERE,
	cb,
	burst_end_ - now);
	state_ = State_BurstFull;
	return;
	}
	PacketType packet_type;
	PacketKey packet_key;
	PacketRef packet = PopNextPacket(&packet_type, &packet_key);
	PacketSendRecord* const send_record = &(send_history_[packet_key]);
	send_record->time = now;

	if (send_record->cancel_count > 0 && packet_type != PacketType_RTCP) {
	VLOG(2) << "PacedSender is sending a packet known to have been CANCELED "
	<< send_record->cancel_count << " times: "
	<< "ssrc=" << packet_key.ssrc
	<< ", frame_id=" << packet_key.frame_id
	<< ", packet_id=" << packet_key.packet_id;
	}

	switch (packet_type) {
	case PacketType_Resend:
	LogPacketEvent(packet->data, PACKET_RETRANSMITTED);
	break;
	case PacketType_Normal:
	LogPacketEvent(packet->data, PACKET_SENT_TO_NETWORK);
	break;
	case PacketType_RTCP:
	break;
	}

	const bool socket_blocked = !transport_->SendPacket(packet, cb);

	// Save the send record.
	send_record->last_byte_sent = transport_->GetBytesSent();
	send_record->last_byte_sent_for_audio = last_byte_sent_for_audio_;
	send_history_buffer_[packet_key] = *send_record;

	auto it = sessions_.find(packet_key.ssrc);
	// The session should always have been registered in \|sessions_\|.
	DCHECK(it != sessions_.end());
	it->second.last_byte_sent = send_record->last_byte_sent;
	if (it->second.is_audio)
	last_byte_sent_for_audio_ = send_record->last_byte_sent;

	if (socket_blocked) {
	state_ = State_TransportBlocked;
	return;
	}
	current_burst_size_++;
	}

	// Keep ~0.5 seconds of data (1000 packets).
	//
	// TODO(miu): This has no relation to the actual size of the frames, and so
	// there's no way to reason whether 1000 is enough or too much, or whatever.
	if (send_history_buffer_.size() >=
	max_burst_size_ * kMaxDedupeWindowMs / kPacingIntervalMs) {
	send_history_.swap(send_history_buffer_);
	send_history_buffer_.clear();
	}
	DCHECK_LE(send_history_buffer_.size(),
	max_burst_size_ * kMaxDedupeWindowMs / kPacingIntervalMs);
	state_ = State_Unblocked;
	}

	void PacedSender::LogPacketEvent(const Packet& packet, CastLoggingEvent type) {
	if (!recent_packet_events_)
	return;

	recent_packet_events_->push_back(PacketEvent());
	PacketEvent& event = recent_packet_events_->back();

	// Populate the new PacketEvent by parsing the wire-format \|packet\|.
	//
	// TODO(miu): This parsing logic belongs in RtpParser.
	event.timestamp = clock_->NowTicks();
	event.type = type;
	base::BigEndianReader reader(reinterpret_cast<const char*>(&packet[0]),
	packet.size());
	bool success = reader.Skip(4);
	uint32_t truncated_rtp_timestamp;
	success &= reader.ReadU32(&truncated_rtp_timestamp);
	uint32_t ssrc;
	success &= reader.ReadU32(&ssrc);

	auto it = sessions_.find(ssrc);
	// The session should always have been registered in \|sessions_\|.
	DCHECK(it != sessions_.end());
	event.rtp_timestamp = it->second.last_logged_rtp_timestamp_ =
	it->second.last_logged_rtp_timestamp_.Expand(truncated_rtp_timestamp);
	event.media_type = it->second.is_audio ? AUDIO_EVENT : VIDEO_EVENT;
	success &= reader.Skip(2);
	success &= reader.ReadU16(&event.packet_id);
	success &= reader.ReadU16(&event.max_packet_id);
	event.size = base::checked_cast<uint32_t>(packet.size());
	DCHECK(success);
	}

	} // namespace cast
	} // namespace media