tnet

package module
v1.0.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Apr 15, 2024 License: Apache-2.0, BSD-2-Clause, BSD-3-Clause, + 1 more Imports: 26 Imported by: 13

README

English | 中文

Introduction

Go Reference Go Report Card LICENSE Releases Tests Coverage

tnet is an event loop networking framework that provides versatile models. The core aims are:

  • Support more connections (millions of)
  • Higher performance (QPS↑, latency↓)
  • Less memory (use only up to ~10% memory compared with golang/net)
  • Easy to use (compatible with golang/net)

Features:

  • Support TCP, UDP
  • Support IPv4, IPv6
  • Provide blocking/nonblocking Read/Write API
  • Provide batch system calls: ReadV/WriteV
  • Support both server and client programming
  • Support Linux / Mac OS
  • Support TLS
  • Support WebSocket

Getting Started

For the tnet network library itself, it provides two classes of usage:

  1. The user handler is not in the poller goroutine
  2. The user handler is in the poller goroutine

As shown below:

The tnet provides the first class of usage by default, and the second class of usage is opened by tnet.WithNonBlocking(true) (see the detailed examples in the examples folder).

On the basis of these two classes, the user can further differentiate into a more fine-grained mode by choosing whether to use the business goroutine pool in the handler. Specifically,

  • Do not use the business goroutine pool in Class 1, which corresponds to the "classical Go/net mode". Features:

Number of goroutines = number of pollers + number of active connections. It is suitable for business scenarios where network IO and CPU processing are balanced. The disadvantage is that connection multiplexing is not supported, and concurrent processing of services cannot be supported on the same connection (IO processing part cannot be concurrently processed.)

For Class 2 where handler is in the poller goroutine, the handler function can usually be divided into the following two modes:

  1. Separated IO processing and business mode
  2. Combined IO processing and business mode
  • Use the business goroutine pool in Class 2, which corresponds to the "separated IO processing and business mode". Features:

Number of goroutines = number of pollers + number of concurrent processing of data packets, suitable for CPU-intensive scenarios

  • Do not use the business goroutine pool in Class 2, which corresponds to the " combined IO processing and business mode". Features:

Number of goroutines = number of pollers, but the use case is scarce. The classic usage is for gateway scenario where most of the logic is just forwarding requests. The processing time of each request is very low, and there will be no blocking.

Supported TCP Option

  • tnet.WithTCPKeepAlive sets the time interval for keep alive. The default value is 15s. When set to 0, keep alive can be closed.
// WithTCPKeepAlive sets the tcp keep alive interval.
func WithTCPKeepAlive(keepAlive time.Duration) Option {
	return Option{func(op *options) {
		op.tcpKeepAlive = keepAlive
	}}
}
  • tnet.WithTCPIdleTimeout sets the idle timeout of the connection, and the connection will be automatically disconnected when the idle time exceeds the given value.
// WithTCPIdleTimeout sets the idle timeout to close tcp connection.
func WithTCPIdleTimeout(idleTimeout time.Duration) Option {
	return Option{func(op *options) {
		op.tcpIdleTimeout = idleTimeout
	}}
}
  • tnet.WithOnTCPOpened can set the operations that need to be performed when the TCP connection is just established.
// WithOnTCPOpened registers the OnTCPOpened method that is fired when connection is established.
func WithOnTCPOpened(onTCPOpened OnTCPOpened) Option {
	return Option{func(op *options) {
		op.onTCPOpened = onTCPOpened
	}}
}
  • tnet.WithOnTCPClosed can set the operations that need to be performed when the TCP connection is disconnected.
// WithOnTCPClosed registers the OnTCPClosed method that is fired when tcp connection is closed.
func WithOnTCPClosed(onTCPClosed OnTCPClosed) Option {
	return Option{func(op *options) {
		op.onTCPClosed = onTCPClosed
	}}
}
  • tnet.WithTCPFlushWrite(true) enables the user to complete the sending of the package directly in the current business goroutine.
// WithTCPFlushWrite sets whether use flush write for TCP
// connection or not. Default is notify.
func WithTCPFlushWrite(flush bool) Option {
	return Option{func(op *options) {
		op.flushWrite = flush
	}}
}

Take separated mode as an example, the procedure is like:

Supported UDP Option

  • tnet.WithOnUDPClosed can set the operation that needs to be performed when UDP is closed.
// WithOnUDPClosed registers the OnUDPClosed method that is fired when udp connection is closed.
func WithOnUDPClosed(onUDPClosed OnUDPClosed) Option {
	return Option{func(op *options) {
		op.onUDPClosed = onUDPClosed
	}}
}
  • tnet.WithMaxUDPPacketSize sets the maximum length of UDP packets.
// WithMaxUDPPacketSize sets maximal UDP packet size when receiving UDP packets.
func WithMaxUDPPacketSize(size int) Option {
	return Option{func(op *options) {
		op.maxUDPPacketSize = size
	}}
}

Supported common Option

  • tnet.WithNonBlocking can set blocking/non-blocking mode, which is also an option to control whether IO processing is in the Poller goroutine, the default is blocking mode with IO processing not in the Poller goroutine.
// WithNonBlocking set conn/packconn to nonblocking mode
func WithNonBlocking(nonblock bool) Option {
	return Option{func(op *options) {
		op.nonblocking = nonblock
	}}
}

Explanation of special APIs

The tnet.Conn interface is extended on the basis of net.Conn.

  • The four zero-copy APIs are as follows:
// Peek returns the next n bytes without advancing the reader. It waits until it has
// read at least n bytes or error occurs such as connection closed or read timeout.
// The bytes stop being valid at the next ReadN or Release call.
// Zero-Copy API.
Peek(n int) ([]byte, error)

// Next returns the next n bytes with advancing the reader, It waits until it has
// read at least n bytes or error occurs such as connection closed or read timeout.
// The bytes stop being valid at the next ReadN or Release call.
// Zero-Copy API.
Next(n int) ([]byte, error)

// Skip the next n bytes and advance the reader. It waits until the underlayer has at
// least n bytes or error occurs such as connection closed or read timeout.
// Zero-Copy API.
Skip(n int) error

// Release releases underlayer buffer when using Peek() and Skip() Zero-Copy APIs.
Release()
  1. Peek: Read a given number of bytes, but do not move the read pointer of the underlying Linked Buffer. The returned byte slice is directly referenced from the Linked Buffer. At this time, it is required that this part of the data cannot be released before it is used up.
  2. Skip: Move the read pointer of the underlying Linked Buffer to skip a given number of bytes, usually used with Peek.
  3. Next: It is equivalent to calling Peek first, then calling Skip, the returned byte slice will be invalid after the call of Release.
  4. Release: Release the read part, usually after using the byte slice. When calling the security API Read/ReadN, Release will be automatically called to release the read buffer.
  • The tnet.Conn interface provides Writev, which can be used to write out multiple data blocks in turn, such as packet header and packet body, without manual data packet splicing. See examples/tcp/classical/main.go for details.
// Writev provides multiple data slice write in order.
Writev(p ...[]byte) (int, error)
  • tnet.Conn also provides a method to sense the connection state:
// IsActive checks whether the connection is active or not.
IsActive() bool
  • tnet.Conn provides SetMetaData/GetMetadata for storing/loading the user's private data on the connection:
// SetMetaData sets meta data. Through this method, users can bind some custom data to a connection.
SetMetaData(m interface{})
// GetMetaData gets meta data.
GetMetaData() interface{}

Use cases

  • tRPC-Go

tRPC-Go has integrated tnet.

Documentation

Overview

Package tnet provides event loop networking framework.

Index

Constants

This section is empty.

Variables

View Source 
var (
	// DefaultCleanUpThrottle is a default connections number throttle to determine
	// whether to enable buffer clean up feature.
	DefaultCleanUpThrottle = 10000
	// ErrConnClosed connection is closed.
	ErrConnClosed = netError{/* contains filtered or unexported fields */}
	// EAGAIN represents error of not enough data.
	EAGAIN = netError{/* contains filtered or unexported fields */}
)
View Source 
var MassiveConnections bool

MassiveConnections denotes whether this is under heavy connections scenario.

Functions

func EnablePollerGoschedAfterEvent 

func EnablePollerGoschedAfterEvent()

EnablePollerGoschedAfterEvent enables calling runtime.Gosched() after processing of each event during epoll wait handling. This function can only be called inside func init().

func Listen 

func Listen(network, address string) (net.Listener, error)

Listen announces on the local network address. The network must be "tcp", "tcp4", "tcp6".

func NumPollers 

func NumPollers() int

NumPollers returns the current number of pollers.

func SetNumPollers 

func SetNumPollers(n int) error

SetNumPollers is used to set the number of pollers. Generally it is not actively used. Note that n can't be smaller than the current poller numbers.

NOTE: the default poller number is 1.

func Submit 

func Submit(task func()) error

Submit submits a task to usrPool.

Users can use this API to submit a task to the default user business goroutine pool.

Types

type BaseConn 

type BaseConn interface {
	// Conn extends net.Conn, just for interface compatibility.
	net.Conn

	// Len returns the total length of the readable data in the reader.
	Len() int

	// IsActive checks whether the connection is active or not.
	IsActive() bool

	// SetNonBlocking sets conn to nonblocking. Read APIs will return EAGAIN when there is no
	// enough data for reading
	SetNonBlocking(nonblock bool)

	// SetFlushWrite sets whether to flush the data or not.
	// Default value is false.
	// Deprecated: whether enable this feature is controlled by system automatically.
	SetFlushWrite(flushWrite bool)

	// SetMetaData sets metadata. Through this method, users can bind some custom data to a connection.
	SetMetaData(m any)

	// GetMetaData gets metadata.
	GetMetaData() any
}

BaseConn is common for stream and packet oriented network connection.

type Conn 

type Conn interface {
	BaseConn

	// Peek returns the next n bytes without advancing the reader. It waits until it has
	// read at least n bytes or error occurs such as connection closed or read timeout.
	// The bytes stop being valid at the next ReadN or Release call.
	// Zero-Copy API.
	Peek(n int) ([]byte, error)

	// Next returns the next n bytes with advancing the reader, It waits until it has
	// read at least n bytes or error occurs such as connection closed or read timeout.
	// The bytes stop being valid at the next ReadN or Release call.
	// Zero-Copy API.
	Next(n int) ([]byte, error)

	// Skip the next n bytes and advance the reader. It waits until the underlayer has at
	// least n bytes or error occurs such as connection closed or read timeout.
	// Zero-Copy API.
	Skip(n int) error

	// Release releases underlayer buffer when using Peek() and Skip() Zero-Copy APIs.
	Release()

	// ReadN is similar to Peek(), except that it will copy the n bytes data from the underlayer,
	// and advance the reader.
	ReadN(n int) ([]byte, error)

	// Writev provides multiple data slice write in order.
	// The default behavior of Write/Writev will hold a reference to the given byte slices p,
	// therefore if the caller want to reuse byte slice p after calling Write/Writev, the
	// SetSafeWrite(true) option is required.
	Writev(p ...[]byte) (int, error)

	// SetKeepAlive sets keep alive time for tcp connection.
	// By default, keep alive is turned on with value defaultKeepAlive.
	// If keepAlive <= 0, keep alive will be turned off.
	// Otherwise, keep alive value will be round up to seconds.
	SetKeepAlive(t time.Duration) error

	// SetOnRequest can set or replace the TCPHandler method for a connection.
	// Generally, on the server side the handler is set when the connection is established.
	// On the client side, if necessary, make sure that TCPHandler is set before sending data.
	SetOnRequest(handle TCPHandler) error

	// SetOnClosed sets the additional close process for a connection.
	// Handle is executed when the connection is closed.
	SetOnClosed(handle OnTCPClosed) error

	// SetIdleTimeout sets the idle timeout to close connection.
	SetIdleTimeout(d time.Duration) error

	// SetSafeWrite sets whether writing on connection is safe or not.
	// Default is unsafe.
	//
	// This option affects the behavior of Write/Writev.
	//   If safeWrite = false: the lifetime of buffers passed into Write/Writev will
	//     be handled by tnet, which means users cannot reuse the buffers after passing
	//     them into Write/Writev.
	//   If safeWrite = true: the given buffers is copied into tnet's own buffer.
	//     Therefore users can reuse the buffers passed into Write/Writev.
	SetSafeWrite(safeWrite bool)
}

Conn is generic for stream oriented network connection.

func DialTCP 

func DialTCP(network, address string, timeout time.Duration) (Conn, error)

DialTCP connects to the address on the named network within the timeout. Valid networks for DialTCP are "tcp", "tcp4" (IPv4-only), "tcp6" (IPv6-only).

type OnTCPClosed 

type OnTCPClosed func(conn Conn) error

OnTCPClosed fires when the tcp connection is closed. In this method, please do not perform read-write operations, because the connection has been closed. But you can still manipulate the MetaData in the connection.

type OnTCPOpened 

type OnTCPOpened func(conn Conn) error

OnTCPOpened fires when the tcp connection is established.

type OnUDPClosed 

type OnUDPClosed func(conn PacketConn) error

OnUDPClosed fires when the udp connection is closed. In this method, please do not perform read-write operations, because the connection has been closed. But you can still manipulate the MetaData in the connection.

type Option 

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

Option tnet service option.

func WithFlushWrite 

func WithFlushWrite(flush bool) Option

WithFlushWrite sets whether use flush write for TCP and UDP connection or not. Default value is false. Deprecated: whether enable this feature is controlled by system automatically.

func WithMaxUDPPacketSize 

func WithMaxUDPPacketSize(size int) Option

WithMaxUDPPacketSize sets maximal UDP packet size when receiving UDP packets.

func WithNonBlocking 

func WithNonBlocking(nonblock bool) Option

WithNonBlocking set conn/packconn to nonblocking mode

func WithOnTCPClosed 

func WithOnTCPClosed(onTCPClosed OnTCPClosed) Option

WithOnTCPClosed registers the OnTCPClosed method that is fired when tcp connection is closed.

func WithOnTCPOpened 

func WithOnTCPOpened(onTCPOpened OnTCPOpened) Option

WithOnTCPOpened registers the OnTCPOpened method that is fired when connection is established.

func WithOnUDPClosed 

func WithOnUDPClosed(onUDPClosed OnUDPClosed) Option

WithOnUDPClosed registers the OnUDPClosed method that is fired when udp connection is closed.

func WithSafeWrite 

func WithSafeWrite(safeWrite bool) Option

WithSafeWrite sets the value of safeWrite for TCP. Default value is false.

This option affects the behavior of Write/Writev. 

If safeWrite = false: the lifetime of buffers passed into Write/Writev will
  be handled by tnet, which means users cannot reuse the buffers after passing
  them into Write/Writev.
If safeWrite = true: the given buffers is copied into tnet's own buffer.
  Therefore users can reuse the buffers passed into Write/Writev.

func WithTCPFlushWrite 

func WithTCPFlushWrite(flush bool) Option

WithTCPFlushWrite sets whether use flush write for TCP connection or not. Default value is false. Deprecated: whether enable this feature is controlled by system automatically.

func WithTCPIdleTimeout 

func WithTCPIdleTimeout(idleTimeout time.Duration) Option

WithTCPIdleTimeout sets the idle timeout to close tcp connection.

func WithTCPKeepAlive 

func WithTCPKeepAlive(keepAlive time.Duration) Option

WithTCPKeepAlive sets the tcp keep alive interval.

type Packet 

type Packet interface {
	// Data returns the data of the packet.
	Data() ([]byte, error)

	// Free will release the underlying buffer.
	// It will recycle the underlying buffer for better performance.
	// The bytes will be invalid after free, so free it only when it is no longer in use.
	Free()
}

Packet represents a UDP packet, created by PacketConn Zero-Copy API ReadPacket.

type PacketConn 

type PacketConn interface {
	BaseConn

	// PacketConn extends net.PacketConn, just for interface compatibility.
	net.PacketConn

	// ReadPacket reads a packet from the connection, without copying the underlying buffer.
	// Get the actual data of packet by Packet.Data().
	// Please call Packet.Free() when it is unused, free will recycle the underlying buffer
	// for better performance.
	// Zero-copy API
	ReadPacket() (Packet, net.Addr, error)

	// SetMaxPacketSize sets maximal UDP packet size when receiving UDP packets.
	SetMaxPacketSize(size int)

	// SetOnRequest can set or replace the UDPHandler method for a connection.
	// However, the handler can't be set to nil.
	// Generally, on the server side the handler is set when the connection is established.
	// On the client side, if necessary, make sure that UDPHandler is set before sending data.
	SetOnRequest(handle UDPHandler) error

	// SetOnClosed sets the additional close process for a connection.
	// Handle is executed when the connection is closed.
	SetOnClosed(handle OnUDPClosed) error
}

PacketConn is generic for packet oriented network connection.

func DialUDP 

func DialUDP(network, address string, timeout time.Duration) (PacketConn, error)

DialUDP connects to the address on the named network within the timeout. Valid networks for DialUDP are "udp", "udp4" (IPv4-only), "udp6" (IPv6-only).

func ListenPackets 

func ListenPackets(network, address string, reuseport bool) ([]PacketConn, error)

ListenPackets announces on the local network address. Reuseport sets whether to enable reuseport when creating PacketConns, it will return multiple PacketConn if reuseprot is true. Generally, enabling reuseport can make effective use of multi-core and improve performance.

func NewPacketConn 

func NewPacketConn(conn net.PacketConn) (PacketConn, error)

NewPacketConn creates a tnet.PacketConn from net.PacketConn. Note that conn must listen on UDP and make sure that conn implements syscall.Conn.

type Service 

type Service interface {
	// Serve registers a listener and runs blockingly to provide service, including listening to ports,
	// accepting connections and reading trans data.
	// Param ctx is used to shutdown the service with all connections gracefully.
	Serve(ctx context.Context) error
}

Service provides startup method to udp/tcp server.

func NewTCPService 

func NewTCPService(listener net.Listener, handler TCPHandler, opt ...Option) (Service, error)

NewTCPService creates a tcp Service and binds it to a listener. It is recommended to create listener by func tnet.Listen, otherwise make sure that listener implements syscall.Conn interface. 

type syscall.Conn interface {
	SyscallConn() (RawConn, error)
}

func NewUDPService 

func NewUDPService(lns []PacketConn, handler UDPHandler, opt ...Option) (Service, error)

NewUDPService creates a udp service. Ensure that all listeners are listening to the same address.

type TCPHandler 

type TCPHandler func(conn Conn) error

TCPHandler fires when the tcp connection receives data.

type UDPHandler 

type UDPHandler func(conn PacketConn) error

UDPHandler fires when the udp connection receives data.

Source Files

View all Source files

Directories

Path Synopsis
examples module
extensions
websocket Module
internal
asynctimer
Package asynctimer provides asynchronous timer function, which is implemented by time wheel.
 Package asynctimer provides asynchronous timer function, which is implemented by time wheel.
autopostpone
Package autopostpone provides utilities to decide whether to postpone write.
 Package autopostpone provides utilities to decide whether to postpone write.
buffer
Package buffer provides linked buffers.
 Package buffer provides linked buffers.
cache/mcache
Package mcache provides cache for byte slice.
 Package mcache provides cache for byte slice.
cache/systype
Package systype provides system type such as unix.Ioves.
 Package systype provides system type such as unix.Ioves.
iovec
Package iovec provides utilities to work with unix.Iovec.
 Package iovec provides utilities to work with unix.Iovec.
locker
Package locker provides locking utilities.
 Package locker provides locking utilities.
netutil
poller
Package poller provides event driven polling system to monitor file description events.
 Package poller provides event driven polling system to monitor file description events.
poller/event
Package event provides definitions of event data.
 Package event provides definitions of event data.
reuseport
safejob
Package safejob provides functions to call job in a concurrent-safe manner.
 Package safejob provides functions to call job in a concurrent-safe manner.
timer
Package timer provides functions of timer.
 Package timer provides functions of timer.
Package log provides logging utilities for tnet.
 Package log provides logging utilities for tnet.
Package metrics provides a lot of tnet runtime monitoring data, such as monitoring the efficiency of batch reads and writes, which is a good tool for performance tuning.
 Package metrics provides a lot of tnet runtime monitoring data, such as monitoring the efficiency of batch reads and writes, which is a good tool for performance tuning.
tls
Package tls provides tls connection utilities.
 Package tls provides tls connection utilities.
examples/echo/client
Package main is the main package.
 Package main is the main package.
examples/echo/server
Package main is the main package.
 Package main is the main package.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.