【Linux】Linux異步I/O -libaio

一、libaio 原理概述

1.1 libaio 介紹

libaio（Linux Asynchronous I/O）是 Linux 內核提供的異步 I/O 庫，其核心原理是：

異步提交：應用程序通過 io_submit 提交 I/O 請求后立即返回，不阻塞進程
事件通知：內核通過完成隊列（completion queue）通知應用程序 I/O 操作結果
零拷貝：結合 O_DIRECT 標志繞過內核緩沖區，實現直接磁盤訪問
批量處理：單次系統調用可提交/完成多個 I/O 請求，減少上下文切換

1.2 `O_DIRECT` 打開方式

在Linux系統中，使用O_DIRECT標志打開文件時，會繞過操作系統的緩存機制，直接與存儲設備進行數據交互。這種模式具有以下特點：

繞過系統緩存

不使用頁緩存（Page Cache）
正常文件操作會將數據先存入系統緩存，再異步寫入磁盤。而O_DIRECT會跳過這一步，直接將數據寫入磁盤或從磁盤讀取，減少了數據在緩存中的拷貝開銷。
應用程序需自行管理緩存
由于系統不再自動緩存數據，應用程序需要自己處理數據的緩存邏輯（如預讀、緩存淘汰等）。

在這里插入圖片描述

性能特點

減少數據拷貝次數
傳統I/O流程（如read()/write()）會經歷“用戶空間→內核空間緩存→磁盤”的多次拷貝，而O_DIRECT直接操作磁盤，降低了CPU和內存帶寬的消耗。
適合大尺寸連續I/O
對數據庫、大數據處理等場景中頻繁的大文件順序讀寫（如日志寫入、數據加載）性能提升明顯。
隨機I/O可能更慢
若操作小文件或隨機讀寫，由于缺乏系統緩存的預讀和合并優化，性能可能反而低于普通模式。

對齊要求

數據緩沖區需按塊對齊
使用O_DIRECT時，數據緩沖區的地址、長度必須與磁盤塊大小（通常為4KB）對齊，否則會導致I/O錯誤。例如：

// 錯誤示例（緩沖區未對齊）
char buf[1024];
write(fd, buf, 1024);// 正確示例（使用malloc對齊或posix_memalign）
char *buf;
posix_memalign((void**)&buf, 4096, 1024);

偏移量需對齊
文件讀寫的偏移量也需是塊大小的整數倍，否則可能觸發部分寫入或讀取錯誤。

同步特性

更接近同步I/O行為
O_DIRECT下的write()操作會直接將數據寫入磁盤，類似fsync()的效果，確保數據持久化。但需注意，這并不完全等同于同步I/O，仍需配合fsync()/fdatasync()保證元數據同步。
降低緩存不一致風險
多進程或多節點訪問同一文件時，避免了因系統緩存未刷新導致的數據不一致問題（如數據庫事務的持久性需求）。

適用場景

數據庫系統（如MySQL、PostgreSQL）：通過O_DIRECT減少緩存干擾，自行管理數據頁緩存。
大數據存儲與處理：處理TB級數據時，大尺寸連續I/O可提升吞吐量。
高性能計算（HPC）：科學計算中對I/O延遲和帶寬敏感的場景。
存儲設備測試工具：如dd命令使用oflag=direct測試磁盤裸性能。

1.3 libaio 數據結構

I/O 上下文（io_context_t）
- 每個異步 I/O 操作都需要關聯一個上下文，用于管理 I/O 請求隊列和完成事件。
1. _I/O 請求（io_prep__ 系列函數）
- 通過io_prep_pread()/io_prep_pwrite()等函數準備 I/O 請求，填充請求參數（如文件描述符、緩沖區、偏移量等）。
提交請求（io_submit）
- 將準備好的請求批量提交到內核，由內核異步執行。
獲取完成事件（io_getevents）
- 應用程序通過輪詢或阻塞方式檢查哪些 I/O 請求已完成，并獲取結果。

1.4 libaio 核心 API 完整介紹

io_setup - 創建 AIO 上下文

int io_setup(int maxevents, io_context_t *ctx);

參數詳解：
- maxevents：指定完成隊列（Completion Queue）的最大容量，即同時能處理的最大異步事件數。該值需根據應用并發需求設置（如設置為 8192 表示最多緩存 8192 個完成事件）。
- ctx：輸出參數，用于存儲創建的異步上下文句柄（本質為 unsigned long long 類型）。
返回值：
- 0：成功。
- 負數：錯誤碼（如 -ENOMEM 表示內存不足，-EINVAL 表示參數無效）。
注意事項：
- maxevents 需大于 0，且通常設為 2 的冪次（如 1024、8192）以優化內核隊列管理。
- 上下文創建后需通過 io_destroy() 釋放資源。

io_submit - 提交異步請求

int io_submit(io_context_t ctx, long nr, struct iocb *cb[]);

參數詳解：
- ctx：異步上下文句柄，由 io_setup() 創建。
- nr：提交的 I/O 控制塊（iocb）數量，即 cb 數組的長度。
- cb：指向 iocb 結構體的指針數組，每個元素對應一個待提交的 I/O 請求。
返回值：
- 非負數：成功提交到內核的請求數量（可能小于 nr，如因內核資源不足）。
- 負數：錯誤碼（如 -EAGAIN 表示資源臨時不可用）。
關鍵行為：
- 內核接收請求后立即返回，應用程序無需阻塞等待 I/O 完成。
- 提交的請求會被內核放入隊列，按調度策略執行（如合并相鄰的讀寫請求）。

io_getevents - 獲取完成事件

int io_getevents(io_context_t ctx, long min_nr, long nr, struct io_event *events, struct timespec *timeout);

參數詳解：
- ctx：異步上下文句柄。
- min_nr：期望獲取的最小完成事件數。若當前事件數不足，函數會阻塞（除非 timeout 非零）。
- nr：最多獲取的事件數，即 events 數組的最大長度。
- events：輸出參數，存儲完成事件的數組，每個元素為 struct io_event 類型。
- timeout：超時時間，NULL 表示無限等待；{0, 0} 表示非阻塞模式。
返回值：
- 正數：實際獲取的事件數（范圍：min_nr ≤ 返回值 ≤ nr）。
- 0：超時且無事件（僅當 timeout 非零時可能）。
- 負數：錯誤碼（如 -EBADF 表示無效上下文）。

事件結構體 struct io_event：

struct io_event {void *data;       // 對應 iocb 中的 data 字段（用戶自定義數據，如塊索引）int res;          // I/O 操作結果（成功為字節數，失敗為負錯誤碼）int res2;         // 保留字段（通常為 0）struct iocb *obj; // 指向發起請求的 iocb 結構體
};

io_prep_pwrite - 準備寫請求

void io_prep_pwrite(struct iocb *iocb, int fd, void *buf, size_t count, long long offset);

參數詳解：
- iocb：待初始化的 I/O 控制塊結構體指針。
- fd：目標文件的描述符（需以 O_RDWR 打開）。
- buf：寫入數據的緩沖區，需按扇區大小對齊（如 4KB），否則配合 O_DIRECT 時會失敗。
- count：寫入的字節數，需為塊大小的整數倍（若使用 O_DIRECT）。
- offset：寫入的文件偏移量，需與塊大小對齊（若使用 O_DIRECT）。
關聯操作：
- 寫請求完成后，可通過 io_getevents() 獲取 res 字段確認寫入字節數。
- 若需確保數據持久化，需配合 fsync() 或 fdatasync()。

io_prep_pread - 準備讀請求

void io_prep_pread(struct iocb *iocb, int fd, void *buf, size_t count, long long offset);

參數與 io_prep_pwrite 類似，區別在于功能為讀取數據：
- buf：用于存儲讀取數據的緩沖區（需對齊）。
- 讀取完成后，res 字段返回實際讀取的字節數（失敗時為負錯誤碼）。

io_destroy - 銷毀上下文

int io_destroy(io_context_t ctx);

參數詳解：
- ctx：待銷毀的異步上下文句柄。
返回值：
- 0：成功釋放上下文及關聯資源（如事件隊列）。
- 負數：錯誤碼（如 -EBUSY 表示仍有未完成的請求）。
注意事項：
- 調用前需確保所有提交的請求已通過 io_getevents() 獲取完成事件，否則可能導致資源泄漏。
- 若存在未完成請求，可先通過 io_cancel() 取消請求再銷毀上下文。

補充：iocb 結構體與常用標志

struct iocb {uint64_t aio_fildes;      // 文件描述符uint64_t aio_offset;      // 偏移量void *aio_buf;            // 數據緩沖區uint64_t aio_nbytes;      // 操作字節數int aio_lio_opcode;       // 操作類型（如 IOCB_CMD_PREAD/IOCB_CMD_PWRITE）uint64_t aio_flags;       // 標志位（常用如下）void *aio_data;           // 用戶自定義數據（可通過 io_event.data 訪問）
};

常用 aio_flags 標志：
- IOCB_FLAG_NOWAIT：非阻塞提交（內核忙時立即返回錯誤）。
- IOCB_FLAG_RESFD：將完成事件寫入指定文件描述符（替代輪詢）。
- IOCB_FLAG_NOFSYNC：不自動執行 fsync（需手動調用）。

核心 API 工作流程示例

初始化：io_setup() → 分配對齊緩沖區 → 構建 iocb 數組
提交請求：io_submit() 批量發送請求到內核
事件處理：io_getevents() 阻塞獲取完成事件，處理 res 結果
資源釋放：io_destroy() 銷毀上下文，釋放緩沖區

二、libaio實現異步讀寫文件

API補充

posix_memalign

posix_memalign是一個用于在 C 語言中分配內存對齊空間的函數，主要用于滿足某些硬件或 API（如libaio）對內存地址對齊的特殊要求。以下是其詳細介紹：

函數原型

int posix_memalign(void **memptr, size_t alignment, size_t size);

參數說明

參數	含義
`memptr`	輸出參數，指向分配內存的指針地址
`alignment`	對齊字節數，必須是 2 的冪（如`512`、`4096`），且通常需大于等于`sizeof(void*)`
`size`	分配的內存大小（字節）

返回值

0：內存分配成功，memptr指向對齊后的內存起始地址。
非 0 錯誤碼：分配失敗，常見錯誤包括：
- ENOMEM：內存不足
- EINVAL：alignment不是 2 的冪或無效值

分批異步寫入文件

這段代碼實現了使用 libaio 庫進行批量異步寫操作的核心邏輯，主要功能是將數據分批次提交給內核進行異步寫入，并等待每批次完成后再處理下一批。

1. 批次循環控制

for (int batch = 0; batch < NUM_BLOCKS; batch += BATCH_SIZE) {// 計算當前批次實際處理的塊數int current_batch = (batch + BATCH_SIZE > NUM_BLOCKS) ? NUM_BLOCKS - batch : BATCH_SIZE;

動態批次大小：處理最后一批時自動調整大小，避免越界（如總塊數不是 BATCH_SIZE 整數倍時）。
循環步進：每次處理完一批后，batch += BATCH_SIZE跳轉到下一批。

2. 構建 I/O 請求

for (i = 0; i < current_batch; i++) {int idx = batch + i;// 初始化寫請求控制塊io_prep_pwrite(&iocbs[i], fd, buffers[idx], BLOCK_SIZE, idx * BLOCK_SIZE);// 存儲塊索引到iocb.data，用于事件處理時關聯iocbs[i].data = (void*)(long)idx;iocb_ptrs[i] = &iocbs[i];
}

io_prep_pwrite：設置寫請求參數（文件描述符fd、緩沖區buffers[idx]、塊大小BLOCK_SIZE、偏移量idx * BLOCK_SIZE）。
上下文關聯：通過iocbs[i].data = (void*)(long)idx將塊索引存入 iocb，以便事件處理時知道哪個塊完成了寫入。

3. 提交請求到內核

ret = io_submit(ctx, current_batch, iocb_ptrs);
if (ret != current_batch) {fprintf(stderr, "io_submit write batch %d failed: %d/%d, %s\n", batch/BATCH_SIZE, ret, current_batch, strerror(-ret));goto cleanup;
}

批量提交：io_submit一次性提交current_batch個請求，返回成功提交的數量。
錯誤檢查：若提交數量不等于請求數量，輸出錯誤信息并跳轉清理資源。

4. 等待事件完成

completed = 0;
while (completed < current_batch) {ret = io_getevents(ctx, 1, current_batch, events, NULL);if (ret < 0) {perror("io_getevents write");verify_ok = 0;goto cleanup;}completed += ret;for (i = 0; i < ret; i++) {int block_idx = (int)(long)events[i].data;printf("Write completed for block %d/%d (batch %d/%d)\r", block_idx + 1, NUM_BLOCKS, batch/BATCH_SIZE + 1, (NUM_BLOCKS+BATCH_SIZE-1)/BATCH_SIZE);fflush(stdout);}
}

阻塞等待事件：io_getevents的min_nr=1表示至少等待 1 個事件完成才返回，timeout=NULL表示無限等待。
進度更新：每次獲取到完成事件后，通過events[i].data獲取塊索引，打印進度信息（使用\r實現行內刷新）。
循環完成條件：當completed等于current_batch時，說明當前批次所有請求已完成。

   for (int batch = 0; batch < NUM_BLOCKS; batch += BATCH_SIZE) {int current_batch = (batch + BATCH_SIZE > NUM_BLOCKS) ? NUM_BLOCKS - batch : BATCH_SIZE;for (i = 0; i < current_batch; i++) {int idx = batch + i;io_prep_pwrite(&iocbs[i], fd, buffers[idx], BLOCK_SIZE, idx * BLOCK_SIZE);iocbs[i].data = (void*)(long)idx;iocb_ptrs[i] = &iocbs[i];}ret = io_submit(ctx, current_batch, iocb_ptrs);if (ret != current_batch) {fprintf(stderr, "io_submit write batch %d failed: %d/%d, %s\n", batch/BATCH_SIZE, ret, current_batch, strerror(-ret));goto cleanup;}// 等待當前批次完成completed = 0;while (completed < current_batch) {ret = io_getevents(ctx, 1, current_batch, events, NULL);if (ret < 0) {perror("io_getevents write");verify_ok = 0;goto cleanup;}completed += ret;for (i = 0; i < ret; i++) {int block_idx = (int)(long)events[i].data;printf("Write completed for block %d/%d (batch %d/%d)\r", block_idx + 1, NUM_BLOCKS, batch/BATCH_SIZE + 1, (NUM_BLOCKS+BATCH_SIZE-1)/BATCH_SIZE);fflush(stdout);}}}

分批異步讀取文件

代碼通過批處理機制分批次提交異步讀請求，每批處理 1024 個 4KB 塊（共 4MB），利用 libaio 的異步特性實現非阻塞讀取，同時通過字符串比對驗證數據準確性。主要流程包括：

1. 批次循環與當前批次計算

for (int batch = 0; batch < NUM_BLOCKS; batch += BATCH_SIZE) {int current_batch = (batch + BATCH_SIZE > NUM_BLOCKS) ? NUM_BLOCKS - batch : BATCH_SIZE;

動態批次大小：處理最后一批時自動調整大小（如總塊數 262144，批次 1024，最后一批仍為 1024；若總塊數 262145，最后一批為 1）。
循環步進：每次處理完一批后，batch += BATCH_SIZE跳轉到下一批，確保所有塊被讀取。

2. 構建異步讀請求

for (i = 0; i < current_batch; i++) {int idx = batch + i;// 初始化讀請求控制塊（與寫請求的區別：io_prep_pread）io_prep_pread(&iocbs[i], fd, buffers[idx], BLOCK_SIZE, idx * BLOCK_SIZE);iocbs[i].data = (void*)(long)idx;iocb_ptrs[i] = &iocbs[i];
}

io_prep_pread：設置讀請求參數（文件描述符fd、緩沖區buffers[idx]、塊大小BLOCK_SIZE、偏移量idx * BLOCK_SIZE）。
與寫操作的區別：讀請求從文件讀取數據到buffers[idx]，而寫請求是從緩沖區寫入文件。

3. 提交讀請求到內核

ret = io_submit(ctx, current_batch, iocb_ptrs);
if (ret != current_batch) {fprintf(stderr, "io_submit read batch %d failed: %d/%d, %s\n", batch/BATCH_SIZE, ret, current_batch, strerror(-ret));goto cleanup;
}

批量提交：io_submit一次性提交current_batch個讀請求，返回成功提交數。
錯誤處理：若提交數不等于請求數，輸出錯誤并跳轉清理資源（如釋放內存、關閉文件）。

4. 等待讀事件完成并驗證數據

completed = 0;
while (completed < current_batch) {ret = io_getevents(ctx, 1, current_batch, events, NULL);if (ret < 0) {perror("io_getevents read");verify_ok = 0;goto cleanup;}completed += ret;for (i = 0; i < ret; i++) {int block_idx = (int)(long)events[i].data;printf("Read completed for block %d/%d (batch %d/%d)\r", block_idx + 1, NUM_BLOCKS, batch/BATCH_SIZE + 1, (NUM_BLOCKS+BATCH_SIZE-1)/BATCH_SIZE);fflush(stdout);// 數據驗證核心邏輯char expected[BLOCK_SIZE];sprintf(expected, "This is block %d of %d\n", block_idx + 1, NUM_BLOCKS);if (strncmp(buffers[block_idx], expected, BLOCK_SIZE) != 0) {printf("\nBlock %d verification failed!\n", block_idx + 1);printf("Expected: %s", expected);printf("Actual:   %s", buffers[block_idx]);verify_ok = 0;}}
}

阻塞等待事件：io_getevents的min_nr=1確保至少等待 1 個事件完成，timeout=NULL無限等待。
進度顯示：通過\r實現行內刷新，實時顯示當前完成的塊編號和批次。
數據驗證邏輯：
1. 用sprintf生成預期數據（如"This is block 1 of 262144\n"）。
2. 用strncmp對比實際讀取數據與預期數據。
3. 若不一致，標記verify_ok=0并輸出差異，確保數據完整性。

完整代碼

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <fcntl.h>
#include <unistd.h>
#include <libaio.h>
#include <sys/stat.h>
#include <errno.h>
#include <time.h>#define BLOCK_SIZE 4096       // 4KB塊大小
#define ALIGN_SIZE 4096       // 內存對齊大小
#define NUM_BLOCKS 262144     // 1GB文件 = 4KB * 262144
#define FILE_SIZE (BLOCK_SIZE * NUM_BLOCKS) // 測試文件大小
#define BATCH_SIZE 1024       // 每批處理1024個請求// 記錄時間點
typedef struct {struct timespec start;struct timespec write_start;struct timespec write_end;struct timespec read_start;struct timespec read_end;struct timespec end;
} TimeRecorder;// 初始化時間記錄器
void init_time_recorder(TimeRecorder *recorder) {memset(recorder, 0, sizeof(TimeRecorder));clock_gettime(CLOCK_MONOTONIC, &recorder->start);
}// 記錄時間點
void record_time(struct timespec *timepoint) {clock_gettime(CLOCK_MONOTONIC, timepoint);
}// 計算時間差（納秒）
long long calculate_time_diff(struct timespec *start, struct timespec *end) {return (end->tv_sec - start->tv_sec) * 1e9 + (end->tv_nsec - start->tv_nsec);
}// 打印性能統計
void print_performance(TimeRecorder *recorder) {long long total_time_ns = calculate_time_diff(&recorder->start, &recorder->end);long long write_time_ns = calculate_time_diff(&recorder->write_start, &recorder->write_end);long long read_time_ns = calculate_time_diff(&recorder->read_start, &recorder->read_end);double total_time_s = total_time_ns / 1e9;double write_time_s = write_time_ns / 1e9;double read_time_s = read_time_ns / 1e9;double total_size_mb = FILE_SIZE / (1024.0 * 1024.0);double total_speed = total_size_mb / total_time_s;double write_speed = total_size_mb / write_time_s;double read_speed = total_size_mb / read_time_s;printf("\n===== Performance Statistics =====\n");printf("Total operation time: %.2f seconds\n", total_time_s);printf("Write operation time: %.2f seconds\n", write_time_s);printf("Read operation time: %.2f seconds\n", read_time_s);printf("Total data size: %.2f MB (%.2f GB)\n", total_size_mb, total_size_mb/1024);printf("Overall transfer speed: %.2f MB/s\n", total_speed);printf("Write speed: %.2f MB/s\n", write_speed);printf("Read speed: %.2f MB/s\n", read_speed);printf("================================\n");
}int main() {io_context_t ctx = 0;int fd;char **buffers = NULL;struct iocb **iocb_ptrs = NULL;struct iocb iocbs[BATCH_SIZE];  // 改為批處理大小struct io_event events[BATCH_SIZE];const char *filename = "aio_test.txt";int i, j, ret;int completed = 0;int verify_ok = 1;TimeRecorder time_rec;// 初始化時間記錄器init_time_recorder(&time_rec);// 1. 初始化異步I/O上下文，最多緩存8192個異步事件if (io_setup(8192, &ctx) < 0) { perror("io_setup");return 1;}// 2. 創建測試文件，使用O_DIRECT方式打開fd = open(filename, O_RDWR | O_CREAT | O_TRUNC | O_DIRECT, 0644);if (fd < 0) {perror("open");io_destroy(ctx);return 1;}// 3. 分配對齊的內存緩沖區 4K大小buffers = (char**)malloc(NUM_BLOCKS * sizeof(char *));if (!buffers) {perror("malloc");close(fd);io_destroy(ctx);return 1;}printf("Allocating memory buffers...\n");for (i = 0; i < NUM_BLOCKS; i++) {if (posix_memalign((void**)&buffers[i], ALIGN_SIZE, BLOCK_SIZE) != 0) { //以4K大小對齊perror("posix_memalign");for (j = 0; j < i; j++) free(buffers[j]);free(buffers);close(fd);io_destroy(ctx);return 1;}memset(buffers[i], 0, BLOCK_SIZE);// 顯示內存分配進度if (i % 10000 == 0) {printf("Allocated %d/%d buffers\r", i, NUM_BLOCKS);fflush(stdout);}}printf("Allocated %d buffers successfully\n", NUM_BLOCKS);// 4. 準備測試數據printf("Preparing test data...\n");for (i = 0; i < NUM_BLOCKS; i++) {sprintf(buffers[i], "This is block %d of %d\n", i + 1, NUM_BLOCKS);// 顯示數據準備進度if (i % 10000 == 0) {printf("Prepared %d/%d data blocks\r", i, NUM_BLOCKS);fflush(stdout);}}printf("Prepared %d data blocks successfully\n", NUM_BLOCKS);// 5. 分配I/O控制塊指針數組，每個數組對應的數量為BATCH_SZIEiocb_ptrs = (struct iocb**)malloc(BATCH_SIZE * sizeof(struct iocb*));if (!iocb_ptrs) {perror("malloc iocb_ptrs");goto cleanup;}// 6. 提交異步寫請求 (分批)record_time(&time_rec.write_start);for (int batch = 0; batch < NUM_BLOCKS; batch += BATCH_SIZE) {int current_batch = (batch + BATCH_SIZE > NUM_BLOCKS) ? NUM_BLOCKS - batch : BATCH_SIZE;for (i = 0; i < current_batch; i++) {int idx = batch + i;io_prep_pwrite(&iocbs[i], fd, buffers[idx], BLOCK_SIZE, idx * BLOCK_SIZE);iocbs[i].data = (void*)(long)idx;iocb_ptrs[i] = &iocbs[i];}ret = io_submit(ctx, current_batch, iocb_ptrs);if (ret != current_batch) {fprintf(stderr, "io_submit write batch %d failed: %d/%d, %s\n", batch/BATCH_SIZE, ret, current_batch, strerror(-ret));goto cleanup;}// 等待當前批次完成completed = 0;while (completed < current_batch) {ret = io_getevents(ctx, 1, current_batch, events, NULL);if (ret < 0) {perror("io_getevents write");verify_ok = 0;goto cleanup;}completed += ret;for (i = 0; i < ret; i++) {int block_idx = (int)(long)events[i].data;printf("Write completed for block %d/%d (batch %d/%d)\r", block_idx + 1, NUM_BLOCKS, batch/BATCH_SIZE + 1, (NUM_BLOCKS+BATCH_SIZE-1)/BATCH_SIZE);fflush(stdout);}}}record_time(&time_rec.write_end);printf("\nAll write operations completed\n");// 7. 提交異步讀請求 (分批)record_time(&time_rec.read_start);for (int batch = 0; batch < NUM_BLOCKS; batch += BATCH_SIZE) {int current_batch = (batch + BATCH_SIZE > NUM_BLOCKS) ? NUM_BLOCKS - batch : BATCH_SIZE;for (i = 0; i < current_batch; i++) {int idx = batch + i;io_prep_pread(&iocbs[i], fd, buffers[idx], BLOCK_SIZE, idx * BLOCK_SIZE); //準備異步讀請求iocbs[i].data = (void*)(long)idx;iocb_ptrs[i] = &iocbs[i];}ret = io_submit(ctx, current_batch, iocb_ptrs); //提交異步讀請求if (ret != current_batch) {fprintf(stderr, "io_submit read batch %d failed: %d/%d, %s\n", batch/BATCH_SIZE, ret, current_batch, strerror(-ret));goto cleanup;}// 等待當前批次完成并驗證completed = 0;while (completed < current_batch) {ret = io_getevents(ctx, 1, current_batch, events, NULL);if (ret < 0) {perror("io_getevents read");verify_ok = 0;goto cleanup;}completed += ret;for (i = 0; i < ret; i++) {int block_idx = (int)(long)events[i].data;printf("Read completed for block %d/%d (batch %d/%d)\r", block_idx + 1, NUM_BLOCKS, batch/BATCH_SIZE + 1, (NUM_BLOCKS+BATCH_SIZE-1)/BATCH_SIZE);fflush(stdout);// 驗證數據char expected[BLOCK_SIZE];sprintf(expected, "This is block %d of %d\n", block_idx + 1, NUM_BLOCKS);if (strncmp(buffers[block_idx], expected, BLOCK_SIZE) != 0) {printf("\nBlock %d verification failed!\n", block_idx + 1);printf("Expected: %s", expected);printf("Actual:   %s", buffers[block_idx]);verify_ok = 0;}}}}record_time(&time_rec.read_end);printf("\nAll read operations completed\n");// 9. 輸出驗證結果if (verify_ok) {printf("All blocks verified successfully!\n");} else {printf("Verification failed for some blocks\n");}cleanup:// 10. 記錄總結束時間record_time(&time_rec.end);// 11. 打印性能統計print_performance(&time_rec);// 12. 清理資源printf("Cleaning up resources...\n");for (i = 0; i < NUM_BLOCKS; i++) {if (buffers[i]) free(buffers[i]);// 顯示清理進度if (i % 10000 == 0) {printf("Freed %d/%d buffers\r", i, NUM_BLOCKS);fflush(stdout);}}if (buffers) free(buffers);if (iocb_ptrs) free(iocb_ptrs);if (fd >= 0) close(fd);if (ctx) io_destroy(ctx);printf("Cleanup completed\n");return verify_ok ? 0 : 1;
}