C++实现低延迟日志系统：利用O_DIRECT I/O与批处理写入优化磁盘访问 - 智猿学院-前后端，数据库，人工智能，云计算等领域前沿技术讲座

C++ 低延迟日志系统：O_DIRECT I/O 与批处理写入优化磁盘访问

大家好，今天我们来探讨如何使用 C++ 构建一个低延迟的日志系统，重点是如何利用 O_DIRECT I/O 和批处理写入来优化磁盘访问，从而显著降低日志写入的延迟。

1. 日志系统面临的挑战与优化目标

一个高效的日志系统对于任何需要审计、调试或者性能分析的应用程序都至关重要。然而，传统的日志写入方式往往会带来明显的性能瓶颈，特别是在高并发或者对延迟敏感的场景下。

主要挑战包括：

频繁的系统调用: 每次日志写入都需要一次系统调用 (e.g., write(), fwrite())，这会带来较大的上下文切换开销。
内核缓存 (Page Cache): 操作系统通常会使用 Page Cache 来缓存磁盘数据，虽然可以提高读取性能，但写入时会先写入缓存，然后由内核异步刷新到磁盘。这会导致数据延迟落盘，在高并发情况下可能丢失数据。
磁盘 I/O 延迟: 即使直接写入磁盘，磁盘本身的物理特性也会引入延迟，比如寻道时间和旋转延迟。

我们的优化目标是：

降低写入延迟: 尽可能缩短日志写入到磁盘的时间，确保数据及时持久化。
减少系统调用开销: 通过批量写入减少系统调用次数。
绕过内核缓存: 使用 O_DIRECT 模式直接写入磁盘，避免内核缓存的影响，保证数据的即时性。
提高吞吐量: 在保证低延迟的同时，尽可能提高日志系统的写入吞吐量。

2. O_DIRECT I/O：绕过内核缓存

O_DIRECT 是 Linux 系统提供的一种文件 I/O 标志，使用它可以绕过内核的 Page Cache，直接将数据写入或者读取到磁盘。

优点:

减少延迟: 数据直接写入磁盘，避免了内核缓存带来的延迟。
数据一致性: 确保数据立即持久化到磁盘，避免因系统崩溃导致数据丢失。
可预测的性能: 由于绕过了内核缓存，I/O 行为更加可预测。

缺点:

对齐要求: 使用 O_DIRECT 进行 I/O 操作，对内存地址、文件偏移量和数据长度都有对齐要求，通常需要按照文件系统块大小（例如 4KB）对齐。
复杂性: 需要手动管理缓冲区和数据对齐，增加了编程的复杂性。
降低读取性能 (可能): 如果读取的数据之前没有缓存在 Page Cache 中，O_DIRECT 的读取性能可能不如使用缓存。但是对于日志系统，我们主要关注写入性能，所以这个缺点可以忽略。

代码示例:

#include <iostream>
#include <fstream>
#include <fcntl.h>
#include <unistd.h>
#include <cstdlib>
#include <cstring>
#include <errno.h>

// 获取文件系统块大小
size_t get_filesystem_block_size(const std::string& filename) {
    struct statfs stat;
    if (statfs(filename.c_str(), &stat) == -1) {
        perror("statfs");
        return 0;
    }
    return stat.f_bsize;
}

// 对齐内存
void* aligned_alloc(size_t alignment, size_t size) {
    void* ptr = nullptr;
#if defined(_POSIX_C_SOURCE) && _POSIX_C_SOURCE >= 200112L
    int result = posix_memalign(&ptr, alignment, size);
    if (result != 0) {
        errno = result; // Set errno according to posix_memalign return value.
        return nullptr;
    }
#else
    ptr = memalign(alignment, size);
#endif
    return ptr;
}

int main() {
    std::string filename = "log.txt";
    size_t block_size = get_filesystem_block_size(filename);
    if (block_size == 0) {
        std::cerr << "Failed to get filesystem block size." << std::endl;
        return 1;
    }

    // 打开文件，使用 O_DIRECT 标志
    int fd = open(filename.c_str(), O_WRONLY | O_CREAT | O_DIRECT | O_APPEND, 0666);
    if (fd == -1) {
        perror("open");
        return 1;
    }

    // 分配对齐的缓冲区
    size_t buffer_size = block_size; // 缓冲区大小必须是块大小的倍数
    char* buffer = static_cast<char*>(aligned_alloc(block_size, buffer_size));
    if (buffer == nullptr) {
        perror("aligned_alloc");
        close(fd);
        return 1;
    }

    // 准备数据
    std::string log_message = "This is a log message written using O_DIRECT.n";
    size_t message_length = log_message.length();

    // 确保消息长度小于缓冲区大小
    if (message_length > buffer_size) {
        std::cerr << "Log message is too long for the buffer." << std::endl;
        free(buffer);
        close(fd);
        return 1;
    }

    // 将消息复制到对齐的缓冲区
    std::memcpy(buffer, log_message.c_str(), message_length);

    // 填充缓冲区剩余部分 (如果需要)
    if (message_length < buffer_size) {
      memset(buffer + message_length, 0, buffer_size - message_length);
    }

    // 写入数据
    ssize_t bytes_written = write(fd, buffer, buffer_size);
    if (bytes_written == -1) {
        perror("write");
        free(buffer);
        close(fd);
        return 1;
    }

    if (bytes_written != buffer_size) {
        std::cerr << "Wrote less bytes than expected." << std::endl;
    }

    // 清理
    free(buffer);
    close(fd);

    std::cout << "Log message written successfully." << std::endl;

    return 0;
}

代码解释:

get_filesystem_block_size(): 获取文件系统块大小，这是使用 O_DIRECT 的必要步骤。statfs 函数可以获取文件系统的信息，包括块大小。
aligned_alloc(): 分配对齐的内存。 O_DIRECT 要求缓冲区地址必须是对齐的。这里使用了 posix_memalign (如果可用) 或者 memalign 来分配对齐的内存。
open(): 使用 O_DIRECT 标志打开文件。还需要 O_WRONLY (只写), O_CREAT (如果文件不存在则创建) 和 O_APPEND (追加写入) 标志。
write(): 使用 write() 系统调用将数据写入文件。注意，写入的数据长度也必须是块大小的倍数。
错误处理: 代码中包含了错误处理，例如检查 open(), aligned_alloc() 和 write() 的返回值。
缓冲区填充: 如果写入的数据长度小于块大小，需要用零填充缓冲区剩余部分，以满足 O_DIRECT 的对齐要求。

3. 批处理写入：减少系统调用

频繁的系统调用是日志系统性能的另一个瓶颈。通过将多个日志消息积累到缓冲区中，然后一次性写入磁盘，可以显著减少系统调用的次数。

实现方式:

缓冲区: 维护一个内存缓冲区，用于存储待写入的日志消息。
积累: 将新的日志消息添加到缓冲区中。
触发写入: 当缓冲区达到一定大小（例如，块大小的倍数）或者达到一定的时间间隔时，触发一次写入操作。
写入: 将缓冲区中的所有数据一次性写入磁盘。
清空: 清空缓冲区，为下一次写入做准备。

代码示例:

#include <iostream>
#include <fstream>
#include <fcntl.h>
#include <unistd.h>
#include <cstdlib>
#include <cstring>
#include <vector>
#include <chrono>
#include <thread>
#include <mutex>
#include <condition_variable>

// 获取文件系统块大小
size_t get_filesystem_block_size(const std::string& filename) {
    struct statfs stat;
    if (statfs(filename.c_str(), &stat) == -1) {
        perror("statfs");
        return 0;
    }
    return stat.f_bsize;
}

// 对齐内存
void* aligned_alloc(size_t alignment, size_t size) {
    void* ptr = nullptr;
#if defined(_POSIX_C_SOURCE) && _POSIX_C_SOURCE >= 200112L
    int result = posix_memalign(&ptr, alignment, size);
    if (result != 0) {
        errno = result; // Set errno according to posix_memalign return value.
        return nullptr;
    }
#else
    ptr = memalign(alignment, size);
#endif
    return ptr;
}

class BatchLogger {
public:
    BatchLogger(const std::string& filename, size_t batch_size) :
        filename_(filename),
        batch_size_(batch_size),
        buffer_(static_cast<char*>(aligned_alloc(get_filesystem_block_size(filename_), batch_size_))),
        buffer_offset_(0),
        fd_(-1),
        is_running_(true)
    {
        if (!buffer_) {
            throw std::runtime_error("Failed to allocate aligned buffer.");
        }

        fd_ = open(filename_.c_str(), O_WRONLY | O_CREAT | O_DIRECT | O_APPEND, 0666);
        if (fd_ == -1) {
            free(buffer_);
            throw std::runtime_error("Failed to open log file.");
        }

        // 启动后台线程
        worker_thread_ = std::thread(&BatchLogger::worker_thread_func, this);
    }

    ~BatchLogger() {
        // 停止后台线程
        {
            std::lock_guard<std::mutex> lock(mutex_);
            is_running_ = false;
        }
        condition_.notify_one();
        worker_thread_.join();

        // 刷新缓冲区
        flush();

        // 关闭文件
        close(fd_);
        free(buffer_);
    }

    void log(const std::string& message) {
        std::unique_lock<std::mutex> lock(mutex_);

        // 如果消息太长，直接写入 (避免阻塞)
        if (message.length() > batch_size_) {
            lock.unlock(); // 释放锁，避免死锁

            // 创建一个临时缓冲区，并对齐
            void* temp_buffer = aligned_alloc(get_filesystem_block_size(filename_), message.length());
            if(!temp_buffer) {
                std::cerr << "Failed to allocate temporary buffer for large log message." << std::endl;
                return;
            }
            std::memcpy(temp_buffer, message.c_str(), message.length());

            ssize_t bytes_written = write(fd_, temp_buffer, message.length());
            if (bytes_written == -1) {
                perror("write");
                std::cerr << "Failed to write large log message." << std::endl;
            }
            free(temp_buffer);
            return;
        }

        // 等待缓冲区有足够的空间
        condition_.wait(lock, [this, &message] { return buffer_offset_ + message.length() <= batch_size_ || !is_running_; });

        if (!is_running_) {
            return; // 如果logger正在关闭，直接返回
        }

        // 将消息复制到缓冲区
        std::memcpy(buffer_ + buffer_offset_, message.c_str(), message.length());
        buffer_offset_ += message.length();

        // 如果缓冲区已满，则通知工作线程
        if (buffer_offset_ == batch_size_) {
            condition_.notify_one();
        }
    }

private:
    void worker_thread_func() {
        while (is_running_) {
            std::unique_lock<std::mutex> lock(mutex_);
            condition_.wait(lock, [this] { return buffer_offset_ > 0 || !is_running_; });

            if (!is_running_ && buffer_offset_ == 0) {
                break; // 如果logger正在关闭且缓冲区为空，则退出
            }

            // 写入缓冲区
            if (buffer_offset_ > 0) {
                ssize_t bytes_written = write(fd_, buffer_, batch_size_); // 始终写入整个batch_size_
                if (bytes_written == -1) {
                    perror("write");
                    std::cerr << "Failed to write log message." << std::endl;
                } else if (bytes_written != batch_size_) {
                    std::cerr << "Wrote less bytes than expected." << std::endl;
                }

                // 清空缓冲区
                buffer_offset_ = 0;
                memset(buffer_, 0, batch_size_); // 清空整个缓冲区
                condition_.notify_one(); // 通知其他线程缓冲区已空
            }
        }
    }

    void flush() {
        std::lock_guard<std::mutex> lock(mutex_);
        if (buffer_offset_ > 0) {
            // 创建一个临时缓冲区，并对齐
            void* temp_buffer = aligned_alloc(get_filesystem_block_size(filename_), batch_size_);
            if(!temp_buffer) {
                std::cerr << "Failed to allocate temporary buffer for flush." << std::endl;
                return;
            }
            std::memcpy(temp_buffer, buffer_, buffer_offset_);
            memset(static_cast<char*>(temp_buffer) + buffer_offset_, 0, batch_size_ - buffer_offset_); // 填充剩余部分

            ssize_t bytes_written = write(fd_, temp_buffer, batch_size_); // 写入整个batch_size_
            if (bytes_written == -1) {
                perror("write");
                std::cerr << "Failed to write remaining log messages during flush." << std::endl;
            } else if (bytes_written != batch_size_) {
                std::cerr << "Wrote less bytes than expected during flush." << std::endl;
            }
            free(temp_buffer);
            buffer_offset_ = 0;
        }
    }

private:
    std::string filename_;
    size_t batch_size_;
    char* buffer_;
    size_t buffer_offset_;
    int fd_;
    std::thread worker_thread_;
    std::mutex mutex_;
    std::condition_variable condition_;
    bool is_running_;
};

int main() {
    try {
        std::string filename = "batch_log.txt";
        size_t block_size = get_filesystem_block_size(filename);
        if (block_size == 0) {
            std::cerr << "Failed to get filesystem block size." << std::endl;
            return 1;
        }
        size_t batch_size = block_size * 4; // 批量写入大小，设置为块大小的倍数

        BatchLogger logger(filename, batch_size);

        for (int i = 0; i < 1000; ++i) {
            logger.log("Log message " + std::to_string(i) + "n");
        }

        // 等待一段时间确保所有日志都写入
        std::this_thread::sleep_for(std::chrono::seconds(2));

        std::cout << "Log messages written successfully." << std::endl;

    } catch (const std::exception& e) {
        std::cerr << "Exception: " << e.what() << std::endl;
        return 1;
    }

    return 0;
}

代码解释:

BatchLogger 类: 封装了批处理写入的逻辑。
batch_size_: 批量写入的大小，必须是块大小的倍数。
buffer_: 用于存储日志消息的缓冲区，使用 aligned_alloc() 分配对齐的内存。
buffer_offset_: 缓冲区中当前已使用的空间。
log() 方法: 将日志消息添加到缓冲区中。如果消息长度超过 batch_size_，则直接写入文件。使用了条件变量 condition_ 来同步日志写入线程和工作线程。
worker_thread_func() 方法: 后台线程，负责将缓冲区中的数据写入文件。使用条件变量 condition_ 来等待缓冲区达到一定大小或者收到关闭信号。
flush() 方法: 将缓冲区中剩余的数据写入文件。在 BatchLogger 析构函数中调用，确保所有数据都落盘。
O_DIRECT: 文件以 O_DIRECT 模式打开，保证数据直接写入磁盘。
同步机制: 使用 std::mutex 和 std::condition_variable 来保证线程安全和同步。主线程负责接收日志消息并将其添加到缓冲区，而后台线程负责将缓冲区中的数据写入文件。
大消息处理: 如果日志消息的大小超过了 batch_size_，则直接绕过缓冲区写入磁盘。这可以避免阻塞主线程。

4. 进一步优化

除了 O_DIRECT 和批处理写入，还可以通过以下方式进一步优化日志系统：

异步写入: 使用线程池或者异步 I/O (e.g., libaio) 将写入操作放入后台线程执行，避免阻塞主线程。上面的代码已经使用了后台线程，可以考虑使用线程池来管理多个写入线程。
零拷贝 (Zero-copy): 使用 sendfile() 系统调用，避免内核空间和用户空间之间的数据拷贝。这对于网络日志系统非常有用。
选择合适的磁盘: 使用 SSD 固态硬盘代替传统的机械硬盘，可以显著降低 I/O 延迟。
日志级别控制: 根据不同的日志级别 (e.g., DEBUG, INFO, WARN, ERROR) 过滤日志消息，减少不必要的写入操作。
压缩: 对日志数据进行压缩，减少磁盘空间占用和 I/O 负载。可以使用 zlib, gzip 或者 lz4 等压缩算法。
循环日志 (Rolling Logs): 将日志文件分割成多个小文件，并定期轮换，避免单个文件过大。
多路复用 (Multiplexing): 将多个日志流合并到一个文件中，减少文件数量，提高 I/O 效率。需要注意同步问题。
使用更高效的日志格式: 例如，使用二进制格式代替文本格式，可以减少日志大小和解析时间。可以使用 Protocol Buffers 或者 FlatBuffers 等序列化框架。

5. 性能测试与评估

优化后的日志系统需要进行性能测试和评估，以验证其效果。

测试工具: 可以使用 fio (Flexible I/O Tester), iostat 或者自定义的 benchmark 程序。
测试指标: 关注以下指标：
- 写入延迟: 平均写入延迟、最大写入延迟、99th percentile 延迟。
- 吞吐量: 每秒写入的日志消息数量或者数据量。
- CPU 使用率: 日志系统对 CPU 的消耗。
- 内存使用率: 日志系统占用的内存空间。
测试场景: 模拟不同的并发级别和日志消息大小，测试日志系统的性能表现。

表格：不同优化策略的对比

优化策略	优点	缺点	适用场景
`O_DIRECT` I/O	减少延迟，数据一致性，可预测的性能	需要对齐，复杂性增加，可能降低读取性能	对延迟敏感，需要保证数据一致性的场景
批处理写入	减少系统调用开销，提高吞吐量	增加代码复杂性，需要维护缓冲区，可能引入额外的延迟	高并发，需要高吞吐量的场景
异步写入	避免阻塞主线程，提高响应速度	增加代码复杂性，需要线程同步机制	需要保证主线程响应速度的场景
零拷贝	减少数据拷贝，提高 I/O 效率	适用场景有限，需要内核支持	网络日志系统
SSD 固态硬盘	降低 I/O 延迟，提高吞吐量	成本较高	对性能有极致要求的场景
日志级别控制	减少不必要的写入操作，降低 I/O 负载	需要合理配置日志级别	所有场景
压缩	减少磁盘空间占用和 I/O 负载	增加 CPU 消耗，需要解压缩	磁盘空间有限，I/O 负载较高，CPU 资源充足的场景
循环日志	避免单个文件过大，方便管理	增加文件数量，需要定期轮换	所有场景
多路复用	减少文件数量，提高 I/O 效率	需要处理同步问题，可能增加复杂性	文件数量限制的场景
更高效的日志格式 (e.g., 二进制)	减少日志大小和解析时间	降低可读性，需要使用专门的工具进行解析	需要高性能解析的场景

总结：利用O_DIRECT和批处理，构建低延迟高性能日志系统

通过结合 O_DIRECT I/O 和批处理写入，我们可以构建一个低延迟、高性能的日志系统。 O_DIRECT 保证了数据直接写入磁盘，避免了内核缓存带来的延迟，而批处理写入则减少了系统调用的次数，提高了吞吐量。同时，还需要根据实际情况选择合适的优化策略，并进行充分的性能测试和评估，以达到最佳的性能表现。

更多IT精英技术系列讲座，到智猿学院

C++ 低延迟日志系统：O_DIRECT I/O 与批处理写入优化磁盘访问

发表回复 取消回复

发表回复取消回复