README
¶
Introduction
gmux makes underlying net.Conn to upper level multiplexing connection, and user could open thousands of logical streams from the upper level multiplexing connection.
gmux ( Named 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
- Token bucket controlled receiving, which provides smoother bandwidth graph(see picture below).
- Session-wide receive buffer, shared among streams, fully controlled overall memory usage.
- Minimized header(8Bytes), maximized payload.
- Well-tested on millions of devices in kcptun.
- Builtin fair queue traffic shaping.
- 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()
}
Thanks
It is based on https://github.com/xtaci/smux/tree/09e2c01560df5aaaed50e48e77547858e2623498
Thank you!
Documentation
¶
Overview ¶
Package gmux 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 over a single channel.
Index ¶
- Variables
- func VerifyConfig(config *Config) error
- type Allocator
- type CloseNotifier
- type Config
- type Frame
- type MuxConnIF
- type Session
- func (s *Session) Accept() (io.ReadWriteCloser, error)
- func (s *Session) AcceptStream() (*Stream, error)
- func (s *Session) Close() error
- func (s *Session) GetNoDataTimeout() time.Duration
- func (s *Session) IsClosed() bool
- func (s *Session) LocalAddr() net.Addr
- func (s *Session) NumStreams() int
- func (s *Session) Open(streamName string) (io.ReadWriteCloser, error)
- func (s *Session) OpenStream(streamName string) (*Stream, error)
- func (s *Session) RemoteAddr() net.Addr
- func (s *Session) SetDeadline(t time.Time) error
- func (s *Session) SetNoDataTimeout(noDataTimeout time.Duration) error
- type Stream
- func (s *Stream) Close() error
- func (s *Stream) GetDieCh() <-chan struct{}
- func (s *Stream) GetNoDataTimeout() time.Duration
- func (s *Stream) ID() uint32
- func (s *Stream) LocalAddr() net.Addr
- func (s *Stream) Name() string
- func (s *Stream) Read(b []byte) (n int, err error)
- func (s *Stream) RemoteAddr() net.Addr
- func (s *Stream) SetCloseNotifier(notifier CloseNotifier, ctx interface{})
- func (s *Stream) SetDeadline(t time.Time) error
- func (s *Stream) SetNoDataTimeout(noDataTimeout time.Duration) 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)
- type StreamIF
- Bugs
Constants ¶
This section is empty.
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 = errors.New("timeout") ErrWouldBlock = errors.New("operation would block on IO") ErrInvalidCommand = "invalid command 0x%x" ErrInvalidVersion = "invalid version %d from %s" )
Functions ¶
func VerifyConfig ¶
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%.
type CloseNotifier ¶
type CloseNotifier func(stream StreamIF, ctx interface{})
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 Frame is friendly to upper layer users, but when sending data over the network, the data is not organized by Frame, but by rawHeader.
type MuxConnIF ¶
type MuxConnIF interface {
Open(streamName string) (io.ReadWriteCloser, error)
Accept() (io.ReadWriteCloser, error)
IsClosed() bool
NumStreams() int
LocalAddr() net.Addr
RemoteAddr() net.Addr
Close() error
}
MuxConnIF is an interface for upper level multiplexing connection which based on underlying net.Conn.
type Session ¶
type Session struct {
// contains filtered or unexported fields
}
TODO: rename to MuxConn Session defines a multiplexed connection for streams
func NewClient ¶
func NewClient(conn io.ReadWriteCloser, config *Config) (*Session, error)
NewClient is used to initialize a new client-side connection. It wraps client side underlying net.Conn to upper level multiplexing connection.
func NewServer ¶
func NewServer(conn io.ReadWriteCloser, config *Config) (*Session, error)
NewServer 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 ¶
AcceptStream is used to block until the next available stream is ready to be accepted.
func (*Session) GetNoDataTimeout ¶
GetNoDataTimeout returns current noDataTimeout.
func (*Session) NumStreams ¶
NumStreams returns the number of currently open streams
func (*Session) Open ¶
func (s *Session) Open(streamName string) (io.ReadWriteCloser, error)
Open returns a generic ReadWriteCloser
func (*Session) OpenStream ¶
OpenStream is used to create a new stream
func (*Session) RemoteAddr ¶
RemoteAddr satisfies net.Conn interface
func (*Session) SetDeadline ¶
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 implements net.Conn, it is a logical stream.
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) GetNoDataTimeout ¶
GetNoDataTimeout returns current noDataTimeout.
func (*Stream) RemoteAddr ¶
RemoteAddr satisfies net.Conn interface
func (*Stream) SetCloseNotifier ¶
func (s *Stream) SetCloseNotifier(notifier CloseNotifier, ctx interface{})
func (*Stream) SetDeadline ¶
SetDeadline sets both read and write deadlines as defined by net.Conn.SetDeadline. A zero time value disables the deadlines.
func (*Stream) SetNoDataTimeout ¶
SetNoDataTimeout could be called multiple times, even if Read/Write/WriteTo is being executed or after.
func (*Stream) SetReadDeadline ¶
SetReadDeadline sets the read deadline as defined by net.Conn.SetReadDeadline. A zero time value disables the deadline.
func (*Stream) SetWriteDeadline ¶
SetWriteDeadline sets the write deadline as defined by net.Conn.SetWriteDeadline. A zero time value disables the deadline.
Source Files