ringbuf

Без блокировки кольцевой буфер

Кольцевой буфер с несколькими продюсерами без блокировки (MPSC), который поддерживает операции смежного диапазона и который может быть удобно использовать для передачи сообщений. Реализация записана в C11 и распределена по лицензии BSD на 2 пункта.

API

• int ringbuf_setup(ringbuf_t *rbuf, unsigned nworkers, size_t length)

  • Настройка нового кольцевого буфера заданной длины . rbuf - это указатель на непрозрачный кольцевой объект; вызывающий абонент несет ответственность за распределение места для этого объекта. Как правило, объект был бы распределен динамически при использовании потоков или зарезервирован в общей памяти, заблокированной при использовании процессов. Размер распределения для объекта должен быть получен с использованием функции ringbuf_get_sizes . Возвращает 0 на успех и -1 при неудаче.

• void ringbuf_get_sizes(unsigned nworkers, size_t *ringbuf_obj_size, size_t *ringbuf_worker_size)

  • Возвращает размер непрозрачного ringbuf_t и, необязательно, ringbuf_worker_t . Размер структуры ringbuf_t зависит от количества работников, указанных параметром nworkers .

• ringbuf_worker_t *ringbuf_register(ringbuf_t *rbuf, unsigned i)

  • Зарегистрируйте текущего работника (поток или процесс) в качестве производителя. Каждый производитель должен зарегистрироваться. i - номер работника, начиная с нуля (то есть будет, чем nworkers используемые в настройке). При успехе возвращает указатель на непрозрачный ringbuf_worker_t structured, который является частью блока памяти ringbuf_t . При неудаче возвращает NULL .

• void ringbuf_unregister(ringbuf_t *rbuf, ringbuf_worker_t *worker)

  • Не регистрируйте указанный работник из списка производителей.

• ssize_t ringbuf_acquire(ringbuf_t *rbuf, ringbuf_worker_t *worker, size_t len)

  • Запросите пространство заданной длины в буфере кольца. Возвращает смещение, в котором пространство доступно или -1 при сбое. После того, как данные будут готовы (обычно, когда записывает в кольцевой буфер), должна быть вызвана функция ringbuf_produce , чтобы указать это. Вложенные приобретенные звонки не допускаются.

• void ringbuf_produce(ringbuf_t *rbuf, ringbuf_worker_t *worker)

  • Укажите, что приобретенный диапазон в буфере производится и готов к употреблению.

• size_t ringbuf_consume(ringbuf_t *rbuf, size_t *offset)

  • Получите смежный диапазон, который готов к употреблению. Возвращает ноль, если нет данных, доступных для потребления. После использования данных (как правило, при завершении чтения из кольцевого буфера) функция ringbuf_release должна быть вызвана, чтобы указать это.

• void ringbuf_release(ringbuf_t *rbuf, size_t nbytes)

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

Примечания

Потребитель вернет смежный блок диапазонов, произведенный т.е. ringbuf_consume Call не вернет частичные диапазоны. Если вы думаете о созданном диапазоне как о сообщении, то потребитель вернет блок сообщений, всегда заканчивая границей сообщений. Такое поведение позволяет нам использовать эту реализацию кольцевого буфера в качестве очереди сообщений.

Реализация была тщательно протестирована на 24-ядерном компьютере x86, см. Стресс-тест для деталей техники. Это также дает пример, как механизм можно использовать для передачи сообщений.

Предостережения

Эта реализация кольцевого буфера всегда обеспечивает смежный диапазон пространства для производителя. Это достигается ранним, если запрашиваемый диапазон не может соответствовать в конце. Смысл этого заключается в том, что вызов ringbuf_acquire может выйти из строя, если запрошенный диапазон превышает половину размера буфера. Следовательно, может потребоваться гарантировать, что размер буфера кольца не менее в два раза больше, чем максимальный размер производственного блока.

Следует также отметить, что одним из компромиссов такого дизайна является то, что потребитель в настоящее время выполняет сканирование O (n) в списке производителей.

Пример

Производители:

 if (( w = ringbuf_register ( r , worker_id )) == NULL )
	err ( EXIT_FAILURE , "ringbuf_register" )

...

if (( off = ringbuf_acquire ( r , w , len )) != -1 ) {
	memcpy ( & buf [ off ], payload , len );
	ringbuf_produce ( r , tls );
}

Потребитель:

 if (( len = ringbuf_consume ( r , & off )) != 0 ) {
	process ( & buf [ off ], len );
	ringbuf_release ( r , len );
}