smux

README

English

Introduction

Smux (Simple MUltipleXing) is a stream-multiplexing library for Go. It lets multiple logical streams share a single underlying connection, relying on that connection—such as TCP or KCP—to provide reliable, ordered delivery. Smux was originally developed to manage connections in kcp-go.

Features

1. Token-bucket-based receive control helps smooth bandwidth usage, as illustrated below.
2. A shared, session-wide receive buffer bounds memory usage across streams.
3. An 8-byte frame header leaves more space for payload.
4. Built-in fair-queue traffic shaping.
5. Per-stream sliding-window flow control in protocol version 2.

Architecture

• Session: Manages the underlying io.ReadWriteCloser, stream creation and acceptance, and the shared receive buffer.
• Stream: Represents a logical connection within a session and implements net.Conn, including buffering and flow control.
• Frame: Defines the wire format used to carry stream data and control messages.

Frame Allocator

alloc.go implements a slab-style allocator for frames up to 64 KiB. Seventeen sync.Pool buckets hold slices with power-of-two capacities. A De Bruijn-based msb() lookup selects the smallest bucket that can satisfy a request in constant time. Reused buffers are not cleared when returned to the pool; incoming frame data overwrites the bytes that are read.

This design helps sessions by:

1. Limiting capacity waste to less than 50% for each allocation.
2. Reusing pre-sized slices to reduce allocation and garbage-collection overhead.
3. Selecting a capacity bucket in constant time.

Documentation

For the complete API documentation, see Godoc.

Benchmarks

Run the benchmarks with:

$ 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)                        /
 |                                                               |
 +---------------------------------------------------------------+

Protocol values:
VERSION:
    1/2
    
CMD:
    cmdSYN(0)
    cmdFIN(1)
    cmdPSH(2)
    cmdNOP(3)
    cmdUPD(4)	// only supported on version 2
    
STREAMID:
    client uses odd numbers, starting at 1
    server uses even numbers, starting at 0
    
cmdUPD:
    | CONSUMED(4B) | WINDOW(4B) |

Usage

func client() {
    // Establish a TCP connection
    conn, err := net.Dial(...)
    if err != nil {
        panic(err)
    }

    // Create a client session
    session, err := smux.Client(conn, nil)
    if err != nil {
        panic(err)
    }

    // Open a stream
    stream, err := session.OpenStream()
    if err != nil {
        panic(err)
    }

    // Stream implements io.ReadWriteCloser
    stream.Write([]byte("ping"))
    stream.Close()
    session.Close()
}

func server() {
    // Accept a TCP connection
    conn, err := listener.Accept()
    if err != nil {
        panic(err)
    }

    // Create a server session
    session, err := smux.Server(conn, nil)
    if err != nil {
        panic(err)
    }

    // Accept a stream
    stream, err := session.AcceptStream()
    if err != nil {
        panic(err)
    }

    // Read a message
    buf := make([]byte, 4)
    stream.Read(buf)
    stream.Close()
    session.Close()
}

Configuration

Use smux.Config to tune session behavior:

• Version: Protocol version (1 or 2).
• KeepAliveDisabled: Disables keep-alive messages and timeout checks.
• KeepAliveInterval: Time between NOP keep-alive frames.
• KeepAliveTimeout: Maximum time without received data before the session is closed.
• MaxFrameSize: Maximum frame payload size.
• MaxReceiveBuffer: Maximum amount of buffered data across all streams in the session.
• MaxStreamBuffer: Per-stream receive-window size in protocol version 2.

Reference

• hashicorp/yamux
• xtaci/kcp-go

Documentation

Index

• Variables
• func NewShaperQueue() *shaperQueue
• func VerifyConfig(config *Config) error
• type Allocator
  • func NewAllocator() *Allocator
  • func (alloc *Allocator) Get(size int) *[]byte
  • func (alloc *Allocator) Put(p *[]byte) error
• type CLASSID
• type Config
  • func DefaultConfig() *Config
• type Frame
• type Session
  • func Client(conn io.ReadWriteCloser, config *Config) (*Session, error)
  • func Server(conn io.ReadWriteCloser, config *Config) (*Session, error)
  • func (s *Session) Accept() (io.ReadWriteCloser, error)
  • func (s *Session) AcceptStream() (*Stream, error)
  • func (s *Session) Close() error
  • func (s *Session) CloseChan() <-chan struct{}
  • func (s *Session) IsClosed() bool
  • func (s *Session) LocalAddr() net.Addr
  • func (s *Session) NumStreams() int
  • func (s *Session) Open() (io.ReadWriteCloser, error)
  • func (s *Session) OpenStream() (*Stream, error)
  • func (s *Session) RemoteAddr() net.Addr
  • func (s *Session) SetDeadline(t time.Time) error
• type Stream
  • func (s Stream) Close() error
  • func (s Stream) CloseWrite() error
  • func (s Stream) GetDieCh() <-chan struct{}
  • func (s Stream) ID() uint32
  • func (s Stream) LocalAddr() net.Addr
  • func (s Stream) Read(b []byte) (n int, err error)
  • func (s Stream) RemoteAddr() net.Addr
  • func (s Stream) SetDeadline(t time.Time) error
  • func (s Stream) SetReadDeadline(t time.Time) error
  • func (s Stream) SetWriteDeadline(t time.Time) error
  • func (s Stream) Write(b []byte) (n int, err error)
  • func (s Stream) WriteTo(w io.Writer) (n int64, err error)

Variables

var (
	ErrInvalidProtocol           = errors.New("invalid protocol")
	ErrConsumed                  = errors.New("peer consumed more than sent")
	ErrGoAway                    = errors.New("stream id overflows, should start a new connection")
	ErrTimeout         net.Error = &timeoutError{}
	ErrWouldBlock                = errors.New("operation would block on IO")
)

Functions

func NewShaperQueue

func NewShaperQueue() *shaperQueue

NewShaperQueue creates an empty queue for fairly scheduling stream writes.

func VerifyConfig

func VerifyConfig(config *Config) error

VerifyConfig reports whether config contains valid session settings.

Types

type Allocator

type Allocator struct {
	// contains filtered or unexported fields
}

Allocator pools byte slices for incoming frames.

func NewAllocator

func NewAllocator() *Allocator

NewAllocator creates a byte-slice allocator for frames up to 65536 bytes. Allocations use power-of-two capacities, wasting less than half the capacity.

func (*Allocator) Get

func (alloc *Allocator) Get(size int) *[]byte

Get returns a pooled byte slice sized for size, or nil if size is outside the supported range.

func (*Allocator) Put

func (alloc *Allocator) Put(p *[]byte) error

Put returns a byte slice to the pool. The slice capacity must be a power of two between 1 and 65536.

type CLASSID

type CLASSID int

CLASSID identifies a frame's scheduling class.

const (
	CLSCTRL CLASSID = iota // prioritized control signal
	CLSDATA
)

type Config

type Config struct {
	// Version is the smux protocol version: 1 or 2.
	Version int

	// KeepAliveDisabled disables keep-alive messages and timeout checks.
	KeepAliveDisabled bool

	// KeepAliveInterval is the interval between NOP keep-alive messages.
	KeepAliveInterval time.Duration

	// KeepAliveTimeout is the maximum time without received data before the
	// session is closed.
	KeepAliveTimeout time.Duration

	// MaxFrameSize is the maximum frame payload size sent to the peer.
	MaxFrameSize int

	// MaxReceiveBuffer is the maximum amount of data buffered across all streams.
	MaxReceiveBuffer int

	// MaxStreamBuffer is the receive-window size advertised for each stream in
	// protocol version 2.
	MaxStreamBuffer int
}

Config contains the settings used to create a smux session.

func DefaultConfig

func DefaultConfig() *Config

DefaultConfig returns the default session configuration.

type Frame

type Frame struct {
	// contains filtered or unexported fields
}

Frame represents a packet multiplexed over a single connection.

type Session

type Session struct {
	// contains filtered or unexported fields
}

Session manages streams multiplexed over a single connection.

func Client

func Client(conn io.ReadWriteCloser, config *Config) (*Session, error)

Client initializes a new client-side session over conn.

func Server

func Server(conn io.ReadWriteCloser, config *Config) (*Session, error)

Server initializes a new server-side session over conn.

func (*Session) Accept

func (s *Session) Accept() (io.ReadWriteCloser, error)

Accept returns the next stream as a generic io.ReadWriteCloser.

func (*Session) AcceptStream

func (s *Session) AcceptStream() (*Stream, error)

AcceptStream waits for and returns the next incoming stream.

func (*Session) Close

func (s *Session) Close() error

Close is used to close the session and all streams.

func (*Session) CloseChan

func (s *Session) CloseChan() <-chan struct{}

CloseChan returns a channel that is closed when the session shuts down.

func (*Session) IsClosed

func (s *Session) IsClosed() bool

IsClosed reports whether the session has shut down.

func (*Session) LocalAddr

func (s *Session) LocalAddr() net.Addr

LocalAddr returns the underlying connection's local address, if available.

func (*Session) NumStreams

func (s *Session) NumStreams() int

NumStreams returns the number of streams currently registered in the session.

func (*Session) Open

func (s *Session) Open() (io.ReadWriteCloser, error)

Open opens a stream and returns it as an io.ReadWriteCloser.

func (*Session) OpenStream

func (s *Session) OpenStream() (*Stream, error)

OpenStream opens a new stream on the session.

func (*Session) RemoteAddr

func (s *Session) RemoteAddr() net.Addr

RemoteAddr returns the underlying connection's remote address, if available.

func (*Session) SetDeadline

func (s *Session) SetDeadline(t time.Time) error

SetDeadline sets a deadline used by Accept* calls. A zero time value disables the deadline.

type Stream

type Stream struct {
	// contains filtered or unexported fields
}

Stream is a wrapper around stream that exposes the public API.

func (Stream) Close

func (s Stream) Close() error

Close implements net.Conn Close fully closes the stream (both read and write sides).

func (Stream) CloseWrite

func (s Stream) CloseWrite() error

CloseWrite implements half-close by closing the write side of the stream. After CloseWrite, the stream can still receive data from the peer, but any further writes will return io.ErrClosedPipe. This is similar to net.TCPConn.CloseWrite().

func (Stream) GetDieCh

func (s Stream) GetDieCh() <-chan struct{}

GetDieCh returns a channel that is closed when the stream is closed.

func (Stream) ID

func (s Stream) ID() uint32

ID returns the stream's unique identifier.

func (Stream) LocalAddr

func (s Stream) LocalAddr() net.Addr

LocalAddr returns the underlying connection's local address, if available.

func (Stream) Read

func (s Stream) Read(b []byte) (n int, err error)

Read reads data from the stream into b.

func (Stream) RemoteAddr

func (s Stream) RemoteAddr() net.Addr

RemoteAddr returns the underlying connection's remote address, if available.

func (Stream) SetDeadline

func (s Stream) SetDeadline(t time.Time) error

SetDeadline sets both read and write deadlines as defined by net.Conn.SetDeadline. A zero time value disables the deadlines.

func (Stream) SetReadDeadline

func (s Stream) SetReadDeadline(t time.Time) error

SetReadDeadline sets the read deadline as defined by net.Conn.SetReadDeadline. A zero time value disables the deadline.

func (Stream) SetWriteDeadline

func (s Stream) SetWriteDeadline(t time.Time) error

SetWriteDeadline sets the write deadline as defined by net.Conn.SetWriteDeadline. A zero time value disables the deadline.

func (Stream) Write

func (s Stream) Write(b []byte) (n int, err error)

Write implements net.Conn

Concurrent Write calls may interleave at frame boundaries, and their ordering is nondeterministic.

func (Stream) WriteTo

func (s Stream) WriteTo(w io.Writer) (n int64, err error)

WriteTo implements io.WriteTo WriteTo writes stream data to w until the stream reaches EOF or an error occurs. It returns the number of bytes written and any error encountered. For protocol version 2, it sends window updates to the peer as needed.