123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459 |
- // Copyright © 2015 Daniel Fu <daniel820313@gmail.com>.
- // Copyright © 2019 Loki 'l0k18' Verloren <stalker.loki@protonmail.ch>.
- // Copyright © 2021 Gridfinity, LLC. <admin@gridfinity.com>.
- //
- // All rights reserved.
- //
- // All use of this code is governed by the MIT license.
- // The complete license is available in the LICENSE file.
- package gfcp // import "go.gridfinity.dev/gfcp"
- import (
- "encoding/binary"
- "sync/atomic"
- "github.com/klauspost/reedsolomon"
- )
- const (
- fecHeaderSize = 6
- fecHeaderSizePlus2 = fecHeaderSize + 2
- // KTypeData ...
- KTypeData = 0xf1
- // KTypeParity ...
- KTypeParity = 0xf2
- )
- // FecPacket ...
- type FecPacket []byte
- func (
- bts FecPacket,
- ) seqid() uint32 {
- return binary.LittleEndian.Uint32(
- bts,
- )
- }
- func (
- bts FecPacket,
- ) flag() uint16 {
- return binary.LittleEndian.Uint16(
- bts[4:],
- )
- }
- func (
- bts FecPacket,
- ) data() []byte {
- return bts[6:]
- }
- // FecDecoder ...
- type FecDecoder struct {
- rxlimit int
- dataShards int
- parityShards int
- shardSize int
- rx []FecPacket
- DecodeCache [][]byte
- flagCache []bool
- zeros []byte
- codec reedsolomon.Encoder
- }
- // NewFECDecoder ...
- func NewFECDecoder(
- rxlimit,
- dataShards,
- parityShards int,
- ) *FecDecoder {
- if dataShards <= 0 || parityShards <= 0 {
- return nil
- }
- if rxlimit < dataShards+parityShards {
- return nil
- }
- dec := new(
- FecDecoder,
- )
- dec.rxlimit = rxlimit
- dec.dataShards = dataShards
- dec.parityShards = parityShards
- dec.shardSize = dataShards + parityShards
- codec, err := reedsolomon.New(
- dataShards,
- parityShards,
- )
- if err != nil {
- return nil
- }
- dec.codec = codec
- dec.DecodeCache = make(
- [][]byte,
- dec.shardSize,
- )
- dec.flagCache = make(
- []bool,
- dec.shardSize,
- )
- dec.zeros = make(
- []byte,
- GFcpMtuLimit,
- )
- return dec
- }
- // Decode ...
- func (
- dec *FecDecoder,
- ) Decode(
- in FecPacket,
- ) (
- recovered [][]byte,
- ) {
- n := len(
- dec.rx,
- ) - 1
- insertIdx := 0
- for i := n; i >= 0; i-- {
- if in.seqid() == dec.rx[i].seqid() {
- return nil
- } else if _itimediff(
- in.seqid(),
- dec.rx[i].seqid(),
- ) > 0 {
- insertIdx = i + 1
- break
- }
- }
- // make a copy
- pkt := FecPacket(KxmitBuf.Get().([]byte)[:len(in)])
- copy(
- pkt,
- in,
- )
- if insertIdx == n+1 {
- dec.rx = append(
- dec.rx,
- pkt,
- )
- } else {
- dec.rx = append(
- dec.rx,
- FecPacket{},
- )
- copy(
- dec.rx[insertIdx+1:],
- dec.rx[insertIdx:],
- )
- dec.rx[insertIdx] = pkt
- }
- shardBegin := pkt.seqid() - pkt.seqid()%uint32(dec.shardSize)
- shardEnd := shardBegin + uint32(dec.shardSize) - 1
- searchBegin := insertIdx - int(pkt.seqid()%uint32(dec.shardSize))
- if searchBegin < 0 {
- searchBegin = 0
- }
- searchEnd := searchBegin + dec.shardSize - 1
- if searchEnd >= len(
- dec.rx,
- ) {
- searchEnd = len(
- dec.rx,
- ) - 1
- }
- if searchEnd-searchBegin+1 >= dec.dataShards {
- var numshard, numDataShard, first, maxlen int
- shards := dec.DecodeCache
- shardsflag := dec.flagCache
- for k := range dec.DecodeCache {
- shards[k] = nil
- shardsflag[k] = false
- }
- for i := searchBegin; i <= searchEnd; i++ {
- seqid := dec.rx[i].seqid()
- if _itimediff(
- seqid,
- shardEnd,
- ) > 0 {
- break
- } else if _itimediff(
- seqid,
- shardBegin,
- ) >= 0 {
- shards[seqid%uint32(
- dec.shardSize,
- )] = dec.rx[i].data()
- shardsflag[seqid%uint32(
- dec.shardSize,
- )] = true
- numshard++
- if dec.rx[i].flag() == KTypeData {
- numDataShard++
- }
- if numshard == 1 {
- first = i
- }
- if len(
- dec.rx[i].data(),
- ) > maxlen {
- maxlen = len(
- dec.rx[i].data(),
- )
- }
- }
- }
- if numDataShard == dec.dataShards {
- dec.rx = dec.freeRange(
- first,
- numshard,
- dec.rx,
- )
- } else if numshard >= dec.dataShards {
- for k := range shards {
- if shards[k] != nil {
- dlen := len(
- shards[k],
- )
- shards[k] = shards[k][:maxlen]
- copy(shards[k][dlen:], dec.zeros)
- } else {
- shards[k] = KxmitBuf.Get().([]byte)[:0]
- }
- }
- if err := dec.codec.ReconstructData(
- shards,
- ); err == nil {
- for k := range shards[:dec.dataShards] {
- if !shardsflag[k] {
- recovered = append(
- recovered,
- shards[k],
- )
- }
- }
- }
- dec.rx = dec.freeRange(
- first,
- numshard,
- dec.rx,
- )
- }
- }
- if len(dec.rx) > dec.rxlimit {
- if dec.rx[0].flag() == KTypeData {
- atomic.AddUint64(
- &DefaultSnsi.GFcpFECRuntShards,
- 1,
- )
- }
- dec.rx = dec.freeRange(
- 0,
- 1,
- dec.rx,
- )
- }
- return
- }
- func (
- dec *FecDecoder,
- ) freeRange(
- first,
- n int,
- q []FecPacket,
- ) []FecPacket {
- for i := first; i < first+n; i++ {
- // TODO(jhj): Switch to pointer to avoid allocation.
- KxmitBuf.Put(
- []byte(
- q[i],
- ),
- )
- }
- if first == 0 && n < (cap(q)/2) {
- return q[n:]
- }
- copy(
- q[first:],
- q[first+n:],
- )
- return q[:len(
- q,
- )-n]
- }
- type (
- // FecEncoder ...
- FecEncoder struct {
- dataShards int
- parityShards int
- shardSize int
- paws uint32 // Protect Against Wrapped Sequence numbers
- next uint32 // next seqid
- shardCount int // count the number of datashards collected
- maxSize int // track maximum data length in datashard
- headerOffset int // FEC header offset
- payloadOffset int // FEC payload offset
- shardCache [][]byte
- EncodeCache [][]byte
- zeros []byte
- codec reedsolomon.Encoder
- }
- )
- // NewFECEncoder ...
- func NewFECEncoder(
- dataShards,
- parityShards,
- offset int,
- ) *FecEncoder {
- if dataShards <= 0 || parityShards <= 0 {
- return nil
- }
- enc := new(
- FecEncoder,
- )
- enc.dataShards = dataShards
- enc.parityShards = parityShards
- enc.shardSize = dataShards + parityShards
- enc.paws = (0xFFFFFFFF/uint32(enc.shardSize) - 1) * uint32(enc.shardSize)
- enc.headerOffset = offset
- enc.payloadOffset = enc.headerOffset + fecHeaderSize
- codec, err := reedsolomon.New(
- dataShards,
- parityShards,
- )
- if err != nil {
- return nil
- }
- enc.codec = codec
- enc.EncodeCache = make(
- [][]byte,
- enc.shardSize,
- )
- enc.shardCache = make(
- [][]byte,
- enc.shardSize,
- )
- for k := range enc.shardCache {
- enc.shardCache[k] = make(
- []byte,
- GFcpMtuLimit,
- )
- }
- enc.zeros = make(
- []byte,
- GFcpMtuLimit,
- )
- return enc
- }
- // Encode ...
- func (
- enc *FecEncoder,
- ) Encode(
- b []byte,
- ) (
- ps [][]byte,
- ) {
- enc.markData(
- b[enc.headerOffset:],
- )
- binary.LittleEndian.PutUint16(
- b[enc.payloadOffset:],
- uint16(
- len(
- b[enc.payloadOffset:],
- ),
- ),
- )
- sz := len(
- b,
- )
- enc.shardCache[enc.shardCount] = enc.shardCache[enc.shardCount][:sz]
- copy(
- enc.shardCache[enc.shardCount][enc.payloadOffset:],
- b[enc.payloadOffset:],
- )
- enc.shardCount++
- if sz > enc.maxSize {
- enc.maxSize = sz
- }
- if enc.shardCount == enc.dataShards {
- for i := 0; i < enc.dataShards; i++ {
- shard := enc.shardCache[i]
- slen := len(
- shard,
- )
- copy(
- shard[slen:enc.maxSize],
- enc.zeros,
- )
- }
- cache := enc.EncodeCache
- for k := range cache {
- cache[k] = enc.shardCache[k][enc.payloadOffset:enc.maxSize]
- }
- if err := enc.codec.Encode(
- cache,
- ); err == nil {
- ps = enc.shardCache[enc.dataShards:]
- for k := range ps {
- enc.markParity(
- ps[k][enc.headerOffset:],
- )
- ps[k] = ps[k][:enc.maxSize]
- }
- }
- enc.shardCount = 0
- enc.maxSize = 0
- }
- return
- }
- func (
- enc *FecEncoder,
- ) markData(
- data []byte,
- ) {
- binary.LittleEndian.PutUint32(
- data,
- enc.next,
- )
- binary.LittleEndian.PutUint16(
- data[4:],
- KTypeData,
- )
- enc.next++
- }
- func (
- enc *FecEncoder,
- ) markParity(
- data []byte,
- ) {
- binary.LittleEndian.PutUint32(
- data,
- enc.next,
- )
- binary.LittleEndian.PutUint16(
- data[4:],
- KTypeParity,
- )
- enc.next = (enc.next + 1) % enc.paws
- }
|