/
buffer_rcv.cpp
944 lines (783 loc) · 28.2 KB
/
buffer_rcv.cpp
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
#if ENABLE_NEW_RCVBUFFER
#include <cmath>
#include "buffer_rcv.h"
#include "logging.h"
using namespace std;
using namespace srt::sync;
using namespace srt_logging;
namespace srt_logging
{
extern Logger brlog;
}
#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
// Check if iFirstNonreadPos is in range [iStartPos, (iStartPos + iMaxPosInc) % iSize].
// The right edge is included because we expect iFirstNonreadPos to be
// right after the last valid packet position if all packets are available.
bool isInRange(int iStartPos, int iMaxPosInc, size_t iSize, int iFirstNonreadPos)
{
if (iFirstNonreadPos == iStartPos)
return true;
const int iLastPos = (iStartPos + iMaxPosInc) % iSize;
const bool isOverrun = iLastPos < iStartPos;
if (isOverrun)
return iFirstNonreadPos > iStartPos || iFirstNonreadPos <= iLastPos;
return iFirstNonreadPos > iStartPos && iFirstNonreadPos <= iLastPos;
}
}
/*
* RcvBufferNew (circular buffer):
*
* |<------------------- m_iSize ----------------------------->|
* | |<----------- m_iMaxPosInc ------------>| |
* | | | |
* +---+---+---+---+---+---+---+---+---+---+---+---+---+ +---+
* | 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_iMaxPosInc: none? (modified on add and ack
*/
CRcvBufferNew::CRcvBufferNew(int initSeqNo, size_t size, CUnitQueue* unitqueue, bool peerRexmit, bool bMessageAPI)
: m_entries(size)
, m_szSize(size) // TODO: maybe just use m_entries.size()
, m_pUnitQueue(unitqueue)
, m_iStartSeqNo(initSeqNo)
, m_iStartPos(0)
, m_iFirstNonreadPos(0)
, m_iMaxPosInc(0)
, m_iNotch(0)
, m_numOutOfOrderPackets(0)
, m_iFirstReadableOutOfOrder(-1)
, m_bPeerRexmitFlag(peerRexmit)
, m_bMessageAPI(bMessageAPI)
, m_iBytesCount(0)
, m_iPktsCount(0)
, m_uAvgPayloadSz(SRT_LIVE_DEF_PLSIZE)
{
SRT_ASSERT(size < INT_MAX); // All position pointers are integers
}
CRcvBufferNew::~CRcvBufferNew()
{
for (size_t i = 0; i < m_szSize; ++i)
{
CUnit* unit = m_entries[i].pUnit;
if (unit != NULL)
{
m_pUnitQueue->makeUnitFree(unit);
m_entries[i].pUnit = NULL;
}
}
}
int CRcvBufferNew::insert(CUnit* unit)
{
SRT_ASSERT(unit != NULL);
const int32_t seqno = unit->m_Packet.getSeqNo();
const int offset = CSeqNo::seqoff(m_iStartSeqNo, seqno);
IF_RCVBUF_DEBUG(ScopedLog scoped_log);
IF_RCVBUF_DEBUG(scoped_log.ss << "CRcvBufferNew::insert: seqno " << seqno << " m_iStartSeqNo " << m_iStartSeqNo << " offset " << offset);
if (offset < 0)
{
IF_RCVBUF_DEBUG(scoped_log.ss << " returns -2");
return -2;
}
if (offset >= (int)capacity())
{
IF_RCVBUF_DEBUG(scoped_log.ss << " returns -3");
return -3;
}
// 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 int pos = (m_iStartPos + offset) % m_szSize;
if (offset >= m_iMaxPosInc)
m_iMaxPosInc = offset + 1;
// Packet already exists
SRT_ASSERT(pos >= 0 && pos < m_szSize);
if (m_entries[pos].status != EntryState_Empty)
{
IF_RCVBUF_DEBUG(scoped_log.ss << " returns -1");
return -1;
}
SRT_ASSERT(m_entries[pos].pUnit == NULL);
m_pUnitQueue->makeUnitGood(unit);
m_entries[pos].pUnit = unit;
m_entries[pos].status = EntryState_Avail;
countBytes(1, (int)unit->m_Packet.getLength());
// 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_numOutOfOrderPackets;
onInsertNotInOrderPacket(pos);
}
updateNonreadPos();
IF_RCVBUF_DEBUG(scoped_log.ss << " returns 0 (OK)");
return 0;
}
int CRcvBufferNew::dropUpTo(int32_t seqno)
{
// Can drop only when nothing to read, and
// first unacknowledged packet is missing.
SRT_ASSERT(m_iStartPos == m_iFirstNonreadPos);
IF_RCVBUF_DEBUG(ScopedLog scoped_log);
IF_RCVBUF_DEBUG(scoped_log.ss << "CRcvBufferNew::dropUpTo: seqno " << seqno << " m_iStartSeqNo " << m_iStartSeqNo);
int len = CSeqNo::seqoff(m_iStartSeqNo, seqno);
SRT_ASSERT(len > 0);
if (len <= 0)
{
IF_RCVBUF_DEBUG(scoped_log.ss << ". Nothing to drop.");
return 0;
}
m_iMaxPosInc -= len;
if (m_iMaxPosInc < 0)
m_iMaxPosInc = 0;
// Check that all packets being dropped are missing.
const int iDropCnt = len;
while (len > 0)
{
if (m_entries[m_iStartPos].pUnit != NULL)
{
releaseUnitInPos(m_iStartPos);
}
if (m_entries[m_iStartPos].status != EntryState_Empty)
{
SRT_ASSERT(m_entries[m_iStartPos].status == EntryState_Drop || m_entries[m_iStartPos].status == EntryState_Read);
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 = seqno;
// Move forward if there are "read" entries.
releaseNextFillerEntries();
// Set nonread position to the starting position before updating,
// because start position was increased, and preceeding packets are invalid.
m_iFirstNonreadPos = m_iStartPos;
updateNonreadPos();
return iDropCnt;
}
void CRcvBufferNew::dropMessage(int32_t seqnolo, int32_t seqnohi, int32_t msgno)
{
IF_RCVBUF_DEBUG(ScopedLog scoped_log);
IF_RCVBUF_DEBUG(scoped_log.ss << "CRcvBufferNew::dropMessage: seqnolo " << seqnolo << " seqnohi " << seqnohi << " m_iStartSeqNo " << m_iStartSeqNo);
// TODO: count bytes as removed?
const int end_pos = incPos(m_iStartPos, m_iMaxPosInc);
if (msgno != 0)
{
for (int i = m_iStartPos; i != end_pos; i = incPos(i))
{
// TODO: Maybe check status?
if (!m_entries[i].pUnit)
continue;
const int32_t msgseq = m_entries[i].pUnit->m_Packet.getMsgSeq(m_bPeerRexmitFlag);
if (msgseq == msgno)
{
releaseUnitInPos(i);
m_entries[i].status = EntryState_Drop;
}
}
return;
}
// Drop by packet seqno range.
const int offset_a = CSeqNo::seqoff(m_iStartSeqNo, seqnolo);
const int offset_b = CSeqNo::seqoff(m_iStartSeqNo, seqnohi);
if (offset_b < 0)
{
LOGC(rbuflog.Warn, log << "CRcvBufferNew.dropMessage(): nothing to drop. Requested [" << seqnolo << "; "
<< seqnohi << "]. Buffer start " << m_iStartSeqNo << ".");
return;
}
const int start_off = max(0, offset_a);
const int last_pos = incPos(m_iStartPos, offset_b);
for (int i = incPos(m_iStartPos, start_off); i != end_pos && i != last_pos; i = incPos(i))
{
if (m_entries[i].pUnit)
{
releaseUnitInPos(i);
}
m_entries[i].status = EntryState_Drop;
}
LOGC(rbuflog.Debug, log << "CRcvBufferNew.dropMessage(): [" << seqnolo << "; "
<< seqnohi << "].");
}
int CRcvBufferNew::readMessage(char* data, size_t len, SRT_MSGCTRL* msgctrl)
{
const bool canReadInOrder = hasReadableInorderPkts();
if (!canReadInOrder && m_iFirstReadableOutOfOrder < 0)
{
LOGC(rbuflog.Warn, log << "CRcvBufferNew.readMessage(): nothing to read. Ignored isRcvDataReady() result?");
return 0;
}
IF_RCVBUF_DEBUG(ScopedLog scoped_log);
IF_RCVBUF_DEBUG(scoped_log.ss << "CRcvBufferNew::readMessage. m_iStartSeqNo " << m_iStartSeqNo);
const int readPos = canReadInOrder ? m_iStartPos : m_iFirstReadableOutOfOrder;
// Remember if we actually read out of order packet.
const bool readingOutOfOrderPacket = !canReadInOrder || m_iStartPos == m_iFirstReadableOutOfOrder;
size_t remain = len;
char* dst = data;
int pkts_read = 0;
int bytes_extracted = 0; // The total number of bytes extracted from the buffer.
const bool updateStartPos = (readPos == m_iStartPos); // Indicates if the m_iStartPos can be changed
for (int i = readPos;; i = incPos(i))
{
SRT_ASSERT(m_entries[i].pUnit);
if (!m_entries[i].pUnit)
{
LOGC(rbuflog.Error, log << "CRcvBufferNew::readMessage(): null packet encountered.");
break;
}
const CPacket& packet = m_entries[i].pUnit->m_Packet;
const size_t pktsize = packet.getLength();
const int32_t pktseqno = packet.getSeqNo();
// 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_numOutOfOrderPackets && !packet.getMsgOrderFlag())
--m_numOutOfOrderPackets;
const bool pbLast = packet.getMsgBoundary() & PB_LAST;
if (msgctrl && (packet.getMsgBoundary() & PB_FIRST))
{
msgctrl->pktseq = pktseqno;
msgctrl->msgno = packet.getMsgSeq(m_bPeerRexmitFlag);
}
if (msgctrl && pbLast)
{
msgctrl->srctime = count_microseconds(getPktTsbPdTime(packet.getMsgTimeStamp()).time_since_epoch());
}
releaseUnitInPos(i);
if (updateStartPos)
{
m_iStartPos = incPos(i);
--m_iMaxPosInc;
SRT_ASSERT(m_iMaxPosInc >= 0);
m_iStartSeqNo = CSeqNo::incseq(pktseqno);
}
else
{
// If out of order, only mark it read.
m_entries[i].status = EntryState_Read;
}
if (pbLast)
{
if (readPos == m_iFirstReadableOutOfOrder)
m_iFirstReadableOutOfOrder = -1;
break;
}
}
countBytes(-pkts_read, -bytes_extracted);
if (!m_tsbpd.isEnabled() && readingOutOfOrderPacket)
updateFirstReadableOutOfOrder();
releaseNextFillerEntries();
if (!isInRange(m_iStartPos, m_iMaxPosInc, m_szSize, m_iFirstNonreadPos))
{
m_iFirstNonreadPos = m_iStartPos;
//updateNonreadPos();
}
const int bytes_read = dst - data;
if (bytes_read < bytes_extracted)
{
LOGC(rbuflog.Error, log << "readMessage: small dst buffer, copied only " << bytes_read << "/" << bytes_extracted << " bytes.");
}
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 CRcvBufferNew::readBufferTo(int len, copy_to_dst_f funcCopyToDst, void* arg)
{
int p = m_iStartPos;
const int 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)
{
p = incPos(p);
LOGC(rbuflog.Error, log << "readBufferTo: IPE: NULL unit found in file transmission");
return -1;
}
const srt::CPacket& pkt = m_entries[p].pUnit->m_Packet;
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_iMaxPosInc;
SRT_ASSERT(m_iMaxPosInc >= 0);
m_iStartSeqNo = CSeqNo::incseq(m_iStartSeqNo);
}
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 preceeding packets are invalid.
if (!isInRange(m_iStartPos, m_iMaxPosInc, m_szSize, 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);
}
return iBytesRead;
}
int CRcvBufferNew::readBuffer(char* dst, int len)
{
return readBufferTo(len, copyBytesToBuf, reinterpret_cast<void*>(dst));
}
int CRcvBufferNew::readBufferToFile(fstream& ofs, int len)
{
return readBufferTo(len, writeBytesToFile, reinterpret_cast<void*>(&ofs));
}
bool CRcvBufferNew::hasAvailablePackets() const
{
return hasReadableInorderPkts() || (m_numOutOfOrderPackets > 0 && m_iFirstReadableOutOfOrder != -1);
}
int CRcvBufferNew::getRcvDataSize() const
{
if (m_iFirstNonreadPos >= m_iStartPos)
return m_iFirstNonreadPos - m_iStartPos;
return m_szSize + m_iFirstNonreadPos - m_iStartPos;
}
int CRcvBufferNew::getTimespan_ms() const
{
if (!m_tsbpd.isEnabled())
return 0;
if (m_iMaxPosInc == 0)
return 0;
const int lastpos = incPos(m_iStartPos, m_iMaxPosInc - 1);
int startpos = m_iStartPos;
while (m_entries[startpos].pUnit == NULL)
{
if (startpos == lastpos)
break;
startpos = incPos(startpos);
}
if (m_entries[startpos].pUnit == NULL)
return 0;
// Should not happen
SRT_ASSERT(m_entries[lastpos].pUnit != NULL);
if (m_entries[lastpos].pUnit == NULL)
return 0;
const steady_clock::time_point startstamp =
getPktTsbPdTime(m_entries[startpos].pUnit->m_Packet.getMsgTimeStamp());
const steady_clock::time_point endstamp = getPktTsbPdTime(m_entries[lastpos].pUnit->m_Packet.getMsgTimeStamp());
if (endstamp < startstamp)
return 0;
// One millisecond is added as a duration of a packet in the buffer.
// If there is only one packet in the buffer, one millisecond is returned.
return count_milliseconds(endstamp - startstamp) + 1;
}
int CRcvBufferNew::getRcvDataSize(int& bytes, int& timespan) const
{
ScopedLock lck(m_BytesCountLock);
bytes = m_iBytesCount;
timespan = getTimespan_ms();
return m_iPktsCount;
}
CRcvBufferNew::PacketInfo CRcvBufferNew::getFirstValidPacketInfo() const
{
const int end_pos = incPos(m_iStartPos, m_iMaxPosInc);
for (int i = m_iStartPos; i != end_pos; i = incPos(i))
{
// TODO: Maybe check status?
if (!m_entries[i].pUnit)
continue;
const CPacket& packet = m_entries[i].pUnit->m_Packet;
const PacketInfo info = { packet.getSeqNo(), i != m_iStartPos, getPktTsbPdTime(packet.getMsgTimeStamp()) };
return info;
}
const PacketInfo info = { -1, false, time_point() };
return info;
}
std::pair<int, int> CRcvBufferNew::getAvailablePacketsRange() const
{
const int seqno_last = CSeqNo::incseq(m_iStartSeqNo, countReadable());
return std::pair<int, int>(m_iStartSeqNo, seqno_last);
}
size_t CRcvBufferNew::countReadable() const
{
if (m_iFirstNonreadPos >= m_iStartPos)
return m_iFirstNonreadPos - m_iStartPos;
return m_szSize + m_iFirstNonreadPos - m_iStartPos;
}
bool CRcvBufferNew::isRcvDataReady(time_point time_now) const
{
const bool haveInorderPackets = hasReadableInorderPkts();
if (!m_tsbpd.isEnabled())
{
if (haveInorderPackets)
return true;
SRT_ASSERT((!m_bMessageAPI && m_numOutOfOrderPackets == 0) || m_bMessageAPI);
return (m_numOutOfOrderPackets > 0 && m_iFirstReadableOutOfOrder != -1);
}
if (!haveInorderPackets)
return false;
const PacketInfo info = getFirstValidPacketInfo();
return info.tsbpd_time <= time_now;
}
void CRcvBufferNew::countBytes(int pkts, int bytes)
{
ScopedLock lock(m_BytesCountLock);
m_iBytesCount += bytes; // added or removed bytes from rcv buffer
m_iPktsCount += pkts;
if (bytes > 0) // Assuming one pkt when adding bytes
m_uAvgPayloadSz = avg_iir<100>(m_uAvgPayloadSz, (unsigned) bytes);
}
void CRcvBufferNew::releaseUnitInPos(int pos)
{
CUnit* tmp = m_entries[pos].pUnit;
m_entries[pos] = Entry(); // pUnit = NULL; status = Empty
if (tmp != NULL)
m_pUnitQueue->makeUnitFree(tmp);
}
void CRcvBufferNew::releaseNextFillerEntries()
{
int pos = m_iStartPos;
while (m_entries[pos].status == EntryState_Read || m_entries[pos].status == EntryState_Drop)
{
m_iStartSeqNo = CSeqNo::incseq(m_iStartSeqNo);
releaseUnitInPos(pos);
pos = incPos(pos);
m_iStartPos = pos;
}
}
// TODO: Is this function complete? There are some comments left inside.
void CRcvBufferNew::updateNonreadPos()
{
if (m_iMaxPosInc == 0)
return;
const int end_pos = incPos(m_iStartPos, m_iMaxPosInc); // The empty position right after the last valid entry.
int pos = m_iFirstNonreadPos;
while (m_entries[pos].pUnit && m_entries[pos].status == EntryState_Avail)
{
if (m_bMessageAPI && (m_entries[pos].pUnit->m_Packet.getMsgBoundary() & PB_FIRST) == 0)
break;
for (int i = pos; i != end_pos; i = incPos(i))
{
if (!m_entries[i].pUnit || m_entries[pos].status != EntryState_Avail)
{
break;
}
// Check PB_LAST only in message mode.
if (!m_bMessageAPI || m_entries[i].pUnit->m_Packet.getMsgBoundary() & PB_LAST)
{
m_iFirstNonreadPos = incPos(i);
break;
}
}
if (pos == m_iFirstNonreadPos || !m_entries[m_iFirstNonreadPos].pUnit)
break;
pos = m_iFirstNonreadPos;
}
}
int CRcvBufferNew::findLastMessagePkt()
{
for (int i = m_iStartPos; i != m_iFirstNonreadPos; i = incPos(i))
{
SRT_ASSERT(m_entries[i].pUnit);
if (m_entries[i].pUnit->m_Packet.getMsgBoundary() & PB_LAST)
{
return i;
}
}
return -1;
}
void CRcvBufferNew::onInsertNotInOrderPacket(int insertPos)
{
if (m_numOutOfOrderPackets == 0)
return;
// If the following condition is true, there is already a packet,
// that can be read out of order. We don't need to search for
// another one. The search should be done when that packet is read out from the buffer.
//
// There might happen that the packet being added precedes the previously found one.
// However, it is allowed to re bead out of order, so no need to update the position.
if (m_iFirstReadableOutOfOrder >= 0)
return;
// Just a sanity check. This function is called when a new packet is added.
// So the should be unacknowledged packets.
SRT_ASSERT(m_iMaxPosInc > 0);
SRT_ASSERT(m_entries[insertPos].pUnit);
const CPacket& pkt = m_entries[insertPos].pUnit->m_Packet;
const PacketBoundary boundary = pkt.getMsgBoundary();
//if ((boundary & PB_FIRST) && (boundary & PB_LAST))
//{
// // This packet can be read out of order
// m_iFirstReadableOutOfOrder = insertPos;
// return;
//}
const int msgNo = pkt.getMsgSeq(m_bPeerRexmitFlag);
// First check last packet, because it is expected to be received last.
const bool hasLast = (boundary & PB_LAST) || (-1 < scanNotInOrderMessageRight(insertPos, msgNo));
if (!hasLast)
return;
const int firstPktPos = (boundary & PB_FIRST)
? insertPos
: scanNotInOrderMessageLeft(insertPos, msgNo);
if (firstPktPos < 0)
return;
m_iFirstReadableOutOfOrder = firstPktPos;
return;
}
void CRcvBufferNew::updateFirstReadableOutOfOrder()
{
if (hasReadableInorderPkts() || m_numOutOfOrderPackets <= 0 || m_iFirstReadableOutOfOrder >= 0)
return;
if (m_iMaxPosInc == 0)
return;
// TODO: unused variable outOfOrderPktsRemain?
int outOfOrderPktsRemain = m_numOutOfOrderPackets;
// Search further packets to the right.
// First check if there are packets to the right.
const int lastPos = (m_iStartPos + m_iMaxPosInc - 1) % m_szSize;
int posFirst = -1;
int posLast = -1;
int msgNo = -1;
for (int pos = m_iStartPos; outOfOrderPktsRemain; pos = incPos(pos))
{
if (!m_entries[pos].pUnit)
{
posFirst = posLast = msgNo = -1;
continue;
}
const CPacket& pkt = m_entries[pos].pUnit->m_Packet;
if (pkt.getMsgOrderFlag()) // Skip in order packet
{
posFirst = posLast = msgNo = -1;
continue;
}
--outOfOrderPktsRemain;
const PacketBoundary boundary = pkt.getMsgBoundary();
if (boundary & PB_FIRST)
{
posFirst = pos;
msgNo = pkt.getMsgSeq(m_bPeerRexmitFlag);
}
if (pkt.getMsgSeq(m_bPeerRexmitFlag) != msgNo)
{
posFirst = posLast = msgNo = -1;
continue;
}
if (boundary & PB_LAST)
{
m_iFirstReadableOutOfOrder = posFirst;
return;
}
if (pos == lastPos)
break;
}
return;
}
int CRcvBufferNew::scanNotInOrderMessageRight(const int startPos, int msgNo) const
{
// Search further packets to the right.
// First check if there are packets to the right.
const int lastPos = (m_iStartPos + m_iMaxPosInc - 1) % m_szSize;
if (startPos == lastPos)
return -1;
int pos = startPos;
do
{
pos = incPos(pos);
if (!m_entries[pos].pUnit)
break;
const CPacket& pkt = m_entries[pos].pUnit->m_Packet;
if (pkt.getMsgSeq(m_bPeerRexmitFlag) != msgNo)
{
LOGC(rbuflog.Error, log << "Missing PB_LAST packet for msgNo " << msgNo);
return -1;
}
const PacketBoundary boundary = pkt.getMsgBoundary();
if (boundary & PB_LAST)
return pos;
} while (pos != lastPos);
return -1;
}
int CRcvBufferNew::scanNotInOrderMessageLeft(const int startPos, int msgNo) const
{
// Search preceeding packets to the left.
// First check if there are packets to the left.
if (startPos == m_iStartPos)
return -1;
int pos = startPos;
do
{
pos = decPos(pos);
if (!m_entries[pos].pUnit)
return -1;
const CPacket& pkt = m_entries[pos].pUnit->m_Packet;
if (pkt.getMsgSeq(m_bPeerRexmitFlag) != msgNo)
{
LOGC(rbuflog.Error, log << "Missing PB_FIRST packet for msgNo " << msgNo);
return -1;
}
const PacketBoundary boundary = pkt.getMsgBoundary();
if (boundary & PB_FIRST)
return pos;
} while (pos != m_iStartPos);
return -1;
}
bool CRcvBufferNew::addRcvTsbPdDriftSample(uint32_t usTimestamp, int usRTTSample)
{
return m_tsbpd.addDriftSample(usTimestamp, usRTTSample);
}
void CRcvBufferNew::setTsbPdMode(const steady_clock::time_point& timebase, bool wrap, duration delay)
{
m_tsbpd.setTsbPdMode(timebase, wrap, delay);
}
void CRcvBufferNew::applyGroupTime(const steady_clock::time_point& timebase,
bool wrp,
uint32_t delay,
const steady_clock::duration& udrift)
{
m_tsbpd.applyGroupTime(timebase, wrp, delay, udrift);
}
void CRcvBufferNew::applyGroupDrift(const steady_clock::time_point& timebase,
bool wrp,
const steady_clock::duration& udrift)
{
m_tsbpd.applyGroupDrift(timebase, wrp, udrift);
}
CRcvBufferNew::time_point CRcvBufferNew::getTsbPdTimeBase(uint32_t usPktTimestamp) const
{
return m_tsbpd.getTsbPdTimeBase(usPktTimestamp);
}
void CRcvBufferNew::updateTsbPdTimeBase(uint32_t usPktTimestamp)
{
m_tsbpd.updateTsbPdTimeBase(usPktTimestamp);
}
string CRcvBufferNew::strFullnessState(int iFirstUnackSeqNo, const time_point& tsNow) const
{
stringstream ss;
ss << "Space avail " << getAvailSize(iFirstUnackSeqNo) << "/" << m_szSize;
ss << " pkts. ";
if (m_tsbpd.isEnabled() && m_iMaxPosInc > 0)
{
ss << " (TSBPD ready in ";
if (m_entries[m_iStartPos].pUnit)
{
const uint32_t usPktTimestamp = m_entries[m_iStartPos].pUnit->m_Packet.getMsgTimeStamp();
ss << count_milliseconds(m_tsbpd.getPktTsbPdTime(usPktTimestamp) - tsNow);
}
else
{
ss << "n/a";
}
const int iLastPos = incPos(m_iStartPos, m_iMaxPosInc - 1);
if (m_entries[iLastPos].pUnit)
{
ss << ":";
const uint32_t usPktTimestamp = m_entries[m_iStartPos].pUnit->m_Packet.getMsgTimeStamp();
ss << count_milliseconds(m_tsbpd.getPktTsbPdTime(usPktTimestamp) - tsNow);
ss << " ms";
}
else
{
ss << ":n/a ms";
}
ss << ". ";
}
ss << SRT_SYNC_CLOCK_STR " drift " << getDrift() / 1000 << " ms.";
return ss.str();
}
CRcvBufferNew::time_point CRcvBufferNew::getPktTsbPdTime(uint32_t usPktTimestamp) const
{
return m_tsbpd.getPktTsbPdTime(usPktTimestamp);
}
/* Return moving average of acked data pkts, bytes, and timespan (ms) of the receive buffer */
int CRcvBufferNew::getRcvAvgDataSize(int& bytes, int& timespan)
{
// Average number of packets and timespan could be small,
// so rounding is beneficial, while for the number of
// bytes in the buffer is a higher value, so rounding can be omitted,
// but probably better to round all three values.
timespan = static_cast<int>(round((m_mavg.timespan_ms())));
bytes = static_cast<int>(round((m_mavg.bytes())));
return static_cast<int>(round(m_mavg.pkts()));
}
/* Update moving average of acked data pkts, bytes, and timespan (ms) of the receive buffer */
void CRcvBufferNew::updRcvAvgDataSize(const steady_clock::time_point& now)
{
if (!m_mavg.isTimeToUpdate(now))
return;
int bytes = 0;
int timespan_ms = 0;
const int pkts = getRcvDataSize(bytes, timespan_ms);
m_mavg.update(now, pkts, bytes, timespan_ms);
}
} // namespace srt
#endif // ENABLE_NEW_RCVBUFFER