From cd06d59cba56e09790784218d7786bf019dd927d Mon Sep 17 00:00:00 2001 From: xtaci Date: Thu, 4 Jan 2024 22:43:12 +0800 Subject: [PATCH] Squashed commit of the following: commit 0e5741b954e9a13cbadf8bccbfd79e9045c79702 Author: fuli Date: Thu Jan 4 16:21:42 2024 +0800 use latest packet compare commit 97999b9dc2ccbe1794cb623a2c5450cc8db1fb6d Author: fuli Date: Thu Jan 4 16:07:08 2024 +0800 face race in test commit 446bead851d14d0486079c5b30aff02fe8c2b833 Author: fuli Date: Thu Jan 4 16:04:42 2024 +0800 upd commit 570e9b006e5619b8875554636f27cfad0302f8d9 Author: fuli Date: Thu Jan 4 16:03:17 2024 +0800 revert some file commit c665db20a5508eee4f458c415055f6bd50e597c8 Author: fuli Date: Thu Jan 4 15:59:16 2024 +0800 Revert "Revert "1. simplify readloop_xxx files & tx_xxx files by merging"" This reverts commit 717b79a1568d5c3756462500f189ed7016d8e1d7. commit 717b79a1568d5c3756462500f189ed7016d8e1d7 Author: fuli Date: Thu Jan 4 15:58:30 2024 +0800 Revert "1. simplify readloop_xxx files & tx_xxx files by merging" This reverts commit 681bda64d12f48c3a94d626da3d155450ae4f3ce. commit 681bda64d12f48c3a94d626da3d155450ae4f3ce Author: fuli Date: Thu Jan 4 15:54:29 2024 +0800 1. simplify readloop_xxx files & tx_xxx files by merging 2. fix the race in owned packet test commit 7974357dd0c41ba8a90919a92e865307c66faaa5 Author: xtaci Date: Thu Jan 4 00:04:06 2024 +0800 use unix.Milli commit 375407b80cab3685752e918c51ebefda965d3ffc Author: xtaci Date: Wed Jan 3 23:40:01 2024 +0800 add passive fec --- fec.go | 50 ++++++++++++++++++++++++------------------ fec_test.go | 4 ++-- sess.go | 6 +++--- sess_test.go | 61 +--------------------------------------------------- 4 files changed, 35 insertions(+), 86 deletions(-) diff --git a/fec.go b/fec.go index 27dd66e..7d43543 100644 --- a/fec.go +++ b/fec.go @@ -3,6 +3,7 @@ package kcp import ( "encoding/binary" "sync/atomic" + "time" "github.com/klauspost/reedsolomon" ) @@ -278,15 +279,16 @@ type ( payloadOffset int // FEC payload offset // caches - shardCache [][]byte - encodeCache [][]byte + shardCache [][]byte + encodeCache [][]byte + tsLatestPacket int64 // RS encoder codec reedsolomon.Encoder } ) -func newFECEncoder(dataShards, parityShards, offset int) *fecEncoder { +func newFECEncoder(dataShards, parityShards, offset, minRTO int) *fecEncoder { if dataShards <= 0 || parityShards <= 0 { return nil } @@ -315,7 +317,7 @@ func newFECEncoder(dataShards, parityShards, offset int) *fecEncoder { // encodes the packet, outputs parity shards if we have collected quorum datashards // notice: the contents of 'ps' will be re-written in successive calling -func (enc *fecEncoder) encode(b []byte) (ps [][]byte) { +func (enc *fecEncoder) encode(b []byte, rto uint32) (ps [][]byte) { // The header format: // | FEC SEQID(4B) | FEC TYPE(2B) | SIZE (2B) | PAYLOAD(SIZE-2) | // |<-headerOffset |<-payloadOffset @@ -334,26 +336,30 @@ func (enc *fecEncoder) encode(b []byte) (ps [][]byte) { } // Generation of Reed-Solomon Erasure Code + now := time.Now().UnixMilli() if enc.shardCount == enc.dataShards { - // fill '0' into the tail of each datashard - for i := 0; i < enc.dataShards; i++ { - shard := enc.shardCache[i] - slen := len(shard) - clear(shard[slen:enc.maxSize]) - } + // generate the rs-code only if the data is continuous. + if now-enc.tsLatestPacket < int64(rto) { + // fill '0' into the tail of each datashard + for i := 0; i < enc.dataShards; i++ { + shard := enc.shardCache[i] + slen := len(shard) + clear(shard[slen:enc.maxSize]) + } - // construct equal-sized slice with stripped header - cache := enc.encodeCache - for k := range cache { - cache[k] = enc.shardCache[k][enc.payloadOffset:enc.maxSize] - } + // construct equal-sized slice with stripped header + cache := enc.encodeCache + for k := range cache { + cache[k] = enc.shardCache[k][enc.payloadOffset:enc.maxSize] + } - // encoding - 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] + // encoding + 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] + } } } @@ -362,6 +368,8 @@ func (enc *fecEncoder) encode(b []byte) (ps [][]byte) { enc.maxSize = 0 } + enc.tsLatestPacket = now + return } diff --git a/fec_test.go b/fec_test.go index 59b64ac..46a6a45 100644 --- a/fec_test.go +++ b/fec_test.go @@ -35,9 +35,9 @@ func BenchmarkFECEncode(b *testing.B) { b.ReportAllocs() b.SetBytes(payLoad) - encoder := newFECEncoder(dataSize, paritySize, 0) + encoder := newFECEncoder(dataSize, paritySize, 0, 200) for i := 0; i < b.N; i++ { data := make([]byte, payLoad) - encoder.encode(data) + encoder.encode(data, 200) } } diff --git a/sess.go b/sess.go index 35e7b80..a4a1b45 100644 --- a/sess.go +++ b/sess.go @@ -154,9 +154,9 @@ func newUDPSession(conv uint32, dataShards, parityShards int, l *Listener, conn // FEC codec initialization sess.fecDecoder = newFECDecoder(dataShards, parityShards) if sess.block != nil { - sess.fecEncoder = newFECEncoder(dataShards, parityShards, cryptHeaderSize) + sess.fecEncoder = newFECEncoder(dataShards, parityShards, cryptHeaderSize, IKCP_RTO_DEF) } else { - sess.fecEncoder = newFECEncoder(dataShards, parityShards, 0) + sess.fecEncoder = newFECEncoder(dataShards, parityShards, 0, IKCP_RTO_DEF) } // calculate additional header size introduced by FEC and encryption @@ -531,7 +531,7 @@ func (s *UDPSession) output(buf []byte) { // 1. FEC encoding if s.fecEncoder != nil { - ecc = s.fecEncoder.encode(buf) + ecc = s.fecEncoder.encode(buf, s.kcp.rx_rto) } // 2&3. crc32 & encryption diff --git a/sess_test.go b/sess_test.go index fbe3ad1..f4cc62c 100644 --- a/sess_test.go +++ b/sess_test.go @@ -87,7 +87,7 @@ func dialTinyBufferEcho(port int) (*UDPSession, error) { return sess, err } -////////////////////////// +// //////////////////////// func listenEcho(port int) (net.Listener, error) { //block, _ := NewNoneBlockCrypt(pass) //block, _ := NewSimpleXORBlockCrypt(pass) @@ -575,34 +575,6 @@ func newClosedFlagPacketConn(c net.PacketConn) *closedFlagPacketConn { return &closedFlagPacketConn{c, false} } -// Listener should close a net.PacketConn that it created. -// https://github.com/xtaci/kcp-go/issues/165 -func TestListenerOwnedPacketConn(t *testing.T) { - // ListenWithOptions creates its own net.PacketConn. - l, err := ListenWithOptions("127.0.0.1:0", nil, 0, 0) - if err != nil { - panic(err) - } - defer l.Close() - // Replace the internal net.PacketConn with one that remembers when it - // has been closed. - pconn := newClosedFlagPacketConn(l.conn) - l.conn = pconn - - if pconn.Closed { - t.Fatal("owned PacketConn closed before Listener.Close()") - } - - err = l.Close() - if err != nil { - panic(err) - } - - if !pconn.Closed { - t.Fatal("owned PacketConn not closed after Listener.Close()") - } -} - // Listener should not close a net.PacketConn that it did not create. // https://github.com/xtaci/kcp-go/issues/165 func TestListenerNonOwnedPacketConn(t *testing.T) { @@ -635,37 +607,6 @@ func TestListenerNonOwnedPacketConn(t *testing.T) { } } -// UDPSession should close a net.PacketConn that it created. -// https://github.com/xtaci/kcp-go/issues/165 -func TestUDPSessionOwnedPacketConn(t *testing.T) { - l := sinkServer(0) - defer l.Close() - - // DialWithOptions creates its own net.PacketConn. - client, err := DialWithOptions(l.Addr().String(), nil, 0, 0) - if err != nil { - panic(err) - } - defer client.Close() - // Replace the internal net.PacketConn with one that remembers when it - // has been closed. - pconn := newClosedFlagPacketConn(client.conn) - client.conn = pconn - - if pconn.Closed { - t.Fatal("owned PacketConn closed before UDPSession.Close()") - } - - err = client.Close() - if err != nil { - panic(err) - } - - if !pconn.Closed { - t.Fatal("owned PacketConn not closed after UDPSession.Close()") - } -} - // UDPSession should not close a net.PacketConn that it did not create. // https://github.com/xtaci/kcp-go/issues/165 func TestUDPSessionNonOwnedPacketConn(t *testing.T) {