Skip to content

Z
ZLMediaKit
Commit 4cc2982a by xia-chu
Options

重写jitter buffer算法，修复seq回环时可能丢包的bug

parent c301ecd2
正在显示 包含 102 行增加 和 73 行删除
src/Rtsp/RtpReceiver.h
...@@ -26,6 +26,8 @@ template<typename T, typename SEQ = uint16_t> ...@@ -26,6 +26,8 @@ template<typename T, typename SEQ = uint16_t>
class PacketSortor { class PacketSortor {
public: public:
static constexpr SEQ SEQ_MAX = (std::numeric_limits<SEQ>::max)(); static constexpr SEQ SEQ_MAX = (std::numeric_limits<SEQ>::max)();
using iterator = typename std::map<SEQ, T>::iterator;
PacketSortor() = default; PacketSortor() = default;
~PacketSortor() = default; ~PacketSortor() = default;
...@@ -36,7 +38,7 @@ public: ...@@ -36,7 +38,7 @@ public:
*/ */
void clear() { void clear() {
_started = false; _started = false;
_seq_cycle_count = 0; _ticker.resetTime();
_pkt_sort_cache_map.clear(); _pkt_sort_cache_map.clear();
} }
...@@ -46,95 +48,128 @@ public: ...@@ -46,95 +48,128 @@ public:
size_t getJitterSize() const { return _pkt_sort_cache_map.size(); } size_t getJitterSize() const { return _pkt_sort_cache_map.size(); }
/** /**
* 获取seq回环次数
*/
size_t getCycleCount() const { return _seq_cycle_count; }
/**
* 输入并排序 * 输入并排序
* @param seq 序列号 * @param seq 序列号
* @param packet 包负载 * @param packet 包负载
*/ */
void sortPacket(SEQ seq, T packet) { void sortPacket(SEQ seq, T packet) {
_latest_seq = seq;
if (!_started) { if (!_started) {
// 记录第一个seq // 记录第一个seq
_started = true; _started = true;
_last_seq_out = seq - 1; _last_seq_out = seq - 1;
} }
if (seq == static_cast<SEQ>(_last_seq_out + 1)) { auto next_seq = static_cast<SEQ>(_last_seq_out + 1);
if (seq == next_seq) {
// 收到下一个seq // 收到下一个seq
output(seq, std::move(packet)); output(seq, std::move(packet));
// 清空连续包列表
flushPacket();
return; return;
} }
if (seq < _last_seq_out && _last_seq_out != SEQ_MAX && seq < 1024 && _last_seq_out > SEQ_MAX - 1024) { if (seq < next_seq && !mayLooped(next_seq, seq)) {
// seq回环,清空回环前缓存 // 无回环风险, 过滤seq回退包
flush();
_last_seq_out = SEQ_MAX;
++_seq_cycle_count;
sortPacket(seq, std::move(packet));
return; return;
} }
_pkt_sort_cache_map.emplace(seq, std::move(packet));
if (seq <= _last_seq_out && _last_seq_out != SEQ_MAX) { if (needForceFlush(seq)) {
// 这个回退包已经不再等待 forceFlush(next_seq);
return;
} }
}
_pkt_sort_cache_map.emplace(seq, std::move(packet)); void flush() {
auto it_min = _pkt_sort_cache_map.begin(); if (!_pkt_sort_cache_map.empty()) {
auto it_max = _pkt_sort_cache_map.rbegin(); forceFlush(static_cast<SEQ>(_last_seq_out + 1));
if (it_max->first - it_min->first > (SEQ_MAX >> 1)) { _pkt_sort_cache_map.clear();
// 回环后，收到回环前的大值seq, 忽略掉
_pkt_sort_cache_map.erase((++it_max).base());
return;
} }
}
tryFlushFrontPacket(); void setParams(size_t max_buffer_size, size_t max_buffer_ms, size_t max_distance) {
_max_buffer_size = max_buffer_size;
_max_buffer_ms = max_buffer_ms;
_max_distance = max_distance;
}
if (_pkt_sort_cache_map.size() > _max_buffer_size || (_ticker.elapsedTime() > _max_buffer_ms && !_pkt_sort_cache_map.empty())) { private:
// buffer太长，强行减小 SEQ distance(SEQ seq) {
WarnL << "packet dropped: " << static_cast<SEQ>(_last_seq_out + 1) << " -> " SEQ ret;
<< static_cast<SEQ>(_pkt_sort_cache_map.begin()->first - 1) auto next_seq = static_cast<SEQ>(_last_seq_out + 1);
<< ", jitter buffer size: " << _pkt_sort_cache_map.size() if (seq > next_seq) {
<< ", jitter buffer ms: " << _ticker.elapsedTime(); ret = seq - next_seq;
popIterator(_pkt_sort_cache_map.begin()); } else {
ret = next_seq - seq;
} }
if (ret > SEQ_MAX >> 1) {
return SEQ_MAX - ret;
}
return ret;
} }
void flush() { bool needForceFlush(SEQ seq) {
// 清空缓存 return !_pkt_sort_cache_map.empty() && (_pkt_sort_cache_map.size() > _max_buffer_size ||
while (!_pkt_sort_cache_map.empty()) { distance(seq) > _max_distance || _ticker.elapsedTime() > _max_buffer_ms);
popIterator(_pkt_sort_cache_map.begin());
}
} }
private: //外部调用代码确保_pkt_sort_cache_map不为空
void tryFlushFrontPacket() { void forceFlush(SEQ next_seq) {
while (!_pkt_sort_cache_map.empty()) { // 寻找距离比next_seq大的最近的seq
auto it = _pkt_sort_cache_map.begin(); auto it = _pkt_sort_cache_map.lower_bound(next_seq);
auto next_seq = static_cast<SEQ>(_last_seq_out + 1); if (it == _pkt_sort_cache_map.end()) {
if (it->first < next_seq) { // 没有比next_seq更大的seq，应该是回环时丢包导致
_pkt_sort_cache_map.erase(it); it = _pkt_sort_cache_map.begin();
continue; }
// 丢包无法恢复，把这个包当做next_seq
popIterator(it);
// 清空连续包列表
flushPacket();
// 删除距离next_seq太大的包
for (auto it = _pkt_sort_cache_map.begin(); it != _pkt_sort_cache_map.end();) {
if (distance(it->first) > _max_distance) {
it = _pkt_sort_cache_map.erase(it);
} else {
++it;
} }
if (it->first == next_seq) { }
// 连续的seq }
popIterator(it);
bool mayLooped(SEQ last_seq, SEQ now_seq) { return last_seq > SEQ_MAX - _max_distance || now_seq < _max_distance; }
void flushPacket() {
if (_pkt_sort_cache_map.empty()) {
return;
}
auto next_seq = static_cast<SEQ>(_last_seq_out + 1);
auto it = _pkt_sort_cache_map.lower_bound(next_seq);
if (!mayLooped(next_seq, next_seq)) {
// 无回环风险, 清空 < next_seq的值
it = _pkt_sort_cache_map.erase(_pkt_sort_cache_map.begin(), it);
}
while (it != _pkt_sort_cache_map.end()) {
// 找到下一个包
if (it->first == static_cast<SEQ>(_last_seq_out + 1)) {
it = popIterator(it);
continue; continue;
} }
break; break;
} }
} }
void popIterator(typename std::map<SEQ, T>::iterator it) { iterator popIterator(iterator it) {
auto seq = it->first; output(it->first, std::move(it->second));
auto data = std::move(it->second); return _pkt_sort_cache_map.erase(it);
_pkt_sort_cache_map.erase(it);
output(seq, std::move(data));
} }
void output(SEQ seq, T packet) { void output(SEQ seq, T packet) {
auto next_seq = static_cast<SEQ>(_last_seq_out + 1);
if (seq != next_seq) {
WarnL << "packet dropped: " << next_seq << " -> " << static_cast<SEQ>(seq - 1)
<< ", latest seq: " << _latest_seq
<< ", jitter buffer size: " << _pkt_sort_cache_map.size()
<< ", jitter buffer ms: " << _ticker.elapsedTime();
}
_last_seq_out = seq; _last_seq_out = seq;
_cb(seq, std::move(packet)); _cb(seq, std::move(packet));
_ticker.resetTime(); _ticker.resetTime();
...@@ -142,23 +177,25 @@ private: ...@@ -142,23 +177,25 @@ private:
private: private:
bool _started = false; bool _started = false;
//排序缓存最大保存数据长度，单位毫秒 // 排序缓存最大保存数据长度，单位毫秒
size_t _max_buffer_ms = 3000; size_t _max_buffer_ms = 1000;
//排序缓存最大保存数据个数 // 排序缓存最大保存数据个数
size_t _max_buffer_size = 1024; size_t _max_buffer_size = 1024;
//记录上次output至今的时间 // seq最大跳跃距离
size_t _max_distance = 256;
// 记录上次output至今的时间
toolkit::Ticker _ticker; toolkit::Ticker _ticker;
//下次应该输出的SEQ // 最近输入的seq
SEQ _latest_seq = 0;
// 下次应该输出的SEQ
SEQ _last_seq_out = 0; SEQ _last_seq_out = 0;
//seq回环次数计数 // pkt排序缓存，根据seq排序
size_t _seq_cycle_count = 0;
//pkt排序缓存，根据seq排序
std::map<SEQ, T> _pkt_sort_cache_map; std::map<SEQ, T> _pkt_sort_cache_map;
//回调 // 回调
std::function<void(SEQ seq, T packet)> _cb; std::function<void(SEQ seq, T packet)> _cb;
}; };
class RtpTrack : private PacketSortor<RtpPacket::Ptr> { class RtpTrack : public PacketSortor<RtpPacket::Ptr> {
public: public:
class BadRtpException : public std::invalid_argument { class BadRtpException : public std::invalid_argument {
public: public:
...@@ -268,11 +305,6 @@ public: ...@@ -268,11 +305,6 @@ public:
return _track[index].getJitterSize(); return _track[index].getJitterSize();
} }
size_t getCycleCount(int index) const {
assert(index < kCount && index >= 0);
return _track[index].getCycleCount();
}
uint32_t getSSRC(int index) const { uint32_t getSSRC(int index) const {
assert(index < kCount && index >= 0); assert(index < kCount && index >= 0);
return _track[index].getSSRC(); return _track[index].getSSRC();
......
tests/test_sortor.cpp
...@@ -111,7 +111,6 @@ void test_real() { ...@@ -111,7 +111,6 @@ void test_real() {
} }
cout << "输入数据个数:" << input_list.size() cout << "输入数据个数:" << input_list.size()
<< " 回环次数:" << sortor.getCycleCount()
<< " 抖动缓冲区大小:" << sortor.getJitterSize(); << " 抖动缓冲区大小:" << sortor.getJitterSize();
//清空缓存 //清空缓存
...@@ -206,7 +205,6 @@ void test_rand(){ ...@@ -206,7 +205,6 @@ void test_rand(){
i += (count + 1); i += (count + 1);
} }
cout << "输入数据个数:" << input_list.size() cout << "输入数据个数:" << input_list.size()
<< " 回环次数:" << sortor.getCycleCount()
<< " 抖动缓冲区大小:" << sortor.getJitterSize() << " 抖动缓冲区大小:" << sortor.getJitterSize()
<< " 丢包个数:" << drop_list.size() << " 丢包个数:" << drop_list.size()
<< " 重复包个数:" << repeat_list.size(); << " 重复包个数:" << repeat_list.size();
......
webrtc/WebRtcTransport.cpp
...@@ -652,15 +652,14 @@ void WebRtcTransportImp::onRtcConfigure(RtcConfigure &configure) const { ...@@ -652,15 +652,14 @@ void WebRtcTransportImp::onRtcConfigure(RtcConfigure &configure) const {
/////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////
class RtpChannel class RtpChannel : public RtpTrackImp, public std::enable_shared_from_this<RtpChannel> {
: public RtpTrackImp
, public std::enable_shared_from_this<RtpChannel> {
public: public:
RtpChannel(EventPoller::Ptr poller, RtpTrackImp::OnSorted cb, function<void(const FCI_NACK &nack)> on_nack) { RtpChannel(EventPoller::Ptr poller, RtpTrackImp::OnSorted cb, function<void(const FCI_NACK &nack)> on_nack) {
_poller = std::move(poller); _poller = std::move(poller);
_on_nack = std::move(on_nack); _on_nack = std::move(on_nack);
setOnSorted(std::move(cb)); setOnSorted(std::move(cb));
//设置jitter buffer参数
RtpTrackImp::setParams(1024, NackContext::kNackMaxMS, 512);
_nack_ctx.setOnNack([this](const FCI_NACK &nack) { onNack(nack); }); _nack_ctx.setOnNack([this](const FCI_NACK &nack) { onNack(nack); });
} }
......
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论