payload_queue.go 3.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182
  1. // SPDX-FileCopyrightText: 2023 The Pion community <https://pion.ly>
  2. // SPDX-License-Identifier: MIT
  3. package sctp
  4. import (
  5. "fmt"
  6. "sort"
  7. )
  8. type payloadQueue struct {
  9. chunkMap map[uint32]*chunkPayloadData
  10. sorted []uint32
  11. dupTSN []uint32
  12. nBytes int
  13. }
  14. func newPayloadQueue() *payloadQueue {
  15. return &payloadQueue{chunkMap: map[uint32]*chunkPayloadData{}}
  16. }
  17. func (q *payloadQueue) updateSortedKeys() {
  18. if q.sorted != nil {
  19. return
  20. }
  21. q.sorted = make([]uint32, len(q.chunkMap))
  22. i := 0
  23. for k := range q.chunkMap {
  24. q.sorted[i] = k
  25. i++
  26. }
  27. sort.Slice(q.sorted, func(i, j int) bool {
  28. return sna32LT(q.sorted[i], q.sorted[j])
  29. })
  30. }
  31. func (q *payloadQueue) canPush(p *chunkPayloadData, cumulativeTSN uint32, maxTSNOffset uint32) bool {
  32. _, ok := q.chunkMap[p.tsn]
  33. if ok || sna32LTE(p.tsn, cumulativeTSN) || sna32GTE(p.tsn, cumulativeTSN+maxTSNOffset) {
  34. return false
  35. }
  36. return true
  37. }
  38. func (q *payloadQueue) pushNoCheck(p *chunkPayloadData) {
  39. q.chunkMap[p.tsn] = p
  40. q.nBytes += len(p.userData)
  41. q.sorted = nil
  42. }
  43. // push pushes a payload data. If the payload data is already in our queue or
  44. // older than our cumulativeTSN marker, it will be recored as duplications,
  45. // which can later be retrieved using popDuplicates.
  46. func (q *payloadQueue) push(p *chunkPayloadData, cumulativeTSN uint32) bool {
  47. _, ok := q.chunkMap[p.tsn]
  48. if ok || sna32LTE(p.tsn, cumulativeTSN) {
  49. // Found the packet, log in dups
  50. q.dupTSN = append(q.dupTSN, p.tsn)
  51. return false
  52. }
  53. q.chunkMap[p.tsn] = p
  54. q.nBytes += len(p.userData)
  55. q.sorted = nil
  56. return true
  57. }
  58. // pop pops only if the oldest chunk's TSN matches the given TSN.
  59. func (q *payloadQueue) pop(tsn uint32) (*chunkPayloadData, bool) {
  60. q.updateSortedKeys()
  61. if len(q.chunkMap) > 0 && tsn == q.sorted[0] {
  62. q.sorted = q.sorted[1:]
  63. if c, ok := q.chunkMap[tsn]; ok {
  64. delete(q.chunkMap, tsn)
  65. q.nBytes -= len(c.userData)
  66. return c, true
  67. }
  68. }
  69. return nil, false
  70. }
  71. // get returns reference to chunkPayloadData with the given TSN value.
  72. func (q *payloadQueue) get(tsn uint32) (*chunkPayloadData, bool) {
  73. c, ok := q.chunkMap[tsn]
  74. return c, ok
  75. }
  76. // popDuplicates returns an array of TSN values that were found duplicate.
  77. func (q *payloadQueue) popDuplicates() []uint32 {
  78. dups := q.dupTSN
  79. q.dupTSN = []uint32{}
  80. return dups
  81. }
  82. func (q *payloadQueue) getGapAckBlocks(cumulativeTSN uint32) (gapAckBlocks []gapAckBlock) {
  83. var b gapAckBlock
  84. if len(q.chunkMap) == 0 {
  85. return []gapAckBlock{}
  86. }
  87. q.updateSortedKeys()
  88. for i, tsn := range q.sorted {
  89. if i == 0 {
  90. b.start = uint16(tsn - cumulativeTSN)
  91. b.end = b.start
  92. continue
  93. }
  94. diff := uint16(tsn - cumulativeTSN)
  95. if b.end+1 == diff {
  96. b.end++
  97. } else {
  98. gapAckBlocks = append(gapAckBlocks, gapAckBlock{
  99. start: b.start,
  100. end: b.end,
  101. })
  102. b.start = diff
  103. b.end = diff
  104. }
  105. }
  106. gapAckBlocks = append(gapAckBlocks, gapAckBlock{
  107. start: b.start,
  108. end: b.end,
  109. })
  110. return gapAckBlocks
  111. }
  112. func (q *payloadQueue) getGapAckBlocksString(cumulativeTSN uint32) string {
  113. gapAckBlocks := q.getGapAckBlocks(cumulativeTSN)
  114. str := fmt.Sprintf("cumTSN=%d", cumulativeTSN)
  115. for _, b := range gapAckBlocks {
  116. str += fmt.Sprintf(",%d-%d", b.start, b.end)
  117. }
  118. return str
  119. }
  120. func (q *payloadQueue) markAsAcked(tsn uint32) int {
  121. var nBytesAcked int
  122. if c, ok := q.chunkMap[tsn]; ok {
  123. c.acked = true
  124. c.retransmit = false
  125. nBytesAcked = len(c.userData)
  126. q.nBytes -= nBytesAcked
  127. c.userData = []byte{}
  128. }
  129. return nBytesAcked
  130. }
  131. func (q *payloadQueue) getLastTSNReceived() (uint32, bool) {
  132. q.updateSortedKeys()
  133. qlen := len(q.sorted)
  134. if qlen == 0 {
  135. return 0, false
  136. }
  137. return q.sorted[qlen-1], true
  138. }
  139. func (q *payloadQueue) markAllToRetrasmit() {
  140. for _, c := range q.chunkMap {
  141. if c.acked || c.abandoned() {
  142. continue
  143. }
  144. c.retransmit = true
  145. }
  146. }
  147. func (q *payloadQueue) getNumBytes() int {
  148. return q.nBytes
  149. }
  150. func (q *payloadQueue) size() int {
  151. return len(q.chunkMap)
  152. }