HBS2 P2P Storage and Platform

Коротко про текущую архитектуру CAS хранилища HBS2

Наверное, что бы было понятно, как мы дошли до текущей структуры придется немного углубиться в историю для контекста.

Первоначальное хранилище HBS2 имело вид наивного первоначального git-like хранилища:

root/XX/YYY..YY

То есть прямо каталог, первый уровень – первый байт хэша, далее файл данных с именем — оставшиеся символы хэша в кодировке Base58.

В git объект это дерево, коммит или блоб, то есть файл. Файл может быть любого размера, например, очень большой. Например, настолько большой, что не помещается в память.

В HBS2 единицей, имеющей хэш, является блок. Размер блока такой, что он не слишком большой и не слишком маленький. Слишком большие блоки делать опасно (память и сеть), слишком маленькие — дорого (считать хэши).

Размер блока около 256Kb, может быть меньше или больше (например, с шифрованием). Данные большего размера представляются как деревья Меркла, где каждый узел является блоком в смысле HBS2, а лист является пользовательскими данными, т.е если склеить листовые блоки в порядке обхода дерева — получим исходный файл или блоб.

В git объекты сжимаются zlib и хэшируются вместе с префиксом типа, в HBS2 сознательно не сжимаются, так как была надежда отдавать их sendfile.

Приведенная выше структура (один блок — один файл) очень проста и работает не так плохо, как может показаться. Лучше, чем в гите, т.к по факту в гите очень много очень маленьких объектов, а тут они, всё же, близки к 256Kb.

Что интересно, эта структура работает лучше, чем если бы мы держали объекты в SQLite.

Эта структура все равно не так быстра на запись, как хотелось бы, кроме того, с ростом числа объектов ведет к деградации скорости работы файловой системы. Потому, что файловая это Btree, и с ростом числа объектов и временем Btree деградирует – скорость записи и перестроения индекса растёт.

Журналы

Таким образом, было принято решение разработать новую версию хранилища, используя иммутабельные журналоподобные структуры с добавлением в конец, без переписывания.

Здесь одним их основных источников опять стал git.

Как известно, git пишет объекты сначала в упомянутый формат (git like loose, см. выше) и потом переписывает их в файлы журналов.

Файлы эти имеют достаточно хитрый дизайн из-за желания сжать данные как можно сильнее, используя xdelta, т.к. git порождает огромную избыточность данных и повторять это дизайн в точности нет смысла – слишком сложно и не нужно в нашем контексте.

Не будем углубляться в специфику git. Файлы журналов HBS2 имеют формат:

(SD)*

то есть множество записей вида SD, где S это размер секции данных в формате W32BE, а D - сами произвольные данные. Как видите формат настолько простой, что даже нет смысла делать для него какие-то библиотеки, очень легко обработать вручную.

На этом уровне это еще не CAS. Что бы это превратилось уже во что-то осмысленное, нужно специфицировать сами данные. Которые в случае HBS2 имеют вид:

KPD

где - K – ключ (обычно хэш), байстрока.

P - префикс, который может быть B|R|T|M.

B - это блоб, он же блок. Адресуемый хэшем блок данных, и его хэш равен K.

R - это ссылка, то есть указатель на блок. Ключ ссылки не равен хэшу данных, а является произвольным значением.

T — Tomb, признак удалённости некоего блока.

M — метаданные, служебные данные хранилища.

На всякий случай, префикс P имеет размер 4 байта, с целью какого-то выравнивания и на случай необходимости какого-то расширения (признак сжатия и т.п.)

Каждый файл данных имеет таймстемп, кроме того, в текущей версии хранилища файлы нумеруются 64-битным монотонно возрастающим счётчиком, таким образом, для неповрежденного хранилища всегда можно однозначно выстроить правильную последовательность файлов.

Поскольку файлы имеют порядок, а записи в файлах естественно упорядочены позицией — файлы append-only — то верным значением данных с ключом K является самое последнее известное. То есть из файла с большим номером и таймстемпом и имеющее большее смещение внутри этого файла.

Индексы

Что бы иметь возможность быстрого поиска данных по ключу, журналы необходимо индексировать.

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

В HBS2 (NCQ) это дисковые страничные хэш-таблицы:

P0: B1|B1|B1|B1
    B2|B2|B2|B2
    ...
    Bn|Bn|Bn|Bn
    ...

Pn: B1|B1|B1|B1
    B2|B2|B2|B2
    ...
    Bk|Bk|Bk|Bk

Поскольку у нас любой ключ это хэш известного размера, мы разбиваем его на компоненты Ki по 32 бита, сначала ищем место в первом бакете — если нашли свободное место, то размещаем, если не нашли – то берем следующие 32 бита ключа и так далее, пока не разместили. Если разместить не удалось – то увеличиваем размеры бакетов и начинаем сначала.

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

Количество слотов в бакете B больше одного – там их 4 с целью выравнивания в кэше.

Бакет B устроен так:

KEY:BYTES(32)|OFFSET:W64BE|SIZE:W32BE

Спойлер

Теперь на самом деле так:

KEY|FILEID|OFFSET|SIZE|PADDING

Дизайн этой хэштаблицы навеян Cuckoo индексами в RocksDB, но проще и быстрее на запись, за счёт худшей плотности индекса.

Типичное значение для населенности лежит в диапазоне 0.4 .. 0.67

Но это не так важно, т.к. мы храним данные в этих файлах их размеры на два порядка меньше, чем размеры журналов.

В git один .idx файл соответствует одному файлу журнала и точно таким же был первоначальный дизайн NCQ.

Поиск одного значения в таком индексе — O(1). Но с ростом числа файлов становится O(K), где K — это число файлов.

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

Чем больше файлы – тем дороже их переписывать. Поэтому принято решение в индекс добавить идентификатор файла, и сливать только индексы, а сами файлы сливать только до достижения определенного размера, после чего не трогать.

Легко видеть, что раз есть ссылки – то они могут стать неконсистентными. Что бы избегать этого, все изменения идут через персистентные состояния, или же States.

Состояние (State)

Все операции фиксируются в файле состояний (state) после выполнения операции.

Каждое состояние записывается в отдельный файл и получает монотонно растущий номер.

В случае краха хранилища — состояние будет взято из последнего успешно записанного файла.

Состояние хранит набор ключей (идентификаторов) файлов, известных системе.

Неактуальные файлы удаляются только, если не существует состояний, которые на них ссылаются.

State атомарно пишется на диск при каждом изменении.

Поиск

При поиске мы сначала ищем в memtable (рассмотрим позже).

Если найдено – возвращаем что нашли, если нет — то рассматриваем индексы, в порядке обратном номеру (таймстемпу).

Первое найденное (последнее проиндексированное) значение является актуальным.

Поиск одного значения – O(1) или же O(k) где k - количество индексов.

Объединение (merge) индекса

Для подряд идущих упорядоченных по номерам (времени) файлов индекса наименьшего имеющегося размера I,J:

Мерж индекса: I, J ∣ T(I) > T(J) ⟹ K = I ∪ (J ∖ I), T(K) = T(I)

В конце DEL I, DEL J, K ⟹ I.

Удаление здесь (и везде) осуществляется через State. Т.е мы удаляем ссылки на ключи индексов из State, и потом сборка мусора удалит файлы, отсутствующие в State.

В итоге у нас получается один файл индекса.

Объединение (merge) журналов

Для подряд идущих упорядоченных по номерам (времени) файлов наименьшего имеющегося размера

I,J | Size(I) + Size(J) < M

Мерж: I, J ∣ T(I) > T(J) ⟹ K = I ∪ (J ∖ I), T(K) = T(I)

В конце DEL I, DEL J, K ⟹ I.

Пишется наиболее актуальное значение – т.е из файла с большим ключом и смещением и только в том случае, если оно присутствует в индексе – отсутствие в индексе будущий признак удаления для сборки мусора.

Сборка мусора

Неактуальные блоки собираются при слиянии — т.к в результат пишутся только актуальные блоки.

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

Удаление записей типа T – потребует скана от текущего файла в глубину, что бы гарантировать, что больше нет записей, которые экранирует томб. Это можно сделать за O(1) памяти, если использовать скан с фильтром Блума или Куку-фильтром.

Запись

Сначала записанное новоек значение пишется в Memtable и очередь записи.

Memtable представляет собой структуру вида:

type Shard = TVar (HashMap HashRef NCQEntry)

memtable :: Vector Shard

...

data  NCQEntry =
  NCQEntry
  { ncqEntryData   :: !ByteString
  , ncqDumped      :: !(TVar (Maybe FileKey))
  }

Номер шарда определяется по остатку от деления первого байта ключа на количество шард.

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

Запись осуществляется в один поток в текущий файл.

После достижения порога Fsync пишем блок метаданных M данными в которых является текущий размер файла с учётом этого заголовка.

Блок M:

K:BYTES(32)|M;;\n|SIZE:W64BE

После записи данного блока вызываем fsync и пишем дальше.

После fsync в state добавляется ссылка на ключ файла и последний известный валидный размер.

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

Старт

При старте читаем последний State, проверяем его, в случае валидности — стартуем с ним, удаляем все неактуальные стейты.

Журнал есть, индекс отсутствует

Проверяем сигнатуру в конце (размер, хэш), в случае успеха индексируем файл.

Файл поврежден — ищем последний валидный M блок, индексируем.

Рутины

Меряем текущую нагрузку, используем экспоненциальное скользящее среднее.

В случае, если меньше порога — делаем merge index, merge data или sweep. Не более одной операции за раз, регулируем семафором.

Прочее

Для обеспечения отсутствия гонок — т.к. состояние хранится в STM структурах, но файлы создаются/удаляются в IO — мы считаем количество ссылок на текущий стейт и не чистим его, пока стейт актуален и количество ссылок на него больше нуля.

Индексы и журналы отображаются в память (mmap). Есть пул mmap-ов, т.е их число ограничивается

Производительность

Запись

Запись с подсчётом хэшей и индексацией, Fsync = 16Mb: 850Mb/s

(sqlite: ~50Mb/s, писать в файл в один поток: 400Mb/s)

Чтение

Чтение: ключ 32 байта, 825K lookup/sec при 1 сегменте индекса и нескольких потоках (8) на чтение

Дефолтный RocksDB – в разы хуже, но использовался SSTable, а не Cuckoo.

SQLite – на порядок хуже.

Разное

Для предотвращения деградации производительности при большом числе операций применяется паттерн следующего вида:

Очереди операций ввода-вывода:

-- эта еще и шардированная
ncqWriteOps  :: Vector (TQueue (IO ()))

ncqSyncOps  :: TQueue (IO ())

...


  waiter <- newEmptyTMVarIO

  let work = do
        let h = fromMaybe (HashRef (hashObject @HbSync bs')) mhref
        let bs = ncqMakeSectionBS mtp h bs'
        let shard = ncqGetShard ncq h
        zero <- newTVarIO Nothing

        atomically do
          upd <- stateTVar shard $ flip HM.alterF h \case
                   Nothing -> (True, Just (NCQEntry bs zero))
                   Just e | ncqEntryData e /= bs  -> (True, Just (NCQEntry bs zero))
                          | otherwise -> (False, Just e)

          when upd  do
            modifyTVar ncqWriteQ (|> h)

          putTMVar waiter h

  -- большой кусок работы ставим в очередь
  -- и говорим, куда писать ответ
  atomically do
    modifyTVar ncqWrites succ
    nw <- readTVar ncqWrites <&> (`mod` V.length ncqWriteOps)
    writeTQueue (ncqWriteOps ! nw) work

  -- ждем ответ --- нельзя это делать в одной транзакции
  atomically $ takeTMVar waiter

Применение очередей IO () очень хорошо сказывается на графике числа операций от числа потоков.

Количество очередей воркеров определяется экспериментально, но скорее всего это будет Nw = 2.

Семафоры никак не помогают сгладить нагрузку.

Очереди помогают почти от всего.

Внимательно следим за циклами FSM при помощи fix или явной рекурсии и убеждаемся, что рекурсия 100% хвостовая.

Один ресурс (файл, например) желательно контролировать одним потоком и только одним.

Posix.Files быстрее стандартного ввода – вывода процентов на 15.

Если Fd использовать из разных потоков — то всё взорвётся.

Заключение

Используя простые примитивы и структуры данных вполне можно разработать маленькое простое хранилище, показывающее производительность уровня RocksDB или даже в чём-то лучше.

Текущая кодовая база NCQ3 (третья попытка сделать хранилище NCQ) — менее 1400 loc.