IO_URING. Часть 3

Всем привет! Симулятор написания echo серверов на связи. Сегодня завершающая часть из цикла посвященного io_uring. В этом материале мы поговорим о настройке io_uring, режимах работы и перформансе. Полученные знания используем чтобы улучшить конкурента netpoller из предыдущей статьи. 

---

В предыдущих сериях

• В первой части цикла мы познакомились с io_uring, посмотрели простенькие примеры его работы и даже написали небольшой tcp-сервер.

• Во второй части мы попытались найти практическое применения для io_uring, а именно переписали кусок GO'шного рантайма с использованием новой технологии. К сожалению, производительность полученного решения оказалась так себе.

Reactor собирает войска

Итак, сегодня, разберём опции io_uring и подходы, которые помогут улучшить производительность. Оптимизируем замену для netpoller'а - reactor и напишем новые бенчмарки. Ну и, конечно, подведем какие-никакие выводы.

Промежуточная цель этой статьи — стать небольшим справочником для разработчика, решившего воспользоваться io_uring при работе с сетью. Так что все возможные опции, настройки и прочие аспекты, влияющие на socket I/O, будут рассмотрены подробно и дополнены субъективным мнением автора. Оставшиеся опции будут упомянуты вскользь, так как не относятся к сегодняшней теме.

Условно разделим возможные оптимизации на несколько групп:

• настройка экземпляра io_uring:
  

  • опции, устанавливаемые при создании io_uring через системный вызов io_uring_setup;

  • флаги, которые устанавливаются для каждого SQE отдельно.

• мультиплексирование I/O с несколькими экземплярами io_uring;

• оптимизация кода event loop (reactor).

Настройка нового экземпляра io_uring

Полный список опций, а также способы их установки:

man io_uring_setup

IORING_FEAT_FAST_POLL

Начнем, правда, не с конфигурируемых пользователем опций, а с фичи FEAT_FAST_POLL доступность которой зависит только от версии ядра, она будет включена для ядер версии 5.7 и выше. Наличие этого механизма обязательно для высокого перформанса при сетевом I/O. FAST_POLL это хитрый алгоритм полинга и его понимание очень пригодится в будущем, поэтому давайте разбираться.

Допустим, необходимо выполнить чтение из сокета посредством операции recv (здесь под poll'ом подразумевается интерфейс ОС, который реализует конкретный драйвер, а не семейство syscall'ов):

1. Добавляем операцию recv через io_uring_enter.

2. io_uring выполняет системный вызов recv в неблокирующем режиме:
   

   1. Вызов recv прошел успешно - можно коммитить соответствующий CQE в CQ.

   2. Вызов вернул EAGAIN - poll'им файловый дескриптор, регистрируем асинхронный обработчик для события POLLIN:
      

      1. Вызов poll для сокета незамедлительно вернул результат - добавляем новую операцию recv в SQ, переходим к (2).

      2. В асинхронный обработчик пришло событие - добавляем новую операцию recv в SQ, переходим к (2).

Так при переходе из 2.2.2 к 2 проходит некоторое время то возможно ситуация, когда повторный вызов recv опять вернул EAGAIN. На этот случай есть fallback механизм: вызов recv и отправляется на выполнение в worker-pool основанный на linux-workers. Важный вывод здесь - основная работа происходит в контексте thread'а (task в терминологии ядра linux) который вызвал io_uring_enter, а worker-pool задействуется очень редко.

Теперь можно перейти к первоначальной конфигурации экземпляра io_uring (далее просто кольца).

IORING_SETUP_SQPOLL

Наверное, самая интересная опция. Как мы помним одно из преимуществ io_uring — возможность выполнять системные вызовы пачками. Вместо самостоятельных вызовов разнообразных syscall'ов можно поместить их в очередь и только единожды выполнить системный вызов io_uring_enter. При таком подходе уменьшается количество переходов user/kernel space, что положительно влият на производительность системы. Так вот, с помощью IORING_SETUP_SQPOLL можно избавиться и от вызова io_uring_enter. Работает это так — создается ядерный тред, который берет на себя работу по мониторингу очереди SQ. Если, на протяжении некоторого времени, в SQ не было новых вхождений, то поток засыпает и в дальнейшем его необходимо разбудить, чтобы продолжить обработку очереди.

Плюсы:

• пропадает необходимость в вызове io_uring_enter, что может быть существенно в случае частых вызовов io_uring_enter. С другой стороны, подобные частые системные вызовы могут быть индикатором того, что батчинг syscall'ов используется недостаточно;

• упрощаем код приложения. Вся работа по батчингу и определение момента, когда нужен вызов io_uring_enter, ложится на ядро и снимается с плеч разработчика.

Минусы:

• осознанно теряем в контролируемости, так как часть работы отдаем на совесть kernel thread'а;

• вместо одного потока, который работает с экземпляром io_uring, получаем два.

Мнение автора: опция хороша для быстрого старта — позволяет получить хорошую производительность с минимальными усилиями, но выжать все соки — не получится.

IORING_SETUP_SQ_AFF

Позволяет "прибить" поток, который был создан опцией IORING_SETUP_SQPOLL, к одному CPU.  Может быть полезна при нескольких одновременно работающих io_uring'ах и соответственно нескольких SQPOLL потоках. 

Мнение автора: по заявлениям коммитеров в т. н. экстремальных кейсах можно получить очень значительный прирост. Правда, на своих тестах я получил скорее деградацию производительности. В общем it depends.

IORING_SETUP_ATTACH_WQ

Ранее было упомянуто о том, что каждое кольцо io_uring использует свой worker-pool для выполнения асинхронных операций. С помощью этой опции можно заставить несколько колец работать на одном, общем, пуле (кстати, также можно заставить один SQPOLL thread обрабатывать SQ нескольких колец).

Мнение автора: интересная опция, на практике можно получить некоторый прирост, уменьшив количество пулов в системе. Как мы помним в контексте socket I/O пулы воркеров используются в fallback механизме, так что нет смысла держать пулов столько, сколько инстансов io_uring в системе, будет разумно разделить один-два пула между всеми кольцами.

Прочее

• IORING_SETUP_IOPOLL - включает активное ожидание для I/O операций, положительно влияет на latency и отрицательно на throughput. Не поддерживается для операций на сокетах.

• IORING_SETUP_CQSIZE - устанавливает размер CQ буфера, который стандартно равен size(SQ)*2.

• IORING_SETUP_CLAMP - защита от дурака неверно заданного размера буферов SQ и CQ.

• IORING_SETUP_R_DISABLED - позволяет создать выключенное кольцо.

Флаги SQE.

Поговорим о флагах, которые устанавливаются для SQE перед вызовом io_uring_enter. Все так же разбираемся с ними в контексте socket I/O. Полный список флагов и способы их установки можно подсмотреть в:

man io_uring_enter

IOSQE_ASYNC

Установка этого флага для SQE позволяет сразу обработать SQE в ассинхронном режиме. Для сетевых запросов это значит сразу перейти к fallback механизму - отправить операцию в work pool. 

Мнение автора: если вы по какой-то причине хотите работать с одним инстансом io_uring, но при этом не против параллельно запущенных worker'ов, то включение этой опции дает значительный прирост производительности. Другое дело, что работать с несколькими экземплярами гораздо лучше, впрочем, об этом позже.

IOSQE_BUFFER_SELECT

Если экземпляр io_uring берет в обработку SQE с установленным флагом IOSQE_BUFFER_SELECT то будет использован один из заранее аллоцированных буферов. Набор буферов регистрируется при помощи операции IORING_OP_PROVIDE_BUFFERS.

Мнение автора: теоретически выглядит очень интересно, так как позволяет расшарить участки памяти между приложением и ядром. К сожалению, на практике эффект [не слишком впечатляющий](https://twitter.com/hielkedv/status/1255492941960949760?s=21), ждем изменений.

IOSQE_IO_LINK

Позволяет связать несколько SQE в цепочку, каждое SQE в цепочке будет выполняться последовательно, после завершения предыдущей. В случае возникновения ошибки в одном из SQE вся цепочка инвладириуется. Эта опция актуальна в связке с операцией IORING_OP_LINK_TIMEOUT которая несколько меняет семантику: когда эта операция будет связана с SQE то они будут работать в паре, то есть либо SQE завершится успешно либо сработает timeout. Такая опция пригодится для выполнения сетевых запросов, поскольку без таймаутов — никак.

Прочее

• IOSQE_FIXED_FILE — разрешает пользоваться файловыми дескрипторами, которые были заранее "прибиты" к экземпляру io_uring. В принципе, тут подходят те же выводы, что и с механизмом предопределенных буферов, теоретически интересно, практически, пока что, разница незаметна.  

• IOSQE_IO_DRAIN — своеобразный барьер, SQE с таким флагом будет запущен после завершения всех текущих операций, новые операции также будут ждать окончания обработки данного SQE.

• IOSQE_IO_HARDLINK — используется в связке с IOSQE_IO_LINK, в случае установки данного флага цепочка не будет разорвана, если одна из операций завершилась с ошибкой.

Нужно больше колец

При рассмотрении настроек красной нитью проходит простая мысль: можно создавать сколько угодно экземпляров io_uring. И действительно, если обычно кольцо выполняет net операции в том же потоке на котором был выполнен системный вызов io_uring_enter, то почему бы не создать несколько колец и не расположить их на разных потоках?

Такой подход действительно дает отличный буст производительности и позволяет по полной нагрузить процессор. Но есть несколько моментов о которых не стоит забывать:

• потокобезопасность — да, liburing и прочие обертки защищают несинхронизированного доступа к SQ и CQ между user space'ом и ядром. Но, никакой коробочной синхронизации при доступе нескольких пользовательских потоков к CQ или SQ нет. К примеру, добавлять операции в SQ буфер и делать submit из него же параллельно в нескольких потоках — не безопасно;

• оптимальное количество колец — будет варьироваться от системы к системе и зависит от того, как вы настроили экземпляры. Например, на моем тестовом стенде оптимальным оказалось (количество ядер * 2) - 2.

Оптимизация reactor

Наконец не будем забывать и об оптимизации кода приложения. Давайте разберем пару оптимизаций на примере компонента reactor из прошлой статьи.

Поддержка нескольких экземпляров io_uring

Как мы решили выше — для полной утилизации ресурсов системы необходимо несколько колец io_uring. Так что reactor должен уметь работать с несколькими экземплярами.

Распределяем запросы между кольцами

Необходим способ распределения операции между кольцами. И здесь очевидное решение маршрутизировать операции по файловому дескриптору. Получается некая функция роутинга/шардирования от файлового дескриптора сокета. Возникает небольшая проблема — если в качестве такой функции выбрать (FD % количество колец) то при малом количестве соединений мы получим очень большое снижение производительности. Дело в том, что event-loop на основе кольца, да и сам io_uring работает довольно медленно при низкой частоте операций. Хочется чтобы при небольшом количестве сокетов нагрузка не размазывалась между всеми кольцами, а оседала на одном - двух экземплярах. Поэтому функция роутинга выглядит вот так:

const granSize= 75

func (r *NetworkReactor) loopForFd(fd int) *ringNetEventLoop {
	// n - количество колец с которыми работает reactor
	n := len(r.loops)
	// h - номер "гранулы" к которой принадлежит файловый дескриптор
	h := fd / granSize
	// h%n - номер кольца который отвечатает за обработку гранулы
	return r.loops[h%n]
}

Регистр обработчиков

В наивной реализации реактора использовался единый регистр обработчиков CQE — reactor.callbacks. Он представлял собой goroutine-safe мапу в которой ключом был уникальный номер SQE, а значением замыкание, вызываемое при появлении CQE. Такое решение крайне не оптимально так как оно использует единый lock для всех операций — добавления, вызова и удаления обработчиков. Да и учитывая специфику нашей области, хотелось бы предалоцировать место для обработчиков, ибо мы знаем, что одновременных соединений будет много и для каждого сокета в кольце может единовременно находиться две операции: чтения и записи в сокет. Получаем следующий компонент, решающий эти проблемы:

cbRegistry

package reactor

import (
	"sync"
	"sync/atomic"
)

type (
  // cbMap - обработчики привязанные к файловому дескриптору, ключ - id операции для файлового дескриптора, Callback - замыкание которое обработает CQE
	cbMap map[uint32]Callback
  // shard - шард обработчиков, отвечающий за хранение колбеков к операциям относящимся к определенным файловым дескрипторам
	shard struct {
    // nonce - id операции в контексте файлового дескриптора
		// noncec - массив идентификаторов, где индекс массива - файловый дескриптор, значение - последний выданный nonce  
    nonces    []uint32
    // callbacks - преалоцированный массив для обработчиков CQE
		callbacks []cbMap
    // boundary - граница после которой обработчик попадает в медленное хранилище slowCallbacks
		boundary  int

    // медленные хранилища для обработчиков и последних выданных nonce'ов
		slowCallbacks map[int]cbMap
		slowNonces    map[int]uint32

		sync.Mutex
	}
  // cbRegistry - общий регист для обработчиков CQE
	cbRegistry struct {
		shards      []*shard
		granularity int
		shardCnt    int
	}
)

func newShard(fCap int) *shard {
	buff := make([]cbMap, fCap)
	for i := range buff {
		buff[i] = make(cbMap, 10)
	}

	return &shard{
		callbacks:     buff,
		nonces:        make([]uint32, fCap),
		boundary:      fCap,
		slowCallbacks: make(map[int]cbMap),
		slowNonces:    make(map[int]uint32),
	}
}

func (sh *shard) add(idx int, cb Callback) (n uint32) {
	if idx < sh.boundary {
		n = atomic.AddUint32(&sh.nonces[idx], 1)
		sh.Lock()
		sh.callbacks[idx][n] = cb
		sh.Unlock()
		return n
	}

	//slow path, for big fd values
	sh.Lock()
	sh.slowNonces[idx]++
	n = sh.slowNonces[idx]

	if _, exists := sh.slowCallbacks[idx]; !exists {
		sh.slowCallbacks[idx] = make(cbMap, 10)
	}

	sh.slowCallbacks[idx][n] = cb
	sh.Unlock()
	return n
}

func (sh *shard) pop(idx int, nonce uint32) Callback {
	if idx < sh.boundary {
		sh.Lock()
		cb := sh.callbacks[idx][nonce]
		delete(sh.callbacks[idx], nonce)
		sh.Unlock()
		return cb
	}

	sh.Lock()
	cb := sh.slowCallbacks[idx][nonce]
	delete(sh.slowCallbacks[idx], nonce)
	sh.Unlock()

	return cb
}

func newCbRegistry(shardCount int, granularity int) *cbRegistry {
	shards := make([]*shard, shardCount)
	for i := 0; i < shardCount; i++ {
		shards[i] = newShard((1 << 16) / shardCount)
	}

	return &cbRegistry{
		granularity: granularity,
		shardCnt:    shardCount,
		shards:      shards,
	}
}

func (r *cbRegistry) shardNumAndFlattenIdx(fd int) (int, int) {
	// fd / r.granularity - granule number
	gNum := fd / r.granularity

	// gNum/r.shardCnt - granule number in shard
	// granule number in shard *r.granularity - index of first el in granule
	// index of granule start + fd % r.granularity - index of file descriptor in shard
	return gNum % r.shardCnt, (gNum/r.shardCnt)*r.granularity + (fd % r.granularity)
}

func (r *cbRegistry) add(fd int, cb Callback) uint32 {
	shardNum, idx := r.shardNumAndFlattenIdx(fd)
	return r.shards[shardNum].
		add(idx, cb)
}

func (r *cbRegistry) pop(fd int, nonce uint32) Callback {
	shardNum, idx := r.shardNumAndFlattenIdx(fd)
	return r.shards[shardNum].
		pop(idx, nonce)
}

Управление потоками

Так как одна горутина в языке GO в разные моменты времени может выполняться на разных тредах, то может возникнуть ситуация, когда на одном из тредов будет выполняться сразу два и более кольца (то есть два и более конкурентных вызовов io_uring_enter). Это не лучший расклад, так как в свою очередь это означает, что могут быть треды не занятые socket I/O. Чтобы избежать подобных ситуаций, можно воспользоваться функцией:

runtime.LockOSThread()

Таким образом, можно быть уверенным, что каждый экземпляр io_uring обрабатывает запросы в своем треде.

Переиспользуем память

Кроме специфичных оптимизаций не забываем и о стандартных вещах, например, вместо подобного кода внутри (net.Conn).Read:

	// помещаем Recv операцию в реактор
	op := uring.Recv(uintptr(c.Fd), b, 0)
	c.reactor.Queue(op, func(event uring.CQEvent) {
		c.readChan <- event
	})

Лучше переиспользовать единожды созданную операцию во избежание повторных аллокаций:

	op := c.readOp // создаем readOp единожды, при создании connection
	op.SetBuffer(b)

	c.reactor.Queue(op, func(event uring.CQEvent) {
		c.readChan <- event
	})

В общем следим за тем чтобы максимально переиспользовать доступные ресурсы: буфера, каналы, горутины.

Осада netpoller

Используем описанные выше идеи, добавляем щепотку микрооптимизаций и получаем новый NetReactor, готовый соперничать с nepoller'ом GO:

NetReactor

//go:build linux

package reactor

import (
	"context"
	"errors"
	"github.com/godzie44/go-uring/uring"
	"math"
	"runtime"
	"sync/atomic"
	"syscall"
	"time"
)

const (
	timeoutNonce = math.MaxUint64
	cancelNonce  = math.MaxUint64 - 1

	cqeBuffSize = 1 << 7
)

//RequestID identifier of operation queued into NetworkReactor.
type RequestID uint64

func packRequestID(fd int, nonce uint32) RequestID {
	return RequestID(uint64(fd) | uint64(nonce)<<32)
}

func (ud RequestID) fd() int {
	var mask = uint64(math.MaxUint32)
	return int(uint64(ud) & mask)
}

func (ud RequestID) nonce() uint32 {
	return uint32(ud >> 32)
}

//NetworkReactor is event loop's manager with main responsibility - handling client requests and return responses asynchronously.
//NetworkReactor optimized for network operations like Accept, Recv, Send.
type NetworkReactor struct {
	loops []*ringNetEventLoop

	registry *cbRegistry

	config *configuration
}

//NewNet create NetworkReactor instance.
func NewNet(rings []*uring.Ring, opts ...Option) (*NetworkReactor, error) {
	for _, ring := range rings {
		if err := checkRingReq(ring, true); err != nil {
			return nil, err
		}
	}

	r := &NetworkReactor{
		config: &configuration{
			tickDuration: time.Millisecond * 1,
			logger:       &nopLogger{},
		},
	}

	r.registry = newCbRegistry(len(rings), fdPerGranule)

	for _, opt := range opts {
		opt(r.config)
	}

	for _, ring := range rings {
		loop := newRingNetEventLoop(ring, r.config.logger, r.registry)
		r.loops = append(r.loops, loop)
	}

	return r, nil
}

//Run start NetworkReactor.
func (r *NetworkReactor) Run(ctx context.Context) {
	for _, loop := range r.loops {
		go loop.runConsumer(r.config.tickDuration)
		go loop.runPublisher()
	}

	<-ctx.Done()

	for _, loop := range r.loops {
		loop.stopConsumer()
		loop.stopPublisher()
	}
}

//NetOperation must be implemented by NetworkReactor supported operations.
type NetOperation interface {
	uring.Operation
	Fd() int
}

type subSqeRequest struct {
	op       uring.Operation
	flags    uint8
	userData uint64

	timeout time.Duration
}

func (r *NetworkReactor) queue(op NetOperation, cb Callback, timeout time.Duration) RequestID {
	ud := packRequestID(op.Fd(), r.registry.add(op.Fd(), cb))

	loop := r.loopForFd(op.Fd())
	loop.reqBuss <- subSqeRequest{op, 0, uint64(ud), timeout}

	return ud
}

const fdPerGranule = 75

func (r *NetworkReactor) loopForFd(fd int) *ringNetEventLoop {
	n := len(r.loops)
	h := fd / fdPerGranule
	return r.loops[h%n]
}

//Queue io_uring operation.
//Return RequestID which can be used as the SQE identifier.
func (r *NetworkReactor) Queue(op NetOperation, cb Callback) RequestID {
	return r.queue(op, cb, time.Duration(0))
}

//QueueWithDeadline io_uring operation.
//After a deadline time, a CQE with the error ECANCELED will be placed in the callback function.
func (r *NetworkReactor) QueueWithDeadline(op NetOperation, cb Callback, deadline time.Time) RequestID {
	if deadline.IsZero() {
		return r.Queue(op, cb)
	}

	return r.queue(op, cb, time.Until(deadline))
}

//Cancel queued operation.
//id - SQE id returned by Queue method.
func (r *NetworkReactor) Cancel(id RequestID) {
	loop := r.loopForFd(id.fd())
	loop.cancel(id)
}

type ringNetEventLoop struct {
	ring *uring.Ring

	registry *cbRegistry

	reqBuss      chan subSqeRequest
	submitSignal chan struct{}

	stopConsumerCh  chan struct{}
	stopPublisherCh chan struct{}

	submitAllowed uint32

	log Logger
}

func newRingNetEventLoop(ring *uring.Ring, logger Logger, registry *cbRegistry) *ringNetEventLoop {
	return &ringNetEventLoop{
		ring:            ring,
		reqBuss:         make(chan subSqeRequest, 1<<8),
		submitSignal:    make(chan struct{}),
		stopConsumerCh:  make(chan struct{}),
		stopPublisherCh: make(chan struct{}),
		registry:        registry,
		log:             logger,
	}
}

func (loop *ringNetEventLoop) runConsumer(tickDuration time.Duration) {
	cqeBuff := make([]*uring.CQEvent, cqeBuffSize)
	for {
		loop.submitSignal <- struct{}{}

		_, err := loop.ring.WaitCQEventsWithTimeout(1, tickDuration)
		if errors.Is(err, syscall.EAGAIN) || errors.Is(err, syscall.EINTR) || errors.Is(err, syscall.ETIME) {
			runtime.Gosched()
			goto CheckCtxAndContinue
		}

		if err != nil {
			loop.log.Log("io_uring", loop.ring.Fd(), "wait cqe", err)
			goto CheckCtxAndContinue
		}

		loop.submitSignal <- struct{}{}

		for n := loop.ring.PeekCQEventBatch(cqeBuff); n > 0; n = loop.ring.PeekCQEventBatch(cqeBuff) {
			for i := 0; i < n; i++ {
				cqe := cqeBuff[i]

				if cqe.UserData == timeoutNonce || cqe.UserData == cancelNonce {
					continue
				}

				id := RequestID(cqe.UserData)
				cb := loop.registry.pop(id.fd(), id.nonce())
				cb(uring.CQEvent{
					UserData: cqe.UserData,
					Res:      cqe.Res,
					Flags:    cqe.Flags,
				})
			}

			loop.ring.AdvanceCQ(uint32(n))
		}

	CheckCtxAndContinue:
		select {
		case <-loop.stopConsumerCh:
			close(loop.stopConsumerCh)
			return
		default:
			continue
		}
	}
}

func (loop *ringNetEventLoop) stopConsumer() {
	loop.stopConsumerCh <- struct{}{}
	<-loop.stopConsumerCh
}

func (loop *ringNetEventLoop) stopPublisher() {
	loop.stopPublisherCh <- struct{}{}
	<-loop.stopPublisherCh
}

func (loop *ringNetEventLoop) cancel(id RequestID) {
	op := uring.Cancel(uint64(id), 0)

	loop.reqBuss <- subSqeRequest{
		op:       op,
		userData: cancelNonce,
	}
}

func (loop *ringNetEventLoop) runPublisher() {
	runtime.LockOSThread()

	defer close(loop.reqBuss)
	defer close(loop.submitSignal)

	var err error
	for {
		select {
		case req := <-loop.reqBuss:
			atomic.StoreUint32(&loop.submitAllowed, 1)

			if req.timeout == 0 {
				err = loop.ring.QueueSQE(req.op, req.flags, req.userData)
			} else {
				err = loop.ring.QueueSQE(req.op, req.flags|uring.SqeIOLinkFlag, req.userData)
				if err == nil {
					err = loop.ring.QueueSQE(uring.LinkTimeout(req.timeout), 0, timeoutNonce)
				}
			}

			if err != nil {
				id := RequestID(req.userData)
				loop.registry.pop(id.fd(), id.nonce())
				loop.log.Log("io_uring", loop.ring.Fd(), "queue operation", err)
			}

		case <-loop.submitSignal:
			if atomic.CompareAndSwapUint32(&loop.submitAllowed, 1, 0) {
				_, err = loop.ring.Submit()
				if err != nil {
					if errors.Is(err, syscall.EBUSY) || errors.Is(err, syscall.EAGAIN) {
						atomic.StoreUint32(&loop.submitAllowed, 1)
					} else {
						loop.log.Log("io_uring", loop.ring.Fd(), "submit", err)
					}
				}
			}

		case <-loop.stopPublisherCh:
			close(loop.stopPublisherCh)
			return
		}
	}
}

Пора в последний раз написать tcp-echo сервер и сделать бенчмарки. Полное описание процесса тестирования есть здесь, а результаты представлены ниже:

	c: 100 bytes: 128	c: 50 bytes: 1024	c: 500 bytes: 128	c: 1000 bytes: 128	c: 1000 bytes: 128	c: 1000 bytes: 1024
net/http	132664	139206	133039	139171	133480	139617
go-uring	34202	33159	147362	139313	158483	154194
go-uring SQ_POLL mode	24406	22847	134863	130668	127896	122601

При более-менее большом количестве соединений получилось выжать производительность даже лучше, чем у привычного runtime. И это при том, что runtime использует недоступные в user-space'е низкоуровневые примитивы синхронизации (например, напрямую жонглирует горутинами), тогда как reactor для этих же целей использует общедоступные вещи - каналы, мьютексы, атомики. С другой стороны, получить достойный rps на небольшом количестве коннектов не вышло. Получилось так, что либо в системе много потоков со слабо нагруженными кольцами (в этом случае теряются преимущества io_uring) либо один-два потока с нагруженными кольцами (что лучше, но не получается полностью утилизировать ресурс CPU).

Снова дома

И наконец-то эта серия закончена. Для меня это первый опыт в подобном формате. Не скрою, что каждый следующий шаг давался сложнее, чем предыдущий, поэтому я действительно рад довести дело до конца. Надеюсь, мне удалось выполнить поставленную цель - продемонстрировать, что io_uring перспективная технология, которая как минимум достойна внимания. Напоследок полезные ссылки:

• первая и вторая части

• https://kernel.dk/io_uring.pdf — whitepaper

• https://unixism.net/loti/index.html — блог с примерами реализаций простых приложений

• https://github.com/axboe/liburing — liburing

• https://github.com/godzie44/go-uring - библиотека для GO которая используется в материалах

• https://github.com/frevib/io_uring-echo-server/ - плэйграунд для бенчмарков на C

• https://github.com/axboe/liburing/issues/487 - о том как работает  FEAT_FAST_POLL

---

Дата-центр ITSOFT — размещение и аренда серверов и стоек в двух дата-центрах в Москве. За последние годы UPTIME 100%. Размещение GPU-ферм и ASIC-майнеров, аренда GPU-серверов, лицензии связи, SSL-сертификаты, администрирование серверов и поддержка сайтов.