perf(udprelay): in-place wrap/unwrap obf без копий payload

WrapInPlace/UnwrapInPlace работают прямо в wire-буфере: ingress читает
payload сразу за RTP-заголовком, egress отдаёт subslice расшифровки.
Убирает 2 memcpy (~1400B) и 2 буфера на пакет в obf-режиме, плюс
atomic-запись адреса пайпа теперь один раз. WrapInto/Unwrap делегируют,
wire-формат не изменён.
This commit is contained in:
samosvalishe
2026-06-01 18:13:50 +03:00
parent 4b26d6361d
commit 89c390cef5
3 changed files with 132 additions and 44 deletions
+25 -13
View File
@@ -269,18 +269,28 @@ func oneTURN(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr,
return
}
const maxPayload = 1600
wg.Go(func() {
defer turncancel()
buf := make([]byte, 1600)
var wireBuf []byte
// При obf читаем payload сразу в buf[HeaderLen:], чтобы WrapInPlace
// дописал заголовок+tag без копии payload.
var buf, readSlot []byte
if obfConn != nil {
wireBuf = make([]byte, rtpopus.MaxWire(len(buf)))
buf = make([]byte, rtpopus.MaxWire(maxPayload))
readSlot = buf[rtpopus.HeaderLen : rtpopus.HeaderLen+maxPayload]
} else {
buf = make([]byte, maxPayload)
readSlot = buf
}
// Адрес внутреннего пайпа константен; фиксируем один раз вместо
// atomic-записи на каждый пакет.
addrStored := false
for {
if turnctx.Err() != nil {
return
}
n, addr1, err1 := conn2.ReadFrom(buf)
n, addr1, err1 := conn2.ReadFrom(readSlot)
if err1 != nil {
return
}
@@ -288,16 +298,19 @@ func oneTURN(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr,
return
}
internalPipeAddr.Store(addr1)
if !addrStored {
internalPipeAddr.Store(addr1)
addrStored = true
}
out := buf[:n]
out := readSlot[:n]
if obfConn != nil {
written, wErr := obfConn.WrapInto(wireBuf, out)
written, wErr := obfConn.WrapInPlace(buf, n)
if wErr != nil {
deps.log().Errorf("[STREAM %d] OBF wrap failed: %v", streamID, wErr)
return
}
out = wireBuf[:written]
out = buf[:written]
}
written, err1 := relayConn.WriteTo(out, peer)
@@ -310,12 +323,11 @@ func oneTURN(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr,
wg.Go(func() {
defer turncancel()
readBufLen := 1600
readBufLen := maxPayload
if obfConn != nil {
readBufLen = rtpopus.MaxWire(1600)
readBufLen = rtpopus.MaxWire(maxPayload)
}
buf := make([]byte, readBufLen)
plain := make([]byte, 1600)
for {
n, _, err1 := relayConn.ReadFrom(buf)
if err1 != nil {
@@ -329,12 +341,12 @@ func oneTURN(ctx context.Context, deps *Deps, params *Params, peer *net.UDPAddr,
if addr, ok := addr1.(net.Addr); ok {
payload := buf[:n]
if obfConn != nil {
m, uErr := obfConn.Unwrap(payload, plain)
p, uErr := obfConn.UnwrapInPlace(buf[:n])
if uErr != nil {
deps.log().Errorf("[STREAM %d] OBF unwrap failed: %v (n=%d)", streamID, uErr, n)
continue
}
payload = plain[:m]
payload = p
}
st.AddRx(len(payload))
if _, err := conn2.WriteTo(payload, addr); err != nil {
+55 -31
View File
@@ -41,15 +41,19 @@ import (
)
const (
KeyLen = 32
rtpHdrLen = 12
nonceLen = 12
tagLen = 16
headerLen = rtpHdrLen + nonceLen // 24
Overhead = headerLen + tagLen // 40
rtpVersion = 0x80 // V=2, P=0, X=0, CC=0
rtpPT = 0x6F // M=0, PT=111 (opus)
tsStep = 960 // 20ms @ 48kHz
KeyLen = 32
rtpHdrLen = 12
nonceLen = 12
tagLen = 16
headerLen = rtpHdrLen + nonceLen // 24
// HeaderLen — offset, с которого начинается plaintext в wire-буфере.
// Экспонирован для in-place API (WrapInPlace/UnwrapInPlace): вызывающий
// читает payload сразу в buf[HeaderLen:], избегая копии.
HeaderLen = headerLen
Overhead = headerLen + tagLen // 40
rtpVersion = 0x80 // V=2, P=0, X=0, CC=0
rtpPT = 0x6F // M=0, PT=111 (opus)
tsStep = 960 // 20ms @ 48kHz
)
func MaxWire(payloadLen int) int { return Overhead + payloadLen }
@@ -125,41 +129,65 @@ func NewConnFromState(state *State, isServer bool) (*Conn, error) {
// WrapInto кодирует payload в dst (минимум MaxWire(len(payload)) байт)
// и возвращает число записанных wire-байт.
func (c *Conn) WrapInto(dst, payload []byte) (int, error) {
wireLen := Overhead + len(payload)
if len(dst) < wireLen {
if len(dst) < Overhead+len(payload) {
return 0, errors.New("rtpopus:dst buffer too small")
}
copy(dst[headerLen:], payload)
return c.WrapInPlace(dst, len(payload))
}
// WrapInPlace кодирует plaintext, который вызывающий уже разместил в
// buf[HeaderLen:HeaderLen+plainLen], дописывая RTP-заголовок+nonce перед ним
// и AEAD-tag после — без копии payload. buf должен вмещать MaxWire(plainLen).
// Возвращает число записанных wire-байт.
func (c *Conn) WrapInPlace(buf []byte, plainLen int) (int, error) {
wireLen := Overhead + plainLen
if len(buf) < wireLen {
return 0, errors.New("rtpopus:dst buffer too small")
}
// RTP-заголовок.
dst[0] = rtpVersion
dst[1] = rtpPT
buf[0] = rtpVersion
buf[1] = rtpPT
seq := uint16(c.seq.Add(1) - 1) //nolint:gosec // RTP sequence number is intentionally mod 2^16
binary.BigEndian.PutUint16(dst[2:4], seq)
binary.BigEndian.PutUint16(buf[2:4], seq)
ts := c.timestamp.Add(tsStep) - tsStep
binary.BigEndian.PutUint32(dst[4:8], ts)
copy(dst[8:12], c.ssrc[:])
binary.BigEndian.PutUint32(buf[4:8], ts)
copy(buf[8:12], c.ssrc[:])
// Явный nonce.
noncePos := rtpHdrLen
copy(dst[noncePos:noncePos+4], c.sessionID[:])
copy(buf[noncePos:noncePos+4], c.sessionID[:])
ctr := c.counter.Add(1) - 1
binary.BigEndian.PutUint64(dst[noncePos+4:noncePos+nonceLen], ctr)
binary.BigEndian.PutUint64(buf[noncePos+4:noncePos+nonceLen], ctr)
nonce := dst[noncePos : noncePos+nonceLen]
aad := dst[:headerLen]
nonce := buf[noncePos : noncePos+nonceLen]
aad := buf[:headerLen]
ctPos := headerLen
copy(dst[ctPos:], payload)
c.state.aead.Seal(dst[ctPos:ctPos], nonce, dst[ctPos:ctPos+len(payload)], aad)
c.state.aead.Seal(buf[ctPos:ctPos], nonce, buf[ctPos:ctPos+plainLen], aad)
return wireLen, nil
}
// Unwrap декодирует wire-пакет в dst и возвращает длину plaintext.
// AEAD открывает in-place внутри wire — вызывающий должен считать wire
// потреблённым после вызова (содержимое нельзя переиспользовать).
func (c *Conn) Unwrap(wire, dst []byte) (int, error) {
plain, err := c.UnwrapInPlace(wire)
if err != nil {
return 0, err
}
if len(plain) > len(dst) {
return 0, errors.New("rtpopus:dst buffer too small")
}
copy(dst[:len(plain)], plain)
return len(plain), nil
}
// UnwrapInPlace декодирует wire-пакет на месте и возвращает subslice plaintext
// внутри wire (без копии в отдельный буфер). AEAD открывает in-place — wire
// после вызова считается потреблённым, результат валиден до следующей записи в wire.
func (c *Conn) UnwrapInPlace(wire []byte) ([]byte, error) {
if len(wire) < Overhead {
return 0, errors.New("rtpopus:packet too short")
return nil, errors.New("rtpopus:packet too short")
}
nonce := wire[rtpHdrLen : rtpHdrLen+nonceLen]
aad := wire[:headerLen]
@@ -167,13 +195,9 @@ func (c *Conn) Unwrap(wire, dst []byte) (int, error) {
plain, err := c.state.aead.Open(ct[:0], nonce, ct, aad)
if err != nil {
return 0, fmt.Errorf("rtpopus:AEAD open: %w", err)
return nil, fmt.Errorf("rtpopus:AEAD open: %w", err)
}
if len(plain) > len(dst) {
return 0, errors.New("rtpopus:dst buffer too small")
}
copy(dst[:len(plain)], plain)
return len(plain), nil
return plain, nil
}
func GenKeyHex() (string, error) {
+52
View File
@@ -63,6 +63,58 @@ func TestConnRoundTrip(t *testing.T) {
}
}
func TestInPlaceRoundTrip(t *testing.T) {
key := bytes.Repeat([]byte{0x42}, KeyLen)
payload := []byte("dtls record bytes in-place")
client, err := NewConn(key, false)
if err != nil {
t.Fatalf("NewConn(client): %v", err)
}
server, err := NewConn(key, true)
if err != nil {
t.Fatalf("NewConn(server): %v", err)
}
// Клиент: payload уже в buf[HeaderLen:], WrapInPlace без копии.
buf := make([]byte, MaxWire(len(payload)))
copy(buf[HeaderLen:], payload)
n, err := client.WrapInPlace(buf, len(payload))
if err != nil {
t.Fatalf("WrapInPlace: %v", err)
}
if n != MaxWire(len(payload)) {
t.Fatalf("WrapInPlace len = %d, want %d", n, MaxWire(len(payload)))
}
if bytes.Contains(buf[:n], payload) {
t.Fatalf("wrapped packet contains plaintext payload")
}
// WrapInPlace и WrapInto должны давать совместимый wire (decode общим Unwrap).
dst := make([]byte, 1600)
m, err := server.Unwrap(buf[:n], dst)
if err != nil {
t.Fatalf("Unwrap: %v", err)
}
if !bytes.Equal(dst[:m], payload) {
t.Fatalf("WrapInPlace→Unwrap mismatch: got %q want %q", dst[:m], payload)
}
// Сервер: WrapInto → клиент UnwrapInPlace (subslice внутрь wire, без копии).
wire := make([]byte, MaxWire(len(payload)))
wn, err := server.WrapInto(wire, payload)
if err != nil {
t.Fatalf("server wrap: %v", err)
}
plain, err := client.UnwrapInPlace(wire[:wn])
if err != nil {
t.Fatalf("UnwrapInPlace: %v", err)
}
if !bytes.Equal(plain, payload) {
t.Fatalf("UnwrapInPlace mismatch: got %q want %q", plain, payload)
}
}
func TestRTPHeaderProgression(t *testing.T) {
key := bytes.Repeat([]byte{0x42}, KeyLen)
c, err := NewConn(key, false)