[core] Fixed updating new RCV buffer on ISN change. (#2309)

srtcore/buffer_rcv.cpp

@@ -200,6 +200,15 @@ int CRcvBufferNew::dropUpTo(int32_t seqno)
     return iDropCnt;
 }
 
+int CRcvBufferNew::dropAll()
+{
+    if (empty())
+        return 0;
+
+    const int end_seqno = CSeqNo::incseq(m_iStartSeqNo, m_iMaxPosInc);
+    return dropUpTo(end_seqno);
+}
+
 int CRcvBufferNew::dropMessage(int32_t seqnolo, int32_t seqnohi, int32_t msgno)
 {
     IF_RCVBUF_DEBUG(ScopedLog scoped_log);

srtcore/buffer_rcv.h

@@ -70,9 +70,13 @@ class CRcvBufferNew
     /// 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 packets.
-    ///
     int dropUpTo(int32_t seqno);
 
+    /// @brief Drop all the packets in the receiver buffer.
+    /// The starting position and seqno are shifted right after the last packet in the buffer.
+    /// @return the number of dropped packets.
+    int dropAll();
+
     /// @brief Drop the whole message from the buffer.
     /// If message number is 0, then use sequence numbers to locate sequence range to drop [seqnolo, seqnohi].
     /// When one packet of the message is in the range of dropping, the whole message is to be dropped.
@@ -109,6 +113,10 @@ class CRcvBufferNew
     /// Get the starting position of the buffer as a packet sequence number.
     int getStartSeqNo() const { return m_iStartSeqNo; }
 
+    /// Sets the start seqno of the buffer.
+    /// Must be used with caution and only when the buffer is empty.
+    void setStartSeqNo(int seqno) { m_iStartSeqNo = seqno; }
+
     /// Given the sequence number of the first unacknowledged packet
     /// tells the size of the buffer available for packets.
     /// Effective returns capacity of the buffer minus acknowledged packet still kept in it.
@@ -325,8 +333,6 @@ class CRcvBufferNew
 
     void setPeerRexmitFlag(bool flag) { m_bPeerRexmitFlag = flag; } 
 
-    void applyGroupISN(int rcv_isn) { m_iStartSeqNo = rcv_isn; }
-
     void applyGroupTime(const time_point& timebase, bool wrp, uint32_t delay, const duration& udrift);
 
     void applyGroupDrift(const time_point& timebase, bool wrp, const duration& udrift);

srtcore/core.cpp

@@ -3333,11 +3333,6 @@ void srt::CUDT::synchronizeWithGroup(CUDTGroup* gp)
                     << " (shift by " << CSeqNo::seqcmp(snd_isn, m_iSndLastAck) << ")");
             setInitialRcvSeq(rcv_isn);
             setInitialSndSeq(snd_isn);
-#if ENABLE_NEW_RCVBUFFER
-            enterCS(m_RecvLock);
-            m_pRcvBuffer->applyGroupISN(rcv_isn);
-            leaveCS(m_RecvLock);
-#endif
         }
         else
         {
@@ -5327,6 +5322,34 @@ void * srt::CUDT::tsbpd(void* param)
     return NULL;
 }
 
+void srt::CUDT::setInitialRcvSeq(int32_t isn)
+{
+    m_iRcvLastAck = isn;
+#ifdef ENABLE_LOGGING
+    m_iDebugPrevLastAck = m_iRcvLastAck;
+#endif
+    m_iRcvLastSkipAck = m_iRcvLastAck;
+    m_iRcvLastAckAck = isn;
+    m_iRcvCurrSeqNo = CSeqNo::decseq(isn);
+
+#if ENABLE_NEW_RCVBUFFER
+    sync::ScopedLock rb(m_RcvBufferLock);
+    if (m_pRcvBuffer)
+    {
+        if (!m_pRcvBuffer->empty())
+        {
+            LOGC(cnlog.Error, log << "IPE: setInitialRcvSeq expected empty RCV buffer. Dropping all.");
+            const int iDropCnt = m_pRcvBuffer->dropAll();
+            const uint64_t avgpayloadsz = m_pRcvBuffer->getRcvAvgPayloadSize();
+            sync::ScopedLock sl(m_StatsLock);
+            m_stats.rcvr.dropped.count(stats::BytesPackets(iDropCnt * avgpayloadsz, (size_t)iDropCnt));
+        }
+
+        m_pRcvBuffer->setStartSeqNo(m_iRcvLastSkipAck);
+    }
+#endif
+}
+
 int srt::CUDT::rcvDropTooLateUpTo(int seqno)
 {
     const int seq_gap_len = CSeqNo::seqoff(m_iRcvLastSkipAck, seqno);

srtcore/core.h

@@ -887,16 +887,7 @@ class CUDT
         m_iSndLastAck2 = isn;
     }
 
-    void setInitialRcvSeq(int32_t isn)
-    {
-        m_iRcvLastAck = isn;
-#ifdef ENABLE_LOGGING
-        m_iDebugPrevLastAck = m_iRcvLastAck;
-#endif
-        m_iRcvLastSkipAck = m_iRcvLastAck;
-        m_iRcvLastAckAck = isn;
-        m_iRcvCurrSeqNo = CSeqNo::decseq(isn);
-    }
+    void setInitialRcvSeq(int32_t isn);
 
     int32_t m_iISN;                              // Initial Sequence Number
     bool m_bPeerTsbPd;                           // Peer accept TimeStamp-Based Rx mode