diff --git a/SHA256SUMS b/SHA256SUMS index 0d6fb03..8199229 100644 --- a/SHA256SUMS +++ b/SHA256SUMS @@ -1,6 +1,6 @@ 969e4a020d5475bfff355f38ca809edb68686b5850af8f9709126ed5bb2d13bb *bin/dragontcp-hybrid-server-linux-amd64 eaf2d07d8edf17840ce7bad4b1c81e8f3ccb1504c18397a773ca0aacb2ee72ef *bin/dragontcp-hybrid-server-linux-arm64 -b6dda0c38b8472b8e648a212daddca5a737c8e0a3457eb3c4f29833ec3d2e1c4 *bin/dragontcp-hybrid-client-linux-amd64 -d70e5cb83cb6109cba2a8dab1b232c5b7545abb26bb9ad71b92972b7e8f830d9 *android/lib/arm64-v8a/libdragontcp_client.so +2ac228a78ad918e859a41a90e667e285b0bf4cfbf2e50cd426ad89a706ffb8b2 *bin/dragontcp-hybrid-client-linux-amd64 +b4a37ee8699880ae49ef97e144e9c8fb2cdcb456045a8d41c1e4809de513b994 *android/lib/arm64-v8a/libdragontcp_client.so 3c45f678da399ecd20c28b8acc631d8c6f18151415ac63109f838a8c68a00e68 *android/lib/armeabi-v7a/libdragontcp_client.so 78806799b1c2949835e854635f6c352b572d3b3b526c4a6de877da9dd4e2d98b *android/lib/x86_64/libdragontcp_client.so diff --git a/android.zip b/android.zip deleted file mode 100644 index db6c119..0000000 Binary files a/android.zip and /dev/null differ diff --git a/android/lib/arm64-v8a/libdragontcp_client.so b/android/lib/arm64-v8a/libdragontcp_client.so index 255db71..198dfcb 100644 Binary files a/android/lib/arm64-v8a/libdragontcp_client.so and b/android/lib/arm64-v8a/libdragontcp_client.so differ diff --git a/bin/dragontcp-hybrid-client-linux-amd64 b/bin/dragontcp-hybrid-client-linux-amd64 index 7e43009..cdc341a 100644 Binary files a/bin/dragontcp-hybrid-client-linux-amd64 and b/bin/dragontcp-hybrid-client-linux-amd64 differ diff --git a/core/cmd/dragontcp-client/bp.go b/core/cmd/dragontcp-client/bp.go index cd38371..2570d13 100644 --- a/core/cmd/dragontcp-client/bp.go +++ b/core/cmd/dragontcp-client/bp.go @@ -6,6 +6,7 @@ import ( "io" "net" "sync" + "sync/atomic" "time" "dragontcp/internal/cover" @@ -163,20 +164,29 @@ func (l *bpLane) single(mode byte, sid wire.SessionID, seq uint64, payload []byt var lastErr error for attempt := 0; attempt < 2; attempt++ { if err := l.ensureLocked(); err != nil { + if isTransportTimeout(err) { + return 0, nil, err + } lastErr = err continue } reused := l.pc.requests > 0 _ = l.pc.conn.SetDeadline(time.Now().Add(timeout)) if err := writeBPRequest(l.pc.conn, mode, sid, seq, payload, downloadHint, l.coverProfile.HeaderMask, l.coverProfile.Clear); err != nil { - lastErr = err l.transportFailureLocked(reused) + if isTransportTimeout(err) { + return 0, nil, err + } + lastErr = err continue } status, body, err := readBPResponse(l.pc.conn, sid, mode, seq, l.coverProfile.HeaderMask, l.coverProfile.Clear) if err != nil { - lastErr = err l.transportFailureLocked(reused) + if isTransportTimeout(err) { + return 0, nil, err + } + lastErr = err continue } l.pc.requests++ @@ -222,14 +232,20 @@ func (l *bpLane) download(sid wire.SessionID, offset uint64, maxChunk, count int var lastErr error for attempt := 0; attempt < 2; attempt++ { if err := l.ensureLocked(); err != nil { + if isTransportTimeout(err) { + return nil, 0, err + } lastErr = err continue } reused := l.pc.requests > 0 _ = l.pc.conn.SetDeadline(time.Now().Add(timeout)) if err := writeBPRequest(l.pc.conn, mode, sid, offset, payload, hint, l.coverProfile.HeaderMask, l.coverProfile.Clear); err != nil { - lastErr = err l.transportFailureLocked(reused) + if isTransportTimeout(err) { + return nil, 0, err + } + lastErr = err continue } @@ -238,8 +254,11 @@ func (l *bpLane) download(sid wire.SessionID, offset uint64, maxChunk, count int for i := 0; i < responses; i++ { status, body, err := readBPResponse(l.pc.conn, sid, mode, offset, l.coverProfile.HeaderMask, l.coverProfile.Clear) if err != nil { - lastErr = err l.transportFailureLocked(reused) + if isTransportTimeout(err) { + return nil, 0, err + } + lastErr = err goto retry } lastStatus = status @@ -292,6 +311,7 @@ type bpConn struct { pipeline int closeOnce sync.Once + dead atomic.Bool } func openBPTunnel(serverAddr, token, targetHost string, targetPort int, opts chunkClientOptions) (net.Conn, error) { @@ -385,6 +405,9 @@ func (c *bpConn) fillReadBuffer() error { if c.consumedOffset > c.lastAck { status, body, err := c.downloadLane.single(bpModeACK, c.sid, c.consumedOffset, nil, 0) if err != nil { + if isTransportTimeout(err) { + c.dead.Store(true) + } return err } if status == wire.StatusError { @@ -402,6 +425,10 @@ func (c *bpConn) fillReadBuffer() error { } data, status, err := c.downloadLane.download(c.sid, c.downloadOffset, chunk, count) if err != nil { + if isTransportTimeout(err) { + c.dead.Store(true) + return err + } old, next := c.downSizer.FailureReason(chunk, err) if old == next && next == c.opts.minSize { return err @@ -462,6 +489,10 @@ func (c *bpConn) Write(p []byte) (int, error) { n := minInt(size, len(p)) status, body, err := c.uploadLane.single(bpModeUpload, c.sid, c.upOffset+2, p[:n], 0) if err != nil { + if isTransportTimeout(err) { + c.dead.Store(true) + return total, err + } old, next := c.upSizer.FailureReason(size, err) if old == next && next == c.opts.minSize { return total, err @@ -485,7 +516,11 @@ func (c *bpConn) Write(p []byte) (int, error) { func (c *bpConn) Close() error { c.closeOnce.Do(func() { - _, _, _ = c.downloadLane.single(bpModeACK, c.sid, c.consumedOffset, bpCloseMagic[:], 0) + // A timeout is a hard tunnel boundary. Do not create a fresh physical + // connection only to close a session that has already been abandoned. + if !c.dead.Load() { + _, _, _ = c.downloadLane.single(bpModeACK, c.sid, c.consumedOffset, bpCloseMagic[:], 0) + } c.uploadLane.Close() c.downloadLane.Close() }) diff --git a/core/cmd/dragontcp-client/chunk.go b/core/cmd/dragontcp-client/chunk.go index 67f1da2..e2b3325 100644 --- a/core/cmd/dragontcp-client/chunk.go +++ b/core/cmd/dragontcp-client/chunk.go @@ -263,20 +263,31 @@ func (l *requestLane) single(mode byte, sid wire.SessionID, seq uint64, payload var lastErr error for attempt := 0; attempt < 2; attempt++ { if err := l.ensureLocked(); err != nil { + // A dial timeout is a dead path for this logical tunnel. Do not + // spend another timeout replaying the same request. + if isTransportTimeout(err) { + return 0, nil, err + } lastErr = err continue } reused := l.pc.requests > 0 _ = l.pc.conn.SetDeadline(time.Now().Add(timeout)) if err := wire.WriteRequestProfileEncoding(l.pc.conn, mode, sid, seq, payload, l.headerMask, l.coverProfile.Clear); err != nil { - lastErr = err l.transportFailureLocked(reused) + if isTransportTimeout(err) { + return 0, nil, err + } + lastErr = err continue } status, body, err := wire.ReadResponseProfile(l.pc.conn, l.headerMask) if err != nil { - lastErr = err l.transportFailureLocked(reused) + if isTransportTimeout(err) { + return 0, nil, err + } + lastErr = err continue } l.pc.requests++ @@ -307,14 +318,20 @@ func (l *requestLane) download(sid wire.SessionID, startOffset, ackOffset uint64 var lastErr error for attempt := 0; attempt < 2; attempt++ { if err := l.ensureLocked(); err != nil { + if isTransportTimeout(err) { + return nil, 0, err + } lastErr = err continue } reused := l.pc.requests > 0 _ = l.pc.conn.SetDeadline(time.Now().Add(timeout)) if err := wire.WriteRequestProfileEncoding(l.pc.conn, wire.ModeDownload, sid, startOffset, payload, l.headerMask, l.coverProfile.Clear); err != nil { - lastErr = err l.transportFailureLocked(reused) + if isTransportTimeout(err) { + return nil, 0, err + } + lastErr = err continue } @@ -324,8 +341,11 @@ func (l *requestLane) download(sid wire.SessionID, startOffset, ackOffset uint64 for i := 0; i < count; i++ { status, body, err := wire.ReadResponseProfile(l.pc.conn, l.headerMask) if err != nil { - lastErr = err l.transportFailureLocked(reused) + if isTransportTimeout(err) { + return nil, 0, err + } + lastErr = err goto retry } lastStatus = status @@ -585,6 +605,7 @@ type chunkConn struct { maxPipeline int closeOnce sync.Once + dead atomic.Bool } // appendChunkParts keeps the single-response fast path zero-copy. For a batch, @@ -755,6 +776,10 @@ func (c *chunkConn) fillReadBuffer() error { minFailures = 0 } if err != nil { + if isTransportTimeout(err) { + c.dead.Store(true) + return err + } if c.pipeline > c.minPipeline { old := c.pipeline c.pipeline /= 2 @@ -829,6 +854,10 @@ func (c *chunkConn) Write(p []byte) (int, error) { n := minInt(size, len(p)) status, body, err := c.uploadLane.single(wire.ModeUpload, c.sid, c.upOffset, p[:n]) if err != nil { + if isTransportTimeout(err) { + c.dead.Store(true) + return total, err + } old, next := c.upSizer.FailureReason(size, err) if old == next && next == c.opts.minSize { minFailures++ @@ -858,9 +887,13 @@ func (c *chunkConn) Write(p []byte) (int, error) { func (c *chunkConn) Close() error { c.closeOnce.Do(func() { - // Reuse the upload lane rather than dialling a connection just to say - // goodbye; that was a second wasted dial per flow. - _, _, _ = c.uploadLane.single(wire.ModeClose, c.sid, 0, nil) + // A timed-out tunnel is already dead. Do not redial merely to send a + // graceful CLOSE for a session that the caller is abandoning. + if !c.dead.Load() { + // Reuse the upload lane rather than dialling a connection just to say + // goodbye; that was a second wasted dial per flow. + _, _, _ = c.uploadLane.single(wire.ModeClose, c.sid, 0, nil) + } c.uploadLane.Close() c.downloadLane.Close() }) diff --git a/core/cmd/dragontcp-client/transport_error.go b/core/cmd/dragontcp-client/transport_error.go new file mode 100644 index 0000000..329c485 --- /dev/null +++ b/core/cmd/dragontcp-client/transport_error.go @@ -0,0 +1,19 @@ +package main + +import ( + "errors" + "net" +) + +// isTransportTimeout reports whether err ultimately came from a network +// deadline/timeout. A timeout is not evidence that the chunk size was too +// large: the physical connection may be dead and replaying the same logical +// request can also duplicate an upload whose response was lost. Treat it as a +// hard tunnel boundary instead of feeding it into the adaptive size controller. +func isTransportTimeout(err error) bool { + if err == nil { + return false + } + var netErr net.Error + return errors.As(err, &netErr) && netErr.Timeout() +} diff --git a/core/cmd/dragontcp-client/transport_error_test.go b/core/cmd/dragontcp-client/transport_error_test.go new file mode 100644 index 0000000..cc45357 --- /dev/null +++ b/core/cmd/dragontcp-client/transport_error_test.go @@ -0,0 +1,134 @@ +package main + +import ( + "net" + "sync" + "sync/atomic" + "testing" + "time" + + "dragontcp/internal/cover" +) + +func startStallingTCPServer(t *testing.T) (string, *atomic.Int32, func()) { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + var accepted atomic.Int32 + var conns sync.WaitGroup + stop := make(chan struct{}) + go func() { + for { + conn, err := ln.Accept() + if err != nil { + return + } + accepted.Add(1) + conns.Add(1) + go func(c net.Conn) { + defer conns.Done() + defer c.Close() + <-stop + }(conn) + } + }() + cleanup := func() { + close(stop) + _ = ln.Close() + conns.Wait() + } + return ln.Addr().String(), &accepted, cleanup +} + +func timeoutTestOptions() chunkClientOptions { + return chunkClientOptions{ + startSize: 1024 * 1024, + minSize: 32, + maxSize: 1024 * 1024, + adaptive: true, + adaptSuccesses: 16, + txnTimeout: 60 * time.Millisecond, + minPipeline: 1, + maxPipeline: 1, + } +} + +func TestBPDownloadTimeoutIsHardTunnelBoundary(t *testing.T) { + addr, accepted, cleanup := startStallingTCPServer(t) + defer cleanup() + + opts := timeoutTestOptions() + lane := newBPLane(addr, 0, 0, opts.txnTimeout, cover.Profile{}) + c := &bpConn{ + opts: opts, + uploadLane: newBPLane(addr, 0, 0, opts.txnTimeout, cover.Profile{}), + downloadLane: lane, + pipeline: 1, + } + c.downSizer = newAdaptiveSizer("BP download", opts.startSize, opts) + + started := time.Now() + err := c.fillReadBuffer() + elapsed := time.Since(started) + if err == nil || !isTransportTimeout(err) { + t.Fatalf("err=%v, want transport timeout", err) + } + if got := c.downSizer.Current(); got != opts.startSize { + t.Fatalf("download chunk changed after timeout: got %d want %d", got, opts.startSize) + } + if got := accepted.Load(); got != 1 { + t.Fatalf("timeout opened %d physical connections, want exactly 1", got) + } + if elapsed > 300*time.Millisecond { + t.Fatalf("timeout path took %s; expected immediate hard failure", elapsed) + } + if err := c.Close(); err != nil { + t.Fatal(err) + } + time.Sleep(20 * time.Millisecond) + if got := accepted.Load(); got != 1 { + t.Fatalf("Close redialed after timeout: physical connections=%d, want 1", got) + } +} + +func TestBinaryDownloadTimeoutIsHardTunnelBoundary(t *testing.T) { + addr, accepted, cleanup := startStallingTCPServer(t) + defer cleanup() + + opts := timeoutTestOptions() + lane := newRequestLane(addr, 0, 0, opts.txnTimeout, 0, cover.Profile{}) + c := &chunkConn{ + opts: opts, + uploadLane: newRequestLane(addr, 0, 0, opts.txnTimeout, 0, cover.Profile{}), + downloadLane: lane, + pipeline: 1, + minPipeline: 1, + maxPipeline: 1, + } + c.downSizer = newAdaptiveSizer("download", opts.startSize, opts) + + started := time.Now() + err := c.fillReadBuffer() + elapsed := time.Since(started) + if err == nil || !isTransportTimeout(err) { + t.Fatalf("err=%v, want transport timeout", err) + } + if got := c.downSizer.Current(); got != opts.startSize { + t.Fatalf("download chunk changed after timeout: got %d want %d", got, opts.startSize) + } + if got := accepted.Load(); got != 1 { + t.Fatalf("timeout opened %d physical connections, want exactly 1", got) + } + if elapsed > 300*time.Millisecond { + t.Fatalf("timeout path took %s; expected immediate hard failure", elapsed) + } + if err := c.Close(); err != nil { + t.Fatal(err) + } + time.Sleep(20 * time.Millisecond) + if got := accepted.Load(); got != 1 { + t.Fatalf("Close redialed after timeout: physical connections=%d, want 1", got) + } +}