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

Коротко

Кольцевой буфер (ring buffer) — очередь на массиве фиксированного размера, где конец «замыкается» на начало.

Зачем нужен

Обычная очередь на динамическом массиве периодически перевыделяет память. Для большинства задач это нормально, но есть области, где недопустимо:

• обработка звука и видео — пауза на выделение памяти даёт слышимый щелчок;
• встраиваемые системы — динамической памяти может не быть вообще;
• системы реального времени — нужна предсказуемая задержка;
• логирование — хранить последние N записей и не расти бесконечно;
• сетевые буферы — приём данных с постоянной скоростью.

Кольцевой буфер выделяет память один раз при создании и больше никогда.

Как работает

Массив фиксированного размера и два указателя: head (откуда читать) и tail (куда писать). При достижении конца массива указатель возвращается в начало.

размер 5, записали 3 элемента:

[ A ][ B ][ C ][   ][   ]
  ↑              ↑
 head           tail

прочитали A, записали D и E:

[   ][ B ][ C ][ D ][ E ]
       ↑                  ↑ tail завернулся на 0
      head

записали F:

[ F ][ B ][ C ][ D ][ E ]
  ↑    ↑
 tail head

Замыкание делается остатком от деления:

tail = (tail + 1) % capacity

Проблема пустого и полного

Тонкость, из-за которой кольцевой буфер сложнее, чем кажется. Когда буфер пуст, head == tail. Когда полон — тоже head == tail. Различить состояния по указателям невозможно.

Три стандартных решения:

1. Хранить счётчик элементов. Самое простое и понятное.

2. Оставлять одну ячейку пустой. Буфер считается полным при (tail + 1) % capacity == head. Теряем одну ячейку, зато не нужен счётчик — удобно для атомарных операций без блокировок.

3. Хранить флаг «полон». Работает, но требует аккуратности при каждой операции.

Ниже реализация со счётчиком — она читается лучше всего.

Реализация

class CircularBuffer:
    def __init__(self, capacity: int, overwrite: bool = False):
        self._data = [None] * capacity
        self._capacity = capacity
        self._head = 0        # откуда читаем
        self._tail = 0        # куда пишем
        self._size = 0
        self._overwrite = overwrite   # затирать старое при переполнении

    def __len__(self) -> int:
        return self._size

    def is_full(self) -> bool:
        return self._size == self._capacity

    def push(self, item) -> None:
        if self.is_full():
            if not self._overwrite:
                raise OverflowError('буфер полон')
            # затираем самый старый: сдвигаем head
            self._head = (self._head + 1) % self._capacity
            self._size -= 1

        self._data[self._tail] = item
        self._tail = (self._tail + 1) % self._capacity
        self._size += 1

    def pop(self):
        if self._size == 0:
            raise IndexError('буфер пуст')

        item = self._data[self._head]
        self._data[self._head] = None      # помогаем сборщику мусора
        self._head = (self._head + 1) % self._capacity
        self._size -= 1
        return item

    def peek(self):
        if self._size == 0:
            raise IndexError('буфер пуст')
        return self._data[self._head]

    def __iter__(self):
        for i in range(self._size):
            yield self._data[(self._head + i) % self._capacity]


buf = CircularBuffer(3)
buf.push('a')
buf.push('b')
buf.push('c')
print(list(buf))     # ['a', 'b', 'c']
print(buf.pop())     # a
buf.push('d')
print(list(buf))     # ['b', 'c', 'd']


# режим перезаписи: всегда последние N элементов
log = CircularBuffer(3, overwrite=True)
for i in range(6):
    log.push(i)
print(list(log))     # [3, 4, 5]

Два режима переполнения

Отказ. При полном буфере новые данные отвергаются. Подходит там, где терять данные нельзя — например, при приёме сетевых пакетов с подтверждением.

Перезапись. Старые данные затираются новыми. Подходит для логов, метрик, буфера последних кадров — там, где актуальность важнее полноты.

Выбор режима — архитектурное решение, и его стоит делать осознанно.

Оптимизация: степень двойки

Операция взятия остатка % относительно медленная. Если размер буфера — степень двойки, её можно заменить битовой маской:

# вместо (tail + 1) % capacity
tail = (tail + 1) & (capacity - 1)

Работает потому, что для capacity = 2^k остаток от деления равен младшим k битам. Приём стандартен в высокопроизводительных реализациях: LMAX Disruptor, сетевые драйверы, звуковые буферы.

В Python это уже есть

Для большинства задач писать руками не нужно:

from collections import deque

buf = deque(maxlen=5)
for i in range(10):
    buf.append(i)

print(list(buf))   # [5, 6, 7, 8, 9]

deque с maxlen — готовый кольцевой буфер в режиме перезаписи, реализованный на C. Собственная реализация оправдана только при нестандартных требованиях или в учебных целях.

Применение в конкурентном коде

Кольцевой буфер — основа паттерна «производитель-потребитель». Когда производитель один и потребитель один, буфер можно сделать вообще без блокировок: производитель трогает только tail, потребитель только head.

Этот вариант называется SPSC-очередью (single producer, single consumer) и используется в аудиодрайверах, где поток обработки звука не имеет права ждать на мьютексе.

При нескольких производителях или потребителях всё усложняется — нужны атомарные операции и продуманный порядок обновления указателей.

Частые ошибки

Неразличение пустого и полного. Разобрано выше — без счётчика или пустой ячейки состояния неотличимы.

Забытое обновление размера. Указатели уезжают, а len() врёт.

Отсутствие обнуления при извлечении. Ссылка на объект остаётся в массиве, и он не собирается сборщиком мусора. На больших объектах это заметная утечка.

Итерация по массиву вместо логического порядка. Обходить нужно от head со сдвигом по модулю, а не по физическим индексам.

Использование % в горячем цикле. При строгих требованиях к скорости берите размер степенью двойки.

Что запомнить

• Кольцевой буфер — очередь на массиве фиксированного размера с замыканием по модулю.
• Память выделяется один раз, в рантайме аллокаций нет.
• Пустое и полное состояние неразличимы по указателям — нужен счётчик или жертва одной ячейкой.
• Два режима переполнения: отказ и перезапись, выбор зависит от задачи.
• В Python готовое решение — collections.deque(maxlen=N).

Пройди собеседование в топ-компанию

Платформа для подготовки

Решай алгоритмические задачи как профи

✓ Популярные алгоритмы✓ Разбор решений✓ AI помощь

Начать сейчас→