From fda4e83ae421e5075dc4b88143639e1a257d9ccb Mon Sep 17 00:00:00 2001 From: xiongguangjie Date: Sat, 18 Jun 2022 13:01:04 +0800 Subject: [PATCH] srt support tlpktdrop=0 feature --- srt/PacketQueue.cpp | 11 ++++++++--- srt/PacketQueue.hpp | 5 ++++- srt/PacketSendQueue.cpp | 11 ++++++++--- srt/PacketSendQueue.hpp | 4 +++- srt/SrtTransport.cpp | 9 +++++---- 5 files changed, 28 insertions(+), 12 deletions(-) diff --git a/srt/PacketQueue.cpp b/srt/PacketQueue.cpp index 37980ae0..9349f050 100644 --- a/srt/PacketQueue.cpp +++ b/srt/PacketQueue.cpp @@ -237,11 +237,16 @@ std::string PacketQueue::dump() { //////////////////// PacketRecvQueue ////////////////////////////////// -PacketRecvQueue::PacketRecvQueue(uint32_t max_size, uint32_t init_seq, uint32_t latency) +PacketRecvQueue::PacketRecvQueue(uint32_t max_size, uint32_t init_seq, uint32_t latency,uint32_t flag) : _pkt_cap(max_size) , _pkt_latency(latency) , _pkt_expected_seq(init_seq) - , _pkt_buf(max_size) {} + , _pkt_buf(max_size) + , _srt_flag(flag) {} + +bool PacketRecvQueue::TLPKTDrop(){ + return (_srt_flag&HSExtMessage::HS_EXT_MSG_TLPKTDROP) && (_srt_flag &HSExtMessage::HS_EXT_MSG_TSBPDRCV); +} bool PacketRecvQueue::inputPacket(DataPacket::Ptr pkt, std::list &out) { // TraceL << dump() << " seq:" << pkt->packet_seq_number; while (_size > 0 && _start == _end) { @@ -265,7 +270,7 @@ bool PacketRecvQueue::inputPacket(DataPacket::Ptr pkt, std::list _pkt_latency) { + while (timeLatency() > _pkt_latency && TLPKTDrop()) { it = _pkt_buf[_start]; if (it) { _pkt_buf[_start] = nullptr; diff --git a/srt/PacketQueue.hpp b/srt/PacketQueue.hpp index d3e224c6..95ae9c66 100644 --- a/srt/PacketQueue.hpp +++ b/srt/PacketQueue.hpp @@ -65,7 +65,7 @@ class PacketRecvQueue : public PacketQueueInterface { public: using Ptr = std::shared_ptr; - PacketRecvQueue(uint32_t max_size, uint32_t init_seq, uint32_t latency); + PacketRecvQueue(uint32_t max_size, uint32_t init_seq, uint32_t latency,uint32_t flag = 0xbf); ~PacketRecvQueue() = default; bool inputPacket(DataPacket::Ptr pkt, std::list &out); @@ -85,12 +85,15 @@ private: void insertToCycleBuf(DataPacket::Ptr pkt, uint32_t diff); DataPacket::Ptr getFirst(); DataPacket::Ptr getLast(); + bool TLPKTDrop(); private: uint32_t _pkt_cap; uint32_t _pkt_latency; uint32_t _pkt_expected_seq; + uint32_t _srt_flag; + std::vector _pkt_buf; uint32_t _start = 0; uint32_t _end = 0; diff --git a/srt/PacketSendQueue.cpp b/srt/PacketSendQueue.cpp index 92730a4a..06beddde 100644 --- a/srt/PacketSendQueue.cpp +++ b/srt/PacketSendQueue.cpp @@ -2,9 +2,10 @@ namespace SRT { -PacketSendQueue::PacketSendQueue(uint32_t max_size, uint32_t latency) +PacketSendQueue::PacketSendQueue(uint32_t max_size, uint32_t latency,uint32_t flag) : _pkt_cap(max_size) - , _pkt_latency(latency) {} + , _pkt_latency(latency) + , _srt_flag(flag) {} bool PacketSendQueue::drop(uint32_t num) { decltype(_pkt_cache.begin()) it; @@ -24,12 +25,16 @@ bool PacketSendQueue::inputPacket(DataPacket::Ptr pkt) { while (_pkt_cache.size() > _pkt_cap) { _pkt_cache.pop_front(); } - while (timeLatency() > _pkt_latency) { + while (timeLatency() > _pkt_latency && TLPKTDrop()) { _pkt_cache.pop_front(); } return true; } +bool PacketSendQueue::TLPKTDrop(){ + return (_srt_flag&HSExtMessage::HS_EXT_MSG_TLPKTDROP) && (_srt_flag &HSExtMessage::HS_EXT_MSG_TSBPDSND); +} + std::list PacketSendQueue::findPacketBySeq(uint32_t start, uint32_t end) { std::list re; decltype(_pkt_cache.begin()) it; diff --git a/srt/PacketSendQueue.hpp b/srt/PacketSendQueue.hpp index be91c663..48bb36b2 100644 --- a/srt/PacketSendQueue.hpp +++ b/srt/PacketSendQueue.hpp @@ -16,7 +16,7 @@ public: using Ptr = std::shared_ptr; using LostPair = std::pair; - PacketSendQueue(uint32_t max_size, uint32_t latency); + PacketSendQueue(uint32_t max_size, uint32_t latency,uint32_t flag = 0xbf); ~PacketSendQueue() = default; bool drop(uint32_t num); @@ -25,7 +25,9 @@ public: private: uint32_t timeLatency(); + bool TLPKTDrop(); private: + uint32_t _srt_flag; uint32_t _pkt_cap; uint32_t _pkt_latency; std::list _pkt_cache; diff --git a/srt/SrtTransport.cpp b/srt/SrtTransport.cpp index 52ecc449..0ef720f2 100644 --- a/srt/SrtTransport.cpp +++ b/srt/SrtTransport.cpp @@ -182,9 +182,9 @@ void SrtTransport::handleHandshakeConclusion(HandshakePacket &pkt, struct sockad } if (req) { if (req->srt_flag != srt_flag) { - WarnL << " not support flag " << req->srt_flag; + WarnL << " flag " << req->srt_flag; } - // srt_flag = req->srt_flag; + srt_flag = req->srt_flag; delay = delay <= req->recv_tsbpd_delay ? req->recv_tsbpd_delay : delay; } TraceL << getIdentifier() << " CONCLUSION Phase "; @@ -214,8 +214,8 @@ void SrtTransport::handleHandshakeConclusion(HandshakePacket &pkt, struct sockad sendControlPacket(res, true); TraceL << " buf size = " << res->max_flow_window_size << " init seq =" << _init_seq_number << " latency=" << delay; - _recv_buf = std::make_shared(getPktBufSize(), _init_seq_number, delay * 1e3); - _send_buf = std::make_shared(getPktBufSize(), delay * 1e3); + _recv_buf = std::make_shared(getPktBufSize(), _init_seq_number, delay * 1e3,srt_flag); + _send_buf = std::make_shared(getPktBufSize(), delay * 1e3,srt_flag); _send_packet_seq_number = _init_seq_number; _buf_delay = delay; onHandShakeFinished(_stream_id, addr); @@ -413,6 +413,7 @@ void SrtTransport::sendACKPacket() { _last_ack_pkt_seq_num = pkt->last_ack_pkt_seq_number; sendControlPacket(pkt, true); // TraceL<<"send ack "<dump(); + // TraceL<<_recv_buf->dump(); } void SrtTransport::sendLightACKPacket() {