问题概述
我正在开发一个项目,需要在单个Linux服务器上以非常高的速度将数据流式传输到磁盘。使用以下命令的fio基准测试显示,我应该能够使用io_uring获得所需的写入速度(> 40 GB / s)。
fio --name=seqwrite --rw=write --direct=1 --ioengine=io_uring --bs=128k --numjobs=4 --size=100G --runtime=300 --directory=/mnt/md0/ --iodepth=128 --buffered=0 --numa_cpu_nodes=0 --sqthread_poll=1 --hipri=1
然而,我无法使用我自己的代码复制这种性能,我的代码使用了liburing帮助库来进行io_uring。我的当前写入速度约为9 GB/s。我怀疑liburing的额外开销可能是瓶颈,但在放弃更漂亮的liburing代码之前,我有一些问题要问。
我的方法
- 使用liburing
- 利用提交队列轮询功能
- 不使用
writev()
排队收集/分散io请求,而是排队请求使用普通的write()
函数写入磁盘。(尝试过收集/分散IO请求,但这似乎对我的写入速度没有太大影响。) - 多线程,每个线程一个环
附加信息
- 运行一个简化版本的此代码,不使用线程,得到了类似的结果。
- 我的调试器显示我创建了
NUM_JOBS
宏指定的线程数,但它并没有告诉我内核为sq轮询创建的线程。 - 当运行多于两个线程时,我的性能下降了。
- Linux服务器有96个CPU可供使用。
- 数据被写入了RAID0配置。
- 我在单独的终端中使用了
bpftrace -e 'tracepoint:io_uring:io_uring_submit_sqe {printf("%s(%d)\n", comm, pid);}'
,它显示了专用于sq轮询的内核线程处于活动状态。 - 我已经验证了写入磁盘的数据与我期望的大小和内容完全相匹配。
- 在设置环形缓冲区时,我尝试过使用
IORING_SETUP_ATTACH_WQ
标志。如果说有什么效果,那就是减慢了速度。 - 我尝试过各种块大小,128k似乎是最佳选择。
问题
- 我期望内核会为每个环路启动一个单独的线程来处理sq轮询。但是,我不知道如何验证这是否真的发生了。我可以假设它是吗?
- 当运行超过两个作业时,为什么我的性能会降低?这是因为线程之间争夺要写入的文件而导致的吗?也许是因为实际上只有一个线程在处理来自多个环路的请求而被拖慢了sq轮询的工作?
- 是否有其他标志或选项可以使用以帮助解决问题?
- 是时候采用直接io_uring调用了吗?
代码
下面的代码是一个简化版本,为了简洁起见删除了很多错误处理代码。然而,这个简化版本的性能和功能与完整功能的代码相同。
主函数
#include <fcntl.h>
#include <liburing.h>
#include <cstring>
#include <thread>
#include <vector>
#include "utilities.h"
#define NUM_JOBS 4 // number of single-ring threads
#define QUEUE_DEPTH 128 // size of each ring
#define IO_BLOCK_SIZE 128 * 1024 // write block size
#define WRITE_SIZE (IO_BLOCK_SIZE * 10000) // Total number of bytes to write
#define FILENAME "/mnt/md0/test.txt" // File to write to
char incomingData[WRITE_SIZE]; // Will contain the data to write to disk
int main()
{
// Initialize variables
std::vector<std::thread> threadPool;
std::vector<io_uring*> ringPool;
io_uring_params params;
int fds[2];
int bytesPerThread = WRITE_SIZE / NUM_JOBS;
int bytesRemaining = WRITE_SIZE % NUM_JOBS;
int bytesAssigned = 0;
utils::generate_data(incomingData, WRITE_SIZE); // this just fills the incomingData buffer with known data
// Open the file, store its descriptor
fds[0] = open(FILENAME, O_WRONLY | O_TRUNC | O_CREAT);
// initialize Rings
ringPool.resize(NUM_JOBS);
for (int i = 0; i < NUM_JOBS; i++)
{
io_uring* ring = new io_uring;
// Configure the io_uring parameters and init the ring
memset(¶ms, 0, sizeof(params));
params.flags |= IORING_SETUP_SQPOLL;
params.sq_thread_idle = 2000;
io_uring_queue_init_params(QUEUE_DEPTH, ring, ¶ms);
io_uring_register_files(ring, fds, 1); // required for sq polling
// Add the ring to the pool
ringPool.at(i) = ring;
}
// Spin up threads to write to the file
threadPool.resize(NUM_JOBS);
for (int i = 0; i < NUM_JOBS; i++)
{
int bytesToAssign = (i != NUM_JOBS - 1) ? bytesPerThread : bytesPerThread + bytesRemaining;
threadPool.at(i) = std::thread(writeToFile, 0, ringPool[i], incomingData + bytesAssigned, bytesToAssign, bytesAssigned);
bytesAssigned += bytesToAssign;
}
// Wait for the threads to finish
for (int i = 0; i < NUM_JOBS; i++)
{
threadPool[i].join();
}
// Cleanup the rings
for (int i = 0; i < NUM_JOBS; i++)
{
io_uring_queue_exit(ringPool[i]);
}
// Close the file
close(fds[0]);
return 0;
}
writeToFile() 函数
void writeToFile(int fd, io_uring* ring, char* buffer, int size, int fileIndex)
{
io_uring_cqe *cqe;
io_uring_sqe *sqe;
int bytesRemaining = size;
int bytesToWrite;
int bytesWritten = 0;
int writesPending = 0;
while (bytesRemaining || writesPending)
{
while(writesPending < QUEUE_DEPTH && bytesRemaining)
{
/* In this first inner loop,
* Write up to QUEUE_DEPTH blocks to the submission queue
*/
bytesToWrite = bytesRemaining > IO_BLOCK_SIZE ? IO_BLOCK_SIZE : bytesRemaining;
sqe = io_uring_get_sqe(ring);
if (!sqe) break; // if can't get a sqe, break out of the loop and wait for the next round
io_uring_prep_write(sqe, fd, buffer + bytesWritten, bytesToWrite, fileIndex + bytesWritten);
sqe->flags |= IOSQE_FIXED_FILE;
writesPending++;
bytesWritten += bytesToWrite;
bytesRemaining -= bytesToWrite;
if (bytesRemaining == 0) break;
}
io_uring_submit(ring);
while(writesPending)
{
/* In this second inner loop,
* Handle completions
* Additional error handling removed for brevity
* The functionality is the same as with errror handling in the case that nothing goes wrong
*/
int status = io_uring_peek_cqe(ring, &cqe);
if (status == -EAGAIN) break; // if no completions are available, break out of the loop and wait for the next round
io_uring_cqe_seen(ring, cqe);
writesPending--;
}
}
}
fio
的源代码了吗? - Paul Sandersfio
的源代码。他们正在使用低级别的 io_uring。这就是为什么我怀疑我可能不得不咬紧牙关,做同样的事情。然而,我希望能够避免这种情况,因为 fio 的io_uring.c
源代码超过1300行,如果有可能,我想避免这种复杂性。 - Smitch