gain

README

Gain

Gain is a high-performance io_uring networking framework written entirely in Go.

Report Bug · Request Feature

Table of Contents

1. About The Project
2. Getting Started
   • Prerequisites
   • Installation
3. Usage
4. Roadmap
5. Performance
6. Contributing
7. License
8. Contact
9. Relevant Articles

About The Project

Gain is a high-performance networking framework written entirely in Go. It uses io_uring - a new asynchronous I/O API for Linux created by Jens Axboe from Facebook. Currently only Linux is supported.

WARNING: This is an alpha version so it is not yet stable enough to use in a production environment.

Articles about the project:

• Medium - Meet Gain - the New Fastest Go TCP Framework
• Medium - Writing High-performance TCP Applications Using the Gain Web Framework

Getting Started

See examples:

• CLI server example
• TCP echo server example

Prerequisites

Gain requires Go 1.19+

Installation

1. Install the framework

   go get -u github.com/pawelgaczynski/gain@v0.4.0-alpha
   

(back to top)

Roadmap

• Go liburing port
• Reactor architecture
• Socket sharding architecture
• Async workers and workers pool
• Lock-free
• Protocols
  • TCP
  • UDP
  • Unix Domain Socket
• Load balancing
  • Round robin
  • Least connection
  • Source IP hash
  • Support for custom implementations
• Support for read and write deadlines
• Further io_uring optimizations
• More flexible connection buffer
• Documentation
• Support for kernels older than 5.15
• Support for Windows - IoRing (documentation)

See the open issues for a full list of proposed features (and known issues).

(back to top)

Performance

AWS EC2 instance: m6i.xlarge
vCPU: 4
RAM: 16GB
OS: Ubuntu 22.04 LTS
Kernel: 5.15.0-1026-aws
Go: go1.19.3 linux/amd64
Number of connections: 512
Benchmark type: see TechEmpower Plaintext

Perfect locality

Run this script on the server machine before starting Gain. The first parameter is the name of network interface (e.g. eth0, ens5).

#!/bin/bash

systemctl stop irqbalance.service

export IRQS=($(grep $1 /proc/interrupts | awk '{print $1}' | tr -d :))
for i in ${!IRQS[@]}; do echo $i > /proc/irq/${IRQS[i]}/smp_affinity_list; done;

export TXQUEUES=($(ls -1qdv /sys/class/net/$1/queues/tx-*))
for i in ${!TXQUEUES[@]}; do printf '%x' $((2**i)) > ${TXQUEUES[i]}/xps_cpus; done;

To understand how it was achieved read an excellent article: Extreme HTTP Performance Tuning

(back to top)

Contributing

Contributions are what make the open source community such an amazing place to learn, inspire, and create. Any contributions you make are greatly appreciated.

If you have a suggestion that would make this better, please fork the repo and create a pull request. You can also simply open an issue with the tag "enhancement". Don't forget to give the project a star! Thanks again!

1. Fork the Project
2. Create your Feature Branch (git checkout -b feature/AmazingFeature)
3. Commit your Changes (git commit -m 'Add some AmazingFeature')
4. Push to the Branch (git push origin feature/AmazingFeature)
5. Open a Pull Request

(back to top)

License

Distributed under the Apache 2.0 License. See LICENSE for more information.

(back to top)

Contact

Paweł Gaczyński - LinkedIn

Project Link: https://github.com/pawelgaczynski/gain

(back to top)

Relevant Articles

• Lord of the io_uring
• Ringing in a new asynchronous I/O API
• The rapid growth of io_uring
• An Introduction to the io_uring Asynchronous I/O Framework
• How io_uring and eBPF Will Revolutionize Programming in Linux
• Missing Manuals - io_uring worker pool
• IO_uring Continues Advancing In 2022 With New Features For Pushing Linux I/O
• Getting Hands on with io_uring using Go
• liburing Github

(back to top)

Documentation

Overview

Copyright (c) 2023 Paweł Gaczyński

Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Index

• func ListenAndServe(address string, eventHandler EventHandler, options ...ConfigOption) error
• type Config
  • func NewConfig(opts ...ConfigOption) Config
• type ConfigOption
  • func WithArchitecture(architecture ServerArchitecture) ConfigOption
  • func WithAsyncHandler(asyncHandler bool) ConfigOption
  • func WithCBPF(cbpf bool) ConfigOption
  • func WithCPUAffinity(cpuAffinity bool) ConfigOption
  • func WithGoroutinePool(goroutinePool bool) ConfigOption
  • func WithLoadBalancing(loadBalancing LoadBalancing) ConfigOption
  • func WithLoggerLevel(loggerLevel zerolog.Level) ConfigOption
  • func WithMaxCQEvents(maxCQEvents uint) ConfigOption
  • func WithMaxSQEntries(maxSQEntries uint) ConfigOption
  • func WithPrettyLogger(prettyLogger bool) ConfigOption
  • func WithProcessPriority(processPriority bool) ConfigOption
  • func WithSocketRecvBufferSize(size int) ConfigOption
  • func WithSocketSendBufferSize(size int) ConfigOption
  • func WithTCPKeepAlive(tcpKeepAlive time.Duration) ConfigOption
  • func WithWorkers(workers int) ConfigOption
• type Conn
• type DefaultEventHandler
  • func (e DefaultEventHandler) OnAccept(c Conn)
  • func (e DefaultEventHandler) OnClose(c Conn, err error)
  • func (e DefaultEventHandler) OnRead(c Conn, n int)
  • func (e DefaultEventHandler) OnStart(server Server)
  • func (e DefaultEventHandler) OnWrite(c Conn, n int)
• type EventHandler
• type LoadBalancing
• type Option
• type Reader
• type Server
  • func NewServer(eventHandler EventHandler, config Config) Server
• type ServerArchitecture
• type Socket
• type Writer

Functions

func ListenAndServe

func ListenAndServe(address string, eventHandler EventHandler, options ...ConfigOption) error

ListenAndServe starts a server with a given address and event handler. The server can be configured with additional options.

Types

type Config

type Config struct {
	// Architecture indicates one of the two available architectures: Reactor and SocketSharding.
	//
	// The Reactor design pattern has one input called Acceptor,
	// which demultiplexes the handling of incoming connections to Consumer workers.
	// The load balancing algorithm can be selected via configuration option.
	//
	// The Socket Sharding allows multiple workers to listen on the same address and port combination.
	// In this case the kernel distributes incoming requests across all the sockets.
	Architecture ServerArchitecture
	// AsyncHandler indicates whether the engine should run the OnRead EventHandler method in a separate goroutines.
	AsyncHandler bool
	// GoroutinePool indicates use of pool of bounded goroutines for OnRead calls.
	// Important: Valid only if AsyncHandler is true
	GoroutinePool bool
	// CPUAffinity determines whether each engine worker is locked to the one CPU.
	CPUAffinity bool
	// ProcessPriority sets the prority of the process to high (-19). Requires root privileges.
	ProcessPriority bool
	// Workers indicates the number of consumers or shard workers. The default is runtime.NumCPU().
	Workers int
	// CBPFilter uses custom BPF filter to improve the performance of the Socket Sharding architecture.
	CBPFilter bool
	// LoadBalancing indicates the load-balancing algorithm to use when assigning a new connection.
	// Important: valid only for Reactor architecture.
	LoadBalancing LoadBalancing
	// SocketRecvBufferSize sets the maximum socket receive buffer in bytes.
	SocketRecvBufferSize int
	// SocketSendBufferSize sets the maximum socket send buffer in bytes.
	SocketSendBufferSize int
	// TCPKeepAlive sets the TCP keep-alive for the socket.
	TCPKeepAlive time.Duration
	// LoggerLevel indicates the logging level.
	LoggerLevel zerolog.Level
	// PrettyLogger sets the pretty-printing zerolog mode.
	// Important: it is inefficient so should be used only for debugging.
	PrettyLogger bool

	// ==============================
	// io_uring related options
	// ==============================
	// MaxSQEntries sets the maximum number of SQEs that can be submitted in one batch.
	// If the number of SQEs exceeds this value, the io_uring will return a SQE overflow error.
	MaxSQEntries uint
	// MaxCQEvents sets the maximum number of CQEs that can be retrieved in one batch.
	MaxCQEvents uint
}

Config is the configuration for the gain engine.

func NewConfig

func NewConfig(opts ...ConfigOption) Config

type ConfigOption

type ConfigOption Option[Config]

func WithArchitecture

func WithArchitecture(architecture ServerArchitecture) ConfigOption

WithArchitecture sets the architecture of gain engine.

func WithAsyncHandler

func WithAsyncHandler(asyncHandler bool) ConfigOption

WithAsyncHandler sets the asynchronous mode for the OnRead callback.

func WithCBPF

func WithCBPF(cbpf bool) ConfigOption

WithCBPF sets the CBPF filter for the gain engine.

func WithCPUAffinity

func WithCPUAffinity(cpuAffinity bool) ConfigOption

WithCPUAffinity sets the CPU affinity option.

func WithGoroutinePool

func WithGoroutinePool(goroutinePool bool) ConfigOption

WithGoroutinePool sets the goroutine pool for asynchronous handler.

func WithLoadBalancing

func WithLoadBalancing(loadBalancing LoadBalancing) ConfigOption

WithLoadBalancing sets the load balancing algorithm.

func WithLoggerLevel

func WithLoggerLevel(loggerLevel zerolog.Level) ConfigOption

WithLoggerLevel sets the logging level.

func WithMaxCQEvents

func WithMaxCQEvents(maxCQEvents uint) ConfigOption

WithMaxCQEvents sets the maximum number of entries in the completion queue.

func WithMaxSQEntries

func WithMaxSQEntries(maxSQEntries uint) ConfigOption

WithMaxSQEntries sets the maximum number of entries in the submission queue.

func WithPrettyLogger

func WithPrettyLogger(prettyLogger bool) ConfigOption

WithPrettyLogger sets the pretty-printing zerolog mode.

func WithProcessPriority

func WithProcessPriority(processPriority bool) ConfigOption

WithProcessPriority sets the high process priority. Note: requires root privileges.

func WithSocketRecvBufferSize

func WithSocketRecvBufferSize(size int) ConfigOption

WithSocketRecvBufferSize sets the maximum socket receive buffer in bytes.

func WithSocketSendBufferSize

func WithSocketSendBufferSize(size int) ConfigOption

WithSocketSendBufferSize sets the maximum socket send buffer in bytes.

func WithTCPKeepAlive

func WithTCPKeepAlive(tcpKeepAlive time.Duration) ConfigOption

WithTCPKeepAlive sets the TCP keep-alive for the socket.

func WithWorkers

func WithWorkers(workers int) ConfigOption

WithWorkers sets the number of workers.

type Conn

type Conn interface {
	Reader
	Writer
	Socket

	// Context returns a user-defined context.
	Context() (ctx interface{})

	// SetContext sets a user-defined context.
	SetContext(ctx interface{})

	// LocalAddr is the connection's local socket address.
	LocalAddr() (addr net.Addr)

	// RemoteAddr is the connection's remote peer address.
	RemoteAddr() (addr net.Addr)

	// Close closes the current connection.
	Close() error
}

Conn is an interface representing a network connection. Structures implementing it are not guaranteed to be thread-safe. All write operations are asynchronous.

type DefaultEventHandler

type DefaultEventHandler struct{}

DefaultEventHandler is a default implementation for all of the EventHandler callbacks (do nothing). Compose it with your own implementation of EventHandler and you won't need to implement all callbacks.

func (DefaultEventHandler) OnAccept

func (e DefaultEventHandler) OnAccept(c Conn)

func (DefaultEventHandler) OnClose

func (e DefaultEventHandler) OnClose(c Conn, err error)

func (DefaultEventHandler) OnRead

func (e DefaultEventHandler) OnRead(c Conn, n int)

func (DefaultEventHandler) OnStart

func (e DefaultEventHandler) OnStart(server Server)

func (DefaultEventHandler) OnWrite

func (e DefaultEventHandler) OnWrite(c Conn, n int)

type EventHandler

type EventHandler interface {
	// OnStart fires when the server is ready for accepting new connections.
	OnStart(server Server)
	// OnAccept fires when a new TCP connection has been opened (will not be fired for UDP protocol).
	// This is good place to set the socket options, such as TCP keepalive.
	// The incoming data buffer of the connection will be empty,
	// so the attempt to read the data will never succeed, but you can perform a write at this point
	// The Conn c has information about the connection such as it's local and remote address.
	OnAccept(c Conn)
	// OnRead fires when a socket receives n bytes of data from the peer.
	//
	// Call c.Read() of Conn c to read incoming data from the peer. call c.Write() to send data to the remote peer.
	OnRead(c Conn, n int)
	// OnWrite fires right after a n bytes is written to the peer socket.
	OnWrite(c Conn, n int)
	// OnClose fires when a TCP connection has been closed (will not be fired for UDP protocol).
	// The parameter err is the last known connection error.
	OnClose(c Conn, err error)
}

type LoadBalancing

type LoadBalancing int

const (
	// RoundRobin forwards accepted connections to dedicated workers sequentially.
	RoundRobin LoadBalancing = iota
	// LeastConnections forwards the next accepted connection to the worker with the least number of active connections.
	LeastConnections
	// SourceAddrHash forwards the next accepted connection to the worker by hashing the remote peer address.
	SourceIPHash
)

type Option

type Option[T any] func(*T)

type Reader

type Reader interface {
	io.Reader
	io.WriterTo

	// Next returns a slice containing the next n bytes from the buffer,
	// advancing the buffer as if the bytes had been returned by Read.
	// If n is less or equal to 0, Next returns the entire buffer.
	// The error is io.ErrShortBuffer if n is larger than the reader buffer size.
	//
	// Note that the []byte buf returned by Next() is not allowed to be passed to a new goroutine,
	// as this []byte will be reused within event-loop.
	// If you have to use buf in a new goroutine, then you need to make a copy of buf and pass this copy
	// to that new goroutine.
	Next(n int) (buf []byte, err error)

	// Peek returns the next n bytes without advancing the reader. The bytes stop
	// being valid at the next read call.
	// If n is less or equal to 0, Peek returns the entire buffer.
	//
	// Note that the []byte buf returned by Peek() is not allowed to be passed to a new goroutine,
	// as this []byte will be reused within event-loop.
	// If you have to use buf in a new goroutine, then you need to make a copy of buf and pass this copy
	// to that new goroutine.
	Peek(n int) ([]byte, error)

	// Discard skips the next n bytes, returning the number of bytes discarded.
	//
	// If n > 0 Discards tries to skip n bytes.
	// If n == 0 Discards doesn't skip any bytes.
	// If n < 0 Discards skips all buffered bytes.
	Discard(n int) (int, error)

	// InboundBuffered returns the number of bytes that can be read from the current buffer.
	InboundBuffered() (n int)
}

Reader is an interface that consists of a number of methods for reading that Conn must implement.

type Server

type Server interface {
	// Start starts a server that will listen on the specified address and
	// waits for SIGTERM or SIGINT signals to close the server.
	StartAsMainProcess(address string) error
	// Start starts a server that will listen on the specified address.
	Start(address string) error
	// Shutdown closes all connections and shuts down server. It's blocking until the server shuts down.
	Shutdown()
	// AsyncShutdown closes all connections and shuts down server in asynchronous manner.
	// It does not wait for the server shutdown to complete.
	AsyncShutdown()
	// ActiveConnections returns the number of active connections.
	ActiveConnections() int
	// IsRunning returns true if server is running and handling requests.
	IsRunning() bool
}

func NewServer

func NewServer(eventHandler EventHandler, config Config) Server

type ServerArchitecture

type ServerArchitecture int

const (
	// Reactor design pattern has one input called Acceptor,
	// which demultiplexes the handling of incoming connections to Consumer workers.
	// The load balancing algorithm can be selected via configuration option.
	Reactor ServerArchitecture = iota
	// The Socket Sharding  allow multiple workers to listen on the same address and port combination.
	// In this case the kernel distributes incoming requests across all the sockets.
	SocketSharding
)

type Socket

type Socket interface {
	// Fd returns underlying file descriptor.
	Fd() int

	// SetReadBuffer sets the size of the operating system's
	// receive buffer associated with the connection.
	SetReadBuffer(bytes int) error

	// SetWriteBuffer sets the size of the operating system's
	// transmit buffer associated with the connection.
	SetWriteBuffer(bytes int) error
	// SetLinger sets the behavior of Close on a connection which still
	// has data waiting to be sent or to be acknowledged.
	//
	// If sec < 0 (the default), the operating system finishes sending the
	// data in the background.
	//
	// If sec == 0, the operating system discards any unsent or
	// unacknowledged data.
	//
	// If sec > 0, the data is sent in the background as with sec < 0. On
	// some operating systems after sec seconds have elapsed any remaining
	// unsent data may be discarded.
	SetLinger(sec int) error

	// SetKeepAlivePeriod tells operating system to send keep-alive messages on the connection
	// and sets period between TCP keep-alive probes.
	SetKeepAlivePeriod(d time.Duration) error

	// SetNoDelay controls whether the operating system should delay
	// packet transmission in hopes of sending fewer packets (Nagle's
	// algorithm).
	// The default is true (no delay), meaning that data is sent as soon as possible after a Write.
	SetNoDelay(noDelay bool) error
}

Socket is an interface that represents a socket.

type Writer

type Writer interface {
	io.Writer
	io.ReaderFrom

	// OutboundBuffered returns the number of bytes that can be read from the current buffer.
	OutboundBuffered() (n int)
}

Writer is an interface that consists of a number of methods for writing that Conn must implement.