Кольцевой буфер: лента, у которой нет конца

Если данные приходят непрерывно, а хранить нужно только свежие, удобно пустить запись по кругу.

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

Проблема бесконечного потока

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

Можно держать обычный список и, добавляя новый элемент, удалять самый старый из начала. Но удаление из начала списка медленное — приходится сдвигать все остальные элементы. На быстром потоке это превращается в проблему.

Идея: замкнуть массив в кольцо

Кольцевой буфер решает это изящно. Берём массив фиксированного размера — скажем, на сто ячеек — и пишем в него по очереди. Когда заполнили последнюю, сотую, ячейку, не расширяем массив, а возвращаемся к первой и пишем поверх неё. Самое старое значение затирается самым новым. Запись будто бежит по кругу, отсюда и название.

Чтобы не запутаться, буфер помнит, куда писать дальше — это указатель, который после каждой записи сдвигается на одну позицию, а дойдя до конца, перескакивает в начало. Никаких сдвигов, никакого роста памяти: и запись, и чтение — за O(1).

Магия остатка от деления

«Перескочить в начало по достижении конца» программируется одной операцией — остатком от деления. Позиция (index + 1) % size идёт 0, 1, 2, ..., размер−1, а потом снова 0. Тот самый оператор %, что закрывает кольцо.

Кольцевой буфер своими руками

class RingBuffer:
    def __init__(self, size):
        self.size = size
        self.data = [None] * size
        self.pos = 0          # куда писать следующий элемент
        self.count = 0        # сколько реально записано

    def add(self, value):
        self.data[self.pos] = value      # затираем старое
        self.pos = (self.pos + 1) % self.size   # шаг по кругу
        self.count = min(self.count + 1, self.size)

    def recent(self):
        # вернуть элементы от старого к новому
        if self.count < self.size:
            return self.data[:self.count]
        return self.data[self.pos:] + self.data[:self.pos]

buf = RingBuffer(3)            # храним только 3 последних
for x in [10, 20, 30, 40, 50]:
    buf.add(x)
print(buf.recent())   # [30, 40, 50] — первые два затёрлись

Мы добавили пять чисел в буфер на три ячейки — и остались только три последних. Числа 10 и 20 были аккуратно затёрты, и память при этом не выросла ни на байт.

Где крутятся кольца

Кольцевые буферы повсюду в технике. Аудио- и видеоплееры держат в них поток данных, чтобы воспроизведение не прерывалось при рывках сети. Сетевые карты складывают сюда пакеты. Системы логирования хранят «последние N строк» именно так. В играх кольцевым буфером записывают последние кадры ввода. Везде, где есть непрерывный поток, а хранить нужно только свежий хвост.

Связь с очередью

По сути кольцевой буфер — это компактная очередь фиксированной ёмкости. Элементы входят с одного конца, выходят с другого, порядок (FIFO) соблюдается. Отличие в том, что ёмкость заранее ограничена, и при переполнении буфер не растёт, а спокойно выбрасывает самое старое. Иногда именно такое поведение — «забывать древнее» — это ровно то, что нужно.

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