5

Всем здравствуйте!

============ UPDATE =============

Решил обновить описание, т.к. оно было написано крайне криво, и обсуждение ушло не в ту сторону

upd: В терминалогии СУБД, как оказалось, это называется "Протокол ведения журнала с опережением записи (WAL)"

Есть 2 очереди:

// 1. Очередь записи бинарных данных в файл (журнал)
ConcurrentQueue<byte[]> Queue1;
// 2. Очередь последующей обработки данных
ConcurrentQueue<byte[]> Queue2;

Данные в очередь (1) записи попадают из разных потоков, данные туда могут попадать 1 раз в день, а могут миллион за одну секунду, причем размер сообщения может быть от 20 байт до 4 гигабайт.

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

После достижения определенного размера файл закрывается, затем открывается новый файл для записи и работа продолжается.

// Схематичный код потока обработки 1 очереди 
while (true)
{
    // Если в очереди есть сообщения, считываем первое (без извлечения)
    if (Queue1.TryPeek(out byte[] message))
    {
        // Увеличиваем переменную с порядковым номером сообщения и присваиваем индекс сообщению (~условно)
        message.Index = ++index;
        // Если стрима для записи нет, создаем его, в качестве имени используем индекс текущего сообщения
        if (fs == null) fs = new FileStream($"{TLOG_PATH}{index}.log", FileMode.Create, FileAccess.Write, FileShare.None);

        // Записываем сообщение в стрим (message в данном случае это byte[])
        fs.Write(message);

        // Перекидываем сообщение в очередь 2, однако это не значит что поток обрабатывающий очередь 2 может этими данными уже пользоваться, пока гарантий записи на диск нет!!!
        Queue2.Enqueue(message);
        // Извлекаем обработанное сообщение
        Queue1.TryDequeue(out _);

        // Если в очереди 1 больше нет данных, вызываем метод принудительной очистки буферов с ожиданием выполнения
        if (Queue1.Count == 0)
        {
            // Ждем пока FS гарантировано запишет данные на диск
            fs.Flush(true);
            // Устанавливаем индекс гарантированно записанного сообщения
            writedIndexMessage = index;
        }

        // Если размер файла журнала больше максимального закрываем файл
        if (fs.Length >= MAX_FS_LEN)
        {
            // Флушим
            fs.Flush(true);
            // Закрываем поток
            fs.Close();
            // Вызываем деструктор 
            fs.Dispose();
            // Освобождаем переменную
            fs = null;
        }
    }
}
// Схематичный код потока обработки 2й очереди
while (true)
{
    // Если в очереди есть транзакции, считываем первый (без извлечения)
    // и сверяем значение индекса сообщения с индексом сообщения гарантированно записанным на диск 1м потоком
    if (Queue2.TryPeek(out byte[] message) && message.Index <= writedIndex)
    {
        // ~~~~ Делаем что-то с данными ~~~~
        AnyDo(message);
        // Убираем успешно обработанное сообщение из очереди 2
        Queue2.TryDequeue(out _);
    }
}

(!!!) Вопросы:

// PS. Пожалуйста, имейте ввиду что код схематичный, тут нету оптимизаций работы с потоками (Thread) и т.п. Я понимаю что эти два потока в текущем примере полностью будут занимать 2 ядра.

  1. Есть ли какой-то простой/штатный механиз (в C# .net core 3.1+), используя который мы гарантировано можем знать, что данные физически записались на блин диска, без использования вызова метода fs.Flush(true), и данный механизм гарантировал бы, что при внезапном выключении питания, и последующей загрузке мы эти данные можем (!) гарантировано прочитать из файла. (разумеется если с физическим диском все в порядке), т.е. все блоки данных гарантировано записались, ссылки на блоки гарантировано записались в файловой системе, и данные точно можно будет прочитать.

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

    • Возможно при наличии большой очереди каждые 100 миллисекунд делать принудительную запись на диск (таким образом, ориентировочная задержка между обработкой сообщений будет +100 мс при большой очереди)
    • Возможно считать размер записанных данных и размер кластера (который программа допустим каким-то образом будет знать) и предполагать что в кэше диска блок сформирован и записан на диск. Но тут очень спорный момент.
    • В плане подготовки файла возможно заранее готовить файлы по 100 мегабайт, в свободное время записывать туда нули (если это не делает файловая система сама) и в дальнейшем диску будет проще работать с файлом т.к. не нужно будет мотать головой для записи данных блоков в заголовок файла.
  3. Возможно ли разное поведение при записи в файл на разных файловых системах ntfs/ext3/ext4/... и разных операционных системах? Либо .NET Core 3.1+ гарантирует что после вызова метода fs.Flush(true), данные гарантировано записались на физический диск (энергонезависисый кэш).

  4. Требуется ли какая-то дополнительная оптимизация кода если предусматривается использование аппаратных рейдов с энергонезависимым кэшем (кэш с батарейкой), и разумеется мы заранее не знаем есть ли он, например при использовании СХД и крутясь на виртуалке. (Предполагаю что не требуется, но все же задам вопрос, вдруг есть подводные камни)

На 100% уверен, что все кто разрабатывал базы данных сталкивался с данной проблемой, т.к. запись данных в таблицы и построение индексов происходит ТОЛЬКО после записи транзакции (если БД поддерживает транзакции И эта функция не отключена специально)

upd: В терминалогии СУБД, как оказалось, это называется "Протокол ведения журнала с опережением записи (WAL)"

upd: По мере получения конкретных данных и опыта, обязуюсь представить оптимальный код на C#, в отдельном сообщении. Возможно кому-то сэкономит массу времени. С последующим вылизыванием, если у кого-то будут объективные поправки.

Буду рад и ссылкам на любой касающийся этой темы материал (на любом языке) и личным комментариям. Заранее Огромное Спасибо.

7
  • Несколько раз перечитал, надеюсь понял :-) А варианты расположения файловой системы не на диске рассматриваются? Взять tmpfs и замонтировать её в память? 11 янв 2021 в 14:15
  • Нет, транзакции (и их хранение) нужны для того, что-бы в случае аварийной перезагрузки системы, система могла восстановить целостность базы данных. В случае использования tmpfs при сбое питания, данные транзакций пропадут. Данные в базу данных записываются по каждому отдельному объекту а не транзакцией целиком.
    – Aleksey G
    11 янв 2021 в 15:21
  • С практической точки зрения лучше взять готовую embedded БД с поддержкой диска (github.com/praeclarum/sqlite-net например) и ее использовать. Самому это правильно реализовать - задача непростая. 11 янв 2021 в 17:33
  • Вопрос не в том, что нужно использовать возможности MySQL! Технология разработана, оно уже работает в продакшне, но не оптимально, код описанный выше это крайне упрощенный пример, журнал транзакций описанный выше гораздо сложнее и используется для много чего еще в конвейере обработки транзакции (репликация и т.п.), плюс ко всему используемая СУБД это не какая-то конкретная СУБД, может быть любая удовлетворяющая определенным требованиям. Вопрос ТОЛЬКО в том, как оптимизировать работу с файлом при потоковой записи, и как узнать о гарантированной записи в минимальные сроки.
    – Aleksey G
    12 янв 2021 в 4:04
  • 2
    В случае рейда с батарейкой способа нет. Посмотрите support.microsoft.com/en-us/help/234656/… , там детально расписано как работает write ahead log в sql server - он решает ту же проблему
    – user177221
    12 янв 2021 в 9:32

2 ответа 2

3

Решение, которое обычно используется в СУБД - это Write Ahead Log, запись в лог до того, как операция (сообщение) будет помечено как обработанное. В вашем случае - это запись в лог до вызова TryDequeue

SQL Server использует при этом флаг FILE_FLAG_WRITE_THROUGH, в случае C# - это флаг FileOptions.WriteThrough.

FileOptions.WriteThrough работает и на Windows и на Unix, так что никаких дополнительных хаков под разные платформы не требуется.

В случае рейда с батарейкой обеспечить реальную запись на диск не получится, придется полагаться на надежность контроллера и батарейки.

1

Я хоть не в теме, но интересно, попробую чуть-чуть вникнуть и на что-то ответить. Вообще всю информацию для ответа я надеюсь добыть по ссылке https://github.com/dotnet/runtime/find/master введя фразу filestream.

Вы задали очень много вопросов в одном, это нарушает правила и ставит в затруднённое положение отвечающего. Ну да ладно, поехали..

Что происходит после выполнения функции fs.Flush(false)

Что происходит после выполнения функции fs.Flush(true)

public virtual void Flush(bool flushToDisk)
{
    if (IsClosed) throw Error.GetFileNotOpen();

    FlushInternalBuffer();

    if (flushToDisk && CanWrite)
    {
        FlushOSBuffer();
    }
}

Первое очищает буфер в памяти приложения, второе вдобавок вызывает специальную очистку буфера ОС. Это очистка производиться средствами самой ОС и, что вполне объяснимо, реализация зависит от собственно типа ОС.

UNIX
if (Interop.Sys.FSync(_fileHandle) < 0) 
{
   //обработка ошибок
} 
WINDOWS
if (!Interop.Kernel32.FlushFileBuffers(_fileHandle))
{
   //обработка ошибок
}

Что примечательно асинхронные версии FlushAsync лишены такой функциональности. Так же не увидел я очистки буфера ОС и в Dispose().

Что происходит после выполнения команды fs.Write(anydata) под капотом

Зависит от режима. Есть асинхронный и синхронный. Асинхронный я изучу когда-нибудь потом. А в синхронном происходит следующее. Вызывается private void WriteSpan(ReadOnlySpan<byte> source) он собственно отвечает за накопление данных в буфере, и при переполнениях по сути проводит аналогичную процедуру Flush(false), в результате которой в конечном итоге вызывается в Windows Interop.Kernel32.WriteFile(), а вот в UNIX, такая конструкция

fixed (byte* bufPtr = &MemoryMarshal.GetReference(source))
{
    int offset = 0;
    int count = source.Length;
    while (count > 0)
    {
        int bytesWritten = CheckFileCall(Interop.Sys.Write(_fileHandle, bufPtr + offset, count));
        _filePosition += bytesWritten;
        offset += bytesWritten;
        count -= bytesWritten;
    }
}

возможно лучшая практика котролировать время между командами fs.FLUSH и например выполнять их каждые 100 мс.

Возможно. Есть смысл провести тесты производительности

Ваш ответ

By clicking “Отправить ответ”, you agree to our terms of service and acknowledge you have read our privacy policy.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.