Содержание
В Linux есть несколько особенностей, которые можно использовать для реализации кольцевого буфера. Существует два набора таких особенностей:
Функции, подходящие для определения информации о буферах размером в степень двойки.
Барьеры памяти для случая, когда производитель и потребитель объектов в этих буферах не желают использовать блокирование.
Для использования этих особенностей, необходим только один производитель и один потребитель. Возможно управление множеством производителей путём их сериализации и управление множеством потребителей, с помощью их сериализации.
Прежде всего, что такое кольцевой буфер? Кольцевым буфером называется буфер фиксированного и конечного размера, содержащий два указателя:
'Головной' указатель - точка, начиная с которой производитель размещает созданные объекты.
'Хвостовой' указатель - точка, начиная с которой потребитель ищет следующий элемент в буфере.
Типичная ситуация, когда хвостовой указатель совпадает с головным, указывает на то, что буфер пуст; и признаком заполнения буфера является ситуация, когда головной указатель на единицу меньше хвостового.
Головной указатель увеличивается на единицу, когда в буфер добавляется элемент, а хвостовой указатель наращивается. когда элемент забирается из буфера. Хвостовой указатель никогда не должен "перепрыгивать" через головной. Оба указателя должны сбрасываться в 0, при достижении конца буфера, позволяя таким образом, бесконечному количеству данных протекать через этот буфер.
Обычно, все элементы одинакового размера, но это не строгое ограничение на использование технологии, описанной ниже. Индекс может наращиваться более, чем на 1, если множество элементов, или элементы переменной длинны, помещаются в буфер, обеспечивая, что ни один из индексов не догонит другого. Реализация должна быть безопасной и мощной, так как регион размером более 1, может быть "завёрнут" с конца буфера на его начало и разделён на два сегмента.
Вычисление занятой или оставшейся ёмкости кольцевого буфера произвольного размера, обычно будет достаточно медленной операцией, так как требует использования команд деления по модулю. Однако, если буфер имеет размер кратный степени двойки, то взамен можно использовать намного более быструю команду двоичного AND.
Linux предоставляет набор макросовдля управления кольцевыми буферами размером, кратным степени двойки. Их можно исполь зовать следующим образом:
#include <linux/circ_buf.h>
Содержит макросы:
CIRC_SPACE(head_index, tail_index, buffer_size);
Возвращает количество пространства, оставшегося в буфере [1] в котором может быть размещён элемент.
CIRC_SPACE_TO_END(head_index, tail_index, buffer_size);
Возвращает количество непрерывного пространства, оставшегося в буфере [1] который может быть непосредственно вставлен элемент, без перехода на начало буфера.
CIRC_CNT(head_index, tail_index, buffer_size);
Возвращает количество элементов, занятых в буфере [2] в настоящий момент.
CIRC_CNT_TO_END(head_index, tail_index, buffer_size);
Врзвращает количество последовательных элементов [2], которые могут быть выбраны из буфера, до того, как произойдёт переход на начало буфера.
Каждый из этих макросов должен возвращать значение между 0 и buffer_size-1, однако:
[1] CIRC_SPACE*() предлагается для использования производителем. Производителю будет возвращаться нижняя граница, поскольку производитель управляет головным указателем, но потребитель может в это время исчерпывать буфер на другом центральном процессоре и переместить указатель хвоста.
Потребетелю будет показана верпхняя граница, так как производитель в это время может заниматься исчерпанием пространства.
[2] CIRC_CNT*() предлагается для использования потребителем. Потребителю будет возвращаться нижняя граница, как потребитель управляет хвостовым индексом, но производитель в это же время может заполнять буфер на другом ЦПУ и переместит указатель головы.
Производителю будет показана верхняя граница так как потребитель может быть зханят в это время опустошением буфера.
[3]Для всех остальных, видимый порядок, в котором выполняется запись в указатели производителем и потребителем, не может быть гарантирован, так как они независимы и могут выполняться на различных ЦПУ, поэтому о результате в такой ситуации можно только догадываться, и он может быть даже обратным.
Если использовать барьеры памяти, то станет ненужным делать следующее:
использовать одну локировку для управления доступом к обоим концам буфера, позоволяя, таким образом, одновременно заполнять и опустошать буфер; и
атомарную инструкцию для счётчика
Здесь есть две стороны: производитель, который заполняет буфер и потребитель, который опустошает его. В любой момент времени только одна сущность может заполнять буфер и только одна сущность может опустошать его, но эти две стороны могут работать одновременно.
Производитель должен выглядет приблизительно так:
spin_lock(&producer_lock);
unsigned long head = buffer->head;
unsigned long tail = ACCESS_ONCE(buffer->tail);
if (CIRC_SPACE(head, tail, buffer->size) >= 1) {
/* insert one item into the buffer */
struct item *item = buffer[head];
produce_item(item);
smp_wmb(); /* commit the item before incrementing the head */
buffer->head = (head + 1) & (buffer->size - 1);
/* wake_up() will make sure that the head is committed before
* waking anyone up */
wake_up(consumer);
}
spin_unlock(&producer_lock);
Это будет сообщать ЦПУ, что содержимое нового элемента должно быть до того, как головной указатель сделает его доступным для потребителя и затем инструктирует ЦПУ, что новое значение указателя головы должно быть записано до того, как потребитель проснётся.
Заметим, что wake_up() не реализует в точности этот механизм, но любой используемый механизм должен гарантировать барьер га запись в память, между обновлением головного указателя и изменением состояния потребителя, если происходит такое изменение.
Потребитель должен выглядеть как-то так:
spin_lock(&consumer_lock);
unsigned long head = ACCESS_ONCE(buffer->head);
unsigned long tail = buffer->tail;
if (CIRC_CNT(head, tail, buffer->size) >= 1) {
/* read index before reading contents at that index */
smp_read_barrier_depends();
/* extract one item from the buffer */
struct item *item = buffer[tail];
consume_item(item);
smp_mb(); /* finish reading descriptor before incrementing tail */
buffer->tail = (tail + 1) & (buffer->size - 1);
}
spin_unlock(&consumer_lock);
Так мы инструктируем ЦПУ проверить, что указатель обновлён и актуален, до чтения нового элемента и после этого, необходимо, что бы ЦПУ проверил, что закончилочь чтение элемента до того, как будет записано новое значение хвостового указателя, который будет стирать элемент.
Обратите внимание на использование ACCESS_ONCE() в обоих алгоритмах для чтения противоположных указателей. Это не даёт возможности компиллятору отбрасывать и перезагружать кешированные значения - что будут делать некоторые компилляторы во время вызова smp_read_barrier_depends(). Это не является обязательным условием, если Вы уверены, что противоположные указатели будут использоваться только по одному.