異步IO實戰之四：異步IO的單個處理和批量處理
異步IO由於它的非阻塞特性和強大的並發能力，非常適合用在要求高並發和高吞吐率的場景，比如用在提供SAN存儲的塊設備讀寫的實現上。和傳統IO模式類似，異步IO提供了一次提交一個IO請求的模式，還提供了一次提交一組IO請求的方式。下面將分別介紹這兩種模式的使用方法和差異。
單個處理模式
下面對一個長度為BUF_LEN的buffer,提交了MAX_IO個單個處理的IO請求，每個請求處理BUF_LEN/MAX_IO這麽長的數據。為了便於讀者實際上機驗證，下面貼出了具體處理的代碼：
..............................................................................................
int main(int argc, char * argv[])
{
io_context_t ctx_id = 0;
struct iocb mycb[MAX_AIO];
struct iovec myiov[MAX_AIO];
struct io_event events[MAX_AIO];
const int aioto = 5000000;
io_callback_t cb = my_callback;
unsigned nr_events = MAX_AIO;
int i, lenperIo, done, cnt = 0;
char buf[BUF_LEN];
memset(mycb, 0, sizeof(mycb));
memset(myiov, 0, sizeof(myiov));
int fd = open("./text.txt", O_CREAT | O_RDWR);
assert(fd >= 0);
memset(buf, ‘a‘, BUF_LEN);
io_setup(nr_events, &ctx_id);
lenperIo = BUF_LEN/MAX_AIO;
for (i = 0; i < MAX_AIO; i++) {
io_prep_pwrite(&mycb[i], fd, lenperIo * i + (char *)buf, lenperIo, lenperIo * i);
io_set_callback(&mycb[i], my_callback);
mycb[i].key = i;
events[i].obj = &mycb[i];
}
struct timespec ts;
ts.tv_sec = 0;
ts.tv_nsec = aioto * 1000;
while (cnt < MAX_AIO) {
done = io_getevents(ctx_id, 1, MAX_AIO, events, &ts);
cnt += done;
printf("Finish %d coming IO!\n", done);
sleep(1);
}
for (i = 0; i < MAX_AIO; i++) {
cb = (io_callback_t)(events[i].data);
cb(ctx_id, (struct iocb *)events[i].obj, i, 0);
printf("events[%d]: obj = %p res = %ld res2 = %ld\n", i, events[i].obj, events[i].res, events[i].res2);
}
printf("My callback address : %p\n", (void *)my_callback);
close(fd);
}
..............................................................................................
可以看到，上面是多次提交io_prep_pwrite()，然後通過檢查是否收到相應數目的event來判斷IO是否完成。
2. 批量處理模式
初始化的代碼差不多，主要的差別代碼在下面
..............................................................................................
io_setup(nr_events, &ctx_id);
lenperIo = BUF_LEN/MAX_AIO;
for (i = 0; i < MAX_AIO; i++) {
myiov[i].iov_len = lenperIo;
myiov[i].iov_base = lenperIo * i + (char *)buf;
pmycb[i] = &mycb[i];
}
io_prep_pwritev(&mycb[0], fd, myiov, MAX_AIO, 32);
io_set_callback(&mycb[0], cb);
struct timespec ts;
ts.tv_sec = 0;
ts.tv_nsec = aioto * 1000;
while (cnt < 1) {
done = io_getevents(ctx_id, 1, 1, events, &ts);
cnt += done;
printf("Finish %d coming IO!\n", done);
sleep(2);
}
..............................................................................................
可以看到，在批量處理模式，先需要初始化一個io vector數組，在數組裏面再指定IO操作在內存中的數據起始地址和長度，然後調用一次io_prep_pweritev()，最後等待唯一的一個event就可以了。
3. 兩種處理方式的比較
下面的表格，總結了使用上面兩種處理方式完成相同IO任務的實現上的差異：
模式 | 使用函數 | 是否必需用iovec | io_setup調用次數 | IO提交函數調用次數 | io_getevents 需調用的次數 | 生成的io events數目 |
單個處理
| io_prep_pwrite io_prep_pread | 不是 | 多次 | 多次 | 很可能多次 | 多個 |
批量處理 | io_prep_pwritev io_prep_preadv | 是 | 1次 | 1次 | 最少一次 | 1 個 |
4. 註意事項
單個IO處理的模式很好理解，而對批量處理，個人認為有一處man手冊和頭文件中都沒有說明白的地方，那就是： io_prep_pwritev/ io_prep_preadv函數的最後一個參數offset的含義，它表示的是io vecotr裏面最早執行的那個IO開始執行時讀寫操作在磁盤或文件上的物理偏移, 而下一個IO在磁盤或文件上讀寫的起始地址，就是這個偏移再加上剛完成IO操作的長度。因此，IO操作的總長度是IO vector裏面所有成員的iov_len字段之和。
本文出自 “存儲之廚” 博客，請務必保留此出處http://xiamachao.blog.51cto.com/10580956/1977588
異步IO實戰之四：異步IO的單個處理和批量處理