Всем привет! Продолжаем разбираться с io_uring. Сегодня попробуем использовать io_uring для решения прикладных задач. Сначала мигрируем приложения из предыдущего материала с C на GO, затем подготовим почву для написания полноценных TCP серверов. Ну и без сравнения производительности не обойдется. Обойдемся без длинного вступления, вперед экспериментировать!
Дисклеймер
Это вторая статья из серии, посвященной io_uring. Чтобы понимать, о чем речь — ознакомьтесь с первым материалом.
Сегодня оставим в стороне теорию io_uring, но cфокусируемся на практических аспектах применения этой технологии, так что кода будет много. Для практических экспериментов была выбрана работа с сокетами, а в качестве эталона — runtime GO. Таким образом, в этом материале, код будет на языке GO, да и применение io_uring рассмотрим в контексте GO.
Если данная тема будет неинтересна, то подходите к статье номер 3. Там снова будет обсуждение фич io_uring, настроек и режимов работы, тюнинга производительности.
Echo-сервер, тушёный с приправами
Прежде чем соперничать с runtime'ом GO, немного поговорим об инструменте, который будем использовать. Для работы с io_uring можно было воспользоваться связкой CGO + liburing, но, так как особых помех для работы с системными вызовами семейства io_uring в GO нет (ну ладно, они есть, но говорить сегодня о них не будем), почему бы не сделать полностью нативную библиотеку? Давайте перепишем echo-сервер из прошлой статьи с помощью библиотеки go-uring:
GO echo-server
package main
import (
"errors"
"flag"
"fmt"
"github.com/godzie44/go-uring/uring"
"log"
"strconv"
"syscall"
)
const MaxConns = 4096
const Backlog = 512
const MaxMsgLen = 2048
type connType int
const (
_ connType = iota
ACCEPT
READ
WRITE
)
// Для каждого активного соединения будем держать в памяти connInfo структуру.
// fd файловый дескриптор сокета.
// typ - состояние в котором находится сокет (ожидает чтения/записи/accept'а).
// sendOp, recvOp - закешированные операции чтения из сокета/записи в сокет.
type connInfo struct {
fd int
typ connType
sendOp *uring.SendOp
recvOp *uring.RecvOp
}
// Для каждого соединения предалоцируем операции чтения и записи.
// Переиспользование операция чтения и записи снизит нагрузку на GC.
func makeConns() [MaxConns]connInfo {
var conns [MaxConns]connInfo
for fd := range conns {
conns[fd].recvOp = uring.Recv(uintptr(fd), nil, 0)
conns[fd].sendOp = uring.Send(uintptr(fd), nil, 0)
}
return conns
}
// Буфер для соединений.
var conns = makeConns()
// Для каждого соединения инициализируем буфера для записи/чтения.
func makeBuffers() [][]byte {
buffs := make([][]byte, MaxConns)
for i := range buffs {
buffs[i] = make([]byte, MaxMsgLen)
}
return buffs
}
var buffs = makeBuffers()
func main() {
flag.Parse()
port, _ := strconv.Atoi(flag.Arg(0))
// Создаем серверный сокет и слушаем порт.
// Отметим, что при создании сокета флаг O_NON_BLOCK не устанавливается,
// при этом все операции read/write и тд преобразуются в неблокирующие системные вызовы внутри io_uring
serverSockFd, err := syscall.Socket(syscall.AF_INET, syscall.SOCK_STREAM, 0)
checkErr(err)
defer syscall.Close(serverSockFd)
checkErr(syscall.SetsockoptInt(serverSockFd, syscall.SOL_SOCKET, syscall.SO_REUSEADDR, 1))
checkErr(syscall.Bind(serverSockFd, &syscall.SockaddrInet4{
Port: port,
}))
checkErr(syscall.Listen(serverSockFd, Backlog))
fmt.Printf("io_uring echo server listening for connections on port: %d\n", port)
// Создаем экземпляр io_uring, не используем никакие кастомные настройки.
// Вместимость SQ/SQ буферов устанавливаем в 4096 элементов.
ring, err := uring.New(4096)
checkErr(err)
defer ring.Close()
// Проверяем наличие фичи IORING_FEAT_FAST_POLL.
// Для нас это наиболее "перформящая" фича в данном приложении,
// фактически это встроенный в io_uring движок для поллинга I/O.
if !ring.Params.FastPollFeature() {
checkErr(errors.New("IORING_FEAT_FAST_POLL not available"))
}
// Добавляем первую операцию в SQ - слушаем серверный сокет на предмет новых входящих соединений.
acceptOp := uring.Accept(uintptr(serverSockFd), 0)
addAccept(ring, acceptOp)
cqes := make([]*uring.CQEvent, Backlog)
var cqe *uring.CQEvent
for {
// Сабмитим все SQE которые были добавлены на предыдущей итерации..
_, err = ring.Submit()
checkErr(err)
// Ждем когда в CQ буфере появится хотя бы одно CQE.
_, err = ring.WaitCQEvents(1)
if errors.Is(err, syscall.EAGAIN) || errors.Is(err, syscall.EINTR) {
continue
}
checkErr(err)
// Помещаем все "готовые" CQE в буфер cqes.
n := ring.PeekCQEventBatch(cqes)
for i := 0; i < n; i++ {
cqe = cqes[i]
// В поле user_data находится индекс соответствующего connInfo
// в которой находится служебная информация по сокету.
ud := conns[cqe.UserData]
typ := ud.typ
res := cqe.Res
ring.SeenCQE(cqe)
// Проверяя тип мы идентифицируем операцию результат которой находится в CQE (accept / recv / send).
switch typ {
case ACCEPT:
addRead(ring, int(res))
addAccept(ring, acceptOp)
case READ:
if res <= 0 {
_ = syscall.Shutdown(ud.fd, syscall.SHUT_RDWR)
} else {
addWrite(ring, ud.fd, res)
}
case WRITE:
addRead(ring, ud.fd)
}
}
}
}
// addAccept - добавляет в SQ accept операцию, fd - файловый дескриптор сервер сокета.
func addAccept(ring *uring.Ring, acceptOp *uring.AcceptOp) {
conns[acceptOp.Fd()].fd = acceptOp.Fd()
conns[acceptOp.Fd()].typ = ACCEPT
err := ring.QueueSQE(acceptOp, 0, uint64(acceptOp.Fd()))
checkErr(err)
}
// addRead - добавляет в SQ read операцию, fd - файловый дескриптор клиентского сокета.
func addRead(ring *uring.Ring, fd int) {
buff := buffs[fd]
ci := &conns[fd]
ci.fd = fd
ci.typ = READ
ci.recvOp.SetBuffer(buff)
err := ring.QueueSQE(ci.recvOp, 0, uint64(fd))
checkErr(err)
}
// addWrite - добавляет в SQ write операцию, fd - файловый дескриптор клиентского сокета.
func addWrite(ring *uring.Ring, fd int, bytesRead int32) {
buff := buffs[fd]
ci := &conns[fd]
ci.fd = fd
ci.typ = WRITE
ci.sendOp.SetBuffer(buff[:bytesRead])
err := ring.QueueSQE(ci.sendOp, 0, uint64(fd))
checkErr(err)
}
func checkErr(err error) {
if err != nil {
log.Fatal(err)
}
}Код на GO получился несколько компактнее чем этот же сервер на plain C. С другой стороны, оба исходника довольно схожи как семантически, так и синтаксически. А что по поводу производительности? С подробным описанием методики сравнения можно ознакомиться здесь. Рассмотрим исключительно результаты:
c: 100 bytes: 128 | c: 50 bytes: 1024 | c: 500 bytes: 128 | c: 500 bytes: 1024 | c: 1000 bytes: 128 | c: 1000 bytes: 1024 | |
io_uring-echo-server (C lang) | 235356 | 224783 | 173670 | 155477 | 149407 | 139987 |
echo-server | 227884 | 222709 | 169001 | 150275 | 143664 | 128783 |
Отлично! GO-код не сильно проигрывает коду на C в контексте I/O нагрузки. Но, есть проблема, реализация транспортного слоя прибита гвоздями к логике приложения, что не позволяет использовать ее где-либо кроме как в echo-сервере. Для того чтобы писать полноценные приложения с io_uring в GO, нам нужно реализовать интерфейсы из пакета net: net.Conn и net.Listener. А для этого нужно разобраться в реализации стандартных net.Conn и net.Listener для протокола TCP и написать свои реализации. Но обо всем по порядку.
В логове net package
Рассмотрим необходимые нам интерфейсы подробнее.
net.Listener
Задача listener'а — слушать определенный порт и при подключении к серверу очередного клиента создать структуру реализующую net.Conn, для работы с этим клиентом. Соответствующий интерфейс:
// A Listener is a generic network listener for stream-oriented protocols.
//
// Multiple goroutines may invoke methods on a Listener simultaneously.
type Listener interface {
// Accept waits for and returns the next connection to the listener.
Accept() (Conn, error)
// Close closes the listener.
// Any blocked Accept operations will be unblocked and return errors.
Close() error
// Addr returns the listener's network address.
Addr() Addr
}net.Conn
net.Conn - абстракция над клиентским соединением, позволяет выполнять чтение и запись в соответствующий клиентский сокет. Интерфейс (для удобства немного упрощен):
type Conn interface{
// Read reads data from the connection.
Read(b []byte) (n int, err error)
// Write writes data to the connection.
Write(b []byte) (n int, err error)
// Close closes the connection.
Close() error
}Соответствующие реализации этих интерфейсов лежат в основе работы транспортного уровня в GO. Имея на руках реализации net.Listner и net.Conn для, например, TCP протокола можно реализовать любой вышестоящий протокол, будь то HTTP или любой другой. К примеру, вот так будет выглядеть echo-server с использованием стандартных net.TCPListener и net.TCPConn:
обычный echo-server
package main
import (
"io"
"log"
"net"
)
const MaxConns = 4096
const MaxMsgLen = 2048
func initBuffs() [][]byte {
buffs := make([][]byte, MaxConns)
for i := range buffs {
buffs[i] = make([]byte, MaxMsgLen)
}
return buffs
}
var buffs = initBuffs()
func main() {
listener, err := net.ListenTCP("tcp", &net.TCPAddr{
Port: 8080,
})
checkErr(err)
for {
conn, err := listener.Accept()
checkErr(err)
go handleConn(conn)
}
}
func handleConn(conn net.Conn) {
f, _ := conn.(*net.TCPConn).File()
buff := buffs[int(f.Fd())]
for {
n, err := conn.Read(buff)
if err == io.EOF || n == 0 {
checkErr(conn.Close())
return
}
checkErr(err)
_, err = conn.Write(buff[:n])
checkErr(err)
}
}
func checkErr(err error) {
if err != nil {
log.Fatal(err)
}
}Все довольно просто, в бесконечном цикле принимаем входящие соединения, а затем, для каждого соединения, создаем горутину в которой начинаем читать данные из клиентского сокета и писать туда же. Но представьте, что какой-то из клиентов решил не передавать нам данные, к примеру, в течение минуты. В таком случае поток, на котором находится горутина выполняющая чтение, заблокируется? Естественно нет! Любой GO разработчик знает — это не проблема для нашего сервера. Но каким образом, казалось бы, блокирующие вызовы conn.Read не блокируют поток исполнения? И тут в дело вступает runtime, вернее его компонент — netpoller.
netpoller - тропа через block I/O
Рассмотрим как горутины работают с I/O операциями над сокетами на примере чтения. В GO горутина (G) желающая прочитать данные из сокета, сначала попытается выполнить чтение в неблокирующем режиме. Если чтение в настоящий момент невозможно (к примеру, в сокете нет данных) — G будет заблокирована. При этом поток выполнения (M) не блокируется, а G инициирующая чтение уходит в состояние сна. Ее место на логическом процессоре (P) займет другая горутина. Наконец, когда сокет будет готов для I/O (в нашем случае появились данные для чтения) - планировщик разбудит G, и наша программа выполнит запланированный системный вызов.
Для реализации подобного механизма необходим компонент, который:
позволяет G добавить себя в очередь для оповещения о готовности I/O
будет использован планировщиком, чтобы понять, каким G требуется процессорное время в связи с появлением I/O events
Этот компонент runtime'а называется netpoller и в том числе, благодаря ему, сервера, написанные на GO, обладают такой скоростью.
А вот его упрощенный интерфейс, который нас интересует:
func poll_runtime_pollWait(pd *pollDesc, mode int) int
func netpoll(delay int64) gListpoll_runtime_pollWait - предоставляет возможность для G подписаться на интересующее I/O событие в сокете. Структура pollDesc описывает файловый дескриптор сокета, а mode - тип события, которого следует ждать (read/write)
netpoll - эта функция используется планировщиком для получения списка G которым нужно дать процессорное время
Реализация netpoller зависит от конкретной OS. В Linux netpoller реализован с помощью семейства системных вызовов epoll. Таким образом, в GO используется паттерн подписки на I/O события:
Итак, с помощью netpoller'а можно делать системные вызовы, не боясь заблокировать приложение. В свою очередь, с помощью io_uring можно добиться тех же целей, но другими средствами. Концепция событий отходит в сторону, ведь с io_uring можно просто поставить необходимую операцию в очередь на исполнение и позднее узнать результат операции.
Тут нужно сделать ремарку: можно точно так же отлавливать I/O события и с помощью io_uring операции IORING_OP_POLL_ADD, правда, на производительности такой подход отражается скорее в негативную сторону.
Приручение reactor
Ну что же, теперь можно написать конкурента netpoller'у с использованием io_uring. Назовем его reactor. Суть идеи проста, реактор выполняет операции поддерживаемые io_uring и дергает callback в userspace по завершению той или иной операции. Таким образом, reactor будет полноценным асинхронным I/O бэкендом. Вот его наивная реализация:
reactor.go
type Callback func(event uring.CQEvent)
type Reactor struct {
ring *uring.Ring
callbacks map[uint64]Callback
callbacksLock sync.Mutex
queueSQELock sync.Mutex
currentNonce uint64
submitSignal chan struct{}
}
func New(ring *uring.Ring) *Reactor {
return &Reactor{
ring: ring,
submitSignal: make(chan struct{}),
callbacks: map[uint64]Callback{},
}
}
func (r *Reactor) Run(ctx context.Context) {
wg := &sync.WaitGroup{}
//запускаем два компонента реактора
//consumer - обрабатывает новые CQE
//publisher - отправляет новые SQE на обработку
wg.Add(2)
go func() {
defer wg.Done()
r.runConsumer(ctx)
}()
go func() {
defer wg.Done()
r.runPublisher(ctx)
}()
wg.Wait()
}
// сабмитим новые SQE в отдельной горутине
func (r *Reactor) runPublisher(ctx context.Context) {
defer close(r.submitSignal)
for {
select {
//при поступлении сигнала обрабатываем новые SQE
case <-r.submitSignal:
r.queueSQELock.Lock()
_, err := r.ring.Submit()
r.queueSQELock.Unlock()
if err != nil {
log.Println("io_uring submit", err)
}
case <-ctx.Done():
return
}
}
}
// получаем результаты выполнения команд и дергаем соответствующие колбеки
func (r *Reactor) runConsumer(ctx context.Context) {
cqeBuff := make([]*uring.CQEvent, 512)
// цикл обработки результатов операций (CQE)
for {
// командуем Publisher'у обработать новые SQE
r.submitSignal <- struct{}{}
// ждем хотябы одну завершенную операцию
_, err := r.ring.WaitCQEventsWithTimeout(1, time.Millisecond)
if errors.Is(err, syscall.EAGAIN) || errors.Is(err, syscall.EINTR) || errors.Is(err, syscall.ETIME) {
runtime.Gosched()
goto CheckCtxAndContinue
}
if err != nil {
log.Println("io_uring wait", err)
goto CheckCtxAndContinue
}
for n := r.ring.PeekCQEventBatch(cqeBuff); n > 0; n = r.ring.PeekCQEventBatch(cqeBuff) {
for i := 0; i < n; i++ {
cqe := cqeBuff[i]
nonce := cqe.UserData
//находим соответствующий callback
r.callbacksLock.Lock()
cb := r.callbacks[nonce]
delete(r.callbacks, nonce)
r.callbacksLock.Unlock()
cb(uring.CQEvent{
UserData: cqe.UserData,
Res: cqe.Res,
Flags: cqe.Flags,
})
}
//сообщаем io_uring о N просмотренных CQE
r.ring.AdvanceCQ(uint32(n))
}
CheckCtxAndContinue:
select {
case <-ctx.Done():
return
default:
continue
}
}
}
//Queue добавляет операцию в SQ.
func (r *Reactor) Queue(op uring.Operation, cb Callback) (uint64, error) {
//генерируем уникальное значение UserData
nonce := r.nextNonce()
r.queueSQELock.Lock()
defer r.queueSQELock.Unlock()
//помещаем операцию в SQ
err := r.ring.QueueSQE(op, 0, nonce)
if err == nil {
r.callbacksLock.Lock()
r.callbacks[nonce] = cb
r.callbacksLock.Unlock()
}
return nonce, err
}
func (r *Reactor) nextNonce() uint64 {
return atomic.AddUint64(&r.currentNonce, 1)
}C помощью подобного компонента можно добавить операцию в SQ очередь io_uring и реактивно обработать результат этой операции. Внутри параллельно работают consumer для добавления операций в SQ и publisher для обработки результатов из CQ.
Теперь, при помощи reactor'а, можно реализовать заветные net.Listener и net.Conn интерфейсы.
net.Listener. Реализация с использованием reactor (очевидные участки кода выброшены, ссылка на полную реализацию будет ниже)
type Listener struct {
fd int
reactor *Reactor
acceptChan chan uring.CQEvent
addr net.Addr
}
func NewListener(addr string, reactor *Reactor) (*Listener, error) {
tcpAddr, err := net.ResolveTCPAddr("tcp", addr)
if err != nil {
return nil, err
}
// serverSocket - создает и биндит серверный сокет
sockFd, err := serverSocket(tcpAddr)
if err != nil {
return nil, err
}
return &Listener{
fd: sockFd,
addr: tcpAddr,
reactor: reactor,
acceptChan: make(chan uring.CQEvent),
}, nil
}
func (l *Listener) Accept() (net.Conn, error) {
//помещаем accept операцию в реактор
op := uring.Accept(uintptr(l.fd), 0)
l.reactor.Queue(op, func(event uring.CQEvent) {
l.acceptChan <- event
})
// ждем появление подключения
cqe := <-l.acceptChan
if err := cqe.Error(); err != nil {
return nil, err
}
fd := int(cqe.Res)
rAddr, _ := op.Addr()
tc := newConn(fd, l.addr, rAddr, l.reactor)
return tc, nil
}net.Conn. Реализация с использованием reactor (очевидные участки кода выброшены, ссылка на полную реализацию будет ниже)
type Conn struct {
Fd int
lAddr, rAddr net.Addr
reactor *Reactor
readChan, writeChan chan uring.CQEvent
readLock, writeLock sync.Mutex
}
func newConn(fd int, lAddr, rAddr net.Addr, r *Reactor) *Conn {
return &Conn{
lAddr: lAddr,
rAddr: rAddr,
Fd: fd,
reactor: r,
readChan: make(chan uring.CQEvent),
writeChan: make(chan uring.CQEvent),
}
}
func (c *Conn) Read(b []byte) (n int, err error) {
c.readLock.Lock()
defer c.readLock.Unlock()
//помещаем Recv операцию в реактор
op := uring.Recv(uintptr(c.Fd), b, 0)
c.reactor.Queue(op, func(event uring.CQEvent) {
c.readChan <- event
})
//ждем завершения чтения из сокета
cqe := <-c.readChan
if err = cqe.Error(); err != nil {
return 0, &net.OpError{Op: "read", Net: "tcp", Source: c.lAddr, Addr: c.rAddr, Err: err}
}
if cqe.Res == 0 {
err = &net.OpError{Op: "read", Net: "tcp", Source: c.lAddr, Addr: c.rAddr, Err: io.EOF}
}
runtime.KeepAlive(b)
return int(cqe.Res), err
}
func (c *Conn) Write(b []byte) (n int, err error) {
c.writeLock.Lock()
defer c.writeLock.Unlock()
//помещаем Send операцию в реактор
op := uring.Send(uintptr(c.Fd), b, 0)
c.reactor.Queue(op, func(event uring.CQEvent) {
c.writeChan <- event
})
//ждем завершения записи в сокет
cqe := <-c.writeChan
if err = cqe.Error(); err != nil {
return 0, &net.OpError{Op: "write", Net: "tcp", Source: c.lAddr, Addr: c.rAddr, Err: err}
}
if cqe.Res == 0 {
err = &net.OpError{Op: "write", Net: "tcp", Source: c.lAddr, Addr: c.rAddr, Err: io.ErrUnexpectedEOF}
}
runtime.KeepAlive(b)
return int(cqe.Res), err
}
func (c *Conn) Close() error {
return syscall.Close(c.Fd)
}Проверим на практике новые реализации, давно мы не писали echo-server! Вот его полный код (вместе с reactor'ом, net.Listener и net.Conn).
Сравним новый сервер с io_uring reactor'ом под капотом, с таким же сервером, который использует net.TCPListener и net.TCPConn из стандартной бибилиотеки (а значит, под капотом там netpoller). Как обычно смотрим на rps.
c: 100 bytes: 128 | c: 50 bytes: 1024 | c: 500 bytes: 128 | c: 500 bytes: 1024 | c: 1000 bytes: 128 | c: 1000 bytes: 1024 | |
net/http | 132664 | 139206 | 133039 | 139171 | 133480 | 139617 |
reactor (io_uring) | 30892 | 30077 | 39192 | 38375 | 46120 | 51204 |
Да уж, сравнение не в нашу пользу. Текущая реализация не использует на полную ресурсы железа и возможности io_uring — расплата за это низкая производительность. Тем не менее в плане функциональности даже такой, наивной, реализации reactor достаточно чтобы заменить собой netpoller. Кроме того, reactor можно использовать не только в контексте socket I/O, но и для всех прочих I/O.
О завершающей части
Конечно, такая низкая производительность не может нас устроить. Поэтому в третьей и заключительной части разберемся, почему решение с "наивным" reactor работает так медленно, рассмотрим способы ускорения (как с помощью настройки io_uring, так и с помощью оптимизаций кода reactor), а также поставим точку (хотя скорее многоточие) в битве с netpoller. Спасибо за внимание, stay tuned.
Дата-центр ITSOFT — размещение и аренда серверов и стоек в двух дата-центрах в Москве. За последние годы UPTIME 100%. Размещение GPU-ферм и ASIC-майнеров, аренда GPU-серверов, лицензии связи, SSL-сертификаты, администрирование серверов и поддержка сайтов.
