amass/ZLMediaKit
1
0
Fork 0
mirror of https://github.com/ZLMediaKit/ZLMediaKit.git synced 2024-10-31 08:39:34 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge pull request #990 from ZLMediaKit/feature/packet-sorter-fix-seq

Browse Source 
调整 packet sorter
This commit is contained in:
夏楚 2021-07-18 19:39:38 +08:00 committed by GitHub
commit a9de3723df
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 19 additions and 19 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
src/Rtsp/RtpReceiver.h
View File

@ -37,7 +37,7 @@ public:
*/ */
void clear() { void clear() {
_seq_cycle_count = 0; _seq_cycle_count = 0;
_rtp_sort_cache_map.clear(); _pkt_sort_cache_map.clear();
_next_seq_out = 0; _next_seq_out = 0;
_max_sort_size = kMin; _max_sort_size = kMin;
} }
@ -46,7 +46,7 @@ public:
* 获取排序缓存长度 * 获取排序缓存长度
*/ */
size_t getJitterSize() const{ size_t getJitterSize() const{
return _rtp_sort_cache_map.size(); return _pkt_sort_cache_map.size();
} }
/** /**
@ -67,27 +67,27 @@ public:
//过滤seq回退包(回环包除外) //过滤seq回退包(回环包除外)
return; return;
} }
} else if (_next_seq_out && seq - _next_seq_out > (0xFFFF >> 1)) { } else if (_next_seq_out && seq - _next_seq_out > (std::numeric_limits<SEQ>::max() >> 1)) {
//过滤seq跳变非常大的包(防止回环时乱序时收到非常大的seq) //过滤seq跳变非常大的包(防止回环时乱序时收到非常大的seq)
return; return;
} }
//放入排序缓存 //放入排序缓存
_rtp_sort_cache_map.emplace(seq, std::move(packet)); _pkt_sort_cache_map.emplace(seq, std::move(packet));
//尝试输出排序后的包 //尝试输出排序后的包
tryPopPacket(); tryPopPacket();
} }
void flush(){ void flush(){
//清空缓存 //清空缓存
while (!_rtp_sort_cache_map.empty()) { while (!_pkt_sort_cache_map.empty()) {
popIterator(_rtp_sort_cache_map.begin()); popIterator(_pkt_sort_cache_map.begin());
} }
} }
private: private:
void popPacket() { void popPacket() {
auto it = _rtp_sort_cache_map.begin(); auto it = _pkt_sort_cache_map.begin();
if (it->first >= _next_seq_out) { if (it->first >= _next_seq_out) {
//过滤回跳包 //过滤回跳包
popIterator(it); popIterator(it);
@ -96,37 +96,37 @@ private:
if (_next_seq_out - it->first > (0xFFFF >> 1)) { if (_next_seq_out - it->first > (0xFFFF >> 1)) {
//产生回环了 //产生回环了
if (_rtp_sort_cache_map.size() < 2 * kMin) { if (_pkt_sort_cache_map.size() < 2 * kMin) {
//等足够多的数据后才处理回环, 因为后面还可能出现大的SEQ //等足够多的数据后才处理回环, 因为后面还可能出现大的SEQ
return; return;
} }
++_seq_cycle_count; ++_seq_cycle_count;
//找到大的SEQ并清空掉，然后从小的SEQ重新开始排序 //找到大的SEQ并清空掉，然后从小的SEQ重新开始排序
auto hit = _rtp_sort_cache_map.upper_bound((SEQ) (_next_seq_out - _rtp_sort_cache_map.size())); auto hit = _pkt_sort_cache_map.upper_bound((SEQ) (_next_seq_out - _pkt_sort_cache_map.size()));
while (hit != _rtp_sort_cache_map.end()) { while (hit != _pkt_sort_cache_map.end()) {
//回环前，清空剩余的大的SEQ的数据 //回环前，清空剩余的大的SEQ的数据
_cb(hit->first, hit->second); _cb(hit->first, hit->second);
hit = _rtp_sort_cache_map.erase(hit); hit = _pkt_sort_cache_map.erase(hit);
} }
//下一个回环的数据 //下一个回环的数据
popIterator(_rtp_sort_cache_map.begin()); popIterator(_pkt_sort_cache_map.begin());
return; return;
} }
//删除回跳的数据包 //删除回跳的数据包
_rtp_sort_cache_map.erase(it); _pkt_sort_cache_map.erase(it);
} }
void popIterator(typename map<SEQ, T>::iterator it) { void popIterator(typename map<SEQ, T>::iterator it) {
auto seq = it->first; auto seq = it->first;
auto data = std::move(it->second); auto data = std::move(it->second);
_rtp_sort_cache_map.erase(it); _pkt_sort_cache_map.erase(it);
_next_seq_out = seq + 1; _next_seq_out = seq + 1;
_cb(seq, data); _cb(seq, data);
} }
void tryPopPacket() { void tryPopPacket() {
int count = 0; int count = 0;
while ((!_rtp_sort_cache_map.empty() && _rtp_sort_cache_map.begin()->first == _next_seq_out)) { while ((!_pkt_sort_cache_map.empty() && _pkt_sort_cache_map.begin()->first == _next_seq_out)) {
//找到下个包，直接输出 //找到下个包，直接输出
popPacket(); popPacket();
++count; ++count;
@ -134,7 +134,7 @@ private:
if (count) { if (count) {
setSortSize(); setSortSize();
} else if (_rtp_sort_cache_map.size() > _max_sort_size) { } else if (_pkt_sort_cache_map.size() > _max_sort_size) {
//排序缓存溢出，不再继续排序 //排序缓存溢出，不再继续排序
popPacket(); popPacket();
setSortSize(); setSortSize();
@ -142,7 +142,7 @@ private:
} }
void setSortSize() { void setSortSize() {
_max_sort_size = kMin + _rtp_sort_cache_map.size(); _max_sort_size = kMin + _pkt_sort_cache_map.size();
if (_max_sort_size > kMax) { if (_max_sort_size > kMax) {
_max_sort_size = kMax; _max_sort_size = kMax;
} }
@ -155,8 +155,8 @@ private:
size_t _seq_cycle_count = 0; size_t _seq_cycle_count = 0;
//排序缓存长度 //排序缓存长度
size_t _max_sort_size = kMin; size_t _max_sort_size = kMin;
//rtp排序缓存，根据seq排序 //pkt排序缓存，根据seq排序
map<SEQ, T> _rtp_sort_cache_map; map<SEQ, T> _pkt_sort_cache_map;
//回调 //回调
function<void(SEQ seq, T &packet)> _cb; function<void(SEQ seq, T &packet)> _cb;
}; };