Всем привет! Продолжаем разбираться с 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) gList

poll_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-сертификаты, администрирование серверов и поддержка сайтов.

IO_URING. Часть 2. Let's GO