diff --git a/go.mod b/go.mod index 0762c9fe6..578740127 100644 --- a/go.mod +++ b/go.mod @@ -30,7 +30,7 @@ require ( require ( github.com/docker/docker v28.5.2+incompatible - github.com/xtaci/smux v1.5.49 + github.com/xtaci/smux v1.5.50 ) require ( diff --git a/go.sum b/go.sum index 927b3f00f..6455ecc47 100644 --- a/go.sum +++ b/go.sum @@ -201,8 +201,8 @@ github.com/valyala/fastrand v1.1.0 h1:f+5HkLW4rsgzdNoleUOB69hyT9IlD2ZQh9GyDMfb5G github.com/valyala/fastrand v1.1.0/go.mod h1:HWqCzkrkg6QXT8V2EXWvXCoow7vLwOFN002oeRzjapQ= github.com/valyala/histogram v1.2.0 h1:wyYGAZZt3CpwUiIb9AU/Zbllg1llXyrtApRS815OLoQ= github.com/valyala/histogram v1.2.0/go.mod h1:Hb4kBwb4UxsaNbbbh+RRz8ZR6pdodR57tzWUS3BUzXY= -github.com/xtaci/smux v1.5.49 h1:V3pdyzGLGDMX4R/rbx+e7yo5nB9ticxCcCkQZMxCWwE= -github.com/xtaci/smux v1.5.49/go.mod h1:IGQ9QYrBphmb/4aTnLEcJby0TNr3NV+OslIOMrX825Q= +github.com/xtaci/smux v1.5.50 h1:y/1DlWQC9bnMeZzsyk4oL2hbLK6uVk4BKTz5BeQqUEA= +github.com/xtaci/smux v1.5.50/go.mod h1:IGQ9QYrBphmb/4aTnLEcJby0TNr3NV+OslIOMrX825Q= go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA= go.opentelemetry.io/auto/sdk v1.1.0/go.mod h1:3wSPjt5PWp2RhlCcmmOial7AvC4DQqZb7a7wCow3W8A= go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.63.0 h1:RbKq8BG0FI8OiXhBfcRtqqHcZcka+gU3cskNuf05R18= diff --git a/pkg/dmsghttp/http_transport_test.go b/pkg/dmsghttp/http_transport_test.go index 92715be03..cc691c5b3 100644 --- a/pkg/dmsghttp/http_transport_test.go +++ b/pkg/dmsghttp/http_transport_test.go @@ -81,6 +81,9 @@ func TestHTTPTransport_RoundTrip(t *testing.T) { Timeout: 10 * time.Second, } + // Allow time for dmsg sessions to stabilize on macOS + time.Sleep(200 * time.Millisecond) + // Act: http clients send requests concurrently. // - client1 sends "/index.html" requests. // - client2 sends "/echo" requests. diff --git a/pkg/dmsgpty/host_test.go b/pkg/dmsgpty/host_test.go index 24954fe31..c52023236 100644 --- a/pkg/dmsgpty/host_test.go +++ b/pkg/dmsgpty/host_test.go @@ -9,6 +9,7 @@ import ( "os" "runtime" "testing" + "time" "github.com/skycoin/skywire/pkg/skywire-utilities/pkg/cipher" "github.com/skycoin/skywire/pkg/skywire-utilities/pkg/logging" @@ -154,6 +155,9 @@ func TestHost(t *testing.T) { }) t.Run("endpoint_proxy", func(t *testing.T) { + // Give hostA time to establish its listener on macOS + time.Sleep(100 * time.Millisecond) + conn, err := cliB.prepareConn() require.NoError(t, err) diff --git a/vendor/github.com/xtaci/smux/AGENTS.md b/vendor/github.com/xtaci/smux/AGENTS.md new file mode 100644 index 000000000..ed707eb3e --- /dev/null +++ b/vendor/github.com/xtaci/smux/AGENTS.md @@ -0,0 +1,93 @@ +# AGENTS.md - smux Project Context + +## Project Overview +**smux** (Simple Multiplexing) is a multiplexing library for Golang. It allows multiple logical streams to share a single underlying connection (like TCP or KCP). It is designed for reliability and ordering, and is a core component of [kcp-go](https://github.com/xtaci/kcp-go) and [kcptun](https://github.com/xtaci/kcptun). + +## Key Features +- **Multiplexing**: Multiple streams over one connection. +- **Flow Control**: Token bucket controlled receiving and per-stream sliding window (protocol v2+). +- **Memory Efficiency**: Shared receive buffer among streams to control overall memory usage. +- **Low Overhead**: Minimized header (8 bytes). +- **Traffic Shaping**: Built-in fair queue traffic shaping. + +## Architecture & Core Components + +### 1. Session (`session.go`) +The `Session` struct is the main manager for a multiplexed connection. +- Manages the underlying `io.ReadWriteCloser`. +- Handles the creation and acceptance of streams. +- Manages the shared receive buffer and token bucket. +- **Key Methods**: `Client`, `Server`, `OpenStream`, `AcceptStream`. + +### 2. Stream (`stream.go`) +The `Stream` struct represents a logical stream within a session. +- Implements `net.Conn` interface (Read, Write, Close, etc.). +- Handles data buffering and flow control. +- **Key Methods**: `Read`, `Write`, `Close`. + +### 3. Frame (`frame.go`) +Defines the wire format for data transmission. +- **Header Format** (8 bytes): + - `VERSION` (1 byte): Protocol version (1 or 2). + - `CMD` (1 byte): Command type (`SYN`, `FIN`, `PSH`, `NOP`, `UPD`). + - `LENGTH` (2 bytes): Payload length. + - `STREAMID` (4 bytes): Stream identifier. +- **Commands**: + - `cmdSYN`: Stream open. + - `cmdFIN`: Stream close (EOF). + - `cmdPSH`: Data push. + - `cmdNOP`: No operation (keep-alive). + - `cmdUPD`: Window update (v2 only). + +### 4. Configuration (`mux.go`) +The `Config` struct allows tuning the session. +- **Key Fields**: `Version`, `KeepAliveInterval`, `MaxFrameSize`, `MaxReceiveBuffer`, `MaxStreamBuffer`. + +### 5. Traffic Shaping (`shaper.go`) +Implements traffic shaping logic to ensure fair bandwidth usage among streams. + +## Development Guidelines + +### Testing +- Run all tests: + ```bash + go test -v . + ``` +- Run benchmarks: + ```bash + go test -v -run=^$ -bench . + ``` + +### Coding Conventions +- Follow standard Go coding conventions (formatting, naming, etc.). +- Ensure backward compatibility when modifying protocol-related code. +- Pay attention to concurrency and locking, as `Session` and `Stream` are heavily concurrent. + +## Common Tasks + +### Creating a Client Session +```go +conn, _ := net.Dial(...) +session, _ := smux.Client(conn, nil) // nil for default config +stream, _ := session.OpenStream() +``` + +### Creating a Server Session +```go +conn, _ := listener.Accept() +session, _ := smux.Server(conn, nil) +stream, _ := session.AcceptStream() +``` + +### Protocol Versions +- **Version 1**: Basic multiplexing. +- **Version 2**: Adds `cmdUPD` for flow control (window updates). + +## File Structure +- `alloc.go`: Memory allocation utilities. +- `frame.go`: Frame definition and parsing. +- `mux.go`: Configuration and entry points. +- `session.go`: Session logic. +- `stream.go`: Stream logic. +- `shaper.go`: Traffic shaping logic. +- `*_test.go`: Tests for respective components. diff --git a/vendor/github.com/xtaci/smux/README.md b/vendor/github.com/xtaci/smux/README.md index 8bce5d65f..140949891 100644 --- a/vendor/github.com/xtaci/smux/README.md +++ b/vendor/github.com/xtaci/smux/README.md @@ -17,6 +17,8 @@ [11]: https://sourcegraph.com/github.com/xtaci/smux/-/badge.svg [12]: https://sourcegraph.com/github.com/xtaci/smux?badge +[English](README.md) | [中文](README_zh-cn.md) + ## Introduction Smux (**S**imple **MU**ltiple**X**ing) is a multiplexing library for Golang. It relies on an underlying connection to provide reliability and ordering, such as TCP or [KCP](https://github.com/xtaci/kcp-go), and provides stream-oriented multiplexing. This library was originally designed to power connection management for [kcp-go](https://github.com/xtaci/kcp-go). @@ -32,6 +34,22 @@ Smux (**S**imple **MU**ltiple**X**ing) is a multiplexing library for Golang. It ![smooth bandwidth curve](assets/curve.jpg) +## Architecture + +* **Session**: The main manager for a multiplexed connection. It manages the underlying `io.ReadWriteCloser`, handles stream creation/acceptance, and manages the shared receive buffer. +* **Stream**: A logical stream within a session. It implements the `net.Conn` interface, handling data buffering and flow control. +* **Frame**: The wire format for data transmission. + +## Frame Allocator + +`alloc.go` implements a slab-style allocator tuned for frames up to 64 KB. Seventeen `sync.Pool` buckets cache power-of-two slice capacities, and a De Bruijn based `msb()` lookup picks the smallest bucket that can satisfy a request in constant time. Buffers are deliberately reused without zeroing, so each new frame simply overwrites the previous payload without paying the Go runtime's memclr cost. + +Benefits for the session: + +1. Bounded fragmentation (each request wastes < 50%) keeps the shared receive buffer predictable under load. +2. Reuse of pre-sized slices drastically lowers GC pressure and removes repeated zeroing work. +3. Constant-time bucket selection avoids locks or searches, so high-throughput sessions keep tail latency steady even with thousands of flows. + ## Documentation For complete documentation, see the associated [Godoc](https://godoc.org/github.com/xtaci/smux). @@ -53,7 +71,15 @@ ok github.com/xtaci/smux 7.811s ## Specification ``` -VERSION(1B) | CMD(1B) | LENGTH(2B) | STREAMID(4B) | DATA(LENGTH) + +---------------+---------------+-------------------------------+ + | VERSION (1B) | CMD (1B) | LENGTH (2B) | + +---------------+---------------+-------------------------------+ + | STREAMID (4B) | + +---------------------------------------------------------------+ + | | + / DATA (Variable) / + | | + +---------------------------------------------------------------+ VALUES FOR LATEST VERSION: VERSION: @@ -131,6 +157,19 @@ func server() { ``` -## Status +## Configuration + +`smux.Config` allows tuning the session parameters: + +* `Version`: Protocol version (1 or 2). +* `KeepAliveInterval`: Interval for sending NOP frames to keep the connection alive. +* `KeepAliveTimeout`: Timeout for closing the session if no data is received. +* `MaxFrameSize`: Maximum size of a frame. +* `MaxReceiveBuffer`: Maximum size of the shared receive buffer. +* `MaxStreamBuffer`: Maximum size of the per-stream buffer. + +## Reference -Stable +* [hashicorp/yamux](https://github.com/hashicorp/yamux) +* [xtaci/kcp-go](https://github.com/xtaci/kcp-go) +* [xtaci/kcptun](https://github.com/xtaci/kcptun) diff --git a/vendor/github.com/xtaci/smux/README_zh-cn.md b/vendor/github.com/xtaci/smux/README_zh-cn.md new file mode 100644 index 000000000..318db4bf8 --- /dev/null +++ b/vendor/github.com/xtaci/smux/README_zh-cn.md @@ -0,0 +1,175 @@ +smux + +[![GoDoc][1]][2] [![MIT licensed][3]][4] [![Build Status][5]][6] [![Go Report Card][7]][8] [![Coverage Statusd][9]][10] [![Sourcegraph][11]][12] + +smux + +[1]: https://godoc.org/github.com/xtaci/smux?status.svg +[2]: https://godoc.org/github.com/xtaci/smux +[3]: https://img.shields.io/badge/license-MIT-blue.svg +[4]: LICENSE +[5]: https://img.shields.io/github/created-at/xtaci/smux +[6]: https://img.shields.io/github/created-at/xtaci/smux +[7]: https://goreportcard.com/badge/github.com/xtaci/smux +[8]: https://goreportcard.com/report/github.com/xtaci/smux +[9]: https://codecov.io/gh/xtaci/smux/branch/master/graph/badge.svg +[10]: https://codecov.io/gh/xtaci/smux +[11]: https://sourcegraph.com/github.com/xtaci/smux/-/badge.svg +[12]: https://sourcegraph.com/github.com/xtaci/smux?badge + +[English](README.md) | [中文](README_zh-cn.md) + +## 简介 + +Smux（**S**imple **MU**ltiple**X**ing）是一个用 Golang 实现的多路复用库，让多个有序、可靠的逻辑流共享同一条底层连接（如 TCP 或 [KCP](https://github.com/xtaci/kcp-go)）。它最初为 [kcp-go](https://github.com/xtaci/kcp-go) 设计，用于在复杂网络环境中维持长连接时的精细流量控制和资源管理。 + +## 特性 + +1. **令牌桶限速**：基于令牌桶的接收控制，输出带宽曲线更平滑（如下图）。 +2. **全局缓冲共享**：会话级接收缓冲在各流之间复用，可精确限制整体内存占用。 +3. **极简协议头**：8 字节帧头最大化有效载荷占比。 +4. **大规模验证**：在 [kcptun](https://github.com/xtaci/kcptun) 中经数百万设备验证，稳定可靠。 +5. **公平队列整形**：内建公平调度，避免单个流独占带宽。 +6. **流级滑动窗口**：协议版本 2 起支持 per-stream 拥塞控制，进一步提升吞吐和延迟表现。 + +![smooth bandwidth curve](assets/curve.jpg) + +## 架构 + +* **Session**：多路复用会话管理器，负责维护底层 `io.ReadWriteCloser`，创建或接受 `Stream`，同时调度共享接收缓冲和限速逻辑。 +* **Stream**：会话中的逻辑连接，实现 `net.Conn` 接口，承担读写缓冲与流量控制。 +* **Frame**：在线协议帧格式，定义指令、流 ID、长度等字段，用于在 Session 与 Stream 之间传输数据/控制信息。 + +## 帧内存分配器 + +`alloc.go` 针对 64 KB 以内的帧实现了分层分配器：17 个 `sync.Pool` 分别缓存 2^n 容量的切片，`msb()` 函数利用 De Bruijn 序列常数在 O(1) 时间内定位到最合适的池子。复用出来的切片不会被额外清零，新帧直接覆盖旧负载，从而避开运行时的 memclr 开销。 + +这样带来的系统收益包括： + +1. 单次分配的浪费率 < 50%，在高并发下也能准确控制会话级缓冲占用。 +2. 重复使用固定容量的切片显著降低 GC 压力，也避免了多次清零的额外成本。 +3. 常数时间的桶选择避免了搜索或额外锁竞争，让高吞吐会话在大量流同时活跃时保持低尾延迟。 + +## 文档 + +更完整的 API 与实现细节可参考 [Godoc](https://godoc.org/github.com/xtaci/smux)。 + +## 基准测试 (Benchmark) +``` +$ go test -v -run=^$ -bench . +goos: darwin +goarch: amd64 +pkg: github.com/xtaci/smux +BenchmarkMSB-4 30000000 51.8 ns/op +BenchmarkAcceptClose-4 50000 36783 ns/op +BenchmarkConnSmux-4 30000 58335 ns/op 2246.88 MB/s 1208 B/op 19 allocs/op +BenchmarkConnTCP-4 50000 25579 ns/op 5124.04 MB/s 0 B/op 0 allocs/op +PASS +ok github.com/xtaci/smux 7.811s +``` + +## 规范 (Specification) + +``` ++---------------+---------------+-------------------------------+ + | VERSION (1B) | CMD (1B) | LENGTH (2B) | + +---------------+---------------+-------------------------------+ + | STREAMID (4B) | + +---------------------------------------------------------------+ + | | + / DATA (Variable) / + | | + +---------------------------------------------------------------+ + +VALUES FOR LATEST VERSION: +VERSION: + 1/2 + +CMD: + cmdSYN(0) + cmdFIN(1) + cmdPSH(2) + cmdNOP(3) + cmdUPD(4) // 仅在版本 2 支持 + +STREAMID: + 客户端使用从 1 开始的奇数 + 服务端使用从 0 开始的偶数 + +cmdUPD: + | CONSUMED(4B) | WINDOW(4B) | +``` + +## 用法 (Usage) + +```go + +func client() { + // 建立一条 TCP 连接 + conn, err := net.Dial(...) + if err != nil { + panic(err) + } + + // 初始化 smux 客户端，会采用默认配置 + session, err := smux.Client(conn, nil) + if err != nil { + panic(err) + } + + // 打开一个新的逻辑流 + stream, err := session.OpenStream() + if err != nil { + panic(err) + } + + // Stream 满足 io.ReadWriteCloser，可直接读写 + stream.Write([]byte("ping")) + stream.Close() + session.Close() +} + +func server() { + // 接收传入的 TCP 连接 + conn, err := listener.Accept() + if err != nil { + panic(err) + } + + // 使用 smux.Server 将连接升级为服务端会话 + session, err := smux.Server(conn, nil) + if err != nil { + panic(err) + } + + // 阻塞等待客户端打开的流 + stream, err := session.AcceptStream() + if err != nil { + panic(err) + } + + // 简单读取一条 4 字节消息 + buf := make([]byte, 4) + stream.Read(buf) + stream.Close() + session.Close() +} + +``` + +## 配置 + +`smux.Config` 提供了常用调优项： + +* `Version`：协议版本（1 或 2）。 +* `KeepAliveInterval`：发送 `cmdNOP` 以维持心跳的间隔。 +* `KeepAliveTimeout`：在无数据时认为连接失效的超时时间。 +* `MaxFrameSize`：单帧数据的最大长度。 +* `MaxReceiveBuffer`：会话级共享接收缓冲的上限。 +* `MaxStreamBuffer`：单个流本地缓冲的上限。 + +## 参考 + +* [hashicorp/yamux](https://github.com/hashicorp/yamux) +* [xtaci/kcp-go](https://github.com/xtaci/kcp-go) +* [xtaci/kcptun](https://github.com/xtaci/kcptun) diff --git a/vendor/github.com/xtaci/smux/shaper.go b/vendor/github.com/xtaci/smux/shaper.go index af00288b2..b53fb889e 100644 --- a/vendor/github.com/xtaci/smux/shaper.go +++ b/vendor/github.com/xtaci/smux/shaper.go @@ -57,6 +57,7 @@ func (h *shaperHeap) Pop() any { old := *h n := len(old) x := old[n-1] + old[n-1] = writeRequest{} // avoid memory leak *h = old[0 : n-1] return x } diff --git a/vendor/modules.txt b/vendor/modules.txt index f4c269538..0a53750f4 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -381,7 +381,7 @@ github.com/valyala/fastrand # github.com/valyala/histogram v1.2.0 ## explicit; go 1.12 github.com/valyala/histogram -# github.com/xtaci/smux v1.5.49 +# github.com/xtaci/smux v1.5.50 ## explicit; go 1.18 github.com/xtaci/smux # go.opentelemetry.io/auto/sdk v1.1.0