smux

README

Introduction

Smux ( Simple MUltipleXing) is a multiplexing library for Golang. It relies on an underlying connection to provide reliability and ordering, such as TCP or KCP, and provides stream-oriented multiplexing. The original intention of this library is to power the connection management for kcp-go.

Features

1. Token bucket controlled receiving, which provides smoother bandwidth graph(see picture below).
2. Session-wide receive buffer, shared among streams, fully controlled overall memory usage.
3. Minimized header(8Bytes), maximized payload.
4. Well-tested on millions of devices in kcptun.
5. Builtin fair queue traffic shaping.
6. Per-stream sliding window to control congestion.(protocol version 2+).

Documentation

For complete documentation, see the associated Godoc.

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(LENGTH)  

VALUES FOR LATEST VERSION:
VERSION:
    1/2
    
CMD:
    cmdSYN(0)
    cmdFIN(1)
    cmdPSH(2)
    cmdNOP(3)
    cmdUPD(4)	// only supported on version 2
    
STREAMID:
    client use odd numbers starts from 1
    server use even numbers starts from 0
    
cmdUPD:
    | CONSUMED(4B) | WINDOW(4B) |

Usage

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

    // Setup client side of smux
    session, err := smux.Client(conn, nil)
    if err != nil {
        panic(err)
    }

    // Open a new 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)
    }

    // Setup server side of smux
    session, err := smux.Server(conn, nil)
    if err != nil {
        panic(err)
    }

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

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

Status

Stable

Documentation

Index

• Variables
• func VerifyConfig(config *Config) error
• type Allocator
  • func NewAllocator() *Allocator
  • func (alloc *Allocator) Get(size int) []byte
  • func (alloc *Allocator) Put(buf []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) 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 VerifyConfig

func VerifyConfig(config *Config) error

VerifyConfig is used to verify the sanity of configuration

Types

type Allocator

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

Allocator for incoming frames, optimized to prevent overwriting after zeroing

func NewAllocator

func NewAllocator() *Allocator

NewAllocator initiates a []byte allocator for frames less than 65536 bytes, the waste(memory fragmentation) of space allocation is guaranteed to be no more than 50%.

func (*Allocator) Get

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

Get a []byte from pool with most appropriate cap

func (*Allocator) Put

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

Put returns a []byte to pool for future use, which the cap must be exactly 2^n

type CLASSID

type CLASSID int

CLASSID represents the class of a frame

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

type Config

type Config struct {
	// SMUX Protocol version, support 1,2
	Version int

	// Disabled keepalive
	KeepAliveDisabled bool

	// KeepAliveInterval is how often to send a NOP command to the remote
	KeepAliveInterval time.Duration

	// KeepAliveTimeout is how long the session
	// will be closed if no data has arrived
	KeepAliveTimeout time.Duration

	// MaxFrameSize is used to control the maximum
	// frame size to sent to the remote
	MaxFrameSize int

	// MaxReceiveBuffer is used to control the maximum
	// number of data in the buffer pool
	MaxReceiveBuffer int

	// MaxStreamBuffer is used to control the maximum
	// number of data per stream
	MaxStreamBuffer int
}

Config is used to tune the Smux session

func DefaultConfig

func DefaultConfig() *Config

DefaultConfig is used to return a default configuration

type Frame

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

Frame defines a packet from or to be multiplexed into a single connection

type Session

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

Session defines a multiplexed connection for streams

func Client

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

Client is used to initialize a new client-side connection.

func Server

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

Server is used to initialize a new server-side connection.

func (*Session) Accept

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

Accept Returns a generic ReadWriteCloser instead of smux.Stream

func (*Session) AcceptStream

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

AcceptStream is used to block until the next available stream is ready to be accepted.

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 can be used by someone who wants to be notified immediately when this session is closed

func (*Session) IsClosed

func (s *Session) IsClosed() bool

IsClosed does a safe check to see if we have shutdown

func (*Session) LocalAddr

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

LocalAddr satisfies net.Conn interface

func (*Session) NumStreams

func (s *Session) NumStreams() int

NumStreams returns the number of currently open streams

func (*Session) Open

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

Open returns a generic ReadWriteCloser

func (*Session) OpenStream

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

OpenStream is used to create a new stream

func (*Session) RemoteAddr

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

RemoteAddr satisfies net.Conn interface

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
}

wrapper for GC

func (Stream) Close

func (s Stream) Close() error

Close implements net.Conn

func (Stream) GetDieCh

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

GetDieCh returns a readonly chan which can be readable when the stream is to be 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 satisfies net.Conn interface

func (Stream) Read

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

Read reads data from the stream into the provided buffer.

func (Stream) RemoteAddr

func (s Stream) RemoteAddr() net.Addr

RemoteAddr satisfies net.Conn interface

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

Note that the behavior when multiple goroutines write concurrently is not deterministic, frames may interleave in random way.

func (Stream) WriteTo

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

WriteTo implements io.WriteTo WriteTo writes data to w until there's no more data to write or when an error occurs. The return value n is the number of bytes written. Any error encountered during the write is also returned. WriteTo calls Write in a loop until there is no more data to write or when an error occurs. If the underlying stream is a v2 stream, it will send window update to peer when necessary. If the underlying stream is a v1 stream, it will not send window update to peer.