upgrade version to 1.0,81

.gitignore

@@ -1,5 +1,6 @@
 # Makefile.in
 src/Makefile
+src/tests/Makefile
 
 # Prerequisites
 *.d

HISTORY

@@ -1,4 +1,9 @@
 
+Version 1.81  2025-10-05
+  * support Linux io_uring
+  * free_queue support parameter: need_shrink and set task->shrinked
+  * IOEventCallback: change event type from short to int
+
 Version 1.80  2025-09-10
   * getIpaddrByNameEx: IPv4 has priority over IPv6
   * shared_func.[hc]: add function fc_ftoa

libfastcommon.spec

@@ -3,7 +3,7 @@
 %define CommitVersion %(echo $COMMIT_VERSION)
 
 Name: libfastcommon
-Version: 1.0.80
+Version: 1.0.81
 Release: 1%{?dist}
 Summary: c common functions library extracted from my open source projects FastDFS
 License: LGPL
@@ -17,6 +17,14 @@ BuildRequires: libcurl-devel
 Requires: libcurl
 Requires: %__cp %__mv %__chmod %__grep %__mkdir %__install %__id
 
+%define kernel_major %(uname -r | cut -d'.' -f1)
+%define kernel_minor %(uname -r | cut -d'.' -f2)
+%define kernel_ver_int %(expr %{kernel_major} \* 100 + %{kernel_minor})
+%if %{kernel_ver_int} >= 514
+BuildRequires: liburing-devel >= 2.5
+Requires: liburing >= 2.5
+%endif
+
 %description
 c common functions library extracted from my open source projects FastDFS.
 this library is very simple and stable. functions including: string, logger,

make.sh

@@ -112,7 +112,23 @@ HAVE_VMMETER_H=0
 HAVE_USER_H=0
 if [ "$uname" = "Linux" ]; then
   OS_NAME=OS_LINUX
+
+  major_version=$(uname -r | awk -F . '{print $1;}')
+  minor_version=$(uname -r | awk -F . '{print $2;}')
+  if [ $major_version -eq 5 ] && [ $minor_version -ge 14 ]; then
+    out=$(grep -F IORING_OP_SEND_ZC /usr/include/liburing/io_uring.h)
+    if [ -n "$out" ]; then
+      IOEVENT_USE=IOEVENT_USE_URING
+      LIBS="$LIBS -luring"
+    else
       IOEVENT_USE=IOEVENT_USE_EPOLL
+    fi
+  elif [ $major_version -gt 5 ]; then
+    IOEVENT_USE=IOEVENT_USE_URING
+  else
+    IOEVENT_USE=IOEVENT_USE_EPOLL
+  fi
+
   if [ $glibc_minor -lt 17 ]; then
     LIBS="$LIBS -lrt"
   fi
@@ -261,3 +277,12 @@ make $1 $2 $3
 if [ "$1" = "clean" ]; then
   /bin/rm -f Makefile _os_define.h
 fi
+
+cd tests
+cp Makefile.in Makefile
+sed_replace "s#\\\$(CC)#gcc#g" Makefile
+sed_replace "s#\\\$(INCS)#$INCS#g" Makefile
+sed_replace "s#\\\$(LIBS)#$LIBS#g" Makefile
+if [ "$1" = "clean" ]; then
+  /bin/rm -f Makefile
+fi

src/common_define.h

@@ -125,7 +125,8 @@ extern int pthread_mutexattr_settype(pthread_mutexattr_t *attr, int kind);
 #define FC_IOV_BATCH_SIZE  IOV_MAX
 #endif
 
-// 由于要支持IPv6，所以将IP_ADDRESS_SIZE的值由16修改为46
+#define IPV4_ADDRESS_SIZE   INET_ADDRSTRLEN   //16
+#define IPV6_ADDRESS_SIZE   INET6_ADDRSTRLEN  //46
 #define IP_ADDRESS_SIZE     INET6_ADDRSTRLEN  //46
 #define FORMATTED_IP_SIZE   (IP_ADDRESS_SIZE + 2)
 #define INFINITE_FILE_SIZE (256 * 1024LL * 1024 * 1024 * 1024 * 1024LL)
@@ -454,6 +455,14 @@ static inline int fc_string_compare(const string_t *s1, const string_t *s2)
     }
 }
 
+static inline bool fc_string_equal_ex(const char *str1,
+        const int len1, const char *str2, const int len2)
+{
+    return (len1 == len2) && (memcmp(str1, str2, len1) == 0);
+}
+#define fc_string_equals_ex(str1, len1, str2, len2) \
+    fc_string_equal_ex(str1, len1, str2, len2)
+
 static inline bool fc_string_equal(const string_t *s1, const string_t *s2)
 {
     return (s1->len == s2->len) && (memcmp(s1->str, s2->str, s1->len) == 0);

src/connection_pool.c

@@ -25,10 +25,20 @@
 #include "server_id_func.h"
 #include "connection_pool.h"
 
+static void conn_pool_disconnect_server_cb(ConnectionInfo *conn)
+{
+    conn_pool_disconnect_server(conn);
+}
+
+static bool conn_pool_is_connected_cb(ConnectionInfo *conn)
+{
+    return conn_pool_is_connected(conn);
+}
+
 ConnectionCallbacks g_connection_callbacks = {
     false, {{conn_pool_connect_server_ex1,
-        conn_pool_disconnect_server,
+        conn_pool_disconnect_server_cb,
-        conn_pool_is_connected},
+        conn_pool_is_connected_cb},
     {NULL, NULL, NULL}}, {NULL}
 };
 
@@ -411,20 +421,6 @@ void conn_pool_destroy(ConnectionPool *cp)
     fast_mblock_destroy(&cp->node_allocator);
 }
 
-void conn_pool_disconnect_server(ConnectionInfo *conn)
-{
-    if (conn->sock >= 0)
-    {
-        close(conn->sock);
-        conn->sock = -1;
-    }
-}
-
-bool conn_pool_is_connected(ConnectionInfo *conn)
-{
-    return (conn->sock >= 0);
-}
-
 int conn_pool_connect_server_ex1(ConnectionInfo *conn,
         const char *service_name, const int connect_timeout_ms,
         const char *bind_ipaddr, const bool log_connect_error)

src/connection_pool.h

@@ -331,9 +331,19 @@ int conn_pool_close_connection_ex(ConnectionPool *cp,
 *      conn: the connection
 *   return 0 for success, != 0 for error
 */
-void conn_pool_disconnect_server(ConnectionInfo *conn);
+static inline void conn_pool_disconnect_server(ConnectionInfo *conn)
+{
+    if (conn->sock >= 0)
+    {
+        close(conn->sock);
+        conn->sock = -1;
+    }
+}
 
-bool conn_pool_is_connected(ConnectionInfo *conn);
+static inline bool conn_pool_is_connected(ConnectionInfo *conn)
+{
+    return (conn->sock >= 0);
+}
 
 /**
 *   connect to the server

src/fast_task_queue.c

@@ -59,11 +59,11 @@ static int task_alloc_init(struct fast_task_info *task,
 }
 
 int free_queue_init_ex2(struct fast_task_queue *queue, const char *name,
-        const bool double_buffers, const int max_connections,
+        const bool double_buffers, const bool need_shrink,
-        const int alloc_task_once, const int min_buff_size,
+        const int max_connections, const int alloc_task_once,
-        const int max_buff_size, const int padding_size,
+        const int min_buff_size, const int max_buff_size,
-        const int arg_size, TaskInitCallback init_callback,
+        const int padding_size, const int arg_size,
-        void *init_arg)
+        TaskInitCallback init_callback, void *init_arg)
 {
 #define MAX_DATA_SIZE  (256 * 1024 * 1024)
     int alloc_once;
@@ -123,6 +123,7 @@ int free_queue_init_ex2(struct fast_task_queue *queue, const char *name,
     }
 
     queue->double_buffers = double_buffers;
+    queue->need_shrink = need_shrink;
 	queue->min_buff_size = aligned_min_size;
 	queue->max_buff_size = aligned_max_size;
 	queue->padding_size = aligned_padding_size;
@@ -183,16 +184,23 @@ void free_queue_push(struct fast_task_info *task)
     task->send.ptr->length = 0;
     task->send.ptr->offset = 0;
     task->req_count = 0;
-    if (task->send.ptr->size > task->free_queue->min_buff_size) {//need thrink
+    if (task->free_queue->need_shrink && task->send.
-        _realloc_buffer(task->send.ptr, task->free_queue->min_buff_size, false);
+            ptr->size > task->free_queue->min_buff_size)
+    {   //need thrink
+        _realloc_buffer(task->send.ptr, task->free_queue->
+                min_buff_size, false);
+        task->shrinked = true;
     }
 
     if (task->free_queue->double_buffers) {
         task->recv.ptr->length = 0;
         task->recv.ptr->offset = 0;
-        if (task->recv.ptr->size > task->free_queue->min_buff_size) {
+        if (task->free_queue->need_shrink && task->recv.
+                ptr->size > task->free_queue->min_buff_size)
+        {
             _realloc_buffer(task->recv.ptr, task->free_queue->
                     min_buff_size, false);
+            task->shrinked = true;
         }
     }
 

src/fast_task_queue.h

@@ -42,16 +42,23 @@ typedef void (*TaskCleanUpCallback) (struct fast_task_info *task);
 typedef int (*TaskInitCallback)(struct fast_task_info *task, void *arg);
 typedef void (*TaskReleaseCallback)(struct fast_task_info *task);
 
-typedef void (*IOEventCallback) (int sock, short event, void *arg);
+typedef void (*IOEventCallback) (int sock, const int event, void *arg);
 typedef int (*TaskContinueCallback)(struct fast_task_info *task);
 
 struct sf_network_handler;
 struct fast_task_info;
 
+#if IOEVENT_USE_URING
+#define FC_URING_OP_TYPE(task)     (task)->uring.op_type
+#define FC_URING_IS_CLIENT(task)   (task)->uring.is_client
+#define FC_URING_IS_SEND_ZC(task)  ((task)->uring.op_type == IORING_OP_SEND_ZC)
+#endif
+
 typedef struct ioevent_entry
 {
     FastTimerEntry timer; //must first
     int fd;
+    int res;  //just for io_uring, since v1.0.81
     IOEventCallback callback;
 } IOEventEntry;
 
@@ -119,12 +126,19 @@ struct fast_task_info
     struct fast_net_buffer_wrapper recv;  //recv buffer
 
     uint16_t port; //peer port
+
+    struct {
+        int8_t  is_client;
+        uint8_t op_type;
+    } uring;  //just for io_uring, since v1.0.81
+
     struct {
         uint8_t current;
         volatile uint8_t notify;
     } nio_stages; //stages for network IO
-    volatile int8_t reffer_count;
     volatile int8_t canceled;  //if task canceled
+    volatile int8_t shrinked;  //if task shrinked, since V1.0.81
+    volatile int reffer_count;
     int pending_send_count;
     int64_t req_count; //request count
     struct {
@@ -153,6 +167,7 @@ struct fast_task_queue
     int block_size;
     bool malloc_whole_block;
     bool double_buffers;  //if send buffer and recv buffer are independent
+    bool need_shrink;
     struct fast_mblock_man allocator;
     TaskInitCallback init_callback;
     void *init_arg;
@@ -164,22 +179,22 @@ extern "C" {
 #endif
 
 int free_queue_init_ex2(struct fast_task_queue *queue, const char *name,
-        const bool double_buffers, const int max_connections,
+        const bool double_buffers, const bool need_shrink,
-        const int alloc_task_once, const int min_buff_size,
+        const int max_connections, const int alloc_task_once,
-        const int max_buff_size, const int padding_size,
+        const int min_buff_size, const int max_buff_size,
-        const int arg_size, TaskInitCallback init_callback,
+        const int padding_size, const int arg_size,
-        void *init_arg);
+        TaskInitCallback init_callback, void *init_arg);
 
 static inline int free_queue_init_ex(struct fast_task_queue *queue,
         const char *name, const bool double_buffers,
-        const int max_connections, const int alloc_task_once,
+        const bool need_shrink, const int max_connections,
-        const int min_buff_size, const int max_buff_size,
+        const int alloc_task_once, const int min_buff_size,
-        const int arg_size)
+        const int max_buff_size, const int arg_size)
 {
     const int padding_size = 0;
-    return free_queue_init_ex2(queue, name, double_buffers, max_connections,
+    return free_queue_init_ex2(queue, name, double_buffers, need_shrink,
-            alloc_task_once, min_buff_size, max_buff_size, padding_size,
+            max_connections, alloc_task_once, min_buff_size, max_buff_size,
-            arg_size, NULL, NULL);
+            padding_size, arg_size, NULL, NULL);
 }
 
 static inline int free_queue_init(struct fast_task_queue *queue,
@@ -188,9 +203,11 @@ static inline int free_queue_init(struct fast_task_queue *queue,
 {
     const char *name = "";
     const bool double_buffers = false;
+    const bool need_shrink = true;
     const int arg_size = 0;
-    return free_queue_init_ex(queue, name, double_buffers, max_connections,
+    return free_queue_init_ex(queue, name, double_buffers,
-            alloc_task_once, min_buff_size, max_buff_size, arg_size);
+            need_shrink, max_connections, alloc_task_once,
+            min_buff_size, max_buff_size, arg_size);
 }
 
 static inline void free_queue_set_release_callback(

src/fast_timer.c

@@ -105,6 +105,7 @@ int fast_timer_modify(FastTimer *timer, FastTimerEntry *entry,
         if ((result=fast_timer_remove(timer, entry)) == 0) {
             fast_timer_add_ex(timer, entry, new_expires, true);
         }
+
         return result;
     }
     return 0;
@@ -185,6 +186,7 @@ int fast_timer_timeouts_get(FastTimer *timer, const int64_t current_time,
                     } else {
                         last->rehash = false;
                     }
+
                     continue;
                 }
             } else {  //expired

src/ioevent.c

@@ -45,15 +45,21 @@ int kqueue_ev_convert(int16_t event, uint16_t flags)
 }
 #endif
 
-int ioevent_init(IOEventPoller *ioevent, const int size,
+int ioevent_init(IOEventPoller *ioevent, const char *service_name,
-    const int timeout_ms, const int extra_events)
+        const int size, const int timeout_ms, const int extra_events)
 {
+#if IOEVENT_USE_URING
+    int result;
+#else
     int bytes;
 
-  ioevent->size = size;
-  ioevent->extra_events = extra_events;
     ioevent->iterator.index = 0;
     ioevent->iterator.count = 0;
+#endif
+
+    ioevent->service_name = service_name;
+    ioevent->size = size;
+    ioevent->extra_events = extra_events;
 
 #if IOEVENT_USE_EPOLL
     ioevent->poll_fd = epoll_create(ioevent->size);
@@ -62,6 +68,14 @@ int ioevent_init(IOEventPoller *ioevent, const int size,
     }
     bytes = sizeof(struct epoll_event) * size;
     ioevent->events = (struct epoll_event *)fc_malloc(bytes);
+#elif IOEVENT_USE_URING
+    if ((result=io_uring_queue_init(size, &ioevent->ring, 0)) < 0) {
+        return -result;
+    }
+    ioevent->cqe = NULL;
+    ioevent->submit_count = 0;
+    ioevent->send_zc_logged = false;
+    ioevent->send_zc_done_notify = false;
 #elif IOEVENT_USE_KQUEUE
     ioevent->poll_fd = kqueue();
     if (ioevent->poll_fd < 0) {
@@ -78,18 +92,25 @@ int ioevent_init(IOEventPoller *ioevent, const int size,
     ioevent->events = (port_event_t *)fc_malloc(bytes);
 #endif
 
+#if IOEVENT_USE_URING
+
+#else
     if (ioevent->events == NULL) {
         close(ioevent->poll_fd);
         ioevent->poll_fd = -1;
         return ENOMEM;
     }
-  ioevent_set_timeout(ioevent, timeout_ms);
+#endif
 
+    ioevent_set_timeout(ioevent, timeout_ms);
     return 0;
 }
 
 void ioevent_destroy(IOEventPoller *ioevent)
 {
+#if IOEVENT_USE_URING
+    io_uring_queue_exit(&ioevent->ring);
+#else
   if (ioevent->events != NULL) {
     free(ioevent->events);
     ioevent->events = NULL;
@@ -99,10 +120,11 @@ void ioevent_destroy(IOEventPoller *ioevent)
     close(ioevent->poll_fd);
     ioevent->poll_fd = -1;
   }
+#endif
 }
 
-int ioevent_attach(IOEventPoller *ioevent, const int fd, const int e,
+int ioevent_attach(IOEventPoller *ioevent, const int fd,
-    void *data)
+        const int e, void *data)
 {
 #if IOEVENT_USE_EPOLL
   struct epoll_event ev;
@@ -110,6 +132,15 @@ int ioevent_attach(IOEventPoller *ioevent, const int fd, const int e,
   ev.events = e | ioevent->extra_events;
   ev.data.ptr = data;
   return epoll_ctl(ioevent->poll_fd, EPOLL_CTL_ADD, fd, &ev);
+#elif IOEVENT_USE_URING
+  struct io_uring_sqe *sqe = io_uring_get_sqe(&ioevent->ring);
+  if (sqe == NULL) {
+      return ENOSPC;
+  }
+  io_uring_prep_poll_multishot(sqe, fd, e | ioevent->extra_events);
+  sqe->user_data = (long)data;
+  ioevent->submit_count++;
+  return 0;
 #elif IOEVENT_USE_KQUEUE
   struct kevent ev[2];
   int n = 0;
@@ -128,8 +159,8 @@ int ioevent_attach(IOEventPoller *ioevent, const int fd, const int e,
 #endif
 }
 
-int ioevent_modify(IOEventPoller *ioevent, const int fd, const int e,
+int ioevent_modify(IOEventPoller *ioevent, const int fd,
-    void *data)
+        const int e, void *data)
 {
 #if IOEVENT_USE_EPOLL
   struct epoll_event ev;
@@ -137,6 +168,16 @@ int ioevent_modify(IOEventPoller *ioevent, const int fd, const int e,
   ev.events = e | ioevent->extra_events;
   ev.data.ptr = data;
   return epoll_ctl(ioevent->poll_fd, EPOLL_CTL_MOD, fd, &ev);
+#elif IOEVENT_USE_URING
+  struct io_uring_sqe *sqe = io_uring_get_sqe(&ioevent->ring);
+  if (sqe == NULL) {
+      return ENOSPC;
+  }
+  io_uring_prep_poll_update(sqe, sqe->user_data, sqe->user_data,
+          e | ioevent->extra_events, IORING_POLL_UPDATE_EVENTS);
+  sqe->user_data = (long)data;
+  ioevent->submit_count++;
+  return 0;
 #elif IOEVENT_USE_KQUEUE
   struct kevent ev[2];
   int result;
@@ -173,6 +214,16 @@ int ioevent_detach(IOEventPoller *ioevent, const int fd)
 {
 #if IOEVENT_USE_EPOLL
   return epoll_ctl(ioevent->poll_fd, EPOLL_CTL_DEL, fd, NULL);
+#elif IOEVENT_USE_URING
+  struct io_uring_sqe *sqe = io_uring_get_sqe(&ioevent->ring);
+  if (sqe == NULL) {
+      return ENOSPC;
+  }
+  io_uring_prep_cancel_fd(sqe, fd, 0);
+  /* set sqe->flags MUST after io_uring_prep_xxx */
+  sqe->flags = IOSQE_CQE_SKIP_SUCCESS;
+  ioevent->submit_count++;
+  return 0;
 #elif IOEVENT_USE_KQUEUE
   struct kevent ev[1];
   int r, w;
@@ -192,9 +243,20 @@ int ioevent_detach(IOEventPoller *ioevent, const int fd)
 int ioevent_poll(IOEventPoller *ioevent)
 {
 #if IOEVENT_USE_EPOLL
-  return epoll_wait(ioevent->poll_fd, ioevent->events, ioevent->size, ioevent->timeout);
+  return epoll_wait(ioevent->poll_fd, ioevent->events,
+          ioevent->size, ioevent->timeout);
+#elif IOEVENT_USE_URING
+  int result;
+  result = io_uring_wait_cqe_timeout(&ioevent->ring,
+          &ioevent->cqe, &ioevent->timeout);
+  if (result < 0) {
+      errno = -result;
+      return -1;
+  }
+  return 0;
 #elif IOEVENT_USE_KQUEUE
-  return kevent(ioevent->poll_fd, NULL, 0, ioevent->events, ioevent->size, &ioevent->timeout);
+  return kevent(ioevent->poll_fd, NULL, 0, ioevent->events,
+          ioevent->size, &ioevent->timeout);
 #elif IOEVENT_USE_PORT
   int result;
   int retval;
@@ -225,4 +287,3 @@ int ioevent_poll(IOEventPoller *ioevent)
 #error port me
 #endif
 }
-

src/ioevent.h

@@ -20,8 +20,10 @@
 #include <poll.h>
 #include <sys/time.h>
 #include "_os_define.h"
+#include "logger.h"
 
-#define IOEVENT_TIMEOUT  0x8000
+#define IOEVENT_TIMEOUT  (1 << 20)
+#define IOEVENT_NOTIFY   (1 << 21)  //for io_uring send_zc done callback
 
 #if IOEVENT_USE_EPOLL
 #include <sys/epoll.h>
@@ -31,6 +33,13 @@
 #define IOEVENT_WRITE EPOLLOUT
 #define IOEVENT_ERROR (EPOLLERR | EPOLLPRI | EPOLLHUP)
 
+#elif IOEVENT_USE_URING
+#include <sys/mount.h>
+#include <liburing.h>
+#define IOEVENT_READ  POLLIN
+#define IOEVENT_WRITE POLLOUT
+#define IOEVENT_ERROR (POLLERR | POLLPRI | POLLHUP)
+
 #elif IOEVENT_USE_KQUEUE
 #include <sys/event.h>
 #include <sys/poll.h>
@@ -65,18 +74,28 @@ int kqueue_ev_convert(int16_t event, uint16_t flags);
 #endif
 
 typedef struct ioevent_puller {
+    const char *service_name;
     int size;  //max events (fd)
     int extra_events;
+#if IOEVENT_USE_URING
+    struct io_uring ring;
+    int submit_count;
+    bool send_zc_logged;
+    bool send_zc_done_notify; //if callback when send_zc done
+#else
     int poll_fd;
-
     struct {
         int index;
         int count;
     } iterator;  //for deal event loop
+#endif
 
 #if IOEVENT_USE_EPOLL
     struct epoll_event *events;
-    int timeout;
+    int timeout;  //in milliseconds
+#elif IOEVENT_USE_URING
+    struct io_uring_cqe *cqe;
+    struct __kernel_timespec timeout;
 #elif IOEVENT_USE_KQUEUE
     struct kevent *events;
     struct timespec timeout;
@@ -84,11 +103,18 @@ typedef struct ioevent_puller {
     port_event_t *events;
     timespec_t timeout;
 #endif
+
+#ifdef OS_LINUX
+    bool zero_timeout;
+#endif
+
 } IOEventPoller;
 
 #if IOEVENT_USE_EPOLL
   #define IOEVENT_GET_EVENTS(ioevent, index) \
       (ioevent)->events[index].events
+#elif IOEVENT_USE_URING
+
 #elif IOEVENT_USE_KQUEUE
   #define IOEVENT_GET_EVENTS(ioevent, index)  kqueue_ev_convert( \
       (ioevent)->events[index].filter, (ioevent)->events[index].flags)
@@ -102,6 +128,8 @@ typedef struct ioevent_puller {
 #if IOEVENT_USE_EPOLL
   #define IOEVENT_GET_DATA(ioevent, index)  \
       (ioevent)->events[index].data.ptr
+#elif IOEVENT_USE_URING
+
 #elif IOEVENT_USE_KQUEUE
   #define IOEVENT_GET_DATA(ioevent, index)  \
       (ioevent)->events[index].udata
@@ -115,6 +143,8 @@ typedef struct ioevent_puller {
 #if IOEVENT_USE_EPOLL
   #define IOEVENT_CLEAR_DATA(ioevent, index)  \
       (ioevent)->events[index].data.ptr = NULL
+#elif IOEVENT_USE_URING
+
 #elif IOEVENT_USE_KQUEUE
   #define IOEVENT_CLEAR_DATA(ioevent, index)  \
       (ioevent)->events[index].udata = NULL
@@ -129,18 +159,19 @@ typedef struct ioevent_puller {
 extern "C" {
 #endif
 
-int ioevent_init(IOEventPoller *ioevent, const int size,
+int ioevent_init(IOEventPoller *ioevent, const char *service_name,
-    const int timeout_ms, const int extra_events);
+        const int size, const int timeout_ms, const int extra_events);
 void ioevent_destroy(IOEventPoller *ioevent);
 
-int ioevent_attach(IOEventPoller *ioevent, const int fd, const int e,
+int ioevent_attach(IOEventPoller *ioevent, const int fd,
-    void *data);
+        const int e, void *data);
-int ioevent_modify(IOEventPoller *ioevent, const int fd, const int e,
+int ioevent_modify(IOEventPoller *ioevent, const int fd,
-    void *data);
+        const int e, void *data);
 int ioevent_detach(IOEventPoller *ioevent, const int fd);
 int ioevent_poll(IOEventPoller *ioevent);
 
-static inline void ioevent_set_timeout(IOEventPoller *ioevent, const int timeout_ms)
+static inline void ioevent_set_timeout(IOEventPoller *ioevent,
+        const int timeout_ms)
 {
 #if IOEVENT_USE_EPOLL
   ioevent->timeout = timeout_ms;
@@ -148,6 +179,11 @@ static inline void ioevent_set_timeout(IOEventPoller *ioevent, const int timeout
   ioevent->timeout.tv_sec = timeout_ms / 1000;
   ioevent->timeout.tv_nsec = 1000000 * (timeout_ms % 1000);
 #endif
+
+#ifdef OS_LINUX
+    ioevent->zero_timeout = (timeout_ms == 0);
+#endif
+
 }
 
 static inline int ioevent_poll_ex(IOEventPoller *ioevent, const int timeout_ms)
@@ -156,6 +192,114 @@ static inline int ioevent_poll_ex(IOEventPoller *ioevent, const int timeout_ms)
   return ioevent_poll(ioevent);
 }
 
+#if IOEVENT_USE_URING
+static inline void ioevent_set_send_zc_done_notify(
+        IOEventPoller *ioevent, const bool need_notify)
+{
+    ioevent->send_zc_done_notify = need_notify;
+}
+
+static inline int ioevent_uring_submit(IOEventPoller *ioevent)
+{
+    int result;
+
+    ioevent->submit_count = 0;
+    while (1) {
+        result = io_uring_submit(&ioevent->ring);
+        if (result < 0) {
+            if (result != -EINTR) {
+                return -result;
+            }
+        } else {
+            return 0;
+        }
+    }
+}
+
+static inline struct io_uring_sqe *ioevent_uring_get_sqe(IOEventPoller *ioevent)
+{
+    struct io_uring_sqe *sqe = io_uring_get_sqe(&ioevent->ring);
+    if (sqe == NULL) {
+        logError("file: "__FILE__", line: %d, "
+                "io_uring_get_sqe fail", __LINE__);
+    }
+    return sqe;
+}
+
+static inline void ioevent_uring_prep_recv(IOEventPoller *ioevent,
+        struct io_uring_sqe *sqe, int sockfd,
+        void *buf, size_t size, void *user_data)
+{
+    io_uring_prep_recv(sqe, sockfd, buf, size, 0);
+    sqe->user_data = (long)user_data;
+    ioevent->submit_count++;
+}
+
+static inline void ioevent_uring_prep_send(IOEventPoller *ioevent,
+        struct io_uring_sqe *sqe, int sockfd,
+        void *buf, size_t len, void *user_data)
+{
+    io_uring_prep_send(sqe, sockfd, buf, len, 0);
+    sqe->user_data = (long)user_data;
+    ioevent->submit_count++;
+}
+
+static inline void ioevent_uring_prep_writev(IOEventPoller *ioevent,
+        struct io_uring_sqe *sqe, int sockfd, const struct iovec *iovecs,
+        unsigned nr_vecs, void *user_data)
+{
+    io_uring_prep_writev(sqe, sockfd, iovecs, nr_vecs, 0);
+    sqe->user_data = (long)user_data;
+    ioevent->submit_count++;
+}
+
+static inline void ioevent_uring_prep_send_zc(IOEventPoller *ioevent,
+        struct io_uring_sqe *sqe, int sockfd,
+        void *buf, size_t len, void *user_data)
+{
+    io_uring_prep_send_zc(sqe, sockfd, buf, len, 0,
+#ifdef IORING_SEND_ZC_REPORT_USAGE
+            IORING_SEND_ZC_REPORT_USAGE
+#else
+            0
+#endif
+            );
+    sqe->user_data = (long)user_data;
+    ioevent->submit_count++;
+}
+
+static inline void ioevent_uring_prep_close(IOEventPoller *ioevent,
+        struct io_uring_sqe *sqe, int fd, void *user_data)
+{
+    io_uring_prep_close(sqe, fd);
+    if (user_data == NULL) {
+        /* set sqe->flags MUST after io_uring_prep_xxx */
+        sqe->flags = IOSQE_CQE_SKIP_SUCCESS;
+    } else {
+        sqe->user_data = (long)user_data;
+    }
+    ioevent->submit_count++;
+}
+
+static inline void ioevent_uring_prep_cancel(IOEventPoller *ioevent,
+        struct io_uring_sqe *sqe, void *user_data)
+{
+    io_uring_prep_cancel(sqe, user_data, 0);
+    sqe->user_data = (long)user_data;
+    ioevent->submit_count++;
+}
+
+static inline void ioevent_uring_prep_connect(IOEventPoller *ioevent,
+        struct io_uring_sqe *sqe, int fd, const struct sockaddr *addr,
+        socklen_t addrlen, void *user_data)
+{
+    io_uring_prep_connect(sqe, fd, addr, addrlen);
+    sqe->user_data = (long)user_data;
+    ioevent->submit_count++;
+}
+
+#endif
+
 #ifdef __cplusplus
 }
 #endif

src/ioevent_loop.c

@@ -17,6 +17,76 @@
 #include "logger.h"
 #include "ioevent_loop.h"
 
+#if IOEVENT_USE_URING
+static int ioevent_process(IOEventPoller *ioevent)
+{
+    int result;
+    unsigned head;
+    unsigned count = 0;
+	IOEventEntry *pEntry;
+
+    result = io_uring_wait_cqe_timeout(&ioevent->ring,
+            &ioevent->cqe, &ioevent->timeout);
+    switch (result) {
+        case 0:
+            break;
+        case -ETIME:
+        case -EINTR:
+            return 0;
+        default:
+            result *= -1;
+            logError("file: "__FILE__", line: %d, "
+                    "io_uring_wait_cqe fail, errno: %d, error info: %s",
+                    __LINE__, result, STRERROR(result));
+            return result;
+    }
+
+    io_uring_for_each_cqe(&ioevent->ring, head, ioevent->cqe) {
+        count++;
+        pEntry = (IOEventEntry *)ioevent->cqe->user_data;
+        if (pEntry != NULL) {
+            if (ioevent->cqe->flags & IORING_CQE_F_NOTIF) {
+                if (ioevent->send_zc_done_notify) {
+                    pEntry->callback(pEntry->fd, IOEVENT_NOTIFY, pEntry);
+                }
+
+#ifdef IORING_NOTIF_USAGE_ZC_COPIED
+                if (!ioevent->send_zc_logged) {
+                    struct fast_task_info *task;
+
+                    task = (struct fast_task_info *)pEntry;
+                    ioevent->send_zc_logged = true;
+                    if (ioevent->cqe->res & IORING_NOTIF_USAGE_ZC_COPIED) {
+                        logWarning("file: "__FILE__", line: %d, %s "
+                                "client %s:%u, io_uring send_zc: memory "
+                                "copy instead of zero copy!", __LINE__,
+                                ioevent->service_name, task->client_ip,
+                                task->port);
+                    } else {
+                        logInfo("file: "__FILE__", line: %d, %s "
+                                "client %s:%u, io_uring send_zc: zero "
+                                "copy OK.", __LINE__, ioevent->service_name,
+                                task->client_ip, task->port);
+                    }
+                }
+#endif
+            } else {
+                pEntry->res = ioevent->cqe->res;
+                pEntry->callback(pEntry->fd, 0, pEntry);
+            }
+        } else {
+            logWarning("file: "__FILE__", line: %d, "
+                    "io_uring unexpected flags: %d, result: %d", __LINE__,
+                    ioevent->cqe->flags, ioevent->cqe->res);
+        }
+    }
+
+    io_uring_cq_advance(&ioevent->ring, count);
+    return 0;
+}
+
+#else
+
 static void deal_ioevents(IOEventPoller *ioevent)
 {
 	int event;
@@ -39,36 +109,28 @@ static void deal_ioevents(IOEventPoller *ioevent)
 	}
 }
 
-int ioevent_remove(IOEventPoller *ioevent, void *data)
+static int ioevent_process(IOEventPoller *ioevent)
 {
-	IOEventEntry *pEntry;
+    int result;
-    int index;
 
-    if (ioevent->iterator.index >= ioevent->iterator.count)
+    ioevent->iterator.count = ioevent_poll(ioevent);
-    {
+    if (ioevent->iterator.count > 0) {
-        return ENOENT;
+        deal_ioevents(ioevent);
+    }
+    else if (ioevent->iterator.count < 0) {
+        result = errno != 0 ? errno : EINVAL;
+        if (result != EINTR) {
+            logError("file: "__FILE__", line: %d, "
+                    "ioevent_poll fail, errno: %d, error info: %s",
+                    __LINE__, result, STRERROR(result));
+            return result;
+        }
     }
 
-    pEntry = (IOEventEntry *)IOEVENT_GET_DATA(ioevent,
-            ioevent->iterator.index);
-    if (pEntry != NULL && (void *)pEntry == data) {
-        return 0;  //do NOT clear current entry
-    }
-
-    for (index=ioevent->iterator.index + 1; index < ioevent->iterator.count;
-            index++)
-    {
-        pEntry = (IOEventEntry *)IOEVENT_GET_DATA(ioevent, index);
-        if (pEntry != NULL && (void *)pEntry == data) {
-            logDebug("file: "__FILE__", line: %d, "
-                    "clear ioevent data: %p", __LINE__, data);
-            IOEVENT_CLEAR_DATA(ioevent, index);
     return 0;
 }
-    }
 
-    return ENOENT;
+#endif
-}
 
 static void deal_timeouts(FastTimerEntry *head)
 {
@@ -82,7 +144,8 @@ static void deal_timeouts(FastTimerEntry *head)
 		current = entry;
 		entry = entry->next;
 
-        current->prev = current->next = NULL; //must set NULL because NOT in time wheel
+        /* must set NULL because NOT in time wheel */
+        current->prev = current->next = NULL;
 		pEventEntry = (IOEventEntry *)current;
 		if (pEventEntry != NULL)
 		{
@@ -131,10 +194,9 @@ int ioevent_loop(struct nio_thread_data *thread_data,
 
     thread_data->deleted_list = NULL;
 	last_check_time = g_current_time;
-	while (*continue_flag)
+	while (*continue_flag) {
-	{
 #ifdef OS_LINUX
-        if (thread_data->ev_puller.timeout == 0) {
+        if (thread_data->ev_puller.zero_timeout) {
             sched_pull = (sched_counter++ & 8) != 0;
         } else {
             sched_pull = true;
@@ -143,43 +205,34 @@ int ioevent_loop(struct nio_thread_data *thread_data,
         sched_pull = true;
 #endif
 
-        if (sched_pull)
+#if IOEVENT_USE_URING
-        {
+        if (thread_data->ev_puller.submit_count > 0) {
-            thread_data->ev_puller.iterator.count = ioevent_poll(
+            if ((result=ioevent_uring_submit(&thread_data->ev_puller)) != 0) {
-                    &thread_data->ev_puller);
+                logError("file: "__FILE__", line: %d, "
-            if (thread_data->ev_puller.iterator.count > 0)
+                        "io_uring_submit fail, errno: %d, error info: %s",
-            {
-                deal_ioevents(&thread_data->ev_puller);
-            }
-            else if (thread_data->ev_puller.iterator.count < 0)
-            {
-                result = errno != 0 ? errno : EINVAL;
-                if (result != EINTR)
-                {
-                    logError("file: "__FILE__", line: %d, " \
-                            "ioevent_poll fail, " \
-                            "errno: %d, error info: %s", \
                         __LINE__, result, STRERROR(result));
                 return result;
             }
         }
+#endif
+
+        if (sched_pull) {
+            if ((result=ioevent_process(&thread_data->ev_puller)) != 0) {
+                return result;
+            }
         }
 
-        if (thread_data->busy_polling_callback != NULL)
+        if (thread_data->busy_polling_callback != NULL) {
-        {
             thread_data->busy_polling_callback(thread_data);
         }
 
-		if (thread_data->deleted_list != NULL)
+		if (thread_data->deleted_list != NULL) {
-		{
 			count = 0;
-			while (thread_data->deleted_list != NULL)
+			while (thread_data->deleted_list != NULL) {
-			{
 				task = thread_data->deleted_list;
 				thread_data->deleted_list = task->next;
 
-                if (task->polling.in_queue)
+                if (task->polling.in_queue) {
-                {
                     fc_list_del_init(&task->polling.dlink);
                     task->polling.in_queue = false;
                     if (fc_list_empty(&task->thread_data->polling_queue)) {
@@ -193,8 +246,7 @@ int ioevent_loop(struct nio_thread_data *thread_data,
 			//logInfo("cleanup task count: %d", count);
 		}
 
-		if (g_current_time - last_check_time > 0)
+		if (g_current_time - last_check_time > 0) {
-		{
 			last_check_time = g_current_time;
 			count = fast_timer_timeouts_get(
 				&thread_data->timer, g_current_time, &head);
@@ -204,8 +256,7 @@ int ioevent_loop(struct nio_thread_data *thread_data,
 			}
 		}
 
-        if (thread_data->notify.enabled)
+        if (thread_data->notify.enabled) {
-        {
             int64_t n;
             if ((n=__sync_fetch_and_add(&thread_data->notify.counter, 0)) != 0)
             {
@@ -219,8 +270,7 @@ int ioevent_loop(struct nio_thread_data *thread_data,
             }
         }
 
-        if (thread_data->thread_loop_callback != NULL)
+        if (thread_data->thread_loop_callback != NULL) {
-        {
             thread_data->thread_loop_callback(thread_data);
         }
 	}
@@ -229,15 +279,36 @@ int ioevent_loop(struct nio_thread_data *thread_data,
 }
 
 int ioevent_set(struct fast_task_info *task, struct nio_thread_data *pThread,
-	int sock, short event, IOEventCallback callback, const int timeout)
+        int sock, short event, IOEventCallback callback,
+        const int timeout, const bool use_iouring)
 {
 	int result;
 
 	task->thread_data = pThread;
 	task->event.fd = sock;
 	task->event.callback = callback;
-	if (ioevent_attach(&pThread->ev_puller, sock, event, task) < 0)
+#if IOEVENT_USE_URING
-	{
+    if (use_iouring) {
+        if (FC_URING_OP_TYPE(task) == IORING_OP_NOP) {
+            if ((result=uring_prep_first_recv(task)) != 0) {
+                logError("file: "__FILE__", line: %d, "
+                        "uring_prep_recv fail, fd: %d, "
+                        "errno: %d, error info: %s",
+                        __LINE__, sock, result, STRERROR(result));
+                return result;
+            }
+        } else {
+            /*
+            logWarning("file: "__FILE__", line: %d, "
+                    "skip uring_prep_recv, fd: %d, port: %d, "
+                    "in progress op type: %d, timeout: %"PRId64,
+                    __LINE__, sock, task->port, FC_URING_OP_TYPE(task),
+                    task->event.timer.expires);
+                    */
+        }
+    } else {
+#endif
+        if (ioevent_attach(&pThread->ev_puller, sock, event, task) < 0) {
             result = errno != 0 ? errno : ENOENT;
             logError("file: "__FILE__", line: %d, "
                     "ioevent_attach fail, fd: %d, "
@@ -245,24 +316,11 @@ int ioevent_set(struct fast_task_info *task, struct nio_thread_data *pThread,
                     __LINE__, sock, result, STRERROR(result));
             return result;
         }
+#if IOEVENT_USE_URING
+    }
+#endif
 
 	task->event.timer.expires = g_current_time + timeout;
 	fast_timer_add(&pThread->timer, &task->event.timer);
 	return 0;
 }
-
-int ioevent_reset(struct fast_task_info *task, int new_fd, short event)
-{
-    if (task->event.fd == new_fd)
-    {
-        return 0;
-    }
-
-    if (task->event.fd >= 0)
-    {
-        ioevent_detach(&task->thread_data->ev_puller, task->event.fd);
-    }
-
-    task->event.fd = new_fd;
-    return ioevent_attach(&task->thread_data->ev_puller, new_fd, event, task);
-}

src/ioevent_loop.h

@@ -17,6 +17,14 @@
 #define _IOEVENT_LOOP_H
 
 #include "fast_task_queue.h"
+#if IOEVENT_USE_URING
+#include "sockopt.h"
+#endif
+
+#define fc_hold_task_ex(task, inc_count) __sync_add_and_fetch( \
+        &task->reffer_count, inc_count)
+
+#define fc_hold_task(task)  fc_hold_task_ex(task, 1)
 
 #ifdef __cplusplus
 extern "C" {
@@ -26,13 +34,9 @@ int ioevent_loop(struct nio_thread_data *thread_data,
 	IOEventCallback recv_notify_callback, TaskCleanUpCallback
 	clean_up_callback, volatile bool *continue_flag);
 
-//remove entry from ready list
+int ioevent_set(struct fast_task_info *task, struct nio_thread_data *pThread,
-int ioevent_remove(IOEventPoller *ioevent, void *data);
+        int sock, short event, IOEventCallback callback,
-
+        const int timeout, const bool use_iouring);
-int ioevent_set(struct fast_task_info *pTask, struct nio_thread_data *pThread,
-	int sock, short event, IOEventCallback callback, const int timeout);
-
-int ioevent_reset(struct fast_task_info *task, int new_fd, short event);
 
 static inline bool ioevent_is_canceled(struct fast_task_info *task)
 {
@@ -75,9 +79,170 @@ static inline int ioevent_notify_thread(struct nio_thread_data *thread_data)
     return 0;
 }
 
+#if IOEVENT_USE_URING
+
+#define SET_OP_TYPE_AND_HOLD_TASK(task, _op_type) \
+    struct io_uring_sqe *sqe; \
+    if ((sqe=ioevent_uring_get_sqe(&task->thread_data->ev_puller)) == NULL) { \
+        return ENOSPC; \
+    } \
+    FC_URING_OP_TYPE(task) = _op_type; \
+    fc_hold_task(task)
+
+static inline int uring_prep_recv_data(struct fast_task_info *task,
+        char *buff, const int len)
+{
+    SET_OP_TYPE_AND_HOLD_TASK(task, IORING_OP_RECV);
+    ioevent_uring_prep_recv(&task->thread_data->ev_puller,
+            sqe, task->event.fd, buff, len, task);
+    return 0;
+}
+
+static inline int uring_prep_first_recv(struct fast_task_info *task)
+{
+    SET_OP_TYPE_AND_HOLD_TASK(task, IORING_OP_RECV);
+    ioevent_uring_prep_recv(&task->thread_data->ev_puller,
+            sqe, task->event.fd, task->recv.ptr->data,
+            task->recv.ptr->size, task);
+    return 0;
+}
+
+static inline int uring_prep_next_recv(struct fast_task_info *task)
+{
+    SET_OP_TYPE_AND_HOLD_TASK(task, IORING_OP_RECV);
+    ioevent_uring_prep_recv(&task->thread_data->ev_puller, sqe,
+            task->event.fd, task->recv.ptr->data + task->recv.ptr->offset,
+            task->recv.ptr->length - task->recv.ptr->offset, task);
+    return 0;
+}
+
+static inline int uring_prep_first_send(struct fast_task_info *task)
+{
+    if (task->iovec_array.iovs != NULL) {
+        SET_OP_TYPE_AND_HOLD_TASK(task, IORING_OP_WRITEV);
+        ioevent_uring_prep_writev(&task->thread_data->ev_puller,
+                sqe, task->event.fd, task->iovec_array.iovs,
+                FC_MIN(task->iovec_array.count, IOV_MAX),
+                task);
+    } else {
+        SET_OP_TYPE_AND_HOLD_TASK(task, IORING_OP_SEND);
+        ioevent_uring_prep_send(&task->thread_data->ev_puller,
+                sqe, task->event.fd, task->send.ptr->data,
+                task->send.ptr->length, task);
+    }
+    return 0;
+}
+
+static inline int uring_prep_next_send(struct fast_task_info *task)
+{
+    if (task->iovec_array.iovs != NULL) {
+        SET_OP_TYPE_AND_HOLD_TASK(task, IORING_OP_WRITEV);
+        ioevent_uring_prep_writev(&task->thread_data->ev_puller,
+                sqe, task->event.fd, task->iovec_array.iovs,
+                FC_MIN(task->iovec_array.count, IOV_MAX),
+                task);
+    } else {
+        SET_OP_TYPE_AND_HOLD_TASK(task, IORING_OP_SEND);
+        ioevent_uring_prep_send(&task->thread_data->ev_puller, sqe,
+                task->event.fd, task->send.ptr->data + task->send.ptr->offset,
+                task->send.ptr->length - task->send.ptr->offset, task);
+    }
+    return 0;
+}
+
+static inline int uring_prep_first_send_zc(struct fast_task_info *task)
+{
+    if (task->iovec_array.iovs != NULL) {
+        SET_OP_TYPE_AND_HOLD_TASK(task, IORING_OP_WRITEV);
+        ioevent_uring_prep_writev(&task->thread_data->ev_puller,
+                sqe, task->event.fd, task->iovec_array.iovs,
+                FC_MIN(task->iovec_array.count, IOV_MAX),
+                task);
+    } else if (task->send.ptr->length < 4096) {
+        SET_OP_TYPE_AND_HOLD_TASK(task, IORING_OP_SEND);
+        ioevent_uring_prep_send(&task->thread_data->ev_puller,
+                sqe, task->event.fd, task->send.ptr->data,
+                task->send.ptr->length, task);
+    } else {
+        SET_OP_TYPE_AND_HOLD_TASK(task, IORING_OP_SEND_ZC);
+        ioevent_uring_prep_send_zc(&task->thread_data->ev_puller,
+                sqe, task->event.fd, task->send.ptr->data,
+                task->send.ptr->length, task);
+    }
+    return 0;
+}
+
+static inline int uring_prep_next_send_zc(struct fast_task_info *task)
+{
+    if (task->iovec_array.iovs != NULL) {
+        SET_OP_TYPE_AND_HOLD_TASK(task, IORING_OP_WRITEV);
+        ioevent_uring_prep_writev(&task->thread_data->ev_puller,
+                sqe, task->event.fd, task->iovec_array.iovs,
+                FC_MIN(task->iovec_array.count, IOV_MAX),
+                task);
+    } else if (task->send.ptr->length - task->send.ptr->offset < 4096) {
+        SET_OP_TYPE_AND_HOLD_TASK(task, IORING_OP_SEND);
+        ioevent_uring_prep_send(&task->thread_data->ev_puller, sqe,
+                task->event.fd, task->send.ptr->data + task->send.ptr->offset,
+                task->send.ptr->length - task->send.ptr->offset, task);
+    } else {
+        SET_OP_TYPE_AND_HOLD_TASK(task, IORING_OP_SEND_ZC);
+        ioevent_uring_prep_send_zc(&task->thread_data->ev_puller, sqe,
+                task->event.fd, task->send.ptr->data + task->send.ptr->offset,
+                task->send.ptr->length - task->send.ptr->offset, task);
+    }
+    return 0;
+}
+
+static inline int uring_prep_close_fd(struct fast_task_info *task)
+{
+    struct io_uring_sqe *sqe;
+
+    if ((sqe=ioevent_uring_get_sqe(&task->thread_data->ev_puller)) == NULL) {
+        return ENOSPC;
+    }
+
+    /* do NOT need callback */
+    ioevent_uring_prep_close(&task->thread_data->
+            ev_puller, sqe, task->event.fd, NULL);
+    return 0;
+}
+
+static inline int uring_prep_cancel(struct fast_task_info *task)
+{
+    SET_OP_TYPE_AND_HOLD_TASK(task, IORING_OP_ASYNC_CANCEL);
+    ioevent_uring_prep_cancel(&task->thread_data->ev_puller, sqe, task);
+    return 0;
+}
+
+static inline int uring_prep_connect(struct fast_task_info *task)
+{
+    int result;
+    sockaddr_convert_t *convert;
+
+    if ((task->event.fd=socketCreateEx2(AF_UNSPEC, task->server_ip,
+                    O_NONBLOCK, NULL, &result)) < 0)
+    {
+        return result;
+    }
+
+    convert = (sockaddr_convert_t *)(task->send.ptr->data +
+            task->send.ptr->size - 2 * sizeof(sockaddr_convert_t));
+    if ((result=setsockaddrbyip(task->server_ip, task->port, convert)) != 0) {
+        return result;
+    }
+
+    do {
+        SET_OP_TYPE_AND_HOLD_TASK(task, IORING_OP_CONNECT);
+        ioevent_uring_prep_connect(&task->thread_data->ev_puller, sqe,
+                task->event.fd, &convert->sa.addr, convert->len, task);
+    } while (0);
+    return 0;
+}
+#endif
+
 #ifdef __cplusplus
 }
 #endif
 
 #endif
-

src/multi_socket_client.c

@@ -65,8 +65,8 @@ int fast_multi_sock_client_init_ex(FastMultiSockClient *client,
         return EINVAL;
     }
 
-    if ((result=ioevent_init(&client->ioevent, entry_count,
+    if ((result=ioevent_init(&client->ioevent, "client",
-                    timeout_ms, 0)) != 0)
+                    entry_count, timeout_ms, 0)) != 0)
     {
         logError("file: "__FILE__", line: %d, "
                 "ioevent_init fail, errno: %d, error info: %s",
@@ -316,9 +316,13 @@ static int fast_multi_sock_client_do_recv(FastMultiSockClient *client,
 static int fast_multi_sock_client_deal_io(FastMultiSockClient *client)
 {
     int result;
-	int event;
     int count;
+#if IOEVENT_USE_URING
+    unsigned head;
+#else
+	int event;
     int index;
+#endif
     int remain_timeout;
     FastMultiSockEntry *entry;
     char formatted_ip[FORMATTED_IP_SIZE];
@@ -330,6 +334,37 @@ static int fast_multi_sock_client_deal_io(FastMultiSockClient *client)
             break;
         }
 
+#if IOEVENT_USE_URING
+        result = io_uring_wait_cqe_timeout(&client->ioevent.ring,
+                &client->ioevent.cqe, &client->ioevent.timeout);
+        switch (result) {
+            case 0:
+                break;
+            case -ETIME:
+            case -EAGAIN:
+            case -EINTR:
+                continue;
+            default:
+                result *= -1;
+                logError("file: "__FILE__", line: %d, "
+                        "io_uring_wait_cqe fail, errno: %d, error info: %s",
+                        __LINE__, result, STRERROR(result));
+                return result;
+        }
+
+        count = 0;
+        io_uring_for_each_cqe(&client->ioevent.ring, head, client->ioevent.cqe) {
+            count++;
+            entry = (FastMultiSockEntry *)client->ioevent.cqe->user_data;
+            //logInfo("sock: %d, event: %d", entry->conn->sock, event);
+            result = entry->io_callback(client, entry);
+            if (result != 0 || entry->remain == 0) {
+                fast_multi_sock_client_finish(client, entry, result);
+            }
+        }
+        io_uring_cq_advance(&client->ioevent.ring, count);
+
+#else
         count = ioevent_poll_ex(&client->ioevent, remain_timeout);
         //logInfo("poll count: %d\n", count);
         for (index=0; index<count; index++) {
@@ -354,6 +389,7 @@ static int fast_multi_sock_client_deal_io(FastMultiSockClient *client)
                 fast_multi_sock_client_finish(client, entry, result);
             }
         }
+#endif
     }
 
 /*

src/tests/Makefile.in

@@ -1,8 +1,8 @@
 .SUFFIXES: .c .o
 
 COMPILE = $(CC) -g -O3 -Wall -D_FILE_OFFSET_BITS=64 -D_GNU_SOURCE -g -DDEBUG_FLAG
-INC_PATH = -I/usr/local/include
+INC_PATH = $(INCS)
-LIB_PATH = -lfastcommon -lpthread
+LIB_PATH = -lfastcommon $(LIBS)
 
 ALL_PRGS = test_allocator test_skiplist test_multi_skiplist test_mblock test_blocked_queue \
            test_id_generator test_ini_parser test_char_convert test_char_convert_loader \