modules/remote_bitrate_estimator/test/packet_sender.cc - src - Git at Google

 /*
  *  Copyright (c) 2015 The WebRTC project authors. All Rights Reserved.
  *
  *  Use of this source code is governed by a BSD-style license
  *  that can be found in the LICENSE file in the root of the source
  *  tree. An additional intellectual property rights grant can be found
  *  in the file PATENTS.  All contributing project authors may
  *  be found in the AUTHORS file in the root of the source tree.
  */

 #include "modules/remote_bitrate_estimator/test/packet_sender.h"

 #include <algorithm>
 #include <list>
 #include <sstream>

 #include "modules/include/module_common_types.h"
 #include "modules/pacing/pacer.h"
 #include "modules/remote_bitrate_estimator/test/bbr_paced_sender.h"
 #include "modules/remote_bitrate_estimator/test/bwe.h"
 #include "modules/remote_bitrate_estimator/test/metric_recorder.h"
 #include "rtc_base/checks.h"

 namespace webrtc {
 namespace testing {
 namespace bwe {
 namespace {
 const float kPaceMultiplier = 2.5f;
 }

 void PacketSender::Pause() {
   running_ = false;
   if (metric_recorder_ != nullptr) {
     metric_recorder_->PauseFlow();
   }
 }

 void PacketSender::Resume(int64_t paused_time_ms) {
   running_ = true;
   if (metric_recorder_ != nullptr) {
     metric_recorder_->ResumeFlow(paused_time_ms);
   }
 }

 void PacketSender::set_metric_recorder(MetricRecorder* metric_recorder) {
   metric_recorder_ = metric_recorder;
 }

 void PacketSender::RecordBitrate() {
   if (metric_recorder_ != nullptr) {
     BWE_TEST_LOGGING_CONTEXT("Sender");
     BWE_TEST_LOGGING_CONTEXT(*flow_ids().begin());
     metric_recorder_->UpdateTimeMs(clock_.TimeInMilliseconds());
     metric_recorder_->UpdateSendingEstimateKbps(TargetBitrateKbps());
   }
 }

 std::list<FeedbackPacket*> GetFeedbackPackets(Packets* in_out,
                                               int64_t end_time_ms,
                                               int flow_id) {
   std::list<FeedbackPacket*> fb_packets;
   for (auto it = in_out->begin(); it != in_out->end();) {
     if ((*it)->send_time_us() > 1000 * end_time_ms)
       break;
     if ((*it)->GetPacketType() == Packet::kFeedback &&
         flow_id == (*it)->flow_id()) {
       fb_packets.push_back(static_cast<FeedbackPacket*>(*it));
       it = in_out->erase(it);
     } else {
       ++it;
     }
   }
   return fb_packets;
 }

 VideoSender::VideoSender(PacketProcessorListener* listener,
                          VideoSource* source,
                          BandwidthEstimatorType estimator_type)
     : PacketSender(listener, source->flow_id()),
       source_(source),
       bwe_(CreateBweSender(estimator_type,
                            source_->bits_per_second() / 1000,
                            this,
                            &clock_)),
       previous_sending_bitrate_(0) {
   modules_.push_back(bwe_.get());
 }

 VideoSender::~VideoSender() {
 }

 void VideoSender::Pause() {
   previous_sending_bitrate_ = TargetBitrateKbps();
   PacketSender::Pause();
 }

 void VideoSender::Resume(int64_t paused_time_ms) {
   source_->SetBitrateBps(previous_sending_bitrate_);
   PacketSender::Resume(paused_time_ms);
 }

 void VideoSender::RunFor(int64_t time_ms, Packets* in_out) {
   std::list<FeedbackPacket*> feedbacks = GetFeedbackPackets(
       in_out, clock_.TimeInMilliseconds() + time_ms, source_->flow_id());
   ProcessFeedbackAndGeneratePackets(time_ms, &feedbacks, in_out);
 }

 void VideoSender::ProcessFeedbackAndGeneratePackets(
     int64_t time_ms,
     std::list<FeedbackPacket*>* feedbacks,
     Packets* packets) {
   do {
     // Make sure to at least run Process() below every 100 ms.
     int64_t time_to_run_ms = std::min<int64_t>(time_ms, 100);
     if (!feedbacks->empty()) {
       int64_t time_until_feedback_ms =
           feedbacks->front()->send_time_ms() - clock_.TimeInMilliseconds();
       time_to_run_ms =
           std::max<int64_t>(std::min(time_ms, time_until_feedback_ms), 0);
     }

     if (!running_) {
       source_->SetBitrateBps(0);
     }

     Packets generated;
     source_->RunFor(time_to_run_ms, &generated);
     bwe_->OnPacketsSent(generated);
     packets->merge(generated, DereferencingComparator<Packet>);

     clock_.AdvanceTimeMilliseconds(time_to_run_ms);

     if (!feedbacks->empty()) {
       bwe_->GiveFeedback(*feedbacks->front());
       delete feedbacks->front();
       feedbacks->pop_front();
     }

     bwe_->Process();

     time_ms -= time_to_run_ms;
   } while (time_ms > 0);
   assert(feedbacks->empty());
 }

 int VideoSender::GetFeedbackIntervalMs() const {
   return bwe_->GetFeedbackIntervalMs();
 }

 void VideoSender::OnNetworkChanged(uint32_t target_bitrate_bps,
                                    uint8_t fraction_lost,
                                    int64_t rtt) {
   source_->SetBitrateBps(target_bitrate_bps);
   RecordBitrate();
 }

 uint32_t VideoSender::TargetBitrateKbps() {
   return (source_->bits_per_second() + 500) / 1000;
 }

 PacedVideoSender::PacedVideoSender(PacketProcessorListener* listener,
                                    VideoSource* source,
                                    BandwidthEstimatorType estimator)
     : VideoSender(listener, source, estimator),
       pacer_(
           estimator == kBbrEstimator
               ? static_cast<Pacer*>(new BbrPacedSender(&clock_, this, nullptr))
               : static_cast<Pacer*>(new PacedSender(&clock_, this, nullptr))) {
   modules_.push_back(pacer_.get());
   pacer_->SetPacingRates(source->bits_per_second() * kPaceMultiplier, 0);
 }

 PacedVideoSender::~PacedVideoSender() {
   for (Packet* packet : pacer_queue_)
     delete packet;
   for (Packet* packet : queue_)
     delete packet;
 }

 void PacedVideoSender::RunFor(int64_t time_ms, Packets* in_out) {
   int64_t end_time_ms = clock_.TimeInMilliseconds() + time_ms;
   // Run process periodically to allow the packets to be paced out.
   std::list<FeedbackPacket*> feedbacks =
       GetFeedbackPackets(in_out, end_time_ms, source_->flow_id());
   int64_t last_run_time_ms = -1;
   BWE_TEST_LOGGING_CONTEXT("Sender");
   BWE_TEST_LOGGING_CONTEXT(source_->flow_id());
   do {
     int64_t time_until_process_ms = TimeUntilNextProcess(modules_);
     int64_t time_until_feedback_ms = time_ms;
     if (!feedbacks.empty())
       time_until_feedback_ms = std::max<int64_t>(
           feedbacks.front()->send_time_ms() - clock_.TimeInMilliseconds(), 0);

     int64_t time_until_next_event_ms =
         std::min(time_until_feedback_ms, time_until_process_ms);

     time_until_next_event_ms =
         std::min(source_->GetTimeUntilNextFrameMs(), time_until_next_event_ms);

     // Never run for longer than we have been asked for.
     if (clock_.TimeInMilliseconds() + time_until_next_event_ms > end_time_ms)
       time_until_next_event_ms = end_time_ms - clock_.TimeInMilliseconds();

     // Make sure we don't get stuck if an event doesn't trigger. This typically
     // happens if the prober wants to probe, but there's no packet to send.
     if (time_until_next_event_ms == 0 && last_run_time_ms == 0)
       time_until_next_event_ms = 1;
     last_run_time_ms = time_until_next_event_ms;

     Packets generated_packets;
     source_->RunFor(time_until_next_event_ms, &generated_packets);
     if (!generated_packets.empty()) {
       for (Packet* packet : generated_packets) {
         MediaPacket* media_packet = static_cast<MediaPacket*>(packet);
         pacer_->InsertPacket(
             PacedSender::kNormalPriority, media_packet->header().ssrc,
             media_packet->header().sequenceNumber, media_packet->send_time_ms(),
             media_packet->payload_size(), false);
         pacer_queue_size_in_bytes_ += media_packet->payload_size();
         pacer_queue_.push_back(packet);
         assert(pacer_queue_.size() < 10000);
       }
     }

     clock_.AdvanceTimeMilliseconds(time_until_next_event_ms);

     if (time_until_next_event_ms == time_until_feedback_ms) {
       if (!feedbacks.empty()) {
         bwe_->GiveFeedback(*feedbacks.front());
         delete feedbacks.front();
         feedbacks.pop_front();
       }
       bwe_->Process();
     }

     if (time_until_next_event_ms == time_until_process_ms) {
       CallProcess(modules_);
     }
   } while (clock_.TimeInMilliseconds() < end_time_ms);
   QueuePackets(in_out, end_time_ms * 1000);
 }

 int64_t PacedVideoSender::TimeUntilNextProcess(
     const std::list<Module*>& modules) {
   int64_t time_until_next_process_ms = 10;
   for (Module* module : modules) {
     int64_t next_process_ms = module->TimeUntilNextProcess();
     if (next_process_ms < time_until_next_process_ms)
       time_until_next_process_ms = next_process_ms;
   }
   if (time_until_next_process_ms < 0)
     time_until_next_process_ms = 0;
   return time_until_next_process_ms;
 }

 void PacedVideoSender::CallProcess(const std::list<Module*>& modules) {
   for (Module* module : modules) {
     if (module->TimeUntilNextProcess() <= 0) {
       module->Process();
     }
   }
 }

 void PacedVideoSender::QueuePackets(Packets* batch,
                                     int64_t end_of_batch_time_us) {
   queue_.merge(*batch, DereferencingComparator<Packet>);
   if (queue_.empty()) {
     return;
   }
   Packets::iterator it = queue_.begin();
   for (; it != queue_.end(); ++it) {
     if ((*it)->send_time_us() > end_of_batch_time_us) {
       break;
     }
   }
   Packets to_transfer;
   to_transfer.splice(to_transfer.begin(), queue_, queue_.begin(), it);
   bwe_->OnPacketsSent(to_transfer);
   batch->merge(to_transfer, DereferencingComparator<Packet>);
 }

 bool PacedVideoSender::TimeToSendPacket(uint32_t ssrc,
                                         uint16_t sequence_number,
                                         int64_t capture_time_ms,
                                         bool retransmission,
                                         const PacedPacketInfo& pacing_info) {
   for (Packets::iterator it = pacer_queue_.begin(); it != pacer_queue_.end();
        ++it) {
     MediaPacket* media_packet = static_cast<MediaPacket*>(*it);
     if (media_packet->header().sequenceNumber == sequence_number) {
       int64_t pace_out_time_ms = clock_.TimeInMilliseconds();

       // Make sure a packet is never paced out earlier than when it was put into
       // the pacer.
       assert(pace_out_time_ms >= media_packet->send_time_ms());
       media_packet->SetAbsSendTimeMs(pace_out_time_ms);
       media_packet->set_send_time_us(1000 * pace_out_time_ms);
       media_packet->set_sender_timestamp_us(1000 * pace_out_time_ms);
       queue_.push_back(media_packet);
       pacer_queue_size_in_bytes_ -= media_packet->payload_size();
       pacer_queue_.erase(it);
       return true;
     }
   }
   return false;
 }

 size_t PacedVideoSender::TimeToSendPadding(size_t bytes,
                                            const PacedPacketInfo& pacing_info) {
   return 0;
 }

 void PacedVideoSender::OnNetworkChanged(uint32_t target_bitrate_bps,
                                         uint8_t fraction_lost,
                                         int64_t rtt) {
   VideoSender::OnNetworkChanged(target_bitrate_bps, fraction_lost, rtt);
   pacer_->SetPacingRates(target_bitrate_bps * kPaceMultiplier, 0);
 }

 void PacedVideoSender::OnNetworkChanged(uint32_t bitrate_for_encoder_bps,
                                         uint32_t bitrate_for_pacer_bps,
                                         bool in_probe_rtt,
                                         int64_t target_set_time,
                                         uint64_t congestion_window) {
   VideoSender::OnNetworkChanged(bitrate_for_encoder_bps, 0u, 0u);
   pacer_->SetEstimatedBitrateAndCongestionWindow(
       bitrate_for_pacer_bps, in_probe_rtt, congestion_window);
 }

 void PacedVideoSender::OnBytesAcked(size_t bytes) {
   pacer_->OnBytesAcked(bytes);
 }

 const int kNoLimit = std::numeric_limits<int>::max();
 const int kPacketSizeBytes = 1200;

 TcpSender::TcpSender(PacketProcessorListener* listener,
                      int flow_id,
                      int64_t offset_ms)
     : TcpSender(listener, flow_id, offset_ms, kNoLimit) {
 }

 TcpSender::TcpSender(PacketProcessorListener* listener,
                      int flow_id,
                      int64_t offset_ms,
                      int send_limit_bytes)
     : PacketSender(listener, flow_id),
       cwnd_(10),
       ssthresh_(kNoLimit),
       ack_received_(false),
       last_acked_seq_num_(0),
       next_sequence_number_(0),
       offset_ms_(offset_ms),
       last_reduction_time_ms_(-1),
       last_rtt_ms_(0),
       total_sent_bytes_(0),
       send_limit_bytes_(send_limit_bytes),
       last_generated_packets_ms_(0),
       num_recent_sent_packets_(0),
       bitrate_kbps_(0) {
 }

 void TcpSender::RunFor(int64_t time_ms, Packets* in_out) {
   if (clock_.TimeInMilliseconds() + time_ms < offset_ms_) {
     clock_.AdvanceTimeMilliseconds(time_ms);
     if (running_) {
       Pause();
     }
     return;
   }

   if (!running_ && total_sent_bytes_ == 0) {
     Resume(offset_ms_);
   }

   int64_t start_time_ms = clock_.TimeInMilliseconds();

   std::list<FeedbackPacket*> feedbacks = GetFeedbackPackets(
       in_out, clock_.TimeInMilliseconds() + time_ms, *flow_ids().begin());
   // The number of packets which are sent in during time_ms depends on the
   // number of packets in_flight_ and the max number of packets in flight
   // (cwnd_). Therefore SendPackets() isn't directly dependent on time_ms.
   for (FeedbackPacket* fb : feedbacks) {
     clock_.AdvanceTimeMilliseconds(fb->send_time_ms() -
                                    clock_.TimeInMilliseconds());
     last_rtt_ms_ = fb->send_time_ms() - fb->latest_send_time_ms();
     UpdateCongestionControl(fb);
     SendPackets(in_out);
   }

   for (auto it = in_flight_.begin(); it != in_flight_.end();) {
     if (it->time_ms < clock_.TimeInMilliseconds() - 1000)
       in_flight_.erase(it++);
     else
       ++it;
   }

   clock_.AdvanceTimeMilliseconds(time_ms -
                                  (clock_.TimeInMilliseconds() - start_time_ms));
   SendPackets(in_out);
 }

 void TcpSender::SendPackets(Packets* in_out) {
   int cwnd = ceil(cwnd_);
   int packets_to_send = std::max(cwnd - static_cast<int>(in_flight_.size()), 0);
   int timed_out = TriggerTimeouts();
   if (timed_out > 0) {
     HandleLoss();
   }
   if (packets_to_send > 0) {
     Packets generated = GeneratePackets(packets_to_send);
     for (Packet* packet : generated)
       in_flight_.insert(InFlight(*static_cast<MediaPacket*>(packet)));

     in_out->merge(generated, DereferencingComparator<Packet>);
   }
 }

 void TcpSender::UpdateCongestionControl(const FeedbackPacket* fb) {
   const TcpFeedback* tcp_fb = static_cast<const TcpFeedback*>(fb);
   RTC_DCHECK(!tcp_fb->acked_packets().empty());
   ack_received_ = true;

   uint16_t expected = tcp_fb->acked_packets().back() - last_acked_seq_num_;
   uint16_t missing =
       expected - static_cast<uint16_t>(tcp_fb->acked_packets().size());

   for (uint16_t ack_seq_num : tcp_fb->acked_packets())
     in_flight_.erase(InFlight(ack_seq_num, clock_.TimeInMilliseconds()));

   if (missing > 0) {
     HandleLoss();
   } else if (cwnd_ <= ssthresh_) {
     cwnd_ += tcp_fb->acked_packets().size();
   } else {
     cwnd_ += 1.0f / cwnd_;
   }

   last_acked_seq_num_ =
       LatestSequenceNumber(tcp_fb->acked_packets().back(), last_acked_seq_num_);
 }

 int TcpSender::TriggerTimeouts() {
   int timed_out = 0;
   for (auto it = in_flight_.begin(); it != in_flight_.end();) {
     if (it->time_ms < clock_.TimeInMilliseconds() - 1000) {
       in_flight_.erase(it++);
       ++timed_out;
     } else {
       ++it;
     }
   }
   return timed_out;
 }

 void TcpSender::HandleLoss() {
   if (clock_.TimeInMilliseconds() - last_reduction_time_ms_ < last_rtt_ms_)
     return;
   last_reduction_time_ms_ = clock_.TimeInMilliseconds();
   ssthresh_ = std::max(static_cast<int>(in_flight_.size() / 2), 2);
   cwnd_ = ssthresh_;
 }

 Packets TcpSender::GeneratePackets(size_t num_packets) {
   Packets generated;

   UpdateSendBitrateEstimate(num_packets);

   for (size_t i = 0; i < num_packets; ++i) {
     if ((total_sent_bytes_ + kPacketSizeBytes) > send_limit_bytes_) {
       if (running_) {
         Pause();
       }
       break;
     }
     generated.push_back(
         new MediaPacket(*flow_ids().begin(), 1000 * clock_.TimeInMilliseconds(),
                         kPacketSizeBytes, next_sequence_number_++));
     generated.back()->set_sender_timestamp_us(
         1000 * clock_.TimeInMilliseconds());

     total_sent_bytes_ += kPacketSizeBytes;
   }

   return generated;
 }

 void TcpSender::UpdateSendBitrateEstimate(size_t num_packets) {
   const int kTimeWindowMs = 500;
   num_recent_sent_packets_ += num_packets;

   int64_t delta_ms = clock_.TimeInMilliseconds() - last_generated_packets_ms_;
   if (delta_ms >= kTimeWindowMs) {
     bitrate_kbps_ =
         static_cast<uint32_t>(8 * num_recent_sent_packets_ * kPacketSizeBytes) /
         delta_ms;
     last_generated_packets_ms_ = clock_.TimeInMilliseconds();
     num_recent_sent_packets_ = 0;
   }

   RecordBitrate();
 }

 uint32_t TcpSender::TargetBitrateKbps() {
   return bitrate_kbps_;
 }

 }  // namespace bwe
 }  // namespace testing
 }  // namespace webrtc
	/*
	* Copyright (c) 2015 The WebRTC project authors. All Rights Reserved.
	*
	* Use of this source code is governed by a BSD-style license
	* that can be found in the LICENSE file in the root of the source
	* tree. An additional intellectual property rights grant can be found
	* in the file PATENTS. All contributing project authors may
	* be found in the AUTHORS file in the root of the source tree.
	*/

	#include "modules/remote_bitrate_estimator/test/packet_sender.h"

	#include <algorithm>
	#include <list>
	#include <sstream>

	#include "modules/include/module_common_types.h"
	#include "modules/pacing/pacer.h"
	#include "modules/remote_bitrate_estimator/test/bbr_paced_sender.h"
	#include "modules/remote_bitrate_estimator/test/bwe.h"
	#include "modules/remote_bitrate_estimator/test/metric_recorder.h"
	#include "rtc_base/checks.h"

	namespace webrtc {
	namespace testing {
	namespace bwe {
	namespace {
	const float kPaceMultiplier = 2.5f;
	}

	void PacketSender::Pause() {
	running_ = false;
	if (metric_recorder_ != nullptr) {
	metric_recorder_->PauseFlow();
	}
	}

	void PacketSender::Resume(int64_t paused_time_ms) {
	running_ = true;
	if (metric_recorder_ != nullptr) {
	metric_recorder_->ResumeFlow(paused_time_ms);
	}
	}

	void PacketSender::set_metric_recorder(MetricRecorder* metric_recorder) {
	metric_recorder_ = metric_recorder;
	}

	void PacketSender::RecordBitrate() {
	if (metric_recorder_ != nullptr) {
	BWE_TEST_LOGGING_CONTEXT("Sender");
	BWE_TEST_LOGGING_CONTEXT(*flow_ids().begin());
	metric_recorder_->UpdateTimeMs(clock_.TimeInMilliseconds());
	metric_recorder_->UpdateSendingEstimateKbps(TargetBitrateKbps());
	}
	}

	std::list<FeedbackPacket> GetFeedbackPackets(Packets in_out,
	int64_t end_time_ms,
	int flow_id) {
	std::list<FeedbackPacket*> fb_packets;
	for (auto it = in_out->begin(); it != in_out->end();) {
	if ((it)->send_time_us() > 1000 end_time_ms)
	break;
	if ((*it)->GetPacketType() == Packet::kFeedback &&
	flow_id == (*it)->flow_id()) {
	fb_packets.push_back(static_cast<FeedbackPacket>(it));
	it = in_out->erase(it);
	} else {
	++it;
	}
	}
	return fb_packets;
	}

	VideoSender::VideoSender(PacketProcessorListener* listener,
	VideoSource* source,
	BandwidthEstimatorType estimator_type)
	: PacketSender(listener, source->flow_id()),
	source_(source),
	bwe_(CreateBweSender(estimator_type,
	source_->bits_per_second() / 1000,
	this,
	&clock_)),
	previous_sending_bitrate_(0) {
	modules_.push_back(bwe_.get());
	}

	VideoSender::~VideoSender() {
	}

	void VideoSender::Pause() {
	previous_sending_bitrate_ = TargetBitrateKbps();
	PacketSender::Pause();
	}

	void VideoSender::Resume(int64_t paused_time_ms) {
	source_->SetBitrateBps(previous_sending_bitrate_);
	PacketSender::Resume(paused_time_ms);
	}

	void VideoSender::RunFor(int64_t time_ms, Packets* in_out) {
	std::list<FeedbackPacket*> feedbacks = GetFeedbackPackets(
	in_out, clock_.TimeInMilliseconds() + time_ms, source_->flow_id());
	ProcessFeedbackAndGeneratePackets(time_ms, &feedbacks, in_out);
	}

	void VideoSender::ProcessFeedbackAndGeneratePackets(
	int64_t time_ms,
	std::list<FeedbackPacket> feedbacks,
	Packets* packets) {
	do {
	// Make sure to at least run Process() below every 100 ms.
	int64_t time_to_run_ms = std::min<int64_t>(time_ms, 100);
	if (!feedbacks->empty()) {
	int64_t time_until_feedback_ms =
	feedbacks->front()->send_time_ms() - clock_.TimeInMilliseconds();
	time_to_run_ms =
	std::max<int64_t>(std::min(time_ms, time_until_feedback_ms), 0);
	}

	if (!running_) {
	source_->SetBitrateBps(0);
	}

	Packets generated;
	source_->RunFor(time_to_run_ms, &generated);
	bwe_->OnPacketsSent(generated);
	packets->merge(generated, DereferencingComparator<Packet>);

	clock_.AdvanceTimeMilliseconds(time_to_run_ms);

	if (!feedbacks->empty()) {
	bwe_->GiveFeedback(*feedbacks->front());
	delete feedbacks->front();
	feedbacks->pop_front();
	}

	bwe_->Process();

	time_ms -= time_to_run_ms;
	} while (time_ms > 0);
	assert(feedbacks->empty());
	}

	int VideoSender::GetFeedbackIntervalMs() const {
	return bwe_->GetFeedbackIntervalMs();
	}

	void VideoSender::OnNetworkChanged(uint32_t target_bitrate_bps,
	uint8_t fraction_lost,
	int64_t rtt) {
	source_->SetBitrateBps(target_bitrate_bps);
	RecordBitrate();
	}

	uint32_t VideoSender::TargetBitrateKbps() {
	return (source_->bits_per_second() + 500) / 1000;
	}

	PacedVideoSender::PacedVideoSender(PacketProcessorListener* listener,
	VideoSource* source,
	BandwidthEstimatorType estimator)
	: VideoSender(listener, source, estimator),
	pacer_(
	estimator == kBbrEstimator
	? static_cast<Pacer*>(new BbrPacedSender(&clock_, this, nullptr))
	: static_cast<Pacer*>(new PacedSender(&clock_, this, nullptr))) {
	modules_.push_back(pacer_.get());
	pacer_->SetPacingRates(source->bits_per_second() * kPaceMultiplier, 0);
	}

	PacedVideoSender::~PacedVideoSender() {
	for (Packet* packet : pacer_queue_)
	delete packet;
	for (Packet* packet : queue_)
	delete packet;
	}

	void PacedVideoSender::RunFor(int64_t time_ms, Packets* in_out) {
	int64_t end_time_ms = clock_.TimeInMilliseconds() + time_ms;
	// Run process periodically to allow the packets to be paced out.
	std::list<FeedbackPacket*> feedbacks =
	GetFeedbackPackets(in_out, end_time_ms, source_->flow_id());
	int64_t last_run_time_ms = -1;
	BWE_TEST_LOGGING_CONTEXT("Sender");
	BWE_TEST_LOGGING_CONTEXT(source_->flow_id());
	do {
	int64_t time_until_process_ms = TimeUntilNextProcess(modules_);
	int64_t time_until_feedback_ms = time_ms;
	if (!feedbacks.empty())
	time_until_feedback_ms = std::max<int64_t>(
	feedbacks.front()->send_time_ms() - clock_.TimeInMilliseconds(), 0);

	int64_t time_until_next_event_ms =
	std::min(time_until_feedback_ms, time_until_process_ms);

	time_until_next_event_ms =
	std::min(source_->GetTimeUntilNextFrameMs(), time_until_next_event_ms);

	// Never run for longer than we have been asked for.
	if (clock_.TimeInMilliseconds() + time_until_next_event_ms > end_time_ms)
	time_until_next_event_ms = end_time_ms - clock_.TimeInMilliseconds();

	// Make sure we don't get stuck if an event doesn't trigger. This typically
	// happens if the prober wants to probe, but there's no packet to send.
	if (time_until_next_event_ms == 0 && last_run_time_ms == 0)
	time_until_next_event_ms = 1;
	last_run_time_ms = time_until_next_event_ms;

	Packets generated_packets;
	source_->RunFor(time_until_next_event_ms, &generated_packets);
	if (!generated_packets.empty()) {
	for (Packet* packet : generated_packets) {
	MediaPacket* media_packet = static_cast<MediaPacket*>(packet);
	pacer_->InsertPacket(
	PacedSender::kNormalPriority, media_packet->header().ssrc,
	media_packet->header().sequenceNumber, media_packet->send_time_ms(),
	media_packet->payload_size(), false);
	pacer_queue_size_in_bytes_ += media_packet->payload_size();
	pacer_queue_.push_back(packet);
	assert(pacer_queue_.size() < 10000);
	}
	}

	clock_.AdvanceTimeMilliseconds(time_until_next_event_ms);

	if (time_until_next_event_ms == time_until_feedback_ms) {
	if (!feedbacks.empty()) {
	bwe_->GiveFeedback(*feedbacks.front());
	delete feedbacks.front();
	feedbacks.pop_front();
	}
	bwe_->Process();
	}

	if (time_until_next_event_ms == time_until_process_ms) {
	CallProcess(modules_);
	}
	} while (clock_.TimeInMilliseconds() < end_time_ms);
	QueuePackets(in_out, end_time_ms * 1000);
	}

	int64_t PacedVideoSender::TimeUntilNextProcess(
	const std::list<Module*>& modules) {
	int64_t time_until_next_process_ms = 10;
	for (Module* module : modules) {
	int64_t next_process_ms = module->TimeUntilNextProcess();
	if (next_process_ms < time_until_next_process_ms)
	time_until_next_process_ms = next_process_ms;
	}
	if (time_until_next_process_ms < 0)
	time_until_next_process_ms = 0;
	return time_until_next_process_ms;
	}

	void PacedVideoSender::CallProcess(const std::list<Module*>& modules) {
	for (Module* module : modules) {
	if (module->TimeUntilNextProcess() <= 0) {
	module->Process();
	}
	}
	}

	void PacedVideoSender::QueuePackets(Packets* batch,
	int64_t end_of_batch_time_us) {
	queue_.merge(*batch, DereferencingComparator<Packet>);
	if (queue_.empty()) {
	return;
	}
	Packets::iterator it = queue_.begin();
	for (; it != queue_.end(); ++it) {
	if ((*it)->send_time_us() > end_of_batch_time_us) {
	break;
	}
	}
	Packets to_transfer;
	to_transfer.splice(to_transfer.begin(), queue_, queue_.begin(), it);
	bwe_->OnPacketsSent(to_transfer);
	batch->merge(to_transfer, DereferencingComparator<Packet>);
	}

	bool PacedVideoSender::TimeToSendPacket(uint32_t ssrc,
	uint16_t sequence_number,
	int64_t capture_time_ms,
	bool retransmission,
	const PacedPacketInfo& pacing_info) {
	for (Packets::iterator it = pacer_queue_.begin(); it != pacer_queue_.end();
	++it) {
	MediaPacket* media_packet = static_cast<MediaPacket>(it);
	if (media_packet->header().sequenceNumber == sequence_number) {
	int64_t pace_out_time_ms = clock_.TimeInMilliseconds();

	// Make sure a packet is never paced out earlier than when it was put into
	// the pacer.
	assert(pace_out_time_ms >= media_packet->send_time_ms());
	media_packet->SetAbsSendTimeMs(pace_out_time_ms);
	media_packet->set_send_time_us(1000 * pace_out_time_ms);
	media_packet->set_sender_timestamp_us(1000 * pace_out_time_ms);
	queue_.push_back(media_packet);
	pacer_queue_size_in_bytes_ -= media_packet->payload_size();
	pacer_queue_.erase(it);
	return true;
	}
	}
	return false;
	}

	size_t PacedVideoSender::TimeToSendPadding(size_t bytes,
	const PacedPacketInfo& pacing_info) {
	return 0;
	}

	void PacedVideoSender::OnNetworkChanged(uint32_t target_bitrate_bps,
	uint8_t fraction_lost,
	int64_t rtt) {
	VideoSender::OnNetworkChanged(target_bitrate_bps, fraction_lost, rtt);
	pacer_->SetPacingRates(target_bitrate_bps * kPaceMultiplier, 0);
	}

	void PacedVideoSender::OnNetworkChanged(uint32_t bitrate_for_encoder_bps,
	uint32_t bitrate_for_pacer_bps,
	bool in_probe_rtt,
	int64_t target_set_time,
	uint64_t congestion_window) {
	VideoSender::OnNetworkChanged(bitrate_for_encoder_bps, 0u, 0u);
	pacer_->SetEstimatedBitrateAndCongestionWindow(
	bitrate_for_pacer_bps, in_probe_rtt, congestion_window);
	}

	void PacedVideoSender::OnBytesAcked(size_t bytes) {
	pacer_->OnBytesAcked(bytes);
	}

	const int kNoLimit = std::numeric_limits<int>::max();
	const int kPacketSizeBytes = 1200;

	TcpSender::TcpSender(PacketProcessorListener* listener,
	int flow_id,
	int64_t offset_ms)
	: TcpSender(listener, flow_id, offset_ms, kNoLimit) {
	}

	TcpSender::TcpSender(PacketProcessorListener* listener,
	int flow_id,
	int64_t offset_ms,
	int send_limit_bytes)
	: PacketSender(listener, flow_id),
	cwnd_(10),
	ssthresh_(kNoLimit),
	ack_received_(false),
	last_acked_seq_num_(0),
	next_sequence_number_(0),
	offset_ms_(offset_ms),
	last_reduction_time_ms_(-1),
	last_rtt_ms_(0),
	total_sent_bytes_(0),
	send_limit_bytes_(send_limit_bytes),
	last_generated_packets_ms_(0),
	num_recent_sent_packets_(0),
	bitrate_kbps_(0) {
	}

	void TcpSender::RunFor(int64_t time_ms, Packets* in_out) {
	if (clock_.TimeInMilliseconds() + time_ms < offset_ms_) {
	clock_.AdvanceTimeMilliseconds(time_ms);
	if (running_) {
	Pause();
	}
	return;
	}

	if (!running_ && total_sent_bytes_ == 0) {
	Resume(offset_ms_);
	}

	int64_t start_time_ms = clock_.TimeInMilliseconds();

	std::list<FeedbackPacket*> feedbacks = GetFeedbackPackets(
	in_out, clock_.TimeInMilliseconds() + time_ms, *flow_ids().begin());
	// The number of packets which are sent in during time_ms depends on the
	// number of packets in_flight_ and the max number of packets in flight
	// (cwnd_). Therefore SendPackets() isn't directly dependent on time_ms.
	for (FeedbackPacket* fb : feedbacks) {
	clock_.AdvanceTimeMilliseconds(fb->send_time_ms() -
	clock_.TimeInMilliseconds());
	last_rtt_ms_ = fb->send_time_ms() - fb->latest_send_time_ms();
	UpdateCongestionControl(fb);
	SendPackets(in_out);
	}

	for (auto it = in_flight_.begin(); it != in_flight_.end();) {
	if (it->time_ms < clock_.TimeInMilliseconds() - 1000)
	in_flight_.erase(it++);
	else
	++it;
	}

	clock_.AdvanceTimeMilliseconds(time_ms -
	(clock_.TimeInMilliseconds() - start_time_ms));
	SendPackets(in_out);
	}

	void TcpSender::SendPackets(Packets* in_out) {
	int cwnd = ceil(cwnd_);
	int packets_to_send = std::max(cwnd - static_cast<int>(in_flight_.size()), 0);
	int timed_out = TriggerTimeouts();
	if (timed_out > 0) {
	HandleLoss();
	}
	if (packets_to_send > 0) {
	Packets generated = GeneratePackets(packets_to_send);
	for (Packet* packet : generated)
	in_flight_.insert(InFlight(static_cast<MediaPacket>(packet)));

	in_out->merge(generated, DereferencingComparator<Packet>);
	}
	}

	void TcpSender::UpdateCongestionControl(const FeedbackPacket* fb) {
	const TcpFeedback* tcp_fb = static_cast<const TcpFeedback*>(fb);
	RTC_DCHECK(!tcp_fb->acked_packets().empty());
	ack_received_ = true;

	uint16_t expected = tcp_fb->acked_packets().back() - last_acked_seq_num_;
	uint16_t missing =
	expected - static_cast<uint16_t>(tcp_fb->acked_packets().size());

	for (uint16_t ack_seq_num : tcp_fb->acked_packets())
	in_flight_.erase(InFlight(ack_seq_num, clock_.TimeInMilliseconds()));

	if (missing > 0) {
	HandleLoss();
	} else if (cwnd_ <= ssthresh_) {
	cwnd_ += tcp_fb->acked_packets().size();
	} else {
	cwnd_ += 1.0f / cwnd_;
	}

	last_acked_seq_num_ =
	LatestSequenceNumber(tcp_fb->acked_packets().back(), last_acked_seq_num_);
	}

	int TcpSender::TriggerTimeouts() {
	int timed_out = 0;
	for (auto it = in_flight_.begin(); it != in_flight_.end();) {
	if (it->time_ms < clock_.TimeInMilliseconds() - 1000) {
	in_flight_.erase(it++);
	++timed_out;
	} else {
	++it;
	}
	}
	return timed_out;
	}

	void TcpSender::HandleLoss() {
	if (clock_.TimeInMilliseconds() - last_reduction_time_ms_ < last_rtt_ms_)
	return;
	last_reduction_time_ms_ = clock_.TimeInMilliseconds();
	ssthresh_ = std::max(static_cast<int>(in_flight_.size() / 2), 2);
	cwnd_ = ssthresh_;
	}

	Packets TcpSender::GeneratePackets(size_t num_packets) {
	Packets generated;

	UpdateSendBitrateEstimate(num_packets);

	for (size_t i = 0; i < num_packets; ++i) {
	if ((total_sent_bytes_ + kPacketSizeBytes) > send_limit_bytes_) {
	if (running_) {
	Pause();
	}
	break;
	}
	generated.push_back(
	new MediaPacket(flow_ids().begin(), 1000 clock_.TimeInMilliseconds(),
	kPacketSizeBytes, next_sequence_number_++));
	generated.back()->set_sender_timestamp_us(
	1000 * clock_.TimeInMilliseconds());

	total_sent_bytes_ += kPacketSizeBytes;
	}

	return generated;
	}

	void TcpSender::UpdateSendBitrateEstimate(size_t num_packets) {
	const int kTimeWindowMs = 500;
	num_recent_sent_packets_ += num_packets;

	int64_t delta_ms = clock_.TimeInMilliseconds() - last_generated_packets_ms_;
	if (delta_ms >= kTimeWindowMs) {
	bitrate_kbps_ =
	static_cast<uint32_t>(8 * num_recent_sent_packets_ * kPacketSizeBytes) /
	delta_ms;
	last_generated_packets_ms_ = clock_.TimeInMilliseconds();
	num_recent_sent_packets_ = 0;
	}

	RecordBitrate();
	}

	uint32_t TcpSender::TargetBitrateKbps() {
	return bitrate_kbps_;
	}

	} // namespace bwe
	} // namespace testing
	} // namespace webrtc