github/kcp-go
4
0
Fork 0
mirror of https://github.com/xtaci/kcp-go.git synced 2024-04-21 12:32:15 +00:00
Code Issues Releases Wiki Activity

Squashed commit of the following:

Browse Source 
commit 0e5741b954
Author: fuli <fuli@rockx.com>
Date:   Thu Jan 4 16:21:42 2024 +0800

    use latest packet compare

commit 97999b9dc2
Author: fuli <fuli@rockx.com>
Date:   Thu Jan 4 16:07:08 2024 +0800

    face race in test

commit 446bead851
Author: fuli <fuli@rockx.com>
Date:   Thu Jan 4 16:04:42 2024 +0800

    upd

commit 570e9b006e
Author: fuli <fuli@rockx.com>
Date:   Thu Jan 4 16:03:17 2024 +0800

    revert some file

commit c665db20a5
Author: fuli <fuli@rockx.com>
Date:   Thu Jan 4 15:59:16 2024 +0800

    Revert "Revert "1. simplify readloop_xxx files & tx_xxx files by merging""

    This reverts commit 717b79a156.

commit 717b79a156
Author: fuli <fuli@rockx.com>
Date:   Thu Jan 4 15:58:30 2024 +0800

    Revert "1. simplify readloop_xxx files & tx_xxx files by merging"

    This reverts commit 681bda64d1.

commit 681bda64d1
Author: fuli <fuli@rockx.com>
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 7974357dd0
Author: xtaci <daniel820313@gmail.com>
Date:   Thu Jan 4 00:04:06 2024 +0800

    use unix.Milli

commit 375407b80c
Author: xtaci <daniel820313@gmail.com>
Date:   Wed Jan 3 23:40:01 2024 +0800

    add passive fec
This commit is contained in:
xtaci
2024-01-04 22:43:12 +08:00
parent 637838a7b2
commit cd06d59cba
4 changed files with 35 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fec.go
+29 -21
View File
@@ -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
}
fec_test.go
+2 -2
View File
@@ -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)
}
}
sess.go
+3 -3
View File
@@ -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
sess_test.go
+1 -60
View File
@@ -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) {