test OK

Author: qinyihust

async_io.cpp

@@ -51,7 +51,6 @@ Libaio::~Libaio() {
 
 int Libaio::SubmitIo(IoTask *task) {
     iocb *iocbp = new(iocb);
-    memset(iocbp, 0, sizeof(iocb));
     if (task->isRead)
         io_prep_preadv(iocbp, task->fd, &(task->iov), 1, task->offset);
     else
@@ -75,7 +74,7 @@ IoTask *Libaio::ReapIo() {
     iocb *iocbp = event.obj;
     assert(task == iocbp->data);
     delete iocbp;
-    task->res = event.res2; // res is r/w length, res2 is operation result
+    task->res = event.res; // res: +x for done length, -x for error code
 
     return task;
-}
+}

async_io.h

@@ -12,6 +12,7 @@
 struct IoTask;
 typedef void (*IocbFunc)(IoTask *task);
 struct IoTask {
+    int index; // thread index
     int fd;
     bool isRead;
     off_t first_offset, offset;
@@ -50,4 +51,4 @@ class Libaio: public AsyncIo {
     io_context_t ctx_;
 };
 
-#endif // ASYNC_IO_H_
+#endif // ASYNC_IO_H_

build.sh

@@ -7,7 +7,7 @@
 
 #define QD	64
 #define BS	(4*1024)
-#define NR_THREAD 2
+#define NR_THREAD 10
 
 struct TestEnv {
     int index;
@@ -20,6 +20,7 @@ int queueIo(TestEnv *env, off_t offset, off_t len, bool isRead, void *buf, IocbF
     if (!task)
         return -1;
 
+    task->index = env->index;
     task->fd = env->fd;
 
     task->isRead = isRead;
@@ -30,7 +31,7 @@ int queueIo(TestEnv *env, off_t offset, off_t len, bool isRead, void *buf, IocbF
     task->iov.iov_len = len;
 
     task->cb = cb;
-    task->arg = arg;
+    task->arg = env->submitter;
     task->res = -1;
 
     env->submitter->Push(task);
@@ -39,36 +40,66 @@ int queueIo(TestEnv *env, off_t offset, off_t len, bool isRead, void *buf, IocbF
 }
 
 void cb1(IoTask *task) {
-    assert(task->isRead);
-    std::cout << "read data: " << (static_cast<char *>(task->iov.iov_base))[0]
-              << std::endl;
+    assert (!task->isRead);
+    unsigned num = ((char *)(task->iov.iov_base))[0];
+    std::cout << "reaper: write for thread " << task->index <<" done, data="
+              << std::hex << num << ", res="<< task->res << std::endl;
 
     delete task;
 }
 
+
+void cb0(IoTask *task) {
+    assert (task->isRead);
+    unsigned num = ((char *)(task->iov.iov_base))[0];
+    std::cout << "reaper: read from thread " << task->index << " done, data="
+              << std::hex << num << ", res=" << task->res << std::endl;
+    num++;
+    std::cout << "reaper: send write I/O for thread " << task->index << ", data="
+	      << std::hex << num << std::endl;
+    // send write I/O
+    task->isRead = false;
+    memset(task->iov.iov_base, num, BS);
+    task->cb = cb1;
+    task->res = -1;
+    Submitter *submitter = (Submitter *)task->arg;
+    submitter->Push(task);
+}
+
 void *SendIo(void *arg) {
     TestEnv *env = (TestEnv *)arg;
     int idx = env->index;
-    int fd = env->fd;
-
-    std::string data(BS, 0);
-    switch (idx) {
-        case 0:
-            data.assign(BS, '1');
-            queueIo(env, 0, BS, false, (void *)(data.c_str()), nullptr, nullptr);
-        case 1:
-            sleep(10);
-            queueIo(env, 0, BS, true, (void *)data.c_str(), cb1, nullptr);
-    }
-    sleep(20);
+    char *data = nullptr;
+    posix_memalign((void **)&data, getpagesize(), BS);
+    memset(data, 0, BS);
+    std::cout << "thread " << env->index << ": read testfile" << std::endl;
+    queueIo(env, 0, BS, true, data, cb0, nullptr);
 
+    sleep(3);
     return nullptr;
 }
 
-
 int main(int argc, const char* argv[]) {
     int ret = 0;
-    Submitter submitter(IoEngine::IO_ENGINE_URING, QD);
+
+    if (argc != 2) {
+	std::cout << "usage: " << argv[0] << " [option]" << std::endl;
+	std::cout << "option: libaio or uring" << std::endl;
+        return -1;
+    }
+    
+    IoEngine engine = IoEngine::IO_ENGINE_NONE;
+    if (!strncmp(argv[1], "libaio", 7))
+	engine = IoEngine::IO_ENGINE_LIBAIO;
+    else if (!strncmp(argv[1], "uring", 6))
+	engine = IoEngine::IO_ENGINE_URING;
+    else {
+	std::cout << "usage: " << argv[0] << " [option]" << std::endl;
+        std::cout << "option: libaio or uring" << std::endl;
+        return -1;
+    }
+    	
+    Submitter submitter(engine, QD);
     if (submitter.Run())
         return -1;
 
@@ -103,6 +134,8 @@ int main(int argc, const char* argv[]) {
         pthread_join(tidp[i], nullptr);
     reaper.Finish();
     submitter.Finish();
+    fsync(fd);
+    close(fd);
 
     return ret;
 }

io_demo.cpp

@@ -14,6 +14,7 @@
 enum class IoEngine {
     IO_ENGINE_LIBAIO,
     IO_ENGINE_URING,
+    IO_ENGINE_NONE,
 };
 
 class Submitter {