diff --git a/srtcore/buffer_rcv.cpp b/srtcore/buffer_rcv.cpp index 88f94103b..45f9ecff6 100644 --- a/srtcore/buffer_rcv.cpp +++ b/srtcore/buffer_rcv.cpp @@ -206,6 +206,52 @@ int CRcvBuffer::insert(CUnit* unit) return 0; } +int CRcvBuffer::erase(int32_t seqno) +{ + const int offset = CSeqNo::seqoff(m_iStartSeqNo, seqno); + if (offset < 0) + { + LOGC(rbuflog.Debug, log << "CRcvBuffer.erase(): nothing to erase. Requested @" << seqno + << ". Buffer start " << m_iStartSeqNo << "."); + return 0; + } + + const int pos = incPos(m_iStartPos, offset); + if (!m_entries[pos].pUnit) + return 0; + + const bool bMsgOrderFlag = packetAt(pos).getMsgOrderFlag(); + // Undo insert()'s accounting. releaseUnitInPos() does not do it, and callers that + // release a unit account for it themselves (see readMessage, dropMessage). Without + // this the retransmission that lands in this slot is counted a second time. + countBytes(-1, -(int)packetAt(pos).getLength()); + // Leaves the entry EntryState_Empty, so that a retransmission of the same + // sequence number can still be inserted into this slot. + releaseUnitInPos(pos); + + if (m_bMessageAPI && !bMsgOrderFlag && !m_tsbpd.isEnabled()) + { + --m_numOutOfOrderPackets; + if (pos == m_iFirstReadableOutOfOrder) + { + m_iFirstReadableOutOfOrder = -1; + updateFirstReadableOutOfOrder(); + } + } + + HLOGC(rbuflog.Debug, log << "CRcvBuffer.erase(): @" << seqno << "."); + + // Check if a unit before m_iFirstNonreadPos was erased. + const bool needUpdateNonreadPos = offset <= getRcvDataSize(); + if (needUpdateNonreadPos) + { + m_iFirstNonreadPos = m_iStartPos; + updateNonreadPos(); + } + + return 1; +} + std::pair CRcvBuffer::dropUpTo(int32_t seqno) { IF_RCVBUF_DEBUG(ScopedLog scoped_log); diff --git a/srtcore/buffer_rcv.h b/srtcore/buffer_rcv.h index eac1c7c47..9fcbe92f7 100644 --- a/srtcore/buffer_rcv.h +++ b/srtcore/buffer_rcv.h @@ -64,6 +64,12 @@ class CRcvBuffer // TODO: Previously '-2' also meant 'already acknowledged'. Check usage of this value. int insert(CUnit* unit); + /// Erase a packet from the buffer based on the packet sequence number. + /// The entry is marked EntryState_Empty. The CUnit is marked free. + /// @param seqno packet sequence number. + /// @return the number of packets erased. + int erase(int32_t seqno); + /// Drop packets in the receiver buffer from the current position up to the seqno (excluding seqno). /// @param [in] seqno drop units up to this sequence number /// @return number of dropped (missing) and discarded (available) packets as a pair(dropped, discarded). diff --git a/srtcore/core.cpp b/srtcore/core.cpp index ece511b3e..eae7c7d95 100644 --- a/srtcore/core.cpp +++ b/srtcore/core.cpp @@ -10631,6 +10631,9 @@ int srt::CUDT::handleSocketPacketReception(const vector& incoming, bool& const bool retransmitted = pktrexmitflag == 1; bool adding_successful = true; + // The packet arrived, but failed AEAD decryption and was erased from the RCV buffer. + // It is therefore still missing and has to be recovered like a lost packet. + bool undecrypted = false; const int32_t bufidx = CSeqNo::seqoff(bufseq, rpkt.seqno()); @@ -10739,6 +10742,7 @@ int srt::CUDT::handleSocketPacketReception(const vector& incoming, bool& if (rc != ENCS_CLEAR) { adding_successful = false; + undecrypted = true; IF_HEAVY_LOGGING(exc_type = "UNDECRYPTED"); // If TSBPD is disabled, then SRT either operates in buffer mode, of in message API without a restriction @@ -10750,20 +10754,24 @@ int srt::CUDT::handleSocketPacketReception(const vector& incoming, bool& // See issue ##2626. SRT_ASSERT(m_bTsbPd); - // Drop the packet from the receiver buffer. - // The packet was added to the buffer based on the sequence number, therefore sequence number should be used to drop it from the buffer. - // A drawback is that it would prevent a valid packet with the same sequence number, if it happens to arrive later, to end up in the buffer. - const int iDropCnt = m_pRcvBuffer->dropMessage(u->m_Packet.getSeqNo(), u->m_Packet.getSeqNo(), SRT_MSGNO_NONE, CRcvBuffer::DROP_EXISTING); + // Erase the packet from the receiver buffer, rather than dropping it. + // Dropping marks the buffer slot as dropped, which would prevent a valid packet + // with the same sequence number, if it happens to arrive later, from ending up + // in the buffer. Erasing leaves the slot fillable. + // The packet was added to the buffer based on the sequence number, therefore the + // sequence number is used to erase it from the buffer. + // The erased sequence is put back into the loss list further below, so that ARQ + // retransmits a good copy of it into the slot that was just freed. + const int iEraseCnt = m_pRcvBuffer->erase(u->m_Packet.getSeqNo()); const steady_clock::time_point tnow = steady_clock::now(); ScopedLock lg(m_StatsLock); - m_stats.rcvr.dropped.count(stats::BytesPackets(iDropCnt * rpkt.getLength(), iDropCnt)); m_stats.rcvr.undecrypted.count(stats::BytesPackets(rpkt.getLength(), 1)); string why; if (frequentLogAllowed(FREQLOGFA_ENCRYPTION_FAILURE, tnow, (why))) { - LOGC(qrlog.Warn, log << CONID() << "Decryption failed (seqno %" << u->m_Packet.getSeqNo() << "), dropped " - << iDropCnt << ". pktRcvUndecryptTotal=" << m_stats.rcvr.undecrypted.total.count() << "." << why); + LOGC(qrlog.Warn, log << CONID() << "Decryption failed (seqno %" << u->m_Packet.getSeqNo() << "), erased " + << iEraseCnt << ", re-requesting. pktRcvUndecryptTotal=" << m_stats.rcvr.undecrypted.total.count() << "." << why); } #if SRT_ENABLE_FREQUENT_LOG_TRACE else @@ -10837,7 +10845,11 @@ int srt::CUDT::handleSocketPacketReception(const vector& incoming, bool& // Decryption should have made the crypto flags EK_NOENC. // Otherwise it's an error. - if (adding_successful) + // Loss detection must also run for an undecrypted packet: the sequence jump preceding it + // is real regardless of whether this particular packet could be decrypted, and + // m_iRcvCurrSeqNo advances below either way. Skipping the check would leave the jumped-over + // sequences in no loss list at all - never NAKed, and the ACK eventually passes over them. + if (adding_successful || undecrypted) { HLOGC(qrlog.Debug, log << CONID() @@ -10852,6 +10864,13 @@ int srt::CUDT::handleSocketPacketReception(const vector& incoming, bool& } } + // Re-request the erased (undecrypted) sequence so that ARQ retransmits a good copy into the + // buffer slot that erase() left fillable. This must come after the loss detection above: + // CRcvLossList::insert() rejects entries below its largest-ever sequence, so a preceding + // jump range has to be recorded first. + if (undecrypted) + w_srt_loss_seqs.push_back(make_pair(rpkt.seqno(), rpkt.seqno())); + // Update the current largest sequence number that has been received. // Or it is a retransmitted packet, remove it from receiver loss list. if (CSeqNo::seqcmp(rpkt.seqno(), m_iRcvCurrSeqNo) > 0) @@ -10864,7 +10883,13 @@ int srt::CUDT::handleSocketPacketReception(const vector& incoming, bool& } else { - unlose(rpkt); // was BELATED or RETRANSMITTED + // An undecrypted packet was erased from the buffer and is therefore still missing. + // Keep it in the receiver loss list: the ACK position (getFirstNoncontSequence) is + // derived from that list, so removing the entry would let the ACK advance past the + // erased hole, after which every retransmission of it is discarded as belated and the + // sequence can never be recovered. + if (!undecrypted) + unlose(rpkt); // was BELATED or RETRANSMITTED w_was_sent_in_order &= 0 != pktrexmitflag; } }