ppn/free-turn-proxy
3
0
Fork 0
mirror of https://github.com/hackdiaz-dev/free-turn-proxy.git synced 2026-10-12 09:41:53 +02:00
Code Issues Packages Projects Releases Wiki Activity

refactor(bondframe): вынести Reorder; разделить между bondclient и bondserver

Browse Source
This commit is contained in:
samosvalishe
2026-05-20 11:55:35 +03:00
parent 7b4d6be1d8
commit 4a962d8264
3 changed files with 90 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/proxy/bondclient/bondclient.go
+7 -51
View File
@@ -216,55 +216,11 @@ func writeBondFrameToNextLane(ctx context.Context, lanes []*lane, typ byte, seq
}
func (h *Handler) copyBondToTCP(ctx context.Context, connID uint64, tcpConn net.Conn, recvCh <-chan bondframe.Frame) {
pending := make(map[uint64][]byte)
var expect uint64
var finSeq *uint64
for {
if finSeq != nil && expect == *finSeq {
bondframe.CloseWrite(tcpConn, h.Deps.log().Debugf)
h.Deps.log().Debugf("[bond %d] download finished chunks=%d", connID, expect)
return
}
select {
case <-ctx.Done():
return
case f, ok := <-recvCh:
if !ok {
return
}
switch f.Type {
case bondframe.FrameData:
if len(pending) >= 1024 {
h.Deps.log().Errorf("[bond %d] pending map overflow (>1024), closing", connID)
return
}
pending[f.Seq] = f.Data
case bondframe.FrameFIN:
v := f.Seq
if finSeq == nil || v < *finSeq {
finSeq = &v
}
default:
h.Deps.log().Errorf("[bond %d] unknown frame type %d", connID, f.Type)
return
}
for {
data, ok := pending[expect]
if !ok {
break
}
delete(pending, expect)
if len(data) > 0 {
if _, err := tcpConn.Write(data); err != nil {
h.Deps.log().Errorf("[bond %d] local TCP write error: %v", connID, err)
return
}
}
expect++
}
}
}
chunks := bondframe.Reorder(ctx, tcpConn, recvCh, bondframe.ReorderHooks{
OnOverflow: func(have int) { h.Deps.log().Errorf("[bond %d] pending map overflow (>%d), closing", connID, bondframe.PendingCap) },
OnUnknownType: func(typ byte) { h.Deps.log().Errorf("[bond %d] unknown frame type %d", connID, typ) },
OnWriteError: func(err error) { h.Deps.log().Errorf("[bond %d] local TCP write error: %v", connID, err) },
OnCloseWrite: h.Deps.log().Debugf,
})
h.Deps.log().Debugf("[bond %d] download finished chunks=%d", connID, chunks)
}
internal/proxy/bondserver/bondserver.go
+7 -51
View File
@@ -265,57 +265,13 @@ func (c *serverConn) run() {
}
func (c *serverConn) copyBondToBackend(backendConn net.Conn) {
pending := make(map[uint64][]byte)
var expect uint64
var finSeq *uint64
for {
if finSeq != nil && expect == *finSeq {
bondframe.CloseWrite(backendConn, c.deps.log().Debugf)
c.deps.log().Debugf("[bond %d] upload to backend finished chunks=%d", c.id, expect)
return
}
select {
case <-c.ctx.Done():
return
case f := <-c.recvCh:
switch f.Type {
case bondframe.FrameData:
// Bound the reorder buffer so a malicious or buggy peer
// that emits seq with permanent gaps cannot exhaust memory.
// Matches bondclient.copyBondToTCP.
if len(pending) >= 1024 {
c.deps.log().Errorf("[bond %d] pending map overflow (>1024), closing", c.id)
return
}
pending[f.Seq] = f.Data
case bondframe.FrameFIN:
v := f.Seq
if finSeq == nil || v < *finSeq {
finSeq = &v
}
default:
c.deps.log().Errorf("[bond %d] unknown frame type %d", c.id, f.Type)
return
}
for {
data, ok := pending[expect]
if !ok {
break
}
delete(pending, expect)
if len(data) > 0 {
if _, err := backendConn.Write(data); err != nil {
c.deps.log().Errorf("[bond %d] backend write error: %v", c.id, err)
return
}
}
expect++
}
}
}
chunks := bondframe.Reorder(c.ctx, backendConn, c.recvCh, bondframe.ReorderHooks{
OnOverflow: func(have int) { c.deps.log().Errorf("[bond %d] pending map overflow (>%d), closing", c.id, bondframe.PendingCap) },
OnUnknownType: func(typ byte) { c.deps.log().Errorf("[bond %d] unknown frame type %d", c.id, typ) },
OnWriteError: func(err error) { c.deps.log().Errorf("[bond %d] backend write error: %v", c.id, err) },
OnCloseWrite: c.deps.log().Debugf,
})
c.deps.log().Debugf("[bond %d] upload to backend finished chunks=%d", c.id, chunks)
}
func (c *serverConn) copyBackendToBond(backendConn net.Conn) {
internal/wire/bondframe/bondframe.go
+76
View File
@@ -4,6 +4,7 @@
package bondframe
import (
"context"
"encoding/binary"
"fmt"
"io"
@@ -117,6 +118,81 @@ func ReadFrame(r io.Reader) (Frame, error) {
return f, nil
}
// PendingCap bounds the per-bond reorder buffer. A peer that emits seq with
// permanent gaps cannot grow it past this many frames.
const PendingCap = 1024
// ReorderHooks plugs caller-specific logging into Reorder. All fields may be nil.
type ReorderHooks struct {
OnOverflow func(have int)
OnUnknownType func(typ byte)
OnWriteError func(err error)
OnCloseWrite func(format string, v ...any)
}
// Reorder consumes Frames from recv, writes payloads to dst in Seq order, and
// returns the number of fully delivered chunks. It returns when:
// - the FIN seq has been reached (CloseWrite is called on dst);
// - recv is closed;
// - ctx is cancelled;
// - an unknown frame type, pending overflow, or write error fires.
func Reorder(ctx context.Context, dst net.Conn, recv <-chan Frame, h ReorderHooks) uint64 {
pending := make(map[uint64][]byte)
var expect uint64
var finSeq *uint64
for {
if finSeq != nil && expect == *finSeq {
CloseWrite(dst, h.OnCloseWrite)
return expect
}
select {
case <-ctx.Done():
return expect
case f, ok := <-recv:
if !ok {
return expect
}
switch f.Type {
case FrameData:
if len(pending) >= PendingCap {
if h.OnOverflow != nil {
h.OnOverflow(len(pending))
}
return expect
}
pending[f.Seq] = f.Data
case FrameFIN:
v := f.Seq
if finSeq == nil || v < *finSeq {
finSeq = &v
}
default:
if h.OnUnknownType != nil {
h.OnUnknownType(f.Type)
}
return expect
}
for {
data, ok := pending[expect]
if !ok {
break
}
delete(pending, expect)
if len(data) > 0 {
if _, err := dst.Write(data); err != nil {
if h.OnWriteError != nil {
h.OnWriteError(err)
}
return expect
}
}
expect++
}
}
}
}
// CloseWrite half-closes the write side of conn if the underlying type
// supports it (TCPConn, smux.Stream, …); otherwise it is a no-op. errf is
// invoked with the error if CloseWrite fails; callers typically pass a