diff --git a/README.md b/README.md index 853aea6..4848d93 100644 --- a/README.md +++ b/README.md @@ -291,6 +291,8 @@ docker build -t vk-turn-proxy . Для более тонкой настройки доступны переменные: `VK_TURN_KCP_NODELAY`, `VK_TURN_KCP_INTERVAL`, `VK_TURN_KCP_RESEND`, `VK_TURN_KCP_NC`, `VK_TURN_KCP_SNDWND`, `VK_TURN_KCP_RCVWND`, `VK_TURN_KCP_ACK_NODELAY`. +**FEC (Reed-Solomon):** `VK_TURN_KCP_FEC=data:parity` (например, `10:3`) — упреждающая коррекция ошибок. На каждые `data` пакетов добавляется `parity` избыточных, потери до `parity` из `data+parity` восстанавливаются без ретрансмита. Overhead по полосе: `parity/data` (для `10:3` — 30%). Должно совпадать на клиенте и сервере. По умолчанию выключено (`0:0`). Включай только при случайных потерях; при шейпе по полосе FEC ухудшит goodput. + ## Яндекс Телемост diff --git a/client/main.go b/client/main.go index 77ecb09..a5c41ca 100644 --- a/client/main.go +++ b/client/main.go @@ -175,15 +175,20 @@ type throughputStats struct { } func (s *throughputStats) addTx(n int) { - if n > 0 { - s.tx.Add(uint64(n)) + // Counters are only consumed by logEvery, which is itself debug-gated. + // Skip the atomic add (and the cache-line traffic across N streams) + // when nobody is going to read the result. + if !isDebug || n <= 0 { + return } + s.tx.Add(uint64(n)) } func (s *throughputStats) addRx(n int) { - if n > 0 { - s.rx.Add(uint64(n)) + if !isDebug || n <= 0 { + return } + s.rx.Add(uint64(n)) } func (s *throughputStats) logEvery(ctx context.Context, label, txName, rxName string) { @@ -1730,7 +1735,7 @@ func oneDtlsConnection(ctx context.Context, peer *net.UDPAddr, listenConn net.Pa } wg := sync.WaitGroup{} - wg.Add(1) + wg.Add(2) context.AfterFunc(dtlsctx, func() { if err := dtlsConn.SetDeadline(time.Now()); err != nil { log.Printf("[STREAM %d] Warning: SetDeadline failed: %v", streamID, err) @@ -1738,6 +1743,7 @@ func oneDtlsConnection(ctx context.Context, peer *net.UDPAddr, listenConn net.Pa }) go func() { + defer wg.Done() defer dtlscancel() for { select { @@ -1786,6 +1792,32 @@ func (c *connectedUDPConn) WriteTo(p []byte, _ net.Addr) (int, error) { return c.Write(p) } +// splitFirstWriteConn wraps a TCP net.Conn and splits the first Write into two +// segments (default 6 bytes + remainder) with a small delay, so the STUN magic +// cookie at offset 4-7 is broken across two TCP segments. Defeats DPI rules +// that match on the first segment without reassembly. +type splitFirstWriteConn struct { + net.Conn + splitAt int + delay time.Duration + done atomic.Bool +} + +func (s *splitFirstWriteConn) Write(b []byte) (int, error) { + if s.done.CompareAndSwap(false, true) && len(b) > s.splitAt { + n1, err := s.Conn.Write(b[:s.splitAt]) + if err != nil { + return n1, err + } + if s.delay > 0 { + time.Sleep(s.delay) + } + n2, err := s.Conn.Write(b[s.splitAt:]) + return n1 + n2, err + } + return s.Conn.Write(b) +} + type turnParams struct { host string port string @@ -1854,7 +1886,8 @@ func oneTurnConnection(ctx context.Context, turnParams *turnParams, peer *net.UD return } }() - turnConn = turn.NewSTUNConn(conn) + wrappedConn := &splitFirstWriteConn{Conn: conn, splitAt: 6, delay: 20 * time.Millisecond} + turnConn = turn.NewSTUNConn(wrappedConn) } var addrFamily turn.RequestedAddressFamily if peer.IP.To4() != nil { @@ -1930,6 +1963,12 @@ func oneTurnConnection(ctx context.Context, turnParams *turnParams, peer *net.UD go func() { defer turncancel() buf := make([]byte, 1600) + // Reusable scratch buffer for wrapped wire bytes; sized once per + // stream so the hot-path TX loop performs zero allocations. + var wireBuf []byte + if useWrap { + wireBuf = make([]byte, wrapNonceLen+len(buf)) + } for { if turnctx.Err() != nil { return @@ -1946,12 +1985,12 @@ func oneTurnConnection(ctx context.Context, turnParams *turnParams, peer *net.UD out := buf[:n] if useWrap { - wrapped, wrapErr := wrapPacket(turnParams.wrapKey, out) + written, wrapErr := wrapInto(wireBuf, turnParams.wrapKey, out) if wrapErr != nil { log.Printf("[STREAM %d] WRAP failed: %v", streamID, wrapErr) return } - out = wrapped + out = wireBuf[:written] } written, err1 := relayConn.WriteTo(out, peer) @@ -2248,6 +2287,14 @@ func main() { inboundChan := make(chan *UDPPacket, 2000) go func() { + // Pointer-cache for the last seen local peer addr. Avoids the + // per-packet addr.String() allocation pair on the hot WG ingest path: + // most packets come from the same UDPAddr instance, so a pointer + // equality check covers the fast path. The slow path (new instance + // from ReadFrom for the same ip:port) does one String compare and + // then refreshes the cache. + var lastAddr net.Addr + var lastAddrStr string for { pktIface := packetPool.Get() pkt, ok := pktIface.(*UDPPacket) @@ -2260,16 +2307,13 @@ func main() { return } - // Save the local WireGuard peer address - current := activeLocalPeer.Load() - if current == nil { - activeLocalPeer.Store(addr) - } else if addrStr, ok := current.(net.Addr); ok { - if addrStr.String() != addr.String() { + if addr != lastAddr { + s := addr.String() + if s != lastAddrStr { activeLocalPeer.Store(addr) + lastAddrStr = s } - } else { - activeLocalPeer.Store(addr) + lastAddr = addr } pkt.N = nRead @@ -2919,7 +2963,8 @@ func createSmuxSession(ctx context.Context, tp *turnParams, peer *net.UDPAddr, i return nil, nil, fmt.Errorf("dial TURN (tcp): %w", err1) } cleanupFns = append(cleanupFns, func() { _ = c.Close() }) - turnConn = turn.NewSTUNConn(c) + wrappedC := &splitFirstWriteConn{Conn: c, splitAt: 6, delay: 20 * time.Millisecond} + turnConn = turn.NewSTUNConn(wrappedC) } // 3. Create TURN client and allocate relay diff --git a/client/wrap.go b/client/wrap.go index f271e13..8ca6a2c 100644 --- a/client/wrap.go +++ b/client/wrap.go @@ -4,9 +4,11 @@ package main import ( "crypto/rand" + "encoding/binary" "encoding/hex" "errors" "fmt" + "sync/atomic" "golang.org/x/crypto/chacha20" ) @@ -16,6 +18,31 @@ const ( wrapKeyLen = 32 ) +// Nonce layout: [4B session prefix | 8B atomic counter, big-endian]. +// Session prefix is randomised once at process start; counter monotonically +// increments per-packet. This eliminates the per-packet crypto/rand syscall +// (which on Linux is a getrandom() round-trip ~hundreds of ns) while keeping +// (key, nonce) pairs unique across 2^64 packets per session and 2^32 sessions +// per key. Wire format is byte-identical to the previous random-nonce scheme, +// so server-side decryption is unchanged. +var ( + nonceSessionPrefix [4]byte + nonceCounter atomic.Uint64 +) + +func init() { + // Best-effort: if rand.Read fails this early, the OS is in trouble and the + // binary is going to crash anyway during DTLS cert generation. Leaving the + // prefix zero would still keep nonces unique within this process. + _, _ = rand.Read(nonceSessionPrefix[:]) +} + +func fillWrapNonce(dst []byte) { + _ = dst[wrapNonceLen-1] // bounds-check hint + copy(dst[:4], nonceSessionPrefix[:]) + binary.BigEndian.PutUint64(dst[4:wrapNonceLen], nonceCounter.Add(1)) +} + func genWrapKeyHex() (string, error) { key := make([]byte, wrapKeyLen) if _, err := rand.Read(key); err != nil { @@ -41,20 +68,36 @@ func decodeWrapKey(enabled bool, raw string) ([]byte, error) { return key, nil } -func wrapPacket(key, payload []byte) ([]byte, error) { +// wrapInto encrypts payload into dst using the shared scheme +// [12B nonce | ChaCha20-XOR(payload)]. dst must have capacity for +// wrapNonceLen+len(payload). Returns total bytes written. Hot-path callers +// should reuse a single dst buffer to avoid per-packet allocation. +func wrapInto(dst, key, payload []byte) (int, error) { if len(key) != wrapKeyLen { - return nil, fmt.Errorf("wrap: key must be %d bytes (got %d)", wrapKeyLen, len(key)) + return 0, fmt.Errorf("wrap: key must be %d bytes (got %d)", wrapKeyLen, len(key)) } - out := make([]byte, wrapNonceLen+len(payload)) - if _, err := rand.Read(out[:wrapNonceLen]); err != nil { - return nil, fmt.Errorf("wrap: nonce gen: %w", err) + total := wrapNonceLen + len(payload) + if len(dst) < total { + return 0, errors.New("wrap: dst buffer too small") } - cipher, err := chacha20.NewUnauthenticatedCipher(key, out[:wrapNonceLen]) + fillWrapNonce(dst[:wrapNonceLen]) + cipher, err := chacha20.NewUnauthenticatedCipher(key, dst[:wrapNonceLen]) if err != nil { - return nil, fmt.Errorf("wrap: cipher init: %w", err) + return 0, fmt.Errorf("wrap: cipher init: %w", err) } - cipher.XORKeyStream(out[wrapNonceLen:], payload) - return out, nil + cipher.XORKeyStream(dst[wrapNonceLen:total], payload) + return total, nil +} + +// wrapPacket allocates a fresh wire buffer. Retained for tests and control +// paths; hot-path code should call wrapInto with a reusable scratch buffer. +func wrapPacket(key, payload []byte) ([]byte, error) { + out := make([]byte, wrapNonceLen+len(payload)) + n, err := wrapInto(out, key, payload) + if err != nil { + return nil, err + } + return out[:n], nil } func unwrapPacket(key, wire, dst []byte) (int, error) { diff --git a/go.mod b/go.mod index e6e5929..618d88c 100644 --- a/go.mod +++ b/go.mod @@ -15,6 +15,7 @@ require ( github.com/pion/turn/v5 v5.0.3 github.com/xtaci/kcp-go/v5 v5.6.18 github.com/xtaci/smux v1.5.34 + golang.org/x/crypto v0.49.0 ) require ( @@ -39,7 +40,6 @@ require ( github.com/templexxx/xorsimd v0.4.3 // indirect github.com/tjfoc/gmsm v1.4.1 // indirect github.com/wlynxg/anet v0.0.5 // indirect - golang.org/x/crypto v0.49.0 // indirect golang.org/x/mod v0.34.0 // indirect golang.org/x/net v0.52.0 // indirect golang.org/x/sync v0.20.0 // indirect diff --git a/server/main.go b/server/main.go index 1ff83f4..6b7fa7e 100644 --- a/server/main.go +++ b/server/main.go @@ -180,15 +180,17 @@ type throughputStats struct { } func (s *throughputStats) addTx(n int) { - if n > 0 { - s.tx.Add(uint64(n)) + if !isDebug || n <= 0 { + return } + s.tx.Add(uint64(n)) } func (s *throughputStats) addRx(n int) { - if n > 0 { - s.rx.Add(uint64(n)) + if !isDebug || n <= 0 { + return } + s.rx.Add(uint64(n)) } func (s *throughputStats) logEvery(ctx context.Context, label, txName, rxName string) { diff --git a/server/wrap.go b/server/wrap.go index be3d62a..3403a07 100644 --- a/server/wrap.go +++ b/server/wrap.go @@ -4,9 +4,11 @@ package main import ( "crypto/rand" + "encoding/binary" "errors" "fmt" "net" + "sync/atomic" "time" dtlsnet "github.com/pion/dtls/v3/pkg/net" @@ -19,6 +21,25 @@ const ( wrapKeyLen = 32 ) +// Nonce layout matches the client: [4B session prefix | 8B atomic counter]. +// Replaces a per-packet crypto/rand syscall (getrandom on Linux) with a +// monotonic counter; (key, nonce) uniqueness preserved within the process. +// Wire format unchanged. +var ( + nonceSessionPrefix [4]byte + nonceCounter atomic.Uint64 +) + +func init() { + _, _ = rand.Read(nonceSessionPrefix[:]) +} + +func fillWrapNonce(dst []byte) { + _ = dst[wrapNonceLen-1] + copy(dst[:4], nonceSessionPrefix[:]) + binary.BigEndian.PutUint64(dst[4:wrapNonceLen], nonceCounter.Add(1)) +} + func listenWrapped(addr *net.UDPAddr, key []byte) (dtlsnet.PacketListener, error) { if len(key) != wrapKeyLen { return nil, fmt.Errorf("wrap: key must be %d bytes (got %d)", wrapKeyLen, len(key)) @@ -78,9 +99,7 @@ func (c *wrapPacketConn) ReadFrom(p []byte) (int, net.Addr, error) { func (c *wrapPacketConn) WriteTo(p []byte, addr net.Addr) (int, error) { out := make([]byte, wrapNonceLen+len(p)) - if _, err := rand.Read(out[:wrapNonceLen]); err != nil { - return 0, fmt.Errorf("wrap: nonce gen: %w", err) - } + fillWrapNonce(out[:wrapNonceLen]) cipher, err := chacha20.NewUnauthenticatedCipher(c.key, out[:wrapNonceLen]) if err != nil { return 0, fmt.Errorf("wrap: cipher init: %w", err) diff --git a/tcputil/tcputil.go b/tcputil/tcputil.go index bb2aab2..9c1ee81 100644 --- a/tcputil/tcputil.go +++ b/tcputil/tcputil.go @@ -95,6 +95,25 @@ func envInt(name string, fallback int) int { return value } +// selectedFEC parses VK_TURN_KCP_FEC as "data:parity" (e.g. "10:3"). +// Returns 0,0 if unset/invalid (FEC disabled). Both sides must match. +func selectedFEC() (dataShards, parityShards int) { + raw := strings.TrimSpace(os.Getenv("VK_TURN_KCP_FEC")) + if raw == "" { + return 0, 0 + } + parts := strings.SplitN(raw, ":", 2) + if len(parts) != 2 { + return 0, 0 + } + d, err1 := strconv.Atoi(strings.TrimSpace(parts[0])) + p, err2 := strconv.Atoi(strings.TrimSpace(parts[1])) + if err1 != nil || err2 != nil || d <= 0 || p <= 0 { + return 0, 0 + } + return d, p +} + func envBool(name string, fallback bool) bool { raw := strings.ToLower(strings.TrimSpace(os.Getenv(name))) switch raw { @@ -158,10 +177,12 @@ func NewKCPOverDTLS(dtlsConn net.Conn, isServer bool) (*kcp.UDPSession, error) { var sess *kcp.UDPSession + dataShards, parityShards := selectedFEC() + if isServer { // Server: listen on the PacketConn and accept one session var listener *kcp.Listener - listener, err = kcp.ServeConn(block, 0, 0, pc) + listener, err = kcp.ServeConn(block, dataShards, parityShards, pc) if err != nil { return nil, err } @@ -174,7 +195,7 @@ func NewKCPOverDTLS(dtlsConn net.Conn, isServer bool) (*kcp.UDPSession, error) { } } else { // Client: dial through the PacketConn - sess, err = kcp.NewConn2(dtlsConn.RemoteAddr(), block, 0, 0, pc) + sess, err = kcp.NewConn2(dtlsConn.RemoteAddr(), block, dataShards, parityShards, pc) if err != nil { return nil, err }