diff --git a/fec.go b/fec.go index de58b15..576e45f 100644 --- a/fec.go +++ b/fec.go @@ -13,6 +13,7 @@ const ( typeData = 0xf1 typeParity = 0xf2 fecExpire = 60000 + rxFECMulti = 3 // FEC keeps rxFECMulti* (dataShard+parityShard) ordered packets in memory ) // fecPacket is a decoded FEC packet @@ -50,19 +51,16 @@ type fecDecoder struct { autoTune autoTune } -func newFECDecoder(rxlimit, dataShards, parityShards int) *fecDecoder { +func newFECDecoder(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 + dec.rxlimit = rxFECMulti * dec.shardSize codec, err := reedsolomon.New(dataShards, parityShards) if err != nil { return nil @@ -76,17 +74,17 @@ func newFECDecoder(rxlimit, dataShards, parityShards int) *fecDecoder { // decode a fec packet func (dec *fecDecoder) decode(in fecPacket) (recovered [][]byte) { - // autotune + // sample to auto FEC tuner if in.flag() == typeData { dec.autoTune.Sample(true, in.seqid()) } else { dec.autoTune.Sample(false, in.seqid()) } - // check if FEC parameters has mismatched + // check if FEC parameters is out of sync var shouldTune bool - if int(in.seqid())%dec.shardSize < dec.dataShards { // should be data - if in.flag() != typeData { + if int(in.seqid())%dec.shardSize < dec.dataShards { + if in.flag() != typeData { // expect typeData shouldTune = true } } else { @@ -100,10 +98,11 @@ func (dec *fecDecoder) decode(in fecPacket) (recovered [][]byte) { autoPS := dec.autoTune.FindPeriod(false) // edges found, we can tune parameters now - if autoDS > 0 && autoPS > 0 { + if autoDS > 0 && autoPS > 0 && autoDS < 256 && autoPS < 256 { dec.dataShards = autoDS dec.parityShards = autoPS dec.shardSize = autoDS + autoPS + dec.rxlimit = rxFECMulti * dec.shardSize codec, err := reedsolomon.New(autoDS, autoPS) if err != nil { return nil diff --git a/sess.go b/sess.go index 21b0dd3..a902783 100644 --- a/sess.go +++ b/sess.go @@ -35,9 +35,6 @@ const ( // maximum packet size mtuLimit = 1500 - // FEC keeps rxFECMulti* (dataShard+parityShard) ordered packets in memory - rxFECMulti = 3 - // accept backlog acceptBacklog = 128 ) @@ -155,7 +152,7 @@ func newUDPSession(conv uint32, dataShards, parityShards int, l *Listener, conn } // FEC codec initialization - sess.fecDecoder = newFECDecoder(rxFECMulti*(dataShards+parityShards), dataShards, parityShards) + sess.fecDecoder = newFECDecoder(dataShards, parityShards) if sess.block != nil { sess.fecEncoder = newFECEncoder(dataShards, parityShards, cryptHeaderSize) } else { @@ -689,6 +686,10 @@ func (s *UDPSession) kcpInput(data []byte) { // lock s.mu.Lock() + // if fecDecoder is not initialized, create one with default parameter + if s.fecDecoder == nil { + s.fecDecoder = newFECDecoder(1, 1) + } recovers := s.fecDecoder.decode(f) if f.flag() == typeData { if ret := s.kcp.Input(data[fecHeaderSizePlus2:], true, s.ackNoDelay); ret != 0 { @@ -1021,7 +1022,7 @@ func serveConn(block BlockCrypt, dataShards, parityShards int, conn net.PacketCo l.dataShards = dataShards l.parityShards = parityShards l.block = block - l.fecDecoder = newFECDecoder(rxFECMulti*(dataShards+parityShards), dataShards, parityShards) + l.fecDecoder = newFECDecoder(dataShards, parityShards) l.chSocketReadError = make(chan struct{}) // calculate header size