blob: 765e43fcf6789da5b1bd18522e54481fba98dd93 [file] [log] [blame]
/*
* Copyright (c) 2017 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/pacing/packet_queue.h"
#include <algorithm>
#include <list>
#include <vector>
#include "modules/include/module_common_types.h"
#include "modules/pacing/alr_detector.h"
#include "modules/pacing/bitrate_prober.h"
#include "modules/pacing/interval_budget.h"
#include "modules/utility/include/process_thread.h"
#include "rtc_base/checks.h"
#include "rtc_base/logging.h"
#include "system_wrappers/include/clock.h"
#include "system_wrappers/include/field_trial.h"
namespace webrtc {
PacketQueue::Packet::Packet(RtpPacketSender::Priority priority,
uint32_t ssrc,
uint16_t seq_number,
int64_t capture_time_ms,
int64_t enqueue_time_ms,
size_t length_in_bytes,
bool retransmission,
uint64_t enqueue_order)
: priority(priority),
ssrc(ssrc),
sequence_number(seq_number),
capture_time_ms(capture_time_ms),
enqueue_time_ms(enqueue_time_ms),
sum_paused_ms(0),
bytes(length_in_bytes),
retransmission(retransmission),
enqueue_order(enqueue_order) {}
PacketQueue::Packet::~Packet() {}
PacketQueue::PacketQueue(const Clock* clock)
: bytes_(0),
clock_(clock),
queue_time_sum_(0),
time_last_updated_(clock_->TimeInMilliseconds()),
paused_(false) {}
PacketQueue::~PacketQueue() {}
void PacketQueue::Push(const Packet& packet) {
if (!AddToDupeSet(packet))
return;
UpdateQueueTime(packet.enqueue_time_ms);
// Store packet in list, use pointers in priority queue for cheaper moves.
// Packets have a handle to its own iterator in the list, for easy removal
// when popping from queue.
packet_list_.push_front(packet);
std::list<Packet>::iterator it = packet_list_.begin();
it->this_it = it; // Handle for direct removal from list.
prio_queue_.push(&(*it)); // Pointer into list.
bytes_ += packet.bytes;
}
const PacketQueue::Packet& PacketQueue::BeginPop() {
const PacketQueue::Packet& packet = *prio_queue_.top();
prio_queue_.pop();
return packet;
}
void PacketQueue::CancelPop(const PacketQueue::Packet& packet) {
prio_queue_.push(&(*packet.this_it));
}
void PacketQueue::FinalizePop(const PacketQueue::Packet& packet) {
RemoveFromDupeSet(packet);
bytes_ -= packet.bytes;
int64_t packet_queue_time_ms = time_last_updated_ - packet.enqueue_time_ms;
RTC_DCHECK_LE(packet.sum_paused_ms, packet_queue_time_ms);
packet_queue_time_ms -= packet.sum_paused_ms;
RTC_DCHECK_LE(packet_queue_time_ms, queue_time_sum_);
queue_time_sum_ -= packet_queue_time_ms;
packet_list_.erase(packet.this_it);
RTC_DCHECK_EQ(packet_list_.size(), prio_queue_.size());
if (packet_list_.empty())
RTC_DCHECK_EQ(0, queue_time_sum_);
}
bool PacketQueue::Empty() const {
return prio_queue_.empty();
}
size_t PacketQueue::SizeInPackets() const {
return prio_queue_.size();
}
uint64_t PacketQueue::SizeInBytes() const {
return bytes_;
}
int64_t PacketQueue::OldestEnqueueTimeMs() const {
auto it = packet_list_.rbegin();
if (it == packet_list_.rend())
return 0;
return it->enqueue_time_ms;
}
void PacketQueue::UpdateQueueTime(int64_t timestamp_ms) {
RTC_DCHECK_GE(timestamp_ms, time_last_updated_);
if (timestamp_ms == time_last_updated_)
return;
int64_t delta_ms = timestamp_ms - time_last_updated_;
if (paused_) {
// Increase per-packet accumulators of time spent in queue while paused,
// so that we can disregard that when subtracting main accumulator when
// popping packet from the queue.
for (auto& it : packet_list_) {
it.sum_paused_ms += delta_ms;
}
} else {
// Use packet packet_list_.size() not prio_queue_.size() here, as there
// might be an outstanding element popped from prio_queue_ currently in
// the SendPacket() call, while packet_list_ will always be correct.
queue_time_sum_ += delta_ms * packet_list_.size();
}
time_last_updated_ = timestamp_ms;
}
void PacketQueue::SetPauseState(bool paused, int64_t timestamp_ms) {
if (paused_ == paused)
return;
UpdateQueueTime(timestamp_ms);
paused_ = paused;
}
int64_t PacketQueue::AverageQueueTimeMs() const {
if (prio_queue_.empty())
return 0;
return queue_time_sum_ / packet_list_.size();
}
bool PacketQueue::AddToDupeSet(const PacketQueue::Packet& packet) {
SsrcSeqNoMap::iterator it = dupe_map_.find(packet.ssrc);
if (it == dupe_map_.end()) {
// First for this ssrc, just insert.
dupe_map_[packet.ssrc].insert(packet.sequence_number);
return true;
}
// Insert returns a pair, where second is a bool set to true if new element.
return it->second.insert(packet.sequence_number).second;
}
void PacketQueue::RemoveFromDupeSet(const PacketQueue::Packet& packet) {
SsrcSeqNoMap::iterator it = dupe_map_.find(packet.ssrc);
RTC_DCHECK(it != dupe_map_.end());
it->second.erase(packet.sequence_number);
if (it->second.empty()) {
dupe_map_.erase(it);
}
}
} // namespace webrtc