/*

* SRT - Secure, Reliable, Transport

* Copyright (c) 2018 Haivision Systems Inc.

*

* This Source Code Form is subject to the terms of the Mozilla Public

* License, v. 2.0. If a copy of the MPL was not distributed with this

* file, You can obtain one at http://mozilla.org/MPL/2.0/.

*

*/

/*****************************************************************************

Copyright (c) 2001 - 2009, The Board of Trustees of the University of Illinois.

All rights reserved.

Redistribution and use in source and binary forms, with or without

modification, are permitted provided that the following conditions are

met:

* Redistributions of source code must retain the above

copyright notice, this list of conditions and the

following disclaimer.

* Redistributions in binary form must reproduce the

above copyright notice, this list of conditions

and the following disclaimer in the documentation

and/or other materials provided with the distribution.

* Neither the name of the University of Illinois

nor the names of its contributors may be used to

endorse or promote products derived from this

software without specific prior written permission.

THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS

IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO,

THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR

PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR

CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,

EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,

PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR

PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF

LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING

NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS

SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

*****************************************************************************/

#include <cmath>

#include <limits>

#include <fstream>

#include "buffer_rcv.h"

#include "logging.h"

#include "logger_fas.h"

using namespace std;

using namespace srt::sync;

using namespace srt::logging;

using namespace hvu;

#define rbuflog brlog

namespace srt {

namespace {

struct ScopedLog

{

ScopedLog() {}

~ScopedLog()

{

LOGC(rbuflog.Warn, log << ss.str());

}

stringstream ss;

};

#define IF_RCVBUF_DEBUG(instr) (void)0

}

/*

* RcvBufferNew (circular buffer):

*

* |<------------------- m_iSize ----------------------------->|

* | |<----------- m_iMaxPosOff ------------>| |

* | | | |

* +---+---+---+---+---+---+---+---+---+---+---+---+---+ +---+

* | 0 | 0 | 1 | 1 | 1 | 0 | 1 | 1 | 1 | 1 | 0 | 1 | 0 |...| 0 | m_pUnit[]

* +---+---+---+---+---+---+---+---+---+---+---+---+---+ +---+

* | |

* | |__last pkt received

* |___ m_iStartPos: first message to read

*

* m_pUnit[i]->m_iFlag: 0:free, 1:good, 2:passack, 3:dropped

*

* thread safety:

* m_iStartPos: CUDT::m_RecvLock

* m_iLastAckPos: CUDT::m_AckLock

* m_iMaxPosOff: none? (modified on add and ack

*/

CRcvBuffer::CRcvBuffer(int initSeqNo, size_t size, CUnitQueue* unitqueue, bool bMessageAPI)

: m_entries(size)

, m_szSize(size) // TODO: maybe just use m_entries.size()

, m_pUnitQueue(unitqueue)

, m_iStartSeqNo(initSeqNo) // NOTE: SRT_SEQNO_NONE is allowed here.

, m_iStartPos(0)

, m_iEndOff(0)

, m_iDropOff(0)

, m_iFirstNonreadPos(0)

, m_iMaxPosOff(0)

, m_iNotch(0)

, m_numNonOrderPackets(0)

, m_iFirstNonOrderMsgPos(CPos_TRAP)

, m_bPeerRexmitFlag(true)

, m_bMessageAPI(bMessageAPI)

, m_iBytesCount(0)

, m_iPktsCount(0)

, m_uAvgPayloadSz(0)

{

SRT_ASSERT(size < size_t(std::numeric_limits<int>::max())); // All position pointers are integers

}

CRcvBuffer::~CRcvBuffer()

{

// Can be optimized by only iterating m_iMaxPosOff from m_iStartPos.

for (FixedArray<Entry>::iterator it = m_entries.begin(); it != m_entries.end(); ++it)

{

if (!it->pUnit)

continue;

m_pUnitQueue->makeUnitFree(it->pUnit);

it->pUnit = NULL;

}

}

void CRcvBuffer::debugShowState(const char* source SRT_ATR_UNUSED)

{

HLOGC(brlog.Debug, log << "RCV-BUF-STATE(" << source

<< ") start=" << m_iStartPos

<< " end=+" << m_iEndOff

<< " drop=+" << m_iDropOff

<< " max-off=+" << m_iMaxPosOff

<< " seq[start]=%" << m_iStartSeqNo.val());

}

CRcvBuffer::InsertInfo CRcvBuffer::insert(CUnit* unit)

{

SRT_ASSERT(unit != NULL);

const int32_t seqno = unit->m_Packet.getSeqNo();

const COff offset = COff(SeqNo(seqno) - m_iStartSeqNo);

IF_RCVBUF_DEBUG(ScopedLog scoped_log);

IF_RCVBUF_DEBUG(scoped_log.ss << "CRcvBuffer::insert: seqno " << seqno);

IF_RCVBUF_DEBUG(scoped_log.ss << " msgno " << unit->m_Packet.getMsgSeq(m_bPeerRexmitFlag));

IF_RCVBUF_DEBUG(scoped_log.ss << " m_iStartSeqNo " << m_iStartSeqNo << " offset " << offset);

if (offset < COff(0))

{

IF_RCVBUF_DEBUG(scoped_log.ss << " returns -2");

return InsertInfo(InsertInfo::BELATED);

}

IF_HEAVY_LOGGING(string debug_source = fmtcat("insert %", seqno));

if (offset >= COff(capacity()))

{

IF_RCVBUF_DEBUG(scoped_log.ss << " returns -3");

InsertInfo ireport (InsertInfo::DISCREPANCY);

getAvailInfo((ireport));

IF_HEAVY_LOGGING(debugShowState((debug_source + " overflow").c_str()));

return ireport;

}

// TODO: Don't do assert here. Process this situation somehow.

// If >= 2, then probably there is a long gap, and buffer needs to be reset.

SRT_ASSERT((m_iStartPos + offset) / m_szSize < 2);

const CPos newpktpos = incPos(m_iStartPos, offset);

const COff prev_max_off = m_iMaxPosOff;

bool extended_end = false;

if (offset >= m_iMaxPosOff)

{

m_iMaxPosOff = offset + COff(1);

extended_end = true;

}

// Packet already exists

// (NOTE: the above extension of m_iMaxPosOff is

// possible even before checking that the packet

// exists because existence of a packet beyond

// the current max position is not possible).

SRT_ASSERT(newpktpos >= 0 && newpktpos < int(m_szSize));

if (m_entries[newpktpos].status != EntryState_Empty)

{

IF_RCVBUF_DEBUG(scoped_log.ss << " returns -1");

IF_HEAVY_LOGGING(debugShowState((debug_source + " redundant").c_str()));

return InsertInfo(InsertInfo::REDUNDANT);

}

SRT_ASSERT(m_entries[newpktpos].pUnit == NULL);

m_pUnitQueue->makeUnitTaken(unit);

m_entries[newpktpos].pUnit = unit;

m_entries[newpktpos].status = EntryState_Avail;

countBytes(1, (int)unit->m_Packet.getLength());

// Set to a value, if due to insertion there was added

// a packet that is earlier to be retrieved than the earliest

// currently available packet.

time_point earlier_time = updatePosInfo(unit, prev_max_off, offset, extended_end);

InsertInfo ireport (InsertInfo::INSERTED);

ireport.first_time = earlier_time;

// If packet "in order" flag is zero, it can be read out of order.

// With TSBPD enabled packets are always assumed in order (the flag is ignored).

if (!m_tsbpd.isEnabled() && m_bMessageAPI && !unit->m_Packet.getMsgOrderFlag())

{

++m_numNonOrderPackets;

onInsertNonOrderPacket(newpktpos);

}

updateNonreadPos();

// This updates only the first_seq and avail_range fields.

getAvailInfo((ireport));

IF_RCVBUF_DEBUG(scoped_log.ss << " returns 0 (OK)");

IF_HEAVY_LOGGING(debugShowState((debug_source + " ok").c_str()));

return ireport;

}

void CRcvBuffer::getAvailInfo(CRcvBuffer::InsertInfo& w_if)

{

// This finds the first possible available packet, which is

// preferably at cell 0, but if not available, try also with

// given fallback position, if it's set

if (m_entries[m_iStartPos].status == EntryState_Avail)

{

const CPacket* pkt = &packetAt(m_iStartPos);

SRT_ASSERT(pkt);

w_if.avail_range = m_iEndOff;

w_if.first_seq = SeqNo(pkt->getSeqNo());

return;

}

// If not the first position, probe the skipped positions:

// - for live mode, check the DROP position

// (for potential after-drop reading)

// - for message mode, check the non-order message position

// (for potential out-of-oder message delivery)

const CPacket* pkt = NULL;

if (m_tsbpd.isEnabled())

{

// With TSBPD you can rely on drop position, if set

// Drop position must point always to a valid packet.

// Drop position must start from +1; 0 means no drop.

if (m_iDropOff)

{

pkt = &packetAt(incPos(m_iStartPos, m_iDropOff));

SRT_ASSERT(pkt);

}

}

else

{

// Message-mode: try non-order read position.

if (m_iFirstNonOrderMsgPos != CPos_TRAP)

{

pkt = &packetAt(m_iFirstNonOrderMsgPos);

SRT_ASSERT(pkt);

}

}

if (!pkt)

{

// This is default, but set just in case

// The default seq is SRT_SEQNO_NONE.

w_if.avail_range = COff(0);

return;

}

// TODO: we know that at least 1 packet is available, but only

// with m_iEndOff we know where the true range is. This could also

// be implemented for message mode, but still this would employ

// a separate begin-end range declared for a complete out-of-order

// message.

w_if.avail_range = COff(1);

w_if.first_seq = SeqNo(pkt->getSeqNo());

}

// This function is called exclusively after packet insertion.

// This will update also m_iEndOff and m_iDropOff fields (the latter

// regardless of the TSBPD mode).

CRcvBuffer::time_point CRcvBuffer::updatePosInfo(const CUnit* unit, const COff prev_max_off,

const COff offset,

const bool extended_end)

{

time_point earlier_time;

// Update flags

// Case [A]: insertion of the packet has extended the busy region.

if (extended_end)

{

// THIS means that the buffer WAS CONTIGUOUS BEFORE.

if (m_iEndOff == prev_max_off)

{

// THIS means that the new packet didn't CAUSE a gap

if (m_iMaxPosOff == prev_max_off + 1)

{

// This means that m_iEndOff now shifts by 1,

// and m_iDropOff is set to 0 as there's no gap.

m_iEndOff = m_iMaxPosOff;

m_iDropOff = 0;

}

else

{

// Otherwise we have a drop-after-gap candidate

// which is the currently inserted packet.

// Therefore m_iEndOff STAYS WHERE IT IS.

m_iDropOff = m_iMaxPosOff - 1;

}

}

}

//

// Since this place, every 'offset' is in the range

// between m_iEndOff (inclusive) and m_iMaxPosOff.

else if (offset == m_iEndOff)

{

// Case [D]: inserted a packet at the first gap following the

// contiguous region. This makes a potential to extend the

// contiguous region and we need to find its end.

// If insertion happened at the very first packet, it is the

// new earliest packet now. In any other situation under this

// condition there's some contiguous packet range preceding

// this position.

if (m_iEndOff == 0)

{

earlier_time = getPktTsbPdTime(unit->m_Packet.getMsgTimeStamp());

}

updateGapInfo();

}

else if (offset < m_iDropOff)

{

// Case [C]: the newly inserted packet precedes the

// previous earliest delivery position after drop,

// that is, there is now a "better" after-drop delivery

// candidate.

// New position updated a valid packet on an earlier

// position than the drop position was before, although still

// following a gap.

//

// We know it because if the position has filled a gap following

// a valid packet, this preceding valid packet would be pointed

// by m_iDropOff, or it would point to some earlier packet in a

// contiguous series of valid packets following a gap, hence

// the above condition wouldn't be satisfied.

m_iDropOff = offset;

// If there's an inserted packet BEFORE drop-pos (which makes it

// a new drop-pos), while the very first packet is absent (the

// below condition), it means we have a new earliest-available

// packet. Otherwise we would have only a newly updated drop

// position, but still following some earlier contiguous range

// of valid packets - so it's earlier than previous drop, but

// not earlier than the earliest packet.

if (m_iEndOff == 0)

{

earlier_time = getPktTsbPdTime(unit->m_Packet.getMsgTimeStamp());

}

}

// OTHERWISE: case [B] in which nothing is to be updated.

return earlier_time;

}

// This function is called when the m_iEndOff has been set to a new

// position and the m_iDropOff should be calculated since that position again.

void CRcvBuffer::updateGapInfo()

{

COff from = m_iEndOff;

SRT_ASSERT(m_entries[incPos(m_iStartPos, m_iMaxPosOff)].status == EntryState_Empty);

CPos pos = incPos(m_iStartPos, from);

if (m_entries[pos].status == EntryState_Avail)

{

CPos end_pos = incPos(m_iStartPos, m_iMaxPosOff);

for (; pos != end_pos; pos = incPos(pos))

{

if (m_entries[pos].status != EntryState_Avail)

break;

}

m_iEndOff = offPos(m_iStartPos, pos);

}

// XXX This should be this way, but there are still inconsistencies

// in the message code.

//USE: SRT_ASSERT(m_entries[incPos(m_iStartPos, m_iEndOff)].status == EntryState_Empty);

SRT_ASSERT(m_entries[incPos(m_iStartPos, m_iEndOff)].status != EntryState_Avail);

// XXX Controversy: m_iDropOff is only used in case when SRTO_TLPKTDROP

// is set. This option is not handled in message mode, only in live mode.

// Dropping by packet makes sense only in case of packetwise reading,

// which isn't the case of neither stream nor message mode.

if (!m_tsbpd.isEnabled())

{

m_iDropOff = 0;

return;

}

// Do not touch m_iDropOff if it's still beside the contiguous

// region. DO NOT SEARCH for m_iDropOff if m_iEndOff is max

// because this means that the whole buffer is contiguous.

// That would simply find nothing and only uselessly burden the

// performance by searching for a not present empty cell.

// Also check if the current drop position is a readable packet.

// If not, start over.

CPos drop_pos = incPos(m_iStartPos, m_iDropOff);

if (m_iDropOff < m_iEndOff || m_entries[drop_pos].status != EntryState_Avail)

{

m_iDropOff = 0;

if (m_iEndOff < m_iMaxPosOff)

{

CPos start = incPos(m_iStartPos, m_iEndOff + 1),

end = incPos(m_iStartPos, m_iEndOff);

for (CPos i = start; i != end; i = incPos(i))

{

if (m_entries[i].status == EntryState_Avail)

{

m_iDropOff = offPos(m_iStartPos, i);

break;

}

}

// Must be found somewhere, worst case at the position

// of m_iMaxPosOff-1. If no finding loop caught it somehow,

// it will remain at 0. The case when you have empty packets

// in the busy range is only with message mode after reading

// packets out-of-order, but this doesn't use tsbpd mode.

SRT_ASSERT(m_iDropOff != 0);

}

}

}

/// Request to remove from the receiver buffer

/// all packets with earlier sequence than @a seqno.

/// (Meaning, the packet with given sequence shall

/// be the first packet in the buffer after the operation).

std::pair<int, int> CRcvBuffer::dropUpTo(int32_t seqno)

{

IF_RCVBUF_DEBUG(ScopedLog scoped_log);

IF_RCVBUF_DEBUG(scoped_log.ss << "CRcvBuffer::dropUpTo: seqno " << seqno << " m_iStartSeqNo " << m_iStartSeqNo);

COff len = COff(SeqNo(seqno) - m_iStartSeqNo);

if (len <= 0)

{

IF_RCVBUF_DEBUG(scoped_log.ss << ". Nothing to drop.");

return std::make_pair(0, 0);

}

m_iMaxPosOff = decOff(m_iMaxPosOff, len);

m_iEndOff = decOff(m_iEndOff, len);

m_iDropOff = decOff(m_iDropOff, len);

int iNumDropped = 0; // Number of dropped packets that were missing.

int iNumDiscarded = 0; // The number of dropped packets that existed in the buffer.

while (len > 0)

{

// Note! Dropping a EntryState_Read must not be counted as a drop because it was read.

// Note! Dropping a EntryState_Drop must not be counted as a drop because it was already dropped and counted earlier.

if (m_entries[m_iStartPos].status == EntryState_Avail)

++iNumDiscarded;

else if (m_entries[m_iStartPos].status == EntryState_Empty)

++iNumDropped;

dropUnitInPos(m_iStartPos);

m_entries[m_iStartPos].status = EntryState_Empty;

SRT_ASSERT(m_entries[m_iStartPos].pUnit == NULL && m_entries[m_iStartPos].status == EntryState_Empty);

m_iStartPos = incPos(m_iStartPos);

--len;

}

// Update positions

m_iStartSeqNo.set(seqno);

// Move forward if there are "read/drop" entries.

// (This call MAY shift m_iStartSeqNo further.)

releaseNextFillerEntries();

updateGapInfo();

// If the nonread position is now behind the starting position, set it to the starting position and update.

// Preceding packets were likely missing, and the non read position can probably be moved further now.

if (!isInUsedRange(m_iFirstNonreadPos))

{

m_iFirstNonreadPos = m_iStartPos;

updateNonreadPos();

}

if (!m_tsbpd.isEnabled() && m_bMessageAPI)

updateFirstReadableNonOrder();

IF_HEAVY_LOGGING(debugShowState(fmtcat("drop %", seqno).c_str()));

return std::make_pair(iNumDropped, iNumDiscarded);

}

int CRcvBuffer::dropAll()

{

if (empty())

return 0;

const int32_t end_seqno = CSeqNo::incseq(m_iStartSeqNo.val(), m_iMaxPosOff);

const std::pair<int, int> numDropped = dropUpTo(end_seqno);

return numDropped.first + numDropped.second;

}

int CRcvBuffer::dropMessage(int32_t seqnolo, int32_t seqnohi, int32_t msgno, DropActionIfExists actionOnExisting)

{

IF_RCVBUF_DEBUG(ScopedLog scoped_log);

IF_RCVBUF_DEBUG(scoped_log.ss << "CRcvBuffer::dropMessage(): %(" << seqnolo << " - " << seqnohi << ")"

<< " #" << msgno << " actionOnExisting=" << actionOnExisting << " m_iStartSeqNo=%"

<< m_iStartSeqNo);

if (msgno < 0) // Note that only SRT_MSGNO_CONTROL is allowed in the protocol.

{

HLOGC(rbuflog.Error, log << "EPE: received UMSG_DROPREQ with msgflag field set to a negative value!");

}

// Drop by packet seqno range to also wipe those packets that do not exist in the buffer.

const int offset_a = SeqNo(seqnolo) - m_iStartSeqNo;

const int offset_b = SeqNo(seqnohi) - m_iStartSeqNo;

if (offset_b < 0)

{

LOGC(rbuflog.Debug, log << "CRcvBuffer.dropMessage(): nothing to drop. Requested [" << seqnolo << "; "

<< seqnohi << "]. Buffer start " << m_iStartSeqNo.val() << ".");

return 0;

}

const bool bKeepExisting = (actionOnExisting == KEEP_EXISTING);

COff minDroppedOffset (-1);

int iDropCnt = 0;

const COff start_off = COff(max(0, offset_a));

const CPos start_pos = incPos(m_iStartPos, start_off);

const COff end_off = COff(min((int) m_szSize - 1, offset_b + 1));

const CPos end_pos = incPos(m_iStartPos, end_off);

bool bDropByMsgNo = msgno > SRT_MSGNO_CONTROL; // Excluding both SRT_MSGNO_NONE (-1) and SRT_MSGNO_CONTROL (0).

for (CPos i = start_pos; i != end_pos; i = incPos(i))

{

// Check if the unit was already dropped earlier.

if (m_entries[i].status == EntryState_Drop)

continue;

if (m_entries[i].pUnit)

{

const PacketBoundary bnd = packetAt(i).getMsgBoundary();

// Don't drop messages, if all its packets are already in the buffer.

// TODO: Don't drop a several-packet message if all packets are in the buffer.

if (bKeepExisting && bnd == PB_SOLO)

{

bDropByMsgNo = false; // Solo packet, don't search for the rest of the message.

HLOGC(rbuflog.Debug,

log << "CRcvBuffer::dropMessage(): Skipped dropping an existing SOLO packet %"

<< packetAt(i).getSeqNo() << ".");

continue;

}

const int32_t msgseq = packetAt(i).getMsgSeq(m_bPeerRexmitFlag);

if (msgno > SRT_MSGNO_CONTROL && msgseq != msgno)

{

LOGC(rbuflog.Warn, log << "CRcvBuffer.dropMessage(): Packet seqno %" << packetAt(i).getSeqNo() << " has msgno " << msgseq << " differs from requested " << msgno);

}

if (bDropByMsgNo && bnd == PB_FIRST)

{

// First packet of the message is about to be dropped. That was the only reason to search for msgno.

bDropByMsgNo = false;

}

}

dropUnitInPos(i);

++iDropCnt;

m_entries[i].status = EntryState_Drop;

if (minDroppedOffset == -1)

minDroppedOffset = offPos(m_iStartPos, i);

}

if (end_off > m_iMaxPosOff)

{

HLOGC(rbuflog.Debug, log << "CRcvBuffer::dropMessage: requested to drop up to %" << seqnohi

<< " with highest in the buffer %" << CSeqNo::incseq(m_iStartSeqNo.val(), end_off)

<< " - updating the busy region");

m_iMaxPosOff = end_off;

}

if (bDropByMsgNo)

{

// If msgno is specified, potentially not the whole message was dropped using seqno range.

// The sender might have removed the first packets of the message, and thus @a seqnolo may point to a packet in the middle.

// The sender should have the last packet of the message it is requesting to be dropped.

// Therefore we don't search forward, but need to check earlier packets in the RCV buffer.

// Try to drop by the message number in case the message starts earlier than @a seqnolo.

const CPos stop_pos = decPos(m_iStartPos);

for (CPos i = start_pos; i != stop_pos; i = decPos(i))

{

// Can't drop if message number is not known.

if (!m_entries[i].pUnit) // also dropped earlier.

continue;

const PacketBoundary bnd = packetAt(i).getMsgBoundary();

const int32_t msgseq = packetAt(i).getMsgSeq(m_bPeerRexmitFlag);

if (msgseq != msgno)

break;

if (bKeepExisting && bnd == PB_SOLO)

{

LOGC(rbuflog.Debug,

log << "CRcvBuffer::dropMessage(): Skipped dropping an existing SOLO message packet %"

<< packetAt(i).getSeqNo() << ".");

break;

}

++iDropCnt;

dropUnitInPos(i);

m_entries[i].status = EntryState_Drop;

// As the search goes backward, i is always earlier than minDroppedOffset.

minDroppedOffset = offPos(m_iStartPos, i);

// Break the loop if the start of the message has been found. No need to search further.

if (bnd == PB_FIRST)

break;

}

IF_RCVBUF_DEBUG(scoped_log.ss << " iDropCnt " << iDropCnt);

}

if (iDropCnt)

{

// We don't need the drop position, if we allow to drop messages by number

// and with that value we risk that drop was pointing to a dropped packet.

// Theoretically to make it consistent we need to shift the value to the

// next found packet, but we don't need this information if we use the message

// mode (because drop-by-packet is not supported in this mode) and this

// will burden the performance for nothing.

m_iDropOff = 0;

}

// Check if units before m_iFirstNonreadPos are dropped.

const bool needUpdateNonreadPos = (minDroppedOffset != -1 && minDroppedOffset <= getRcvDataSize());

releaseNextFillerEntries();

updateGapInfo();

IF_HEAVY_LOGGING(debugShowState(

("dropmsg off %" + fmts(seqnolo) + " #" + fmts(msgno)).c_str()));

if (needUpdateNonreadPos)

{

m_iFirstNonreadPos = m_iStartPos;

updateNonreadPos();

}

if (!m_tsbpd.isEnabled() && m_bMessageAPI)

{

if (!checkFirstReadableNonOrder())

m_iFirstNonOrderMsgPos = CPos_TRAP;

updateFirstReadableNonOrder();

}

IF_HEAVY_LOGGING(debugShowState(("dropmsg off %" + fmts(seqnolo)).c_str()));

return iDropCnt;

}

bool CRcvBuffer::getContiguousEnd(int32_t& w_seq) const

{

if (m_iEndOff == 0)

{

// Initial contiguous region empty (including empty buffer).

HLOGC(rbuflog.Debug, log << "CONTIG: empty, give up base=%" << m_iStartSeqNo.val());

w_seq = m_iStartSeqNo.val();

return m_iMaxPosOff > 0;

}

w_seq = CSeqNo::incseq(m_iStartSeqNo.val(), m_iEndOff);

HLOGC(rbuflog.Debug, log << "CONTIG: endD=" << m_iEndOff

<< " maxD=" << m_iMaxPosOff

<< " base=%" << m_iStartSeqNo.val()

<< " end=%" << w_seq);

return (m_iEndOff < m_iMaxPosOff);

}

int CRcvBuffer::readMessage(char* data, size_t len, SRT_MSGCTRL* msgctrl, pair<int32_t, int32_t>* pw_seqrange)

{

const bool canReadInOrder = hasReadableInorderPkts();

if (!canReadInOrder && m_iFirstNonOrderMsgPos == CPos_TRAP)

{

LOGC(rbuflog.Warn, log << "CRcvBuffer.readMessage(): nothing to read. Ignored isRcvDataReady() result?");

return 0;

}

const CPos readPos = canReadInOrder ? m_iStartPos : m_iFirstNonOrderMsgPos;

const bool isReadingFromStart = (readPos == m_iStartPos); // Indicates if the m_iStartPos can be changed

IF_RCVBUF_DEBUG(ScopedLog scoped_log);

IF_RCVBUF_DEBUG(scoped_log.ss << "CRcvBuffer::readMessage. m_iStartSeqNo " << m_iStartSeqNo << " m_iStartPos " << m_iStartPos << " readPos " << readPos);

size_t remain = len;

char* dst = data;

int pkts_read = 0;

int bytes_extracted = 0; // The total number of bytes extracted from the buffer.

int32_t out_seqlo = SRT_SEQNO_NONE;

int32_t out_seqhi = SRT_SEQNO_NONE;

// As we have a green light for reading, it is already known that

// we're going to either remove or extract packets from the buffer,

// so drop position won't count anymore.

//

// The END position should be updated, that is:

// - remain just updated by the shifted start position if it's still ahead

// - recalculated from 0 again otherwise

m_iDropOff = 0;

int nskipped = 0;

for (CPos i = readPos;; i = incPos(i))

{

SRT_ASSERT(m_entries[i].pUnit);

if (!m_entries[i].pUnit)

{

LOGC(rbuflog.Error, log << "CRcvBuffer::readMessage(): null packet encountered.");

break;

}

const CPacket& packet = packetAt(i);

const size_t pktsize = packet.getLength();

const int32_t pktseqno = packet.getSeqNo();

if (out_seqlo == SRT_SEQNO_NONE)

out_seqlo = pktseqno;

out_seqhi = pktseqno;

// unitsize can be zero

const size_t unitsize = std::min(remain, pktsize);

memcpy(dst, packet.m_pcData, unitsize);

remain -= unitsize;

dst += unitsize;

++pkts_read;

bytes_extracted += (int) pktsize;

if (m_tsbpd.isEnabled())

updateTsbPdTimeBase(packet.getMsgTimeStamp());

if (m_numNonOrderPackets && !packet.getMsgOrderFlag())

--m_numNonOrderPackets;

const bool pbLast = packet.getMsgBoundary() & PB_LAST;

if (msgctrl && (packet.getMsgBoundary() & PB_FIRST))

{

msgctrl->msgno = packet.getMsgSeq(m_bPeerRexmitFlag);

}

if (msgctrl && pbLast)

{

msgctrl->srctime = count_microseconds(getPktTsbPdTime(packet.getMsgTimeStamp()).time_since_epoch());

}

if (msgctrl)

msgctrl->pktseq = pktseqno;

releaseUnitInPos(i);

if (isReadingFromStart)

{

m_iStartPos = incPos(i);

m_iStartSeqNo = SeqNo(pktseqno) + 1;

++nskipped;

}

else

{

// If out of order, only mark it read.

m_entries[i].status = EntryState_Read;

}

if (pbLast)

{

if (readPos == m_iFirstNonOrderMsgPos)

{

m_iFirstNonOrderMsgPos = CPos_TRAP;

m_iDropOff = 0; // always set to 0 in this mode.

}

break;

}

}

if (nskipped)

{

// This means that m_iStartPos HAS BEEN shifted by that many packets.

// Update offset variables

m_iMaxPosOff = m_iMaxPosOff - nskipped;

// This is checked as the PB_LAST flag marked packet should still

// be extracted in the existing period.

SRT_ASSERT(m_iMaxPosOff >= 0);

m_iEndOff = decOff(m_iEndOff, len);

}

countBytes(-pkts_read, -bytes_extracted);

releaseNextFillerEntries();

// This will update the end position

updateGapInfo();

if (!isInUsedRange(m_iFirstNonreadPos))

{

m_iFirstNonreadPos = m_iStartPos;

//updateNonreadPos();

}

if (!m_tsbpd.isEnabled())

// We need updateFirstReadableOutOfOrder() here even if we are reading inorder,

// in case readable inorder packets are all read out.

updateFirstReadableNonOrder();

const int bytes_read = int(dst - data);

if (bytes_read < bytes_extracted)

{

LOGC(rbuflog.Error, log << "readMessage: small dst buffer, copied only " << bytes_read << "/" << bytes_extracted << " bytes.");

}

IF_RCVBUF_DEBUG(scoped_log.ss << " pldi64 " << *reinterpret_cast<uint64_t*>(data));

if (pw_seqrange)

*pw_seqrange = make_pair(out_seqlo, out_seqhi);

IF_HEAVY_LOGGING(debugShowState("readmsg"));

return bytes_read;

}

namespace {

/// @brief Writes bytes to file stream.

/// @param data pointer to data to write.

/// @param len the number of bytes to write

/// @param dst_offset ignored

/// @param arg a void pointer to the fstream to write to.

/// @return true on success, false on failure

bool writeBytesToFile(char* data, int len, int dst_offset SRT_ATR_UNUSED, void* arg)

{

fstream* pofs = reinterpret_cast<fstream*>(arg);

pofs->write(data, len);

return !pofs->fail();

}

/// @brief Copies bytes to the destination buffer.

/// @param data pointer to data to copy.

/// @param len the number of bytes to copy

/// @param dst_offset offset in destination buffer

/// @param arg A pointer to the destination buffer

/// @return true on success, false on failure

bool copyBytesToBuf(char* data, int len, int dst_offset, void* arg)

{

char* dst = reinterpret_cast<char*>(arg) + dst_offset;

memcpy(dst, data, len);

return true;

}

}

int CRcvBuffer::readBufferTo(int len, copy_to_dst_f funcCopyToDst, void* arg)

{

CPos p = m_iStartPos;

const CPos end_pos = m_iFirstNonreadPos;

const bool bTsbPdEnabled = m_tsbpd.isEnabled();

const steady_clock::time_point now = (bTsbPdEnabled ? steady_clock::now() : steady_clock::time_point());

int rs = len;

while ((p != end_pos) && (rs > 0))

{

if (!m_entries[p].pUnit)

{

// REDUNDANT? p = incPos(p); // Return abandons the loop anyway.

LOGC(rbuflog.Error, log << "readBufferTo: IPE: NULL unit found in file transmission");

return -1;

}

const CPacket& pkt = packetAt(p);

if (bTsbPdEnabled)

{

const steady_clock::time_point tsPlay = getPktTsbPdTime(pkt.getMsgTimeStamp());

HLOGC(rbuflog.Debug,

log << "readBuffer: check if time to play:"

<< " NOW=" << FormatTime(now)

<< " PKT TS=" << FormatTime(tsPlay));

if ((tsPlay > now))

break; // too early for this unit, return whatever was copied

}

const int pktlen = (int)pkt.getLength();

const int remain_pktlen = pktlen - m_iNotch;

const int unitsize = std::min(remain_pktlen, rs);

if (!funcCopyToDst(pkt.m_pcData + m_iNotch, unitsize, len - rs, arg))

break;

if (rs >= remain_pktlen)

{

releaseUnitInPos(p);

p = incPos(p);

m_iNotch = 0;

m_iStartPos = p;

--m_iMaxPosOff;

SRT_ASSERT(m_iMaxPosOff >= 0);

m_iEndOff = decOff(m_iEndOff, 1);

m_iDropOff = decOff(m_iDropOff, 1);

m_iStartSeqNo = m_iStartSeqNo.inc();

}

else

m_iNotch += rs;

rs -= unitsize;

}

const int iBytesRead = len - rs;

/* we removed acked bytes form receive buffer */

countBytes(-1, -iBytesRead);

// Update positions

// Set nonread position to the starting position before updating,

// because start position was increased, and preceding packets are invalid.

if (!isInUsedRange( m_iFirstNonreadPos))

{

m_iFirstNonreadPos = m_iStartPos;

}

if (iBytesRead == 0)

{

LOGC(rbuflog.Error, log << "readBufferTo: 0 bytes read. m_iStartPos=" << m_iStartPos

<< ", m_iFirstNonreadPos=" << m_iFirstNonreadPos);

}

IF_HEAVY_LOGGING(debugShowState("readbuf"));

return iBytesRead;

}

int CRcvBuffer::readBuffer(char* dst, int len)

{

return readBufferTo(len, copyBytesToBuf, reinterpret_cast<void*>(dst));

}

int CRcvBuffer::readBufferToFile(fstream& ofs, int len)

{

return readBufferTo(len, writeBytesToFile, reinterpret_cast<void*>(&ofs));

}

bool CRcvBuffer::hasAvailablePackets() const

{

return hasReadableInorderPkts() || (m_numNonOrderPackets > 0 && m_iFirstNonOrderMsgPos != CPos_TRAP);

}

int CRcvBuffer::getRcvDataSize() const

{

return offPos(m_iStartPos, m_iFirstNonreadPos);

}

int CRcvBuffer::getTimespan_ms() const

{

if (!m_tsbpd.isEnabled())

return 0;

if (m_iMaxPosOff == 0)

return 0;

CPos lastpos = incPos(m_iStartPos, m_iMaxPosOff - 1);

// Normally the last position should always be non empty

// if TSBPD is enabled (reading out of order is not allowed).

// However if decryption of the last packet fails, it may be dropped

// from the buffer (AES-GCM), and the position will be empty.

SRT_ASSERT(m_entries[lastpos].pUnit != NULL || m_entries[lastpos].status == EntryState_Drop);

while (m_entries[lastpos].pUnit == NULL && lastpos != m_iStartPos)

{

lastpos = decPos(lastpos);

}

if (m_entries[lastpos].pUnit == NULL)

return 0;

CPos startpos = m_iStartPos;

while (m_entries[startpos].pUnit == NULL && startpos != lastpos)

{

startpos = incPos(startpos);