async_connect use io_uring

make.sh

@@ -9,10 +9,13 @@ DEBUG_FLAG=0
 
 if [ -f /usr/include/fastcommon/_os_define.h ]; then
   OS_BITS=$(grep -F OS_BITS /usr/include/fastcommon/_os_define.h | awk '{print $NF;}')
+  USE_URING=$(grep -F IOEVENT_USE_URING /usr/include/fastcommon/_os_define.h | awk '{print $NF;}')
 elif [ -f /usr/local/include/fastcommon/_os_define.h ]; then
   OS_BITS=$(grep -F OS_BITS /usr/local/include/fastcommon/_os_define.h | awk '{print $NF;}')
+  USE_URING=$(grep -F IOEVENT_USE_URING /usr/local/include/fastcommon/_os_define.h | awk '{print $NF;}')
 else
   OS_BITS=64
+  USE_URING=''
 fi
 
 uname=$(uname)
@@ -49,6 +52,9 @@ LIBS=''
 uname=$(uname)
 if [ "$uname" = "Linux" ]; then
   CFLAGS="$CFLAGS"
+  if [ -n "$USE_URING" ]; then
+    LIBS="$LIBS -luring"
+  fi
 elif [ "$uname" = "FreeBSD" ] || [ "$uname" = "Darwin" ]; then
   CFLAGS="$CFLAGS"
   if [ "$uname" = "Darwin" ]; then

src/sf_nio.c

@@ -220,10 +220,17 @@ static inline int set_read_event(struct fast_task_info *task)
 
 #if IOEVENT_USE_URING
     if (task->handler->use_io_uring) {
+        if (FC_URING_OP_TYPE(task) != IORING_OP_NOP) {
             if (FC_URING_OP_TYPE(task) == IORING_OP_RECV) {
                 logWarning("file: "__FILE__", line: %d, "
                         "trigger recv again!", __LINE__);
                 return 0;
+            } else {
+                logWarning("file: "__FILE__", line: %d, "
+                        "another operation in progress, op_type: %d!",
+                        __LINE__, FC_URING_OP_TYPE(task));
+                return EBUSY;
+            }
         }
 
         if (task->event.callback != (IOEventCallback)sf_client_sock_read) {
@@ -316,16 +323,31 @@ static int sf_client_connect_done(int sock, short event, void *arg)
 
     task = (struct fast_task_info *)arg;
     if (task->canceled) {
+#if IOEVENT_USE_URING
+        if (task->handler->use_io_uring && event != IOEVENT_TIMEOUT) {
+            CLEAR_OP_TYPE_AND_RELEASE_TASK(task);
+        }
+#endif
         return ENOTCONN;
     }
 
     if (event & IOEVENT_TIMEOUT) {
         result = ETIMEDOUT;
     } else {
+#if IOEVENT_USE_URING
+        if (task->handler->use_io_uring) {
+            CLEAR_OP_TYPE_AND_RELEASE_TASK(task);
+            result = (task->event.res < 0 ? -1 * task->event.res :
+                    task->event.res);
+        } else {
+#endif
             result = task->handler->async_connect_check(task);
             if (result == EINPROGRESS) {
                 return 0;
             }
+#if IOEVENT_USE_URING
+        }
+#endif
     }
 
     if (SF_CTX->callbacks.connect_done != NULL) {
@@ -356,6 +378,15 @@ static int sf_client_connect_done(int sock, short event, void *arg)
 int sf_socket_async_connect_server(struct fast_task_info *task)
 {
     int result;
+
+#if IOEVENT_USE_URING
+    if (task->handler->use_io_uring) {
+        if ((result=uring_prep_connect(task)) != 0) {
+            return result;
+        }
+        return EINPROGRESS;
+    } else {
+#endif
         if ((task->event.fd=socketCreateEx2(AF_UNSPEC, task->server_ip,
                         O_NONBLOCK, NULL, &result)) < 0)
         {
@@ -364,6 +395,9 @@ int sf_socket_async_connect_server(struct fast_task_info *task)
 
         return asyncconnectserverbyip(task->event.fd,
                 task->server_ip, task->port);
+#if IOEVENT_USE_URING
+    }
+#endif
 }
 
 static int sf_async_connect_server(struct fast_task_info *task)