cloud/redapricot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

break: replace muxed workers with 1:1 tunnels

Browse Source 
Worker frames are now FrameType + payload; there is no stream id.
Each player gets its own worker conn. maxTunnels (default 256)
caps concurrent tunnels. The old maxConn pool size is ignored so
existing configs do not silently admit only a handful of players.

Resume, per-direction windows, the control session, and the
DATA-only shaper stay. A dropped worker still hangs that one
player and reattaches over a fresh conn.

Add a hub-side per-IP limiter for player intents only (default
8/s, burst 16, 64 concurrent). Unmatched hostnames consume a
token; Intent 17 is never counted. 0 disables each knob.
This commit is contained in:
iceBear67
2026-08-15 18:32:51 +08:00
parent da17140583
commit 4df2560331
27 changed files with 1174 additions and 856 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/client.go
+41 -31
View File
@@ -14,8 +14,8 @@ import (
"github.com/iceBear67/redapricot/client/wire"
)
// Client is a redapricot client: it holds a control session with the hub and a
// pool of worker connections used to serve player streams.
// Client is a redapricot client: it holds a control session with the hub and
// dials one worker connection per player.
type Client struct {
cfg *Config
pskBytes []byte
@@ -25,7 +25,7 @@ type Client struct {
mappings map[string]Mapping // normalized pattern -> mapping
pool *WorkerPool
// ctx/cancel own every pool dial: Close cancels it so an in-flight dial
// ctx/cancel own every worker dial: Close cancels it so an in-flight dial
// aborts instead of holding a goroutine for the whole handshake timeout.
// The control path uses the caller's context from Start, which is the same
// shutdown signal by convention.
@@ -34,10 +34,10 @@ type Client struct {
// closing is set by Close; a conn-loss teardown checks it and closes
// streams outright rather than parking them for a reattach that is never
// coming. Allocate also consults it through the pool's own flag.
// coming. Dial also consults it through the pool's own flag.
closing atomic.Bool
streamWnd int // our advertised per-stream receive window (bytes)
streamWnd int // our advertised per-connection receive window (bytes)
shaper *Shaper // caps aggregate egress to the hub; nil when unlimited
chunk int // DATA payload cap; shrinks below DataChunkSize at low rates
@@ -78,10 +78,20 @@ func New(cfg *Config) *Client {
log.Printf("egress shaped to %d B/s (burst %d B, chunk %d B)",
bps, int64(c.shaper.burst), c.shaper.chunk)
}
c.pool = newWorkerPool(c, cfg.MaxConn)
c.pool = newWorkerPool(c, clampMaxTunnels(cfg.MaxTunnels))
return c
}
func clampMaxTunnels(n int) int {
if n < 1 {
return DefaultMaxTunnels
}
if n > MaxMaxTunnels {
return MaxMaxTunnels
}
return n
}
func clampWindow(w int) int {
if w <= 0 {
return DefaultStreamWindow
@@ -99,8 +109,8 @@ func clampWindow(w int) int {
// was negotiated during establishment.
type session struct {
fc *wire.FramedConn
peerWnd int // hub's advertised per-stream receive window
heartbeat bool // hub accepted mux-level PING/PONG on worker conns
peerWnd int // hub's advertised per-connection receive window
heartbeat bool // hub accepted connection-level PING/PONG on worker conns
resume bool // hub accepted stream resumption (§7.5)
// hubGrace is how long the hub will hang a parked player, as advertised in
// SessionReady. Zero when resumption was not negotiated.
@@ -108,7 +118,7 @@ type session struct {
}
// dialSession opens a TCP connection, performs the Intent-17 handshake, the
// Phase-A rekey, and reads SessionReady. Per-stream flow control is mandatory:
// Phase-A rekey, and reads SessionReady. Per-connection flow control is mandatory:
// a hub that does not echo the STREAM_FC flag is rejected.
//
// The whole exchange is bounded by HandshakeTimeout. A hub that accepts the
@@ -116,7 +126,7 @@ type session struct {
// behalf) must fail fast rather than park the caller forever.
//
// ctx bounds the dial: the control path passes the caller's context so a
// shutdown mid-handshake aborts the attempt, and the pool passes the client's
// shutdown mid-handshake aborts the attempt, and worker dials pass the client's
// own context so Close cancels in-flight dials. After the dial, a cancelled
// ctx keeps aborting by closing the conn underneath the deadline-bounded
// handshake I/O.
@@ -158,7 +168,7 @@ func (c *Client) dialSession(ctx context.Context, magic byte) (sess *session, er
)
// 3. Rekey frame (Phase A), including the mandatory feature flags and our
// per-stream receive window.
// per-connection receive window.
rnd := make([]byte, 16)
if _, err := crand.Read(rnd); err != nil {
return nil, err
@@ -186,7 +196,7 @@ func (c *Client) dialSession(ctx context.Context, magic byte) (sess *session, er
)
// 5. SessionReady: the type byte followed by the hub's accepted flags and
// its per-stream receive window. Both are required.
// its per-connection receive window. Both are required.
payload, err := fc.ReadFrame()
if err != nil {
return nil, err
@@ -198,7 +208,7 @@ func (c *Client) dialSession(ctx context.Context, magic byte) (sess *session, er
flags, ferr := r.VarInt()
hubWnd, werr := r.VarInt()
if ferr != nil || werr != nil || flags&FlagStreamFC == 0 || hubWnd <= 0 {
return nil, fmt.Errorf("hub did not accept per-stream flow control (unsupported hub version?)")
return nil, fmt.Errorf("hub did not accept per-connection flow control (unsupported hub version?)")
}
if hubWnd > MaxStreamWindow {
hubWnd = MaxStreamWindow
@@ -398,8 +408,8 @@ func (c *Client) pingLoop(ctx context.Context, ctrl *ctrlSession) {
}
}
// handleControlRequest reacts to a matched player: allocate a worker stream,
// SYN it, and bridge it to the mapped destination.
// handleControlRequest reacts to a matched player: dial a dedicated worker
// conn, SYN it, and bridge it to the mapped destination.
func (c *Client) handleControlRequest(cid []byte, pattern, ip string, port int) {
mapping, ok := c.mappings[NormalizeAddress(pattern)]
if !ok {
@@ -408,37 +418,37 @@ func (c *Client) handleControlRequest(cid []byte, pattern, ip string, port int)
}
log.Printf("player %s:%d joined via pattern %q -> %s", ip, port, pattern, mapping.Destination)
// Allocate and publish must agree on a live conn: Allocate hands out a
// (conn, sid) pair that can die before we register on it, which would strand
// the stream in a map nothing iterates. registerStream reports that, and we
// simply pick another conn.
// Dial and attach must agree on a live conn: Dial hands out a conn that can
// die before we attach on it, which would strand the stream on a conn
// nothing iterates. attach reports that, and we simply dial another.
var st *Stream
var lg *leg
for attempt := 0; attempt < allocateAttempts; attempt++ {
wc, sid, err := c.pool.Allocate()
var wc *WorkerConn
for attempt := 0; attempt < dialAttempts; attempt++ {
var err error
wc, err = c.pool.Dial()
if err != nil {
log.Printf("worker allocate failed: %v", err)
log.Printf("worker dial failed: %v", err)
return
}
st = newStream(c, wc, sid, cid, mapping, ip, port)
// Register before SYN so inbound DATA can never race ahead of the table,
st = newStream(c, wc, cid, mapping, ip, port)
// Attach before SYN so inbound DATA can never race ahead of the binding,
// and start the pumps before the (bounded) SYN write so a failed or slow
// SYN cannot strand a stream that nothing would ever tear down.
if wc.registerStream(sid, st) {
lg = st.conn()
if wc.attach(st) {
break
}
_ = wc.fc.Close()
st = nil
}
if st == nil {
log.Printf("worker allocate failed: no live conn after %d attempts", allocateAttempts)
log.Printf("worker dial failed: no live conn after %d attempts", dialAttempts)
return
}
go st.writeLoop()
go st.run()
if err := lg.wc.sendSyn(lg.sid, cid); err != nil {
log.Printf("stream %s: SYN failed: %v", lg, err)
if err := wc.sendSyn(cid); err != nil {
log.Printf("stream %s: SYN failed: %v", st.name(), err)
st.teardown(false)
}
}
@@ -452,7 +462,7 @@ func (c *Client) WorkerConnCount() int { return c.pool.count() }
// open.
func (c *Client) Close() {
c.closing.Store(true)
c.cancel() // aborts in-flight pool dials
c.cancel() // aborts in-flight worker dials
c.mu.Lock()
fc := c.ctrl
c.mu.Unlock()
client/config.example.json
+1 -1
View File
@@ -1,7 +1,7 @@
{
"server": "hub.example.com:25565",
"psk": "change-me-to-a-long-random-passphrase",
"maxConn": 4,
"maxTunnels": 256,
"pingIntervalMs": 20000,
"streamWindowBytes": 262144,
"maxBandwidth": "",
client/config.go
+41 -34
View File
@@ -43,19 +43,15 @@ const (
MuxWnd = 0x04
MuxPing = 0x05
MuxPong = 0x06
// MuxResume reattaches a parked stream to this conn (CID + our accepted
// MuxResume reattaches a parked player to this conn (CID + our accepted
// offset); MuxResumeAck carries the hub's accepted offset and a fresh CID.
MuxResume = 0x07
MuxResumeAck = 0x08
// MuxCtlSid is the reserved stream id carrying connection-scoped mux frames
// (PING/PONG). Real streams are numbered from 1.
MuxCtlSid = 0
// RST reason codes (optional trailing byte; absence means "unspecified").
// Distinguishing them matters for resume: an unknown stream is terminal,
// while "already bound" means the hub has the stream on another conn — the
// reattach retries until that bind dies and the hub re-parks the stream.
// while "already bound" means the hub has the player on another conn — the
// reattach retries until that bind dies and the hub re-parks the player.
RstUnspecified = 0x00
RstUnknownStream = 0x01
RstAlreadyBound = 0x02
@@ -65,33 +61,33 @@ const (
FrameError = 0x7F
// SaturationThreshold caps how many streams share one worker conn once the
// pool has grown to maxConn. Below maxConn the pool grows first (§7.1), so a
// single connection is never a shared point of failure for every player.
SaturationThreshold = 8
// DefaultMaxTunnels / MaxMaxTunnels bound concurrent 1:1 worker conns.
// The old mux-era maxConn cap of 8 would silently become "8 players".
DefaultMaxTunnels = 256
MaxMaxTunnels = 4096
// Session-establishment feature flags (trailing VarInt on the Rekey message).
FlagStreamFC = 0x01
// FlagWorkerHeartbeat enables mux-level PING/PONG on worker conns. Without
// it a worker conn whose path is silently blackholed (NAT/conntrack drop,
// firewall) is never detected: the read loop parks forever, the dead conn
// stays in the pool, and no player can be served until the client restarts.
// FlagWorkerHeartbeat enables connection-level PING/PONG on worker conns.
// Without it a worker conn whose path is silently blackholed (NAT/conntrack
// drop, firewall) is never detected: the read loop parks forever and that
// player is stuck until the client restarts.
FlagWorkerHeartbeat = 0x02
// FlagStreamResume enables stream resumption (PROTOCOL.md §7.5): a worker
// conn drop parks its streams instead of killing them, the hub hangs the
// player sockets, and the client reattaches each stream byte-exactly over a
// fresh conn. Negotiated, so either side may decline and get today's
// behaviour (immediate teardown) unchanged.
// conn drop parks the player instead of killing them, the hub hangs the
// player socket, and the client reattaches byte-exactly over a fresh conn.
// Negotiated, so either side may decline and get today's behaviour
// (immediate teardown) unchanged.
FlagStreamResume = 0x04
// Per-stream flow-control window bounds (bytes). The advertised window is the
// receiver's promise of how much un-credited DATA it will buffer per stream.
// Per-connection flow-control window bounds (bytes). The advertised window
// is the receiver's promise of how much un-credited DATA it will buffer.
DefaultStreamWindow = 256 * 1024
MinStreamWindow = 32 * 1024
MaxStreamWindow = 8 << 20
// DataChunkSize caps a single DATA frame's payload so no stream monopolizes
// the shared worker connection for long.
// DataChunkSize caps a single DATA frame's payload so one write cannot
// occupy the link for a full 1-MiB frame.
DataChunkSize = 32 * 1024
)
@@ -211,14 +207,20 @@ type Mapping struct {
// Config is the client configuration (PROTOCOL.md §9.2).
type Config struct {
Server string `json:"server"`
PSK string `json:"psk"`
MaxConn int `json:"maxConn"`
PingIntervalMs int `json:"pingIntervalMs"`
StreamWindowBytes int `json:"streamWindowBytes"` // per-stream receive window; 0 = default
Server string `json:"server"`
PSK string `json:"psk"`
// MaxTunnels is the max concurrent 1:1 worker connections (PROTOCOL.md §7.1).
// 0 means the default. Clamped to [1, 4096].
MaxTunnels int `json:"maxTunnels"`
// MaxConn is the retired mux-era pool size. Ignored when loading a file:
// honouring a value of 4 as a player cap would silently break existing
// configs. Tests that construct a Config should set MaxTunnels instead.
MaxConn int `json:"maxConn"`
PingIntervalMs int `json:"pingIntervalMs"`
StreamWindowBytes int `json:"streamWindowBytes"` // per-connection receive window; 0 = default
// MaxBandwidth caps what the client sends to the hub, aggregated over every
// stream on every worker conn — the direction that carries the game server's
// output to the players, and the one a residential uplink runs out of first.
// worker conn — the direction that carries the game server's output to the
// players, and the one a residential uplink runs out of first.
// Empty means no limit. See parseBandwidth for the accepted syntax.
MaxBandwidth string `json:"maxBandwidth"`
// StreamResume enables stream resumption (PROTOCOL.md §7.5). A pointer so an
@@ -310,11 +312,16 @@ func LoadConfig(path string) (*Config, error) {
if c.PSK == "" {
return nil, fmt.Errorf("psk is required")
}
if c.MaxConn < 1 {
c.MaxConn = 1
if c.MaxConn != 0 && c.MaxTunnels == 0 {
// Old mux pool size. Must not become the player cap: a previously-working
// maxConn: 4 would admit only four players.
fmt.Fprintf(os.Stderr, "redapricot-client: maxConn is ignored (it was the mux pool size); use maxTunnels (default %d)\n", DefaultMaxTunnels)
}
if c.MaxConn > 8 {
c.MaxConn = 8
if c.MaxTunnels < 1 {
c.MaxTunnels = DefaultMaxTunnels
}
if c.MaxTunnels > MaxMaxTunnels {
c.MaxTunnels = MaxMaxTunnels
}
if c.PingIntervalMs <= 0 {
c.PingIntervalMs = DefaultPingIntervalMs
client/pool_test.go
+19 -47
View File
@@ -40,24 +40,25 @@ func stalledHub(t *testing.T) string {
return ln.Addr().String()
}
// TestAllocateDoesNotWedgePoolOnStalledHub is the regression guard for the
// worst failure mode found in the stability audit: Allocate used to dial while
// holding the pool mutex, and the handshake read had no deadline. One
// TestDialDoesNotWedgeOnStalledHub is the regression guard for the worst
// failure mode found in the stability audit: Dial used to share a single
// in-flight handshake, and the handshake read had no deadline. One
// unresponsive hub therefore parked every present and future allocation
// forever, so no player could be served again until the process restarted.
func TestAllocateDoesNotWedgePoolOnStalledHub(t *testing.T) {
// forever. 1:1 dials independently, but each must still fail on its own
// HandshakeTimeout rather than block the other.
func TestDialDoesNotWedgeOnStalledHub(t *testing.T) {
c := New(&Config{
Server: stalledHub(t),
PSK: "pool-test",
MaxConn: 8,
MaxTunnels: 8,
PingIntervalMs: 20000,
Mappings: []Mapping{{Pattern: "mc.local", Destination: "127.0.0.1:1"}},
})
done := make(chan error, 2)
go func() { _, _, err := c.pool.Allocate(); done <- err }()
go func() { _, err := c.pool.Dial(); done <- err }()
time.Sleep(200 * time.Millisecond) // let the first caller get into the dial
go func() { _, _, err := c.pool.Allocate(); done <- err }()
go func() { _, err := c.pool.Dial(); done <- err }()
// Both must give up on their own; neither may be stuck behind the other.
limit := time.After(HandshakeTimeout + 15*time.Second)
@@ -65,50 +66,21 @@ func TestAllocateDoesNotWedgePoolOnStalledHub(t *testing.T) {
select {
case err := <-done:
if err == nil {
t.Fatal("Allocate succeeded against a hub that never answers")
t.Fatal("Dial succeeded against a hub that never answers")
}
case <-limit:
t.Fatalf("Allocate #%d never returned: the pool is wedged again", i+1)
t.Fatalf("Dial #%d never returned: a stalled hub wedged the other caller", i+1)
}
}
}
// TestAllocateSpreadsAcrossConns guards the allocation rule: the pool must fan
// out to maxConn before stacking streams, so a single worker connection is
// never the shared point of failure for every player. Seven players used to all
// land on one conn, which meant one dead TCP connection dropped everybody.
func TestAllocateSpreadsAcrossConns(t *testing.T) {
p := &WorkerPool{maxConn: 4}
p.cond = sync.NewCond(&p.mu)
newConn := func() *WorkerConn {
return &WorkerConn{pool: p, streams: make(map[int]*Stream), nextSid: 1, done: make(chan struct{})}
}
p.conns = []*WorkerConn{newConn()}
if !p.conns[0].registerStream(1, &Stream{}) {
t.Fatal("registerStream refused on a live conn")
}
// One conn holding a stream, pool below maxConn: growth is warranted.
_, bestCount := p.leastLoadedLocked()
if bestCount < StreamsBeforeGrowing {
t.Fatalf("a conn with %d stream(s) should trigger growth", bestCount)
}
// Once the pool is at maxConn, growth stops and streams stack on the
// least-loaded conn instead.
for len(p.conns) < p.maxConn {
p.conns = append(p.conns, newConn())
}
best, bestCount := p.leastLoadedLocked()
if bestCount != 0 {
t.Fatalf("expected an empty conn to be least-loaded, got %d streams", bestCount)
}
p.maybeGrowLocked(bestCount)
if p.dialing != 0 {
t.Fatalf("pool dialed past maxConn=%d", p.maxConn)
}
if best == nil {
t.Fatal("no conn selected")
// TestDialRespectsMaxTunnels pins the concurrency cap: once live+dialing
// equals maxTunnels, further Dial calls fail immediately rather than stacking.
func TestDialRespectsMaxTunnels(t *testing.T) {
p := newWorkerPool(&Client{}, 2)
p.conns[&WorkerConn{id: 1}] = struct{}{}
p.conns[&WorkerConn{id: 2}] = struct{}{}
if _, err := p.Dial(); err != errTooManyTunnels {
t.Fatalf("Dial at cap: got %v, want %v", err, errTooManyTunnels)
}
}
client/resume.go
+41 -41
View File
@@ -10,11 +10,10 @@ import (
// Stream resumption (PROTOCOL.md §7.5).
//
// A worker conn carries many players but is only the middle leg of each: when
// it dies, both terminal sockets are usually still perfectly healthy. Tearing
// the streams down therefore throws away working connections because a
// replaceable transport failed — one conntrack expiry disconnects everyone on
// that conn.
// A worker conn is only the middle leg of the player it carries: when it dies,
// both terminal sockets are usually still perfectly healthy. Tearing the
// tunnel down therefore throws away working connections because a replaceable
// transport failed.
//
// Instead the stream parks: the destination socket stays open, the hub hangs the
// player socket, and the client reattaches over a fresh conn. Correctness rests
@@ -104,7 +103,7 @@ func (s *Stream) resumeLoop(grace time.Duration) {
return
}
// errResumeRaced falls through to the retry below. The hub has this
// stream bound to a conn that is not ours — a half-open conn whose
// player bound to a conn that is not ours — a half-open conn whose
// death the hub has not yet learned, or a bind left behind by a racing
// attempt on a now-dead conn. There is no other live attempt: park is
// the only resumeLoop starter and it refuses to double-start. Retrying
@@ -131,10 +130,10 @@ func (s *Stream) resumeLoop(grace time.Duration) {
s.teardown(false)
}
// tryResume performs one reattach attempt: find a live conn, claim a stream id
// on it, send RESUME, and replay from wherever the hub says it got to.
// tryResume performs one reattach attempt: dial a fresh worker conn, send
// RESUME, and replay from wherever the hub says it got to.
func (s *Stream) tryResume() error {
wc, sid, err := s.allocateForResume()
wc, err := s.allocateForResume()
if err != nil {
return err
}
@@ -147,10 +146,10 @@ func (s *Stream) tryResume() error {
s.resumeWait = wait
s.mu.Unlock()
msg := wire.NewWriter().U8(MuxResume).VarInt(sid).Bytes(cid).
msg := wire.NewWriter().U8(MuxResume).Bytes(cid).
I64(accepted).I64(delivered).Out()
if err := wc.fc.WriteFrame(msg); err != nil {
s.abandonAttempt(wc, sid)
s.abandonAttempt(wc)
return err
}
@@ -158,47 +157,51 @@ func (s *Stream) tryResume() error {
select {
case res = <-wait:
case <-s.done:
// Torn down while waiting. teardown only deregisters the leg the stream
// Torn down while waiting. teardown only detaches the conn the stream
// was bound to, which is not this one, so the claim made above has to be
// withdrawn here or it stays in the new conn's table forever.
s.abandonAttempt(wc, sid)
// withdrawn here or the new conn stays bound forever.
s.abandonAttempt(wc)
return errResumeRefused
case <-time.After(ResumeAckTimeout):
s.abandonAttempt(wc, sid)
s.abandonAttempt(wc)
return errResumeTimeout
}
if res.err != nil {
s.abandonAttempt(wc, sid)
s.abandonAttempt(wc)
return res.err
}
return s.completeResume(wc, sid, res)
return s.completeResume(wc, res)
}
// allocateForResume picks a live conn that will honour a reattach.
func (s *Stream) allocateForResume() (*WorkerConn, int, error) {
for attempt := 0; attempt < allocateAttempts; attempt++ {
wc, sid, err := s.client.pool.Allocate()
// allocateForResume dials a dedicated worker conn that will honour a reattach.
// 1:1: this must never land on someone else's tunnel.
func (s *Stream) allocateForResume() (*WorkerConn, error) {
for attempt := 0; attempt < dialAttempts; attempt++ {
wc, err := s.client.pool.Dial()
if err != nil {
return nil, 0, err
return nil, err
}
// Re-checked per conn, not assumed from the dead one: this may be a
// different or restarted hub. Sending RESUME to a hub that does not know
// the frame type would hang the player for the rest of the grace waiting
// for an answer that is never coming.
if !wc.resume {
return nil, 0, errResumeNoResume
_ = wc.fc.Close()
return nil, errResumeNoResume
}
if wc.registerStream(sid, s) {
return wc, sid, nil
if wc.attach(s) {
return wc, nil
}
_ = wc.fc.Close()
}
return nil, 0, errResumeRefused
return nil, errResumeRefused
}
// abandonAttempt withdraws a failed attempt from the conn it was made on, so a
// retry can never leave two RESUMEs outstanding for one stream.
func (s *Stream) abandonAttempt(wc *WorkerConn, sid int) {
wc.removeStream(sid)
// abandonAttempt withdraws a failed attempt from the conn it was made on and
// closes that conn — 1:1, it exists only for this attempt.
func (s *Stream) abandonAttempt(wc *WorkerConn) {
wc.detach()
_ = wc.fc.Close()
s.mu.Lock()
s.resumeWait = nil
s.mu.Unlock()
@@ -206,7 +209,7 @@ func (s *Stream) abandonAttempt(wc *WorkerConn, sid int) {
// completeResume rebinds the stream to its new conn and replays what the hub is
// missing, holding sendMu throughout so live traffic cannot overtake the replay.
func (s *Stream) completeResume(wc *WorkerConn, sid int, res resumeResult) error {
func (s *Stream) completeResume(wc *WorkerConn, res resumeResult) error {
s.sendMu.Lock()
// Delivery is a strictly stronger fact than credit — the hub only credits what
// it has delivered — so the reported offset can be adopted wholesale. Doing so
@@ -218,7 +221,7 @@ func (s *Stream) completeResume(wc *WorkerConn, sid int, res resumeResult) error
replay := s.un.from(res.accepted)
if replay == nil {
s.sendMu.Unlock()
s.abandonAttempt(wc, sid)
s.abandonAttempt(wc)
return errResumeTooOld
}
// Three offsets, three jobs, and conflating any two of them breaks something
@@ -235,10 +238,8 @@ func (s *Stream) completeResume(wc *WorkerConn, sid int, res resumeResult) error
outstanding := s.un.length()
replayed := s.un.end() - res.accepted
// Publish the new binding before any frame goes out on it, and as one value:
// stream ids restart at 1 per conn, so a half-updated pair would address a
// different player's stream.
s.leg.Store(&leg{wc: wc, sid: sid})
// Publish the new binding before any frame goes out on it.
s.wc.Store(wc)
s.mu.Lock()
// Restated, not patched. The window is a delta ledger and the outage tore a
@@ -251,7 +252,6 @@ func (s *Stream) completeResume(wc *WorkerConn, sid int, res resumeResult) error
// Symmetrically, our own pending credit is discarded rather than flushed:
// the delivered offset we reported already tells the hub everything those
// deltas would have, and sending both would grant the same bytes twice.
// Counting resumes from this baseline.
s.consumed = 0
if len(res.cid) == CIDLen {
s.cid = res.cid // fresh capability, so a CID is never reusable twice
@@ -267,9 +267,9 @@ func (s *Stream) completeResume(wc *WorkerConn, sid int, res resumeResult) error
if n > s.client.chunk {
n = s.client.chunk
}
if err := wc.sendData(sid, replay[:n]); err != nil {
if err := wc.sendData(replay[:n]); err != nil {
// The conn died mid-replay. The stream is still resumable, but not
// from this leg — put it back in the parked state before returning
// from this conn — put it back in the parked state before returning
// so the conn's teardown takes park()'s already-branch instead of
// starting a second resumeLoop. The loop we came from keeps
// retrying with the fresh CID, which the hub re-parked alongside
@@ -300,11 +300,11 @@ func (s *Stream) completeResume(wc *WorkerConn, sid int, res resumeResult) error
// A destination that closed while we were parked owed the hub a FIN that had
// nowhere to go at the time.
if owedFin {
wc.sendFin(sid)
wc.sendFin()
s.teardown(false)
return nil
}
log.Printf("stream conn%d/sid%d resumed (%d bytes replayed, %d outstanding)", wc.id, sid, replayed, outstanding)
log.Printf("stream %s resumed (%d bytes replayed, %d outstanding)", s.name(), replayed, outstanding)
return nil
}
client/stats.go
+16 -18
View File
@@ -57,11 +57,9 @@ type connStats struct {
framesOut atomic.Int64
writeErrs atomic.Int64
// Round-trip time of the mux heartbeat. The probe already carries a
// Round-trip time of the worker heartbeat. The probe already carries a
// timestamp that the peer echoes and both sides currently throw away, so
// this measures tunnel latency for no added cost — and it is the best signal
// available for head-of-line blocking, where one stream's backlog delays
// every other stream sharing the connection.
// this measures tunnel latency for no added cost.
mu sync.Mutex
rttLast time.Duration
rttMin time.Duration
@@ -137,22 +135,19 @@ func (c *Client) StatsLine() string {
live, parked := 0, 0
for _, wc := range conns {
wc.mu.Lock()
streams := make([]*Stream, 0, len(wc.streams))
for _, s := range wc.streams {
streams = append(streams, s)
}
wc.mu.Unlock()
live += len(streams)
for _, s := range streams {
s.mu.Lock()
if s.parked {
st := wc.getStream()
bound := 0
if st != nil {
bound = 1
live++
st.mu.Lock()
if st.parked {
parked++
}
s.mu.Unlock()
st.mu.Unlock()
}
fmt.Fprintf(&b, " | conn%d streams=%d", wc.id, len(streams))
fmt.Fprintf(&b, " | conn%d bound=%d", wc.id, bound)
if cs := wc.stats; cs != nil {
_, mn, avg, mx := cs.rtt()
fmt.Fprintf(&b, " frames=%d/%d rtt=%s/%s/%s",
@@ -162,7 +157,7 @@ func (c *Client) StatsLine() string {
}
}
}
fmt.Fprintf(&b, " | streams=%d parked=%d", live, parked)
fmt.Fprintf(&b, " | tunnels=%d parked=%d", live, parked)
return b.String()
}
@@ -189,7 +184,10 @@ func (s *Stream) logSummary() {
func (p *WorkerPool) snapshot() []*WorkerConn {
p.mu.Lock()
conns := append([]*WorkerConn(nil), p.conns...)
conns := make([]*WorkerConn, 0, len(p.conns))
for wc := range p.conns {
conns = append(conns, wc)
}
p.mu.Unlock()
sort.Slice(conns, func(i, j int) bool { return conns[i].id < conns[j].id })
return conns
client/wire/framedconn.go
+3 -3
View File
@@ -15,9 +15,9 @@ import (
const MaxFrame = 1 << 20
// WriteTimeout bounds a single frame write. A peer that stops reading must not
// be able to park every stream on the connection inside WriteFrame forever: the
// write mutex is held for the whole socket write, so one stalled write would
// otherwise wedge the entire multiplexed connection.
// be able to park this connection inside WriteFrame forever: the write mutex
// is held for the whole socket write, so one stalled write would otherwise
// wedge liveness (PING/PONG) and WND/FIN on this tunnel.
const WriteTimeout = 30 * time.Second
var (
client/worker.go
+172 -291
View File
@@ -12,166 +12,87 @@ import (
"github.com/iceBear67/redapricot/client/wire"
)
// errPoolClosed is returned by Allocate after Close: the pool is shutting down
// errPoolClosed is returned by Dial after Close: the set is shutting down
// and must not start new dials, so a caller (handleControlRequest,
// allocateForResume) gives up rather than wait on a cond no one will satisfy.
var errPoolClosed = errors.New("worker pool closed")
// allocateForResume) gives up rather than waiting on a hub that will never
// be used.
var errPoolClosed = errors.New("worker set closed")
// StreamsBeforeGrowing is how many streams a worker conn may carry before the
// pool starts opening another one. Set to 1 so the pool fans out to maxConn
// under load *before* stacking streams: concentrating every player on a single
// TCP connection makes that connection a shared point of failure, which is
// exactly how a whole server's worth of players used to drop at once.
const StreamsBeforeGrowing = 1
// errTooManyTunnels is returned when live+in-flight worker conns already
// equal maxTunnels. The ControlRequest is dropped; the hub closes the player
// when pendingTimeoutMs fires.
var errTooManyTunnels = errors.New("maxTunnels reached")
// allocateAttempts bounds how many times a caller retries Allocate when the
// conn it was handed dies before the stream could be registered on it. The race
// is narrow and each retry picks a different conn, so a small bound is enough;
// an unbounded loop would spin against a hub that is refusing every connection.
const allocateAttempts = 3
// dialAttempts bounds how many times a caller retries Dial when the conn it
// was handed dies before the stream could be attached to it. The race is
// narrow; an unbounded loop would spin against a hub that is refusing every
// connection.
const dialAttempts = 3
// WorkerPool manages up to maxConn worker connections and allocates streams
// using the least-loaded strategy (PROTOCOL.md §7.1).
// WorkerPool tracks live 1:1 worker connections up to maxTunnels
// (PROTOCOL.md §7.1). There is no least-loaded placement and no sharing:
// every player gets its own TCP connection.
type WorkerPool struct {
client *Client
maxConn int
client *Client
maxTunnels int
connSeq atomic.Int64 // conn ids, for log correlation
mu sync.Mutex
cond *sync.Cond
conns []*WorkerConn
dialing int // dials currently in flight (foreground + background)
dialGen uint64
dialErr error // most recent dial failure
closed bool // closeAll ran; no new conns may join the pool
conns map[*WorkerConn]struct{}
dialing int // dials currently in flight
closed bool // closeAll ran; no new conns may join
}
func newWorkerPool(c *Client, maxConn int) *WorkerPool {
p := &WorkerPool{client: c, maxConn: maxConn}
p.cond = sync.NewCond(&p.mu)
return p
func newWorkerPool(c *Client, maxTunnels int) *WorkerPool {
return &WorkerPool{
client: c,
maxTunnels: maxTunnels,
conns: make(map[*WorkerConn]struct{}),
}
}
// Allocate returns a worker conn and a fresh stream id to place a new stream on.
// Dial opens a dedicated worker conn for one player.
//
// A dial is never performed while holding p.mu: session establishment involves
// network I/O, and holding the pool lock across it would park every other
// player behind one unresponsive hub. When the pool is empty exactly one caller
// dials and the rest wait on the condition variable; when the pool is merely
// below maxConn, growth happens in the background and the caller is served
// immediately by an existing conn.
func (p *WorkerPool) Allocate() (*WorkerConn, int, error) {
// network I/O, and holding the lock across it would park every other player
// behind one unresponsive hub. Each caller dials independently; unlike the
// old mux pool there is no shared conn to wait for.
func (p *WorkerPool) Dial() (*WorkerConn, error) {
p.mu.Lock()
for {
if p.closed {
// Close won. No new conn may join the pool, so no stream may be
// placed; waiting on the cond could only be satisfied by a dial we
// must not start.
p.mu.Unlock()
return nil, 0, errPoolClosed
}
best, bestCount := p.leastLoadedLocked()
if best != nil {
p.maybeGrowLocked(bestCount)
p.mu.Unlock()
return best, best.newSid(), nil
}
if p.dialing > 0 {
// Someone is already dialing the first conn; wait for it rather
// than piling up redundant connections.
gen := p.dialGen
p.cond.Wait()
if len(p.conns) == 0 && p.dialGen != gen && p.dialErr != nil {
err := p.dialErr
p.mu.Unlock()
return nil, 0, err
}
continue
}
p.dialing++
p.mu.Unlock()
wc, err := p.dialWorker()
p.mu.Lock()
p.dialing--
p.dialGen++
p.dialErr = err
if err != nil {
p.cond.Broadcast()
p.mu.Unlock()
return nil, 0, err
}
if p.closed {
// Close raced this dial: the conn must not enter the pool. Closing
// it here, under p.mu, is a raw socket close — fine, and it makes
// the shutdown atomic with the pool state.
p.cond.Broadcast()
p.mu.Unlock()
_ = wc.fc.Close()
return nil, 0, errPoolClosed
}
p.conns = append(p.conns, wc)
p.cond.Broadcast()
}
}
// leastLoadedLocked returns the worker conn carrying the fewest streams.
func (p *WorkerPool) leastLoadedLocked() (*WorkerConn, int) {
var best *WorkerConn
bestCount := 0
for _, wc := range p.conns {
n := wc.streamCount()
if best == nil || n < bestCount {
best = wc
bestCount = n
}
}
return best, bestCount
}
// maybeGrowLocked opens one more worker conn in the background when the pool is
// below maxConn and the least-loaded conn is already carrying streams. The
// caller does not wait for it: it keeps using the conn it already has, and the
// new one picks up subsequent players.
func (p *WorkerPool) maybeGrowLocked(bestCount int) {
if bestCount < StreamsBeforeGrowing {
return
}
if p.closed {
return // shutdown; do not start dials nobody will join the pool
p.mu.Unlock()
return nil, errPoolClosed
}
if len(p.conns)+p.dialing >= p.maxConn {
if bestCount > SaturationThreshold {
log.Printf("worker pool at maxConn=%d with %d streams on the least-loaded conn", p.maxConn, bestCount)
}
return
if len(p.conns)+p.dialing >= p.maxTunnels {
p.mu.Unlock()
return nil, errTooManyTunnels
}
p.dialing++
go func() {
wc, err := p.dialWorker()
var surplus *WorkerConn
p.mu.Lock()
p.dialing--
p.dialGen++
p.dialErr = err
switch {
case err != nil:
case p.closed:
surplus = wc // Close raced this background dial
case len(p.conns) < p.maxConn:
p.conns = append(p.conns, wc)
default:
surplus = wc // raced with another dial
}
p.cond.Broadcast()
p.mu.Unlock()
wc, err := p.dialWorker()
p.mu.Lock()
p.dialing--
if err != nil {
p.mu.Unlock()
if err != nil {
log.Printf("worker pool: background dial failed: %v", err)
}
if surplus != nil {
_ = surplus.fc.Close()
}
}()
return nil, err
}
if p.closed {
// Close raced this dial: the conn must not enter the set.
p.mu.Unlock()
_ = wc.fc.Close()
return nil, errPoolClosed
}
if len(p.conns) >= p.maxTunnels {
p.mu.Unlock()
_ = wc.fc.Close()
return nil, errTooManyTunnels
}
p.conns[wc] = struct{}{}
p.mu.Unlock()
return wc, nil
}
// dialWorker establishes one worker conn. It must be called without p.mu held.
@@ -188,8 +109,6 @@ func (p *WorkerPool) dialWorker() (*WorkerConn, error) {
resume: sess.resume,
grace: p.client.resumeGrace(sess.hubGrace),
id: int(p.connSeq.Add(1)),
streams: make(map[int]*Stream),
nextSid: 1,
done: make(chan struct{}),
}
if p.client.statsOn() {
@@ -200,7 +119,7 @@ func (p *WorkerPool) dialWorker() (*WorkerConn, error) {
if sess.heartbeat {
go wc.heartbeatLoop(p.client.cfg.pingInterval(), p.client.cfg.heartbeatTimeout())
} else {
log.Printf("worker conn: hub does not support the mux heartbeat; " +
log.Printf("worker conn: hub does not support the worker heartbeat; " +
"a silently dropped path will only be caught by TCP keepalive")
}
log.Printf("opened worker conn (send window %d, recv window %d, heartbeat %v, resume %v)",
@@ -217,41 +136,32 @@ func (p *WorkerPool) count() int {
func (p *WorkerPool) remove(wc *WorkerConn) {
p.mu.Lock()
defer p.mu.Unlock()
for i, c := range p.conns {
if c == wc {
p.conns = append(p.conns[:i], p.conns[i+1:]...)
// A waiter parked on an empty pool must re-evaluate: it may now
// need to dial rather than keep waiting for this conn.
p.cond.Broadcast()
return
}
}
delete(p.conns, wc)
}
func (p *WorkerPool) closeAll() {
p.mu.Lock()
p.closed = true
conns := append([]*WorkerConn(nil), p.conns...)
// Wake waiters parked in Allocate: the closed flag they re-check is the
// only thing that can release them now that no conn will ever join.
p.cond.Broadcast()
conns := make([]*WorkerConn, 0, len(p.conns))
for wc := range p.conns {
conns = append(conns, wc)
}
p.mu.Unlock()
for _, wc := range conns {
_ = wc.fc.Close()
}
}
// WorkerConn is one multiplexed worker connection to the hub.
//
// Only genuinely per-connection state lives here. Client-wide values (the
// shaper, the advertised receive window, the DATA chunk cap) belong to Client:
// reading them through a connection pointer would make every such read a
// re-parenting hazard once a stream can migrate between conns.
// WorkerConn is one 1:1 worker connection to the hub: it carries exactly one
// player. Only genuinely per-connection state lives here. Client-wide values
// (the shaper, the advertised receive window, the DATA chunk cap) belong to
// Client: reading them through a connection pointer would make every such
// read a re-parenting hazard once a stream can migrate between conns.
type WorkerConn struct {
pool *WorkerPool
fc *wire.FramedConn
sendWndInit int // hub's advertised per-stream receive window (our send budget)
sendWndInit int // hub's advertised receive window (our send budget)
// resume is whether this conn negotiated stream resumption, and grace how
// long a stream parked from it may keep trying to reattach. Both are
@@ -266,17 +176,16 @@ type WorkerConn struct {
id int // for log correlation only
stats *connStats // nil unless diagnostics are enabled
mu sync.Mutex
streams map[int]*Stream
nextSid int
closed bool // readLoop has exited; registerStream must refuse
mu sync.Mutex
stream *Stream
closed bool // readLoop has exited; attach must refuse
}
// heartbeatLoop proves the worker conn is still carrying frames end to end. TCP
// alone cannot tell us: a middlebox that drops an established flow (conntrack
// expiry, firewall state loss) sends no FIN or RST, so the read loop would park
// forever, the dead conn would stay in the pool, and every player routed to it
// would silently fail until the process restarted.
// forever and the player on this conn would silently fail until the process
// restarted.
func (wc *WorkerConn) heartbeatLoop(interval, timeout time.Duration) {
ticker := time.NewTicker(interval)
defer ticker.Stop()
@@ -286,12 +195,11 @@ func (wc *WorkerConn) heartbeatLoop(interval, timeout time.Duration) {
return
case <-ticker.C:
if silent := time.Since(time.UnixMilli(wc.lastPong.Load())); silent > timeout {
log.Printf("worker conn silent for %s; dropping it and its %d stream(s)",
silent.Round(time.Second), wc.streamCount())
log.Printf("worker conn silent for %s; dropping it", silent.Round(time.Second))
_ = wc.fc.Close() // readLoop unblocks and tears everything down
return
}
msg := wire.NewWriter().U8(MuxPing).VarInt(MuxCtlSid).I64(time.Now().UnixMilli()).Out()
msg := wire.NewWriter().U8(MuxPing).I64(time.Now().UnixMilli()).Out()
if err := wc.fc.WriteFrame(msg); err != nil {
return
}
@@ -299,56 +207,39 @@ func (wc *WorkerConn) heartbeatLoop(interval, timeout time.Duration) {
}
}
func (wc *WorkerConn) streamCount() int {
wc.mu.Lock()
defer wc.mu.Unlock()
return len(wc.streams)
}
func (wc *WorkerConn) newSid() int {
wc.mu.Lock()
defer wc.mu.Unlock()
sid := wc.nextSid
wc.nextSid++
return sid
}
// registerStream publishes a stream in the conn's table, or reports false if
// the conn has already died.
// attach publishes a stream on this conn, or reports false if the conn has
// already died (or is already bound — which would be a caller bug).
//
// The check is not advisory. Allocate hands out a (conn, sid) pair under the
// pool lock, and the conn's readLoop can exit before the caller gets here — it
// has then already swapped the stream map, so a blind insert would land in a map
// nothing iterates and the stream would never be torn down. That normally hides
// behind a failing SYN, but not on a half-open conn whose readLoop died on a
// framing error while the socket is still writable. Callers must re-Allocate.
func (wc *WorkerConn) registerStream(sid int, st *Stream) bool {
// The check is not advisory. Dial hands out a conn, and the conn's readLoop
// can exit before the caller gets here. A blind store would land on a conn
// nothing iterates and the stream would never be torn down.
func (wc *WorkerConn) attach(st *Stream) bool {
wc.mu.Lock()
defer wc.mu.Unlock()
if wc.closed {
if wc.closed || wc.stream != nil {
return false
}
wc.streams[sid] = st
wc.stream = st
return true
}
func (wc *WorkerConn) getStream(sid int) *Stream {
func (wc *WorkerConn) getStream() *Stream {
wc.mu.Lock()
defer wc.mu.Unlock()
return wc.streams[sid]
return wc.stream
}
func (wc *WorkerConn) removeStream(sid int) *Stream {
func (wc *WorkerConn) detach() *Stream {
wc.mu.Lock()
defer wc.mu.Unlock()
st := wc.streams[sid]
delete(wc.streams, sid)
st := wc.stream
wc.stream = nil
return st
}
// readLoop dispatches inbound mux frames. It must never block on a stream's
// destination: DATA is only enqueued (the per-stream writeLoop does the actual
// destination writes), so one slow destination cannot stall other streams.
// readLoop dispatches inbound tunnel frames. DATA is only enqueued (the
// stream's writeLoop does the actual destination writes), so a stalled
// destination cannot stall liveness / WND / FIN dispatch on this conn.
func (wc *WorkerConn) readLoop() {
for {
payload, err := wc.fc.ReadFrame()
@@ -363,24 +254,20 @@ func (wc *WorkerConn) readLoop() {
if err != nil {
continue
}
sid, err := r.VarInt()
if err != nil {
continue
}
switch ftype {
case MuxData:
if st := wc.getStream(sid); st != nil {
if st := wc.getStream(); st != nil {
st.deliverFromHub(r.Remaining())
}
case MuxWnd:
if delta, err := r.VarInt(); err == nil && delta > 0 {
if st := wc.getStream(sid); st != nil {
if st := wc.getStream(); st != nil {
st.grantSendWnd(delta)
}
}
case MuxFin:
// Graceful: drain what is already queued to the destination first.
if st := wc.removeStream(sid); st != nil {
if st := wc.detach(); st != nil {
st.gracefulFin()
}
case MuxRst:
@@ -389,7 +276,7 @@ func (wc *WorkerConn) readLoop() {
if b, err := r.U8(); err == nil {
reason = int(b)
}
if st := wc.removeStream(sid); st != nil {
if st := wc.detach(); st != nil {
st.onRst(reason)
}
case MuxResumeAck:
@@ -399,12 +286,12 @@ func (wc *WorkerConn) readLoop() {
if aerr != nil || derr != nil || cerr != nil {
continue
}
if st := wc.getStream(sid); st != nil {
if st := wc.getStream(); st != nil {
st.deliverResume(resumeResult{accepted: accepted, delivered: delivered, cid: cid})
}
case MuxPing:
nonce, _ := r.I64()
_ = wc.fc.WriteFrame(wire.NewWriter().U8(MuxPong).VarInt(MuxCtlSid).I64(nonce).Out())
_ = wc.fc.WriteFrame(wire.NewWriter().U8(MuxPong).I64(nonce).Out())
case MuxPong:
now := time.Now()
wc.lastPong.Store(now.UnixMilli())
@@ -416,46 +303,42 @@ func (wc *WorkerConn) readLoop() {
}
}
default:
log.Printf("worker: unknown mux type %d", ftype)
log.Printf("worker: unknown frame type %d", ftype)
}
}
// Connection lost: tear down all streams and drop from pool.
// Connection lost: tear down the bound stream and drop from the set.
close(wc.done)
wc.pool.remove(wc)
wc.mu.Lock()
// Marked before the map is swapped, under the same lock, so a concurrent
// registerStream either lands in the map we are about to drain or is refused.
// Marked before the pointer is cleared, under the same lock, so a concurrent
// attach either lands in the field we are about to drain or is refused.
wc.closed = true
streams := make([]*Stream, 0, len(wc.streams))
for _, st := range wc.streams {
streams = append(streams, st)
}
wc.streams = make(map[int]*Stream)
st := wc.stream
wc.stream = nil
wc.mu.Unlock()
if st == nil {
return
}
// Only the tunnel leg died. Where the session negotiated resumption the
// destination sockets are kept open and each stream reattaches over a fresh
// destination socket is kept open and the stream reattaches over a fresh
// conn (§7.5); otherwise this is the old, unconditional teardown.
// During Close there is no reattach to come: a stream that parked now would
// hold its destination socket open past shutdown, so teardown instead.
if wc.pool.client.closing.Load() {
for _, st := range streams {
st.teardown(false)
}
st.teardown(false)
return
}
for _, st := range streams {
if !st.park(wc.grace) {
st.teardown(false)
}
if !st.park(wc.grace) {
st.teardown(false)
}
}
func (wc *WorkerConn) sendSyn(sid int, cid []byte) error {
return wc.fc.WriteFrame(wire.NewWriter().U8(MuxSyn).VarInt(sid).Bytes(cid).Out())
func (wc *WorkerConn) sendSyn(cid []byte) error {
return wc.fc.WriteFrame(wire.NewWriter().U8(MuxSyn).Bytes(cid).Out())
}
func (wc *WorkerConn) sendData(sid int, data []byte) error {
err := wc.fc.WriteFrame(wire.NewWriter().U8(MuxData).VarInt(sid).Bytes(data).Out())
func (wc *WorkerConn) sendData(data []byte) error {
err := wc.fc.WriteFrame(wire.NewWriter().U8(MuxData).Bytes(data).Out())
if wc.stats != nil {
wc.stats.framesOut.Add(1)
if err != nil {
@@ -465,42 +348,27 @@ func (wc *WorkerConn) sendData(sid int, data []byte) error {
return err
}
func (wc *WorkerConn) sendFin(sid int) {
_ = wc.fc.WriteFrame(wire.NewWriter().U8(MuxFin).VarInt(sid).Out())
func (wc *WorkerConn) sendFin() {
_ = wc.fc.WriteFrame(wire.NewWriter().U8(MuxFin).Out())
}
func (wc *WorkerConn) sendRst(sid int) {
_ = wc.fc.WriteFrame(wire.NewWriter().U8(MuxRst).VarInt(sid).Out())
func (wc *WorkerConn) sendRst() {
_ = wc.fc.WriteFrame(wire.NewWriter().U8(MuxRst).Out())
}
func (wc *WorkerConn) sendWndUpdate(sid, delta int) {
_ = wc.fc.WriteFrame(wire.NewWriter().U8(MuxWnd).VarInt(sid).VarInt(delta).Out())
func (wc *WorkerConn) sendWndUpdate(delta int) {
_ = wc.fc.WriteFrame(wire.NewWriter().U8(MuxWnd).VarInt(delta).Out())
}
// leg binds a stream to one worker conn. The two fields are only meaningful
// together: stream ids are per-conn and restart at 1, so conn A's sid 3 and
// conn B's sid 3 belong to different players. A torn read across the two would
// credit, reset or FIN a stranger's stream, so the pair is swapped as one
// immutable value rather than as two fields.
type leg struct {
wc *WorkerConn
sid int
}
// String formats one (conn, sid) snapshot for log correlation. Stream ids
// restart at 1 per conn, so a bare sid cannot be traced across a reattach —
// the conn id is what ties the log lines together.
func (lg *leg) String() string { return fmt.Sprintf("conn%d/sid%d", lg.wc.id, lg.sid) }
// Stream bridges one player (via the hub) to one destination connection.
//
// Data from the hub is queued and written to the destination by a dedicated
// writeLoop goroutine. The queue is bounded by the advertised receive window —
// the hub never sends more un-credited bytes, so overflow is a protocol
// violation and resets the stream.
// violation and resets the tunnel.
type Stream struct {
client *Client
leg atomic.Pointer[leg]
client *Client
wc atomic.Pointer[WorkerConn]
mapping Mapping
srcIP string
srcPort int
@@ -527,8 +395,8 @@ type Stream struct {
// never interleave with live ones.
//
// Deliberately not s.mu: deliverFromHub takes s.mu from the worker readLoop,
// and holding s.mu across a WriteFrame would stall frame dispatch for every
// other stream on the connection.
// and holding s.mu across a WriteFrame would stall frame dispatch (WND/FIN
// /heartbeat replies) on this conn.
sendMu sync.Mutex
un unackedBuf // guarded by sendMu
@@ -578,10 +446,10 @@ type qentry struct {
fromHub bool
}
func newStream(c *Client, wc *WorkerConn, sid int, cid []byte, m Mapping, ip string, port int) *Stream {
func newStream(c *Client, wc *WorkerConn, cid []byte, m Mapping, ip string, port int) *Stream {
s := &Stream{client: c, cid: cid, mapping: m, srcIP: ip, srcPort: port,
resumable: wc.resume, sendWnd: wc.sendWndInit, done: make(chan struct{})}
s.leg.Store(&leg{wc: wc, sid: sid})
s.wc.Store(wc)
if c.statsOn() {
s.stats = &streamStats{opened: time.Now()}
}
@@ -592,20 +460,28 @@ func newStream(c *Client, wc *WorkerConn, sid int, cid []byte, m Mapping, ip str
return s
}
// conn returns the stream's current binding. Every caller must take exactly one
// snapshot and use both fields from it; re-loading mid-operation reintroduces
// the torn-pair hazard the leg exists to prevent.
func (s *Stream) conn() *leg { return s.leg.Load() }
// conn returns the stream's current worker conn. May be the original or a
// reattach; callers that send must take one snapshot and use it for the
// whole operation so a concurrent rebind cannot split a write across conns.
func (s *Stream) conn() *WorkerConn { return s.wc.Load() }
func (s *Stream) name() string {
if wc := s.conn(); wc != nil {
return fmt.Sprintf("conn%d", wc.id)
}
return "conn?"
}
// run dials the destination, optionally writes the PROXY v2 header, then pumps
// destination -> hub (respecting the stream send window when negotiated).
// destination -> hub (respecting the send window when negotiated).
func (s *Stream) run() {
dest, err := net.DialTimeout("tcp", s.mapping.Destination, 10*time.Second)
if err != nil {
lg := s.conn()
log.Printf("stream %s: dial %s failed: %v", lg, s.mapping.Destination, err)
lg.wc.removeStream(lg.sid)
lg.wc.sendRst(lg.sid)
log.Printf("stream %s: dial %s failed: %v", s.name(), s.mapping.Destination, err)
if wc := s.conn(); wc != nil {
wc.detach()
wc.sendRst()
}
s.teardown(false)
return
}
@@ -616,7 +492,7 @@ func (s *Stream) run() {
if s.mapping.ProxyProtocol {
if hdr := s.buildProxyHeader(dest); hdr != nil {
if _, err := dest.Write(hdr); err != nil {
log.Printf("stream %s: proxy header write: %v", s.conn(), err)
log.Printf("stream %s: proxy header write: %v", s.name(), err)
}
}
}
@@ -663,7 +539,7 @@ func (s *Stream) run() {
}
// sendToHub forwards destination bytes to the hub in bounded DATA frames,
// honoring both the stream send window and the client-wide bandwidth cap.
// honoring both the send window and the client-wide bandwidth cap.
// Returns false once the stream closed or the worker conn failed.
func (s *Stream) sendToHub(data []byte) bool {
for len(data) > 0 {
@@ -674,7 +550,7 @@ func (s *Stream) sendToHub(data []byte) bool {
// Credit first, bandwidth second. The reverse order would charge the
// budget for bytes still parked on an exhausted window, so the client
// would throttle itself below the configured rate. Holding credit while
// waiting for tokens is free — credit is per-stream, and the hub returns
// waiting for tokens is free — credit is per-tunnel, and the hub returns
// it as it drains data to the player, independent of our pacing.
if !s.acquireSendWnd(n) {
return false
@@ -714,9 +590,6 @@ func (s *Stream) sendToHub(data []byte) bool {
// one taken before the attempt. That also makes a failure survivable: while the
// stream can still be resumed the bytes are already safe, and the reattach
// replays them from wherever the hub says it got to.
//
// Note the old code let a failed write drop the rest of the chunk on the floor —
// the caller's slice advance sat after the error return.
func (s *Stream) emit(chunk []byte) bool {
s.sendMu.Lock()
if s.resumable {
@@ -725,8 +598,13 @@ func (s *Stream) emit(chunk []byte) bool {
s.un.advance(s.ackedOffset.Load())
s.un.append(chunk)
}
lg := s.conn()
err := lg.wc.sendData(lg.sid, chunk)
wc := s.conn()
var err error
if wc != nil {
err = wc.sendData(chunk)
} else {
err = errPoolClosed
}
s.sendMu.Unlock()
if err == nil {
@@ -792,7 +670,7 @@ func (s *Stream) writeLoop() {
// deliverFromHub enqueues hub bytes for the destination. Called from the worker
// readLoop; it never blocks — a peer that exceeds the advertised window is a
// protocol violator and gets the stream reset.
// protocol violator and gets the tunnel reset.
func (s *Stream) deliverFromHub(data []byte) {
if s.vel != nil {
s.vel.ObserveC2S(data) // observation only; bytes still forwarded verbatim
@@ -804,10 +682,11 @@ func (s *Stream) deliverFromHub(data []byte) {
}
if s.qBytes+len(data) > s.client.streamWnd {
s.mu.Unlock()
lg := s.conn()
log.Printf("stream %s: peer exceeded flow-control window; resetting", lg)
lg.wc.removeStream(lg.sid)
lg.wc.sendRst(lg.sid)
log.Printf("stream %s: peer exceeded flow-control window; resetting", s.name())
if wc := s.conn(); wc != nil {
wc.detach()
wc.sendRst()
}
s.teardown(false)
return
}
@@ -891,8 +770,9 @@ func (s *Stream) credit(n int) {
delta := s.consumed
s.consumed = 0
s.mu.Unlock()
lg := s.conn()
lg.wc.sendWndUpdate(lg.sid, delta)
if wc := s.conn(); wc != nil {
wc.sendWndUpdate(delta)
}
}
func (s *Stream) buildProxyHeader(dest net.Conn) []byte {
@@ -957,10 +837,11 @@ func (s *Stream) teardown(notifyHub bool) {
if dest != nil {
_ = dest.Close()
}
lg := s.conn()
lg.wc.removeStream(lg.sid)
if notifyHub {
lg.wc.sendFin(lg.sid)
if wc := s.conn(); wc != nil {
wc.detach()
if notifyHub {
wc.sendFin()
}
}
s.logSummary()
}