当前位置：首页 > ds >正文

使用O_DIRECT + 批量写数据到磁盘对丢包率的优化

ds 2025/4/30 10:15:44

背景

当前项目是基于DPDK的全流量存储系统，需要将收集到的网络数据包保存成PCAP文件（类似tcpdump），并实时生成metadata信息（MAC，IP，Port，Offset等），并将metadata保存到数据库（ClickHouse）。通过查询数据库，能够将查询结果组装成新的PCAP文件，提供下载和分析。网络流量20Gbps，数据包平均大小~865 bytes。

当前方案及问题

全流量存储的场景有以下特点，

数据顺序大量写入，写入后不会立即全部访问（即不用Cache数据）
丢包率对写入数据到磁盘的延迟很敏感，如果在某个时刻写入数据的延迟很高，那么当前时刻的丢包率将很高

当前在保存网络包数据到PCAP文件时，使用的是标准C库函数fwrite，

size_t fwrite(const void *ptr, size_t size, size_t count, FILE *stream);

fwrite是C中最常用的写文件函数，大部分场景下使用此函数都没有问题。但是，针对当前全流量存储场景，使用fwrite存在两个问题，

每次获取到网络数据包时，需要先写入包头，再写入数据本身。当流量为20Gbps，数据包平均大小为865 bytes时，每个线程每秒钟会处理355000个数据包。处理每个数据包都需要调用两次fwrite，相当于每个线程每秒钟会调用710000次fwrite。虽然fwrite内部有缓存机制，但频繁的调用也会影响整体性能。
fwrite依赖于操作系统Page Cache，Page Cache会缓存所有需要写入到磁盘的数据。Page Cache是整个OS共用的，OS会控制Cache中数据的生成，写入以及删除。由于OS的介入，导致Page Cache中的数据变得不可控，什么时候写入，什么时候删除等都是不确定的。这在一定程度上增加了整个全流量存储系统的不确定性，从而影响到丢包率，因为丢包率对磁盘写入的延迟非常敏感。

优化方案

针对当前方案存在的两个问题，优化方案是使用 O_DIRECT模式 + 批量写入。

`O_DIRECT模式`

O_DIRECT 是一种文件打开模式，用于在 Linux 和其他类 Unix 系统上执行文件 I/O 操作时绕过操作系统的页缓存。它允许应用程序直接从用户空间读写数据到磁盘。使用 O_DIRECT 需要满足一些对齐要求，通常数据缓冲区和文件偏移需要与文件系统块大小对齐。

O_DIRECT模式的好处

减少缓存污染:
- 避免缓存干扰: 对于某些应用（如数据库），使用 O_DIRECT 可以避免将大量数据加载到页缓存中，从而减少对其他应用程序的缓存干扰。
更可预测的性能:
- 稳定的I/O性能: 由于不使用页缓存，I/O操作的性能更加可预测，尤其是在处理大量数据时。这对于需要稳定性能的应用程序非常重要。
减少内存使用:
- 直接数据传输: 数据直接从用户空间传输到磁盘，减少了内存的使用，因为不需要在内存中保留额外的缓存数据。
适合特定工作负载:
- 大数据处理: 对于需要处理大量数据的应用程序，O_DIRECT 可以减少缓存的开销，提高数据传输效率。
降低延迟:
- 直接写入: 通过直接写入磁盘，可以减少数据在缓存中的停留时间，降低写操作的延迟。

使用 O_DIRECT 的注意事项

对齐要求:
- 数据对齐: 使用 O_DIRECT 时，数据缓冲区和文件偏移通常需要与文件系统块大小对齐。这增加了编程复杂性。
编程复杂性:
- 手动管理缓存: 由于绕过了操作系统的缓存，应用程序需要手动管理数据的缓存和一致性。
不适合所有工作负载:
- 小型随机I/O: 对于小型随机I/O操作，O_DIRECT 可能会导致性能下降，因为它无法利用页缓存的优势。

`批量写入`

批量写入是指在写入数据时先写入数据到一个缓存，缓存大小可以设定，例如10MB。当缓存占满或达到一定阈值时（例如：8MB），调用系统函数将缓存中的数据写入到磁盘。这样能避免频繁的调用系统函数，特别当有大量小IO的场景时，批量写入非常有用。

主要实现代码

file_util.h

#ifndef NTR_FILE_H
#define NTR_FILE_H#include <stdbool.h>
#include <stdio.h>
#include <sys/time.h>
#include <unistd.h>#define FILE_CACHE_FLUSH_SIZE (1024 * 1024 * 8)
#define FILE_CACHE_MAX_SIZE (1024 * 1024 * 9)
#define FS_BLOCK_SIZE 4096struct cache_file {int fd;                   // File fdvoid *cache;              // Cache pointerint cache_size;           // Cache max sizeint cache_flush_size;     // Cache flush sizeint cache_offset;         // Off set in cacheint block_size;           // FS block sizelong int file_size;       // Current file sizelong int file_size_pre;   // File size at last write timestruct timeval flush_tv;  // Flush time
};int init_file(struct cache_file *file);
void open_file(struct cache_file *file, char *output_file);
void close_file(struct cache_file *file);
int write_file(struct cache_file *file, void *buffer, size_t size);
int flush_file(struct cache_file *file, bool force);#endif

file_util.c

#include "file_util.h"#define __USE_GNU#include <fcntl.h>
#include <stdlib.h>
#include <string.h>
#include <sys/mman.h>
#include <sys/stat.h>int init_file(struct cache_file *file) {if (posix_memalign(&file->cache, FS_BLOCK_SIZE, FILE_CACHE_MAX_SIZE) != 0) {return -1;}file->fd = 0;file->cache_size = FILE_CACHE_MAX_SIZE;file->cache_flush_size = FILE_CACHE_FLUSH_SIZE;file->cache_offset = 0;file->file_size = 0;file->file_size_pre = 0;file->block_size = FS_BLOCK_SIZE;gettimeofday(&file->flush_tv, NULL);return 0;
}void open_file(struct cache_file *file, char *output_file) {file->fd = open(output_file, O_WRONLY | O_CREAT | O_DIRECT, 0664);file->cache_offset = 0;file->file_size = 0;file->file_size_pre = 0;gettimeofday(&file->flush_tv, NULL);
}void close_file(struct cache_file *file) {flush_file(file, true);close(file->fd);file->fd = 0;file->cache_offset = 0;file->file_size = 0;file->file_size_pre = 0;
}int write_file(struct cache_file *file, void *buffer, size_t size) {memcpy(file->cache + file->cache_offset, buffer, size);file->cache_offset += size;file->file_size += size;// Flush if reaches to flush sizeif (file->cache_offset >= file->cache_flush_size) {ssize_t bytes_written = flush_file(file, false);if (bytes_written < 0) {return bytes_written;}}return size;
}int flush_file(struct cache_file *file, bool force) {gettimeofday(&file->flush_tv, NULL);// Return if there is no new dataif (file->file_size == file->file_size_pre) {return 0;}size_t remain_data = file->cache_offset % file->block_size;size_t write_data = file->cache_offset - remain_data;size_t algined_data = (remain_data == 0 ? write_data : (write_data + file->block_size));ssize_t bytes_written = -1;// Flush all dataif (force) {bytes_written = write(file->fd, file->cache, algined_data);if (remain_data > 0) {if (ftruncate(file->fd, file->file_size) == -1) {  // truncate file to the real sizeperror("ftruncate failed");return -1;}if (lseek(file->fd, file->file_size - remain_data,SEEK_SET) == -1) {  // lseek to the previous block pointperror("lseek failed");return -1;}}file->file_size_pre = file->file_size;} else if (write_data > 0) {  // Flush part databytes_written = write(file->fd, file->cache, write_data);file->file_size_pre = file->file_size - remain_data;}// Move remain data to the beginning of the cachememmove(file->cache, file->cache + write_data, remain_data);file->cache_offset = remain_data;return bytes_written;
}
}

最重要的是两个函数是write_file和flush_file。write_file首先会将写入的数据写入到cache，并每次进行判断，如果cache中的数据量达到了阈值，则调用flush_file将缓存中的数据写入到磁盘。由于O_DIRECT模式下，要求每次写入的数据都必须是FS block大小的整数倍，因此，cache中的数据可能无法全部写入，最后一部分数据（结尾的小于FS block大小的那部分数据）还存在于缓存中。flush_file的第二个参数force用于指示是否将最后一部分数据也写入到磁盘，在关闭文件，或者超时后需要强制写入时使用。如果进行force写入，那么需要考虑文件的truncate和lseek，要保证数据能完全写入，同时文件大小符合要求（不会在文件尾部写入多余的数据），以及下次还能正常写入（调整lseek的位置）。