zqz/platform-external-webrtc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Prepares PacingController for scheduled send tasks.

Browse Source 
This CL is in preparation for a dynamic (possible TaskQueue-driven)
pacer that instead of processing blindly every 5ms, posts delayed
tasks to be executed when it is actually time to send packs.

This means we need the pacing controller to be able to figure out when
those execution times shall be, and be able to correctly update budget
levels as IntervalBudget only works correctly with periodic processing.

Bug: webrtc:10809
Change-Id: Idd12acaabfb24cc2e6bcc589aac206cd04beb6e4
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/158790
Commit-Queue: Erik Språng <sprang@webrtc.org>
Reviewed-by: Sebastian Jansson <srte@webrtc.org>
Cr-Commit-Position: refs/heads/master@{#29800}
This commit is contained in:
Erik Språng
2019-11-14 14:15:15 +01:00
committed by Commit Bot
parent 3527a4fe55
commit eb48799ec5
6 changed files with 885 additions and 332 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
modules/pacing/paced_sender.cc
View File

@ -35,7 +35,8 @@ PacedSender::PacedSender(Clock* clock,
: pacing_controller_(clock,
static_cast<PacingController::PacketSender*>(this),
event_log,
field_trials),
field_trials,
PacingController::ProcessMode::kPeriodic),
clock_(clock),
packet_router_(packet_router),
process_thread_(process_thread) {
@ -128,22 +129,9 @@ TimeDelta PacedSender::OldestPacketWaitTime() const {
int64_t PacedSender::TimeUntilNextProcess() {
rtc::CritScope cs(&critsect_);
// When paused we wake up every 500 ms to send a padding packet to ensure
// we won't get stuck in the paused state due to no feedback being received.
TimeDelta elapsed_time = pacing_controller_.TimeElapsedSinceLastProcess();
if (pacing_controller_.IsPaused()) {
return std::max(PacingController::kPausedProcessInterval - elapsed_time,
TimeDelta::Zero())
Timestamp next_send_time = pacing_controller_.NextSendTime();
return std::max(TimeDelta::Zero(), next_send_time - clock_->CurrentTime())
.ms();
}
Timestamp next_probe = pacing_controller_.NextProbeTime();
if (next_probe != Timestamp::PlusInfinity()) {
return std::max(TimeDelta::Zero(), next_probe - clock_->CurrentTime()).ms();
}
const TimeDelta min_packet_limit = TimeDelta::ms(5);
return std::max(min_packet_limit - elapsed_time, TimeDelta::Zero()).ms();
}
void PacedSender::Process() {

251
modules/pacing/pacing_controller.cc
View File

@ -28,7 +28,11 @@ namespace {
// Time limit in milliseconds between packet bursts.
constexpr TimeDelta kDefaultMinPacketLimit = TimeDelta::Millis<5>();
constexpr TimeDelta kCongestedPacketInterval = TimeDelta::Millis<500>();
// TODO(sprang): Consider dropping this limit.
// The maximum debt level, in terms of time, capped when sending packets.
constexpr TimeDelta kMaxDebtInTime = TimeDelta::Millis<500>();
constexpr TimeDelta kMaxElapsedTime = TimeDelta::Seconds<2>();
constexpr DataSize kDefaultPaddingTarget = DataSize::Bytes<50>();
// Upper cap on process interval, in case process has not been called in a long
// time.
@ -75,12 +79,15 @@ const TimeDelta PacingController::kMaxExpectedQueueLength =
const float PacingController::kDefaultPaceMultiplier = 2.5f;
const TimeDelta PacingController::kPausedProcessInterval =
kCongestedPacketInterval;
const TimeDelta PacingController::kMinSleepTime = TimeDelta::Millis<1>();
PacingController::PacingController(Clock* clock,
PacketSender* packet_sender,
RtcEventLog* event_log,
const WebRtcKeyValueConfig* field_trials)
: clock_(clock),
const WebRtcKeyValueConfig* field_trials,
ProcessMode mode)
: mode_(mode),
clock_(clock),
packet_sender_(packet_sender),
fallback_field_trials_(
!field_trials ? std::make_unique<FieldTrialBasedConfig>() : nullptr),
@ -97,13 +104,16 @@ PacingController::PacingController(Clock* clock,
paused_(false),
media_budget_(0),
padding_budget_(0),
media_debt_(DataSize::Zero()),
padding_debt_(DataSize::Zero()),
media_rate_(DataRate::Zero()),
padding_rate_(DataRate::Zero()),
prober_(*field_trials_),
probing_send_failure_(false),
padding_failure_state_(false),
pacing_bitrate_(DataRate::Zero()),
time_last_process_(clock->CurrentTime()),
last_send_time_(time_last_process_),
packet_queue_(time_last_process_, field_trials),
last_process_time_(clock->CurrentTime()),
last_send_time_(last_process_time_),
packet_queue_(last_process_time_, field_trials),
packet_counter_(0),
congestion_window_size_(DataSize::PlusInfinity()),
outstanding_data_(DataSize::Zero()),
@ -145,11 +155,21 @@ bool PacingController::IsPaused() const {
}
void PacingController::SetCongestionWindow(DataSize congestion_window_size) {
const bool was_congested = Congested();
congestion_window_size_ = congestion_window_size;
if (was_congested && !Congested()) {
TimeDelta elapsed_time = UpdateTimeAndGetElapsed(CurrentTime());
UpdateBudgetWithElapsedTime(elapsed_time);
}
}
void PacingController::UpdateOutstandingData(DataSize outstanding_data) {
const bool was_congested = Congested();
outstanding_data_ = outstanding_data;
if (was_congested && !Congested()) {
TimeDelta elapsed_time = UpdateTimeAndGetElapsed(CurrentTime());
UpdateBudgetWithElapsedTime(elapsed_time);
}
}
bool PacingController::Congested() const {
@ -180,6 +200,8 @@ void PacingController::SetProbingEnabled(bool enabled) {
void PacingController::SetPacingRates(DataRate pacing_rate,
DataRate padding_rate) {
RTC_DCHECK_GT(pacing_rate, DataRate::Zero());
media_rate_ = pacing_rate;
padding_rate_ = padding_rate;
pacing_bitrate_ = pacing_rate;
padding_budget_.set_target_rate_kbps(padding_rate.kbps());
@ -241,12 +263,19 @@ void PacingController::EnqueuePacketInternal(
packet->set_capture_time_ms(now.ms());
}
if (mode_ == ProcessMode::kDynamic && packet_queue_.Empty() &&
media_debt_ == DataSize::Zero()) {
last_process_time_ = CurrentTime();
}
packet_queue_.Push(priority, now, packet_counter_++, std::move(packet));
}
TimeDelta PacingController::UpdateTimeAndGetElapsed(Timestamp now) {
TimeDelta elapsed_time = now - time_last_process_;
time_last_process_ = now;
if (last_process_time_.IsMinusInfinity()) {
return TimeDelta::Zero();
}
TimeDelta elapsed_time = now - last_process_time_;
last_process_time_ = now;
if (elapsed_time > kMaxElapsedTime) {
RTC_LOG(LS_WARNING) << "Elapsed time (" << elapsed_time.ms()
<< " ms) longer than expected, limiting to "
@ -257,47 +286,92 @@ TimeDelta PacingController::UpdateTimeAndGetElapsed(Timestamp now) {
}
bool PacingController::ShouldSendKeepalive(Timestamp now) const {
if (send_padding_if_silent_ || paused_ || Congested()) {
if (send_padding_if_silent_ || paused_ || Congested() ||
packet_counter_ == 0) {
// We send a padding packet every 500 ms to ensure we won't get stuck in
// congested state due to no feedback being received.
TimeDelta elapsed_since_last_send = now - last_send_time_;
if (elapsed_since_last_send >= kCongestedPacketInterval) {
// We can not send padding unless a normal packet has first been sent. If
// we do, timestamps get messed up.
if (packet_counter_ > 0) {
return true;
}
}
}
return false;
}
Timestamp PacingController::NextProbeTime() {
if (!prober_.IsProbing()) {
return Timestamp::PlusInfinity();
}
Timestamp PacingController::NextSendTime() const {
Timestamp now = CurrentTime();
// If probing is active, that always takes priority.
if (prober_.IsProbing()) {
Timestamp probe_time = prober_.NextProbeTime(now);
if (probe_time.IsInfinite()) {
// |probe_time| == PlusInfinity indicates no probe scheduled.
if (probe_time != Timestamp::PlusInfinity() && !probing_send_failure_) {
return probe_time;
}
if (probe_time <= now && probing_send_failure_) {
return Timestamp::PlusInfinity();
}
return probe_time;
}
if (mode_ == ProcessMode::kPeriodic) {
// In periodc non-probing mode, we just have a fixed interval.
if (paused_) {
return last_send_time_ + kPausedProcessInterval;
}
return last_process_time_ + min_packet_limit_;
}
TimeDelta PacingController::TimeElapsedSinceLastProcess() const {
return CurrentTime() - time_last_process_;
// In dynamic mode, figure out when the next packet should be sent,
// given the current conditions.
if (Congested() || packet_counter_ == 0) {
// If congested, we only send keep-alive or audio (if audio is
// configured in pass-through mode).
if (!pace_audio_ && packet_queue_.NextPacketIsAudio()) {
return now;
}
// We need to at least send keep-alive packets with some interval.
return last_send_time_ + kCongestedPacketInterval;
}
// If there are pending packets, check how long it will take until buffers
// have emptied.
if (media_rate_ > DataRate::Zero() && !packet_queue_.Empty()) {
return std::min(last_send_time_ + kPausedProcessInterval,
last_process_time_ + media_debt_ / media_rate_);
}
// If we _don't_ have pending packets, check how long until we have
// bandwidth for padding packets.
if (padding_rate_ > DataRate::Zero() && packet_queue_.Empty()) {
return std::min(last_send_time_ + kPausedProcessInterval,
last_process_time_ + padding_debt_ / padding_rate_);
}
return last_send_time_ + kPausedProcessInterval;
}
void PacingController::ProcessPackets() {
Timestamp now = CurrentTime();
RTC_DCHECK_GE(now, last_process_time_);
Timestamp target_send_time = now;
if (mode_ == ProcessMode::kDynamic) {
target_send_time = NextSendTime();
if (target_send_time.IsMinusInfinity()) {
target_send_time = now;
} else if (now + kMinSleepTime < target_send_time) {
// We are too early, abort and regroup!
return;
}
}
Timestamp previous_process_time = last_process_time_;
TimeDelta elapsed_time = UpdateTimeAndGetElapsed(now);
if (ShouldSendKeepalive(now)) {
// We can not send padding unless a normal packet has first been sent. If
// we do, timestamps get messed up.
if (packet_counter_ == 0) {
last_send_time_ = now;
} else {
DataSize keepalive_data_sent = DataSize::Zero();
std::vector<std::unique_ptr<RtpPacketToSend>> keepalive_packets =
packet_sender_->GeneratePadding(DataSize::bytes(1));
@ -308,9 +382,11 @@ void PacingController::ProcessPackets() {
}
OnPaddingSent(keepalive_data_sent);
}
}
if (paused_)
if (paused_) {
return;
}
if (elapsed_time > TimeDelta::Zero()) {
DataRate target_rate = pacing_bitrate_;
@ -319,7 +395,7 @@ void PacingController::ProcessPackets() {
// Assuming equal size packets and input/output rate, the average packet
// has avg_time_left_ms left to get queue_size_bytes out of the queue, if
// time constraint shall be met. Determine bitrate needed for that.
packet_queue_.UpdateQueueTime(CurrentTime());
packet_queue_.UpdateQueueTime(now);
if (drain_large_queues_) {
TimeDelta avg_time_left =
std::max(TimeDelta::ms(1),
@ -333,8 +409,15 @@ void PacingController::ProcessPackets() {
}
}
if (mode_ == ProcessMode::kPeriodic) {
// In periodic processing mode, the IntevalBudget allows positive budget
// up to (process interval duration) * (target rate), so we only need to
// update it once before the packet sending loop.
media_budget_.set_target_rate_kbps(target_rate.kbps());
UpdateBudgetWithElapsedTime(elapsed_time);
} else {
media_rate_ = target_rate;
}
}
bool first_packet_in_probe = false;
@ -348,6 +431,7 @@ void PacingController::ProcessPackets() {
}
DataSize data_sent = DataSize::Zero();
// The paused state is checked in the loop since it leaves the critical
// section allowing the paused state to be changed from other code.
while (!paused_) {
@ -367,7 +451,19 @@ void PacingController::ProcessPackets() {
first_packet_in_probe = false;
}
auto* packet = GetPendingPacket(pacing_info);
if (mode_ == ProcessMode::kDynamic &&
previous_process_time < target_send_time) {
// Reduce buffer levels with amount corresponding to time between last
// process and target send time for the next packet.
// If the process call is late, that may be the time between the optimal
// send times for two packets we should already have sent.
UpdateBudgetWithElapsedTime(target_send_time - previous_process_time);
previous_process_time = target_send_time;
}
// Fetch the next packet, so long as queue is not empty or budget is not
// exhausted.
auto* packet = GetPendingPacket(pacing_info, target_send_time, now);
if (packet == nullptr) {
// No packet available to send, check if we should send padding.
DataSize padding_to_add = PaddingToAdd(recommended_probe_size, data_sent);
@ -394,10 +490,22 @@ void PacingController::ProcessPackets() {
packet_sender_->SendRtpPacket(std::move(rtp_packet), pacing_info);
data_sent += packet->size();
// Send succeeded, remove it from the queue.
OnPacketSent(packet);
// Send succeeded, remove it from the queue and update send/process time to
// the target send time.
OnPacketSent(packet, target_send_time);
if (recommended_probe_size && data_sent > *recommended_probe_size)
break;
if (mode_ == ProcessMode::kDynamic) {
// Update target send time in case that are more packets that we are late
// in processing.
Timestamp next_send_time = NextSendTime();
if (next_send_time.IsMinusInfinity()) {
target_send_time = now;
} else {
target_send_time = std::min(now, next_send_time);
}
}
}
if (is_probing) {
@ -410,7 +518,7 @@ void PacingController::ProcessPackets() {
DataSize PacingController::PaddingToAdd(
absl::optional<DataSize> recommended_probe_size,
DataSize data_sent) {
DataSize data_sent) const {
if (!packet_queue_.Empty()) {
// Actual payload available, no need to add padding.
return DataSize::Zero();
@ -434,66 +542,105 @@ DataSize PacingController::PaddingToAdd(
return DataSize::Zero();
}
if (mode_ == ProcessMode::kPeriodic) {
return DataSize::bytes(padding_budget_.bytes_remaining());
} else if (padding_rate_ > DataRate::Zero() &&
padding_debt_ == DataSize::Zero()) {
return kDefaultPaddingTarget;
}
return DataSize::Zero();
}
RoundRobinPacketQueue::QueuedPacket* PacingController::GetPendingPacket(
const PacedPacketInfo& pacing_info) {
const PacedPacketInfo& pacing_info,
Timestamp target_send_time,
Timestamp now) {
if (packet_queue_.Empty()) {
return nullptr;
}
// Since we need to release the lock in order to send, we first pop the
// element from the priority queue but keep it in storage, so that we can
// reinsert it if send fails.
RoundRobinPacketQueue::QueuedPacket* packet = packet_queue_.BeginPop();
bool audio_packet = packet->type() == RtpPacketToSend::Type::kAudio;
bool apply_pacing = !audio_packet || pace_audio_;
if (apply_pacing && (Congested() || (media_budget_.bytes_remaining() == 0 &&
pacing_info.probe_cluster_id ==
PacedPacketInfo::kNotAProbe))) {
packet_queue_.CancelPop();
// First, check if there is any reason _not_ to send the next queued packet.
// Unpaced audio packets and probes are exempted from send checks.
bool unpaced_audio_packet = !pace_audio_ && packet_queue_.NextPacketIsAudio();
bool is_probe = pacing_info.probe_cluster_id != PacedPacketInfo::kNotAProbe;
if (!unpaced_audio_packet && !is_probe) {
if (Congested()) {
// Don't send anyting if congested.
return nullptr;
}
return packet;
if (mode_ == ProcessMode::kPeriodic) {
if (media_budget_.bytes_remaining() <= 0) {
// Not enough budget.
return nullptr;
}
} else {
if (now <= target_send_time) {
// We allow sending slightly early if we think that we would actually
// had been able to, had we been right on time - i.e. the current debt
// is not more than would be reduced to zero at the target sent time.
TimeDelta flush_time = media_debt_ / media_rate_;
if (now + flush_time > target_send_time) {
return nullptr;
}
} else {
// In dynamic mode we should never try get a non-probe packet until
// the media debt is actually zero.
RTC_DCHECK(media_debt_.IsZero());
}
}
}
return packet_queue_.BeginPop();
}
void PacingController::OnPacketSent(
RoundRobinPacketQueue::QueuedPacket* packet) {
Timestamp now = CurrentTime();
void PacingController::OnPacketSent(RoundRobinPacketQueue::QueuedPacket* packet,
Timestamp send_time) {
if (!first_sent_packet_time_) {
first_sent_packet_time_ = now;
first_sent_packet_time_ = send_time;
}
bool audio_packet = packet->type() == RtpPacketToSend::Type::kAudio;
if (!audio_packet || account_for_audio_) {
// Update media bytes sent.
UpdateBudgetWithSentData(packet->size());
last_send_time_ = now;
}
last_send_time_ = send_time;
last_process_time_ = send_time;
// Send succeeded, remove it from the queue.
packet_queue_.FinalizePop();
padding_failure_state_ = false;
}
void PacingController::OnPaddingSent(DataSize data_sent) {
if (data_sent > DataSize::Zero()) {
UpdateBudgetWithSentData(data_sent);
} else {
padding_failure_state_ = true;
}
last_send_time_ = CurrentTime();
last_process_time_ = CurrentTime();
}
void PacingController::UpdateBudgetWithElapsedTime(TimeDelta delta) {
if (mode_ == ProcessMode::kPeriodic) {
delta = std::min(kMaxProcessingInterval, delta);
media_budget_.IncreaseBudget(delta.ms());
padding_budget_.IncreaseBudget(delta.ms());
} else {
media_debt_ -= std::min(media_debt_, media_rate_ * delta);
padding_debt_ -= std::min(padding_debt_, padding_rate_ * delta);
}
}
void PacingController::UpdateBudgetWithSentData(DataSize size) {
outstanding_data_ += size;
if (mode_ == ProcessMode::kPeriodic) {
media_budget_.UseBudget(size.bytes());
padding_budget_.UseBudget(size.bytes());
} else {
media_debt_ += size;
media_debt_ = std::min(media_debt_, media_rate_ * kMaxDebtInTime);
padding_debt_ += size;
padding_debt_ = std::min(padding_debt_, padding_rate_ * kMaxDebtInTime);
}
}
void PacingController::SetQueueTimeLimit(TimeDelta limit) {

48
modules/pacing/pacing_controller.h
View File

@ -44,6 +44,13 @@ namespace webrtc {
//
class PacingController {
public:
// Periodic mode uses the IntervalBudget class for tracking bitrate
// budgets, and expected ProcessPackets() to be called a fixed rate,
// e.g. every 5ms as implemented by PacedSender.
// Dynamic mode allows for arbitrary time delta between calls to
// ProcessPackets.
enum class ProcessMode { kPeriodic, kDynamic };
class PacketSender {
public:
virtual ~PacketSender() = default;
@ -69,10 +76,13 @@ class PacingController {
// to lack of feedback.
static const TimeDelta kPausedProcessInterval;
static const TimeDelta kMinSleepTime;
PacingController(Clock* clock,
PacketSender* packet_sender,
RtcEventLog* event_log,
const WebRtcKeyValueConfig* field_trials);
const WebRtcKeyValueConfig* field_trials,
ProcessMode mode);
~PacingController();
@ -118,16 +128,8 @@ class PacingController {
// effect.
void SetProbingEnabled(bool enabled);
// Time at which next probe should be sent. If this value is set, it should be
// respected - i.e. don't call ProcessPackets() before this specified time as
// that can have unintended side effects.
// If no scheduled probe, Timestamp::PlusInifinity() is returned.
Timestamp NextProbeTime();
// Time since ProcessPackets() was last executed.
TimeDelta TimeElapsedSinceLastProcess() const;
TimeDelta TimeUntilAvailableBudget() const;
// Returns the next time we expect ProcessPackets() to be called.
Timestamp NextSendTime() const;
// Check queue of pending packets and send them or padding packets, if budget
// is available.
@ -146,15 +148,19 @@ class PacingController {
void UpdateBudgetWithSentData(DataSize size);
DataSize PaddingToAdd(absl::optional<DataSize> recommended_probe_size,
DataSize data_sent);
DataSize data_sent) const;
RoundRobinPacketQueue::QueuedPacket* GetPendingPacket(
const PacedPacketInfo& pacing_info);
void OnPacketSent(RoundRobinPacketQueue::QueuedPacket* packet);
const PacedPacketInfo& pacing_info,
Timestamp target_send_time,
Timestamp now);
void OnPacketSent(RoundRobinPacketQueue::QueuedPacket* packet,
Timestamp send_time);
void OnPaddingSent(DataSize padding_sent);
Timestamp CurrentTime() const;
const ProcessMode mode_;
Clock* const clock_;
PacketSender* const packet_sender_;
const std::unique_ptr<FieldTrialBasedConfig> fallback_field_trials_;
@ -164,12 +170,18 @@ class PacingController {
const bool send_padding_if_silent_;
const bool pace_audio_;
const bool small_first_probe_packet_;
TimeDelta min_packet_limit_;
// TODO(webrtc:9716): Remove this when we are certain clocks are monotonic.
// The last millisecond timestamp returned by |clock_|.
mutable Timestamp last_timestamp_;
bool paused_;
// If |use_interval_budget_| is true, |media_budget_| and |padding_budget_|
// will be used to track when packets can be sent. Otherwise the media and
// padding debt counters will be used together with the target rates.
// This is the media budget, keeping track of how many bits of media
// we can pace out during the current interval.
IntervalBudget media_budget_;
@ -178,13 +190,17 @@ class PacingController {
// utilized when there's no media to send.
IntervalBudget padding_budget_;
DataSize media_debt_;
DataSize padding_debt_;
DataRate media_rate_;
DataRate padding_rate_;
BitrateProber prober_;
bool probing_send_failure_;
bool padding_failure_state_;
DataRate pacing_bitrate_;
Timestamp time_last_process_;
Timestamp last_process_time_;
Timestamp last_send_time_;
absl::optional<Timestamp> first_sent_packet_time_;

772
modules/pacing/pacing_controller_unittest.cc
View File

File diff suppressed because it is too large Load Diff

11
modules/pacing/round_robin_packet_queue.cc
View File

@ -216,6 +216,17 @@ DataSize RoundRobinPacketQueue::Size() const {
return size_;
}
bool RoundRobinPacketQueue::NextPacketIsAudio() const {
if (stream_priorities_.empty()) {
return false;
}
uint32_t ssrc = stream_priorities_.begin()->second;
auto stream_info_it = streams_.find(ssrc);
return stream_info_it->second.packet_queue.top().type() ==
RtpPacketToSend::Type::kAudio;
}
Timestamp RoundRobinPacketQueue::OldestEnqueueTime() const {
if (Empty())
return Timestamp::MinusInfinity();

1
modules/pacing/round_robin_packet_queue.h
View File

@ -115,6 +115,7 @@ class RoundRobinPacketQueue {
bool Empty() const;
size_t SizeInPackets() const;
DataSize Size() const;
bool NextPacketIsAudio() const;
Timestamp OldestEnqueueTime() const;
TimeDelta AverageQueueTime() const;