package quic
import (
"errors"
"sync"
"github.com/tmc/go-iroh/internal/qng/internal/protocol"
list "github.com/tmc/go-iroh/internal/qng/internal/utils/linkedlist"
"github.com/tmc/go-iroh/internal/qng/internal/wire"
)
type byteInterval struct {
Start protocol .ByteCount
End protocol .ByteCount
}
var byteIntervalElementPool sync .Pool
func init() {
byteIntervalElementPool = *list .NewPool [byteInterval ]()
}
type frameSorterEntry struct {
Data []byte
Frame *wire .StreamFrame
}
type frameSorter struct {
queue map [protocol .ByteCount ]frameSorterEntry
readPos protocol .ByteCount
gaps *list .List [byteInterval ]
}
var errDuplicateStreamData = errors .New ("duplicate stream data" )
func newFrameSorter() *frameSorter {
s := frameSorter {
gaps : list .NewWithPool [byteInterval ](&byteIntervalElementPool ),
queue : make (map [protocol .ByteCount ]frameSorterEntry ),
}
s .gaps .PushFront (byteInterval {Start : 0 , End : protocol .MaxByteCount })
return &s
}
func (s *frameSorter ) Push (data []byte , offset protocol .ByteCount , frame *wire .StreamFrame ) error {
err := s .push (data , offset , frame )
if err == errDuplicateStreamData {
if frame != nil {
frame .PutBack ()
}
return nil
}
return err
}
func (s *frameSorter ) push (data []byte , offset protocol .ByteCount , frame *wire .StreamFrame ) error {
if len (data ) == 0 {
return errDuplicateStreamData
}
start := offset
end := offset + protocol .ByteCount (len (data ))
if start == s .readPos {
if _ , ok := s .queue [start ]; !ok {
if gap := s .gaps .Front (); gap != nil && gap .Value .Start == start && end <= gap .Value .End {
if end == gap .Value .End {
s .gaps .Remove (gap )
} else {
gap .Value .Start = end
}
s .queue [start ] = frameSorterEntry {Data : data , Frame : frame }
return nil
}
}
}
if end <= s .gaps .Front ().Value .Start {
return errDuplicateStreamData
}
startGap , startsInGap := s .findStartGap (start )
endGap , endsInGap := s .findEndGap (startGap , end )
startGapEqualsEndGap := startGap == endGap
if (startGapEqualsEndGap && end <= startGap .Value .Start ) ||
(!startGapEqualsEndGap && startGap .Value .End >= endGap .Value .Start && end <= startGap .Value .Start ) {
return errDuplicateStreamData
}
startGapNext := startGap .Next ()
startGapEnd := startGap .Value .End
endGapStart := endGap .Value .Start
endGapEnd := endGap .Value .End
var adjustedStartGapEnd bool
var wasCut bool
pos := start
var hasReplacedAtLeastOne bool
for {
oldEntry , ok := s .queue [pos ]
if !ok {
break
}
oldEntryLen := protocol .ByteCount (len (oldEntry .Data ))
if end -pos > oldEntryLen || (hasReplacedAtLeastOne && end -pos == oldEntryLen ) {
delete (s .queue , pos )
pos += oldEntryLen
hasReplacedAtLeastOne = true
if oldEntry .Frame != nil {
oldEntry .Frame .PutBack ()
}
} else {
if !hasReplacedAtLeastOne {
return errDuplicateStreamData
}
data = data [:pos -start ]
end = pos
wasCut = true
break
}
}
if !startsInGap && !hasReplacedAtLeastOne {
data = data [startGap .Value .Start -start :]
start = startGap .Value .Start
wasCut = true
}
if start <= startGap .Value .Start {
if end >= startGap .Value .End {
s .gaps .Remove (startGap )
} else {
startGap .Value .Start = end
}
} else if !hasReplacedAtLeastOne {
startGap .Value .End = start
adjustedStartGapEnd = true
}
if !startGapEqualsEndGap {
s .deleteConsecutive (startGapEnd )
var nextGap *list .Element [byteInterval ]
for gap := startGapNext ; gap .Value .End < endGapStart ; gap = nextGap {
nextGap = gap .Next ()
s .deleteConsecutive (gap .Value .End )
s .gaps .Remove (gap )
}
}
if !endsInGap && start != endGapEnd && end > endGapEnd {
data = data [:endGapEnd -start ]
end = endGapEnd
wasCut = true
}
if end == endGapEnd {
if !startGapEqualsEndGap {
s .gaps .Remove (endGap )
}
} else {
if startGapEqualsEndGap && adjustedStartGapEnd {
s .gaps .InsertAfter (byteInterval {Start : end , End : startGapEnd }, startGap )
} else if !startGapEqualsEndGap {
endGap .Value .Start = end
}
}
if wasCut && len (data ) < protocol .MinStreamFrameBufferSize {
newData := make ([]byte , len (data ))
copy (newData , data )
data = newData
if frame != nil {
frame .PutBack ()
frame = nil
}
}
if s .gaps .Len () > protocol .MaxStreamFrameSorterGaps {
return errors .New ("too many gaps in received data" )
}
s .queue [start ] = frameSorterEntry {Data : data , Frame : frame }
return nil
}
func (s *frameSorter ) findStartGap (offset protocol .ByteCount ) (*list .Element [byteInterval ], bool ) {
for gap := s .gaps .Front (); gap != nil ; gap = gap .Next () {
if offset >= gap .Value .Start && offset <= gap .Value .End {
return gap , true
}
if offset < gap .Value .Start {
return gap , false
}
}
panic ("no gap found" )
}
func (s *frameSorter ) findEndGap (startGap *list .Element [byteInterval ], offset protocol .ByteCount ) (*list .Element [byteInterval ], bool ) {
for gap := startGap ; gap != nil ; gap = gap .Next () {
if offset >= gap .Value .Start && offset < gap .Value .End {
return gap , true
}
if offset < gap .Value .Start {
return gap .Prev (), false
}
}
panic ("no gap found" )
}
func (s *frameSorter ) deleteConsecutive (pos protocol .ByteCount ) {
for {
oldEntry , ok := s .queue [pos ]
if !ok {
break
}
oldEntryLen := protocol .ByteCount (len (oldEntry .Data ))
delete (s .queue , pos )
if oldEntry .Frame != nil {
oldEntry .Frame .PutBack ()
}
pos += oldEntryLen
}
}
func (s *frameSorter ) Pop () (protocol .ByteCount , []byte , *wire .StreamFrame ) {
entry , ok := s .queue [s .readPos ]
if !ok {
return s .readPos , nil , nil
}
delete (s .queue , s .readPos )
offset := s .readPos
s .readPos += protocol .ByteCount (len (entry .Data ))
if s .gaps .Front ().Value .End <= s .readPos {
panic ("frame sorter BUG: read position higher than a gap" )
}
return offset , entry .Data , entry .Frame
}
func (s *frameSorter ) HasMoreData () bool {
return len (s .queue ) > 0
}
var errTooLittleData = errors .New ("too little data" )
func (s *frameSorter ) Peek (offset protocol .ByteCount , p []byte ) error {
if len (p ) == 0 {
return nil
}
pos := offset
remaining := len (p )
for remaining > 0 {
entry , ok := s .queue [pos ]
if !ok {
return errTooLittleData
}
entryLen := len (entry .Data )
if remaining <= entryLen {
break
}
remaining -= entryLen
pos += protocol .ByteCount (entryLen )
}
pos = offset
var copied int
for copied < len (p ) {
entry := s .queue [pos ]
copied += copy (p [copied :], entry .Data )
pos += protocol .ByteCount (len (entry .Data ))
}
return nil
}
The pages are generated with Golds v0.8.4 . (GOOS=linux GOARCH=amd64)
Golds is a Go 101 project developed by Tapir Liu .
PR and bug reports are welcome and can be submitted to the issue list .
Please follow @zigo_101 (reachable from the left QR code) to get the latest news of Golds .