Compare commits

...

31 Commits

Author SHA1 Message Date
vazmin 795b328bd6 gh actions: upgrade to 1.0.83-1 2025-11-23 10:47:37 +00:00
vazmin ac5f4a584f gh actions: upgrade to 1.0.83-1 2025-11-23 10:00:00 +00:00
vazmin c044906e63 gh actions: upgrade to 1.0.83-1 2025-11-23 09:05:57 +00:00
YuQing ccc84945d3 INSTALL changed 2025-11-18 14:47:17 +08:00
YuQing c4b6f1fcb5 upgrade version to 1.0.83 2025-11-16 17:00:25 +08:00
YuQing 017ca1efe7 comment out useless codes 2025-11-16 16:54:16 +08:00
YuQing 71b2229427 fast_task_queue.h: remove field finish_callback 2025-11-15 11:13:15 +08:00
YuQing 446fa6b815 libfastcommon.spec: correct expr statement 2025-11-12 16:39:36 +08:00
YuQing 98948c11bf downgrade required liburing from V2.5 to V2.4 2025-11-12 10:28:44 +08:00
YuQing d60b141a21 upgrade version to 1.0.82 2025-11-09 19:00:41 +08:00
YuQing 86bab518c6 remove compile warning under RockLinux 10 2025-11-09 18:19:20 +08:00
YuQing 4b9ef52da2 fix compile warning 2025-11-05 10:47:44 +08:00
YuQing e8a9967801 set use_io_uring explicitly 2025-11-04 15:34:25 +08:00
YuQing 96c896b09a INSTALL changed for v1.0.81 2025-10-30 15:38:27 +08:00
YuQing 8d9feff6e2 restore function ioevent_reset 2025-10-29 11:18:56 +08:00
YuQing cbcd38a9af upgrade version to 1.0,81 2025-10-26 12:26:07 +08:00
YuQing cd55792a89 Merge remote-tracking branch 'origin/use_iouring' 2025-10-26 12:24:55 +08:00
YuQing 511b1066c4 add macro function fc_string_equals_ex 2025-10-24 11:59:43 +08:00
YuQing ddf6b5dfe9 send zc done notify callback for recycling buffer 2025-10-19 20:10:12 +08:00
YuQing 23cd03bc76 add macro IPV4_ADDRESS_SIZE and IPV6_ADDRESS_SIZE 2025-10-18 15:48:31 +08:00
YuQing 065184a203 batch ioevent_uring_submit for RDMA network 2025-10-07 19:52:48 +08:00
YuQing de80dc19dc struct ioevent_puller add field service_name 2025-10-06 20:53:14 +08:00
YuQing dac653d694 IOEventCallback: change event type from short to int 2025-10-05 09:44:24 +08:00
YuQing d5dbe3d030 free_queue support parameter: need_shrink and set task->shrinked 2025-10-03 21:03:31 +08:00
YuQing 7973d81b69 struct fast_task_info add fields: is_client and op_type for io_uring 2025-10-03 10:18:25 +08:00
YuQing 4576f22e24 add function uring_prep_connect 2025-09-30 10:12:41 +08:00
YuQing cb6f6f13f3 support Linux io_uring OK 2025-09-27 15:37:56 +08:00
YuQing 012b2038ee add functions uring_prep_xxx 2025-09-25 14:49:37 +08:00
YuQing 48a0ea2e30 ioevent_set support io_uring 2025-09-24 15:54:03 +08:00
YuQing aa48e3cd9a upgrade version in spec file 2025-09-20 17:12:30 +08:00
YuQing 47fa7f99df ioevent.[hc] and ioevent_loop.[hc] support io_uring 2025-09-17 03:49:21 +08:00
19 changed files with 878 additions and 215 deletions

1
.gitignore vendored
View File

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

11
HISTORY
View File

@ -1,4 +1,15 @@
Version 1.83 2025-11-15
* fast_task_queue.h: remove field finish_callback
Version 1.82 2025-11-04
* set use_io_uring explicitly
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 Version 1.80 2025-09-10
* getIpaddrByNameEx: IPv4 has priority over IPv6 * getIpaddrByNameEx: IPv4 has priority over IPv6
* shared_func.[hc]: add function fc_ftoa * shared_func.[hc]: add function fc_ftoa

13
INSTALL
View File

@ -6,11 +6,20 @@ Please visit the libfastcommon Home Page for more detail.
English language: https://github.com/happyfish100/libfastcommon English language: https://github.com/happyfish100/libfastcommon
Chinese language: http://www.fastken.com/ Chinese language: http://www.fastken.com/
[Optional Step]
You can enable io_uring for higher performance when Linux kernel version >= 6.2,
CentOS, RockyLinux, AlmaLinux, RHEL etc.:
sudo yum install liburing-devel -y
Debian, Ubuntu etc.:
sudo apt install liburing-dev -y
# download libfastcommon source codes and install it, # download libfastcommon source codes and install it,
# github address: https://github.com/happyfish100/libfastcommon.git # github address: https://github.com/happyfish100/libfastcommon.git
# gitee address: https://gitee.com/fastdfs100/libfastcommon.git # gitee address: https://gitee.com/fastdfs100/libfastcommon.git
# the command lines as: # the command lines as:
git clone https://github.com/happyfish100/libfastcommon.git git clone https://github.com/happyfish100/libfastcommon.git
cd libfastcommon; git checkout V1.0.72 cd libfastcommon; git checkout V1.0.83
./make.sh clean && ./make.sh && ./make.sh install ./make.sh clean && ./make.sh && sudo ./make.sh install

18
debian/changelog vendored
View File

@ -1,3 +1,21 @@
libfastcommon (1.0.83-1) unstable; urgency=medium
* upgrade to 1.0.83-1
-- YuQing <384681@qq.com> Sun, 23 Nov 2025 10:47:37 +0000
libfastcommon (1.0.83-1) unstable; urgency=medium
* upgrade to 1.0.83-1
-- YuQing <384681@qq.com> Sun, 23 Nov 2025 10:00:00 +0000
libfastcommon (1.0.83-1) unstable; urgency=medium
* upgrade to 1.0.83-1
-- YuQing <384681@qq.com> Sun, 23 Nov 2025 09:05:57 +0000
libfastcommon (1.0.78-1) unstable; urgency=medium libfastcommon (1.0.78-1) unstable; urgency=medium
* upgrade to 1.0.78-1 * upgrade to 1.0.78-1

View File

@ -3,7 +3,7 @@
%define CommitVersion %(echo $COMMIT_VERSION) %define CommitVersion %(echo $COMMIT_VERSION)
Name: libfastcommon Name: libfastcommon
Version: 1.0.79 Version: 1.0.83
Release: 1%{?dist} Release: 1%{?dist}
Summary: c common functions library extracted from my open source projects FastDFS Summary: c common functions library extracted from my open source projects FastDFS
License: LGPL License: LGPL
@ -17,6 +17,14 @@ BuildRequires: libcurl-devel
Requires: libcurl Requires: libcurl
Requires: %__cp %__mv %__chmod %__grep %__mkdir %__install %__id 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.4
Requires: liburing >= 2.4
%endif
%description %description
c common functions library extracted from my open source projects FastDFS. c common functions library extracted from my open source projects FastDFS.
this library is very simple and stable. functions including: string, logger, this library is very simple and stable. functions including: string, logger,

25
make.sh
View File

@ -112,7 +112,21 @@ HAVE_VMMETER_H=0
HAVE_USER_H=0 HAVE_USER_H=0
if [ "$uname" = "Linux" ]; then if [ "$uname" = "Linux" ]; then
OS_NAME=OS_LINUX OS_NAME=OS_LINUX
IOEVENT_USE=IOEVENT_USE_EPOLL
major_version=$(uname -r | awk -F . '{print $1;}')
minor_version=$(uname -r | awk -F . '{print $2;}')
if [ $major_version -eq 5 -a $minor_version -ge 14 ] || [ $major_version -gt 5 ]; then
out=$(grep -F IORING_OP_SEND_ZC /usr/include/liburing/io_uring.h 2>/dev/null)
if [ -n "$out" ]; then
IOEVENT_USE=IOEVENT_USE_URING
LIBS="$LIBS -luring"
else
IOEVENT_USE=IOEVENT_USE_EPOLL
fi
else
IOEVENT_USE=IOEVENT_USE_EPOLL
fi
if [ $glibc_minor -lt 17 ]; then if [ $glibc_minor -lt 17 ]; then
LIBS="$LIBS -lrt" LIBS="$LIBS -lrt"
fi fi
@ -261,3 +275,12 @@ make $1 $2 $3
if [ "$1" = "clean" ]; then if [ "$1" = "clean" ]; then
/bin/rm -f Makefile _os_define.h /bin/rm -f Makefile _os_define.h
fi 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

View File

@ -125,7 +125,8 @@ extern int pthread_mutexattr_settype(pthread_mutexattr_t *attr, int kind);
#define FC_IOV_BATCH_SIZE IOV_MAX #define FC_IOV_BATCH_SIZE IOV_MAX
#endif #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 IP_ADDRESS_SIZE INET6_ADDRSTRLEN //46
#define FORMATTED_IP_SIZE (IP_ADDRESS_SIZE + 2) #define FORMATTED_IP_SIZE (IP_ADDRESS_SIZE + 2)
#define INFINITE_FILE_SIZE (256 * 1024LL * 1024 * 1024 * 1024 * 1024LL) #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) 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); return (s1->len == s2->len) && (memcmp(s1->str, s2->str, s1->len) == 0);

View File

@ -25,10 +25,20 @@
#include "server_id_func.h" #include "server_id_func.h"
#include "connection_pool.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 = { ConnectionCallbacks g_connection_callbacks = {
false, {{conn_pool_connect_server_ex1, 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} {NULL, NULL, NULL}}, {NULL}
}; };
@ -411,20 +421,6 @@ void conn_pool_destroy(ConnectionPool *cp)
fast_mblock_destroy(&cp->node_allocator); 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, int conn_pool_connect_server_ex1(ConnectionInfo *conn,
const char *service_name, const int connect_timeout_ms, const char *service_name, const int connect_timeout_ms,
const char *bind_ipaddr, const bool log_connect_error) const char *bind_ipaddr, const bool log_connect_error)

View File

@ -331,9 +331,19 @@ int conn_pool_close_connection_ex(ConnectionPool *cp,
* conn: the connection * conn: the connection
* return 0 for success, != 0 for error * 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 * connect to the server

View File

@ -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, 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) #define MAX_DATA_SIZE (256 * 1024 * 1024)
int alloc_once; 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->double_buffers = double_buffers;
queue->need_shrink = need_shrink;
queue->min_buff_size = aligned_min_size; queue->min_buff_size = aligned_min_size;
queue->max_buff_size = aligned_max_size; queue->max_buff_size = aligned_max_size;
queue->padding_size = aligned_padding_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->length = 0;
task->send.ptr->offset = 0; task->send.ptr->offset = 0;
task->req_count = 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) { if (task->free_queue->double_buffers) {
task->recv.ptr->length = 0; task->recv.ptr->length = 0;
task->recv.ptr->offset = 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-> _realloc_buffer(task->recv.ptr, task->free_queue->
min_buff_size, false); min_buff_size, false);
task->shrinked = true;
} }
} }

View File

@ -37,21 +37,27 @@ struct nio_thread_data;
struct fast_task_info; struct fast_task_info;
typedef int (*ThreadLoopCallback) (struct nio_thread_data *pThreadData); typedef int (*ThreadLoopCallback) (struct nio_thread_data *pThreadData);
typedef int (*TaskFinishCallback) (struct fast_task_info *task);
typedef void (*TaskCleanUpCallback) (struct fast_task_info *task); typedef void (*TaskCleanUpCallback) (struct fast_task_info *task);
typedef int (*TaskInitCallback)(struct fast_task_info *task, void *arg); typedef int (*TaskInitCallback)(struct fast_task_info *task, void *arg);
typedef void (*TaskReleaseCallback)(struct fast_task_info *task); 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); typedef int (*TaskContinueCallback)(struct fast_task_info *task);
struct sf_network_handler; struct sf_network_handler;
struct fast_task_info; 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 typedef struct ioevent_entry
{ {
FastTimerEntry timer; //must first FastTimerEntry timer; //must first
int fd; int fd;
int res; //just for io_uring, since v1.0.81
IOEventCallback callback; IOEventCallback callback;
} IOEventEntry; } IOEventEntry;
@ -119,12 +125,19 @@ struct fast_task_info
struct fast_net_buffer_wrapper recv; //recv buffer struct fast_net_buffer_wrapper recv; //recv buffer
uint16_t port; //peer port uint16_t port; //peer port
struct {
int8_t is_client;
uint8_t op_type;
} uring; //just for io_uring, since v1.0.81
struct { struct {
uint8_t current; uint8_t current;
volatile uint8_t notify; volatile uint8_t notify;
} nio_stages; //stages for network IO } nio_stages; //stages for network IO
volatile int8_t reffer_count;
volatile int8_t canceled; //if task canceled 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; int pending_send_count;
int64_t req_count; //request count int64_t req_count; //request count
struct { struct {
@ -135,7 +148,6 @@ struct fast_task_info
struct fc_list_head dlink; //for polling queue struct fc_list_head dlink; //for polling queue
} polling; //for RDMA busy polling } polling; //for RDMA busy polling
TaskContinueCallback continue_callback; //for continue stage TaskContinueCallback continue_callback; //for continue stage
TaskFinishCallback finish_callback;
struct nio_thread_data *thread_data; struct nio_thread_data *thread_data;
struct sf_network_handler *handler; //network handler for libserverframe nio struct sf_network_handler *handler; //network handler for libserverframe nio
struct fast_task_info *next; //for free queue and deleted list struct fast_task_info *next; //for free queue and deleted list
@ -153,6 +165,7 @@ struct fast_task_queue
int block_size; int block_size;
bool malloc_whole_block; bool malloc_whole_block;
bool double_buffers; //if send buffer and recv buffer are independent bool double_buffers; //if send buffer and recv buffer are independent
bool need_shrink;
struct fast_mblock_man allocator; struct fast_mblock_man allocator;
TaskInitCallback init_callback; TaskInitCallback init_callback;
void *init_arg; void *init_arg;
@ -164,22 +177,22 @@ extern "C" {
#endif #endif
int free_queue_init_ex2(struct fast_task_queue *queue, const char *name, 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, static inline int free_queue_init_ex(struct fast_task_queue *queue,
const char *name, const bool double_buffers, 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; 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, static inline int free_queue_init(struct fast_task_queue *queue,
@ -188,9 +201,11 @@ static inline int free_queue_init(struct fast_task_queue *queue,
{ {
const char *name = ""; const char *name = "";
const bool double_buffers = false; const bool double_buffers = false;
const bool need_shrink = true;
const int arg_size = 0; 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( static inline void free_queue_set_release_callback(

View File

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

View File

@ -45,71 +45,119 @@ int kqueue_ev_convert(int16_t event, uint16_t flags)
} }
#endif #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 bool use_io_uring, const int size, const int timeout_ms,
const int extra_events)
{ {
int bytes; int bytes;
ioevent->size = size; ioevent->iterator.index = 0;
ioevent->extra_events = extra_events; ioevent->iterator.count = 0;
ioevent->iterator.index = 0; ioevent->service_name = service_name;
ioevent->iterator.count = 0; ioevent->size = size;
ioevent->extra_events = extra_events;
#if IOEVENT_USE_EPOLL #ifdef OS_LINUX
ioevent->poll_fd = epoll_create(ioevent->size); #if IOEVENT_USE_URING
if (ioevent->poll_fd < 0) { ioevent->use_io_uring = use_io_uring;
return errno != 0 ? errno : ENOMEM; if (use_io_uring) {
} int result;
bytes = sizeof(struct epoll_event) * size; if ((result=io_uring_queue_init(size, &ioevent->ring, 0)) < 0) {
ioevent->events = (struct epoll_event *)fc_malloc(bytes); return -result;
}
ioevent->cqe = NULL;
ioevent->submit_count = 0;
ioevent->send_zc_logged = false;
ioevent->send_zc_done_notify = false;
} else {
#endif
ioevent->poll_fd = epoll_create(ioevent->size);
if (ioevent->poll_fd < 0) {
return errno != 0 ? errno : ENOMEM;
}
bytes = sizeof(struct epoll_event) * size;
ioevent->events = (struct epoll_event *)fc_malloc(bytes);
#if IOEVENT_USE_URING
}
#endif
#elif IOEVENT_USE_KQUEUE #elif IOEVENT_USE_KQUEUE
ioevent->poll_fd = kqueue(); ioevent->poll_fd = kqueue();
if (ioevent->poll_fd < 0) { if (ioevent->poll_fd < 0) {
return errno != 0 ? errno : ENOMEM; return errno != 0 ? errno : ENOMEM;
} }
bytes = sizeof(struct kevent) * size; bytes = sizeof(struct kevent) * size;
ioevent->events = (struct kevent *)fc_malloc(bytes); ioevent->events = (struct kevent *)fc_malloc(bytes);
#elif IOEVENT_USE_PORT #elif IOEVENT_USE_PORT
ioevent->poll_fd = port_create(); ioevent->poll_fd = port_create();
if (ioevent->poll_fd < 0) { if (ioevent->poll_fd < 0) {
return errno != 0 ? errno : ENOMEM; return errno != 0 ? errno : ENOMEM;
} }
bytes = sizeof(port_event_t) * size; bytes = sizeof(port_event_t) * size;
ioevent->events = (port_event_t *)fc_malloc(bytes); ioevent->events = (port_event_t *)fc_malloc(bytes);
#endif #endif
if (ioevent->events == NULL) { #if IOEVENT_USE_URING
close(ioevent->poll_fd); if (!ioevent->use_io_uring) {
ioevent->poll_fd = -1; #endif
return ENOMEM; if (ioevent->events == NULL) {
} close(ioevent->poll_fd);
ioevent_set_timeout(ioevent, timeout_ms); ioevent->poll_fd = -1;
return ENOMEM;
}
#if IOEVENT_USE_URING
}
#endif
return 0; ioevent_set_timeout(ioevent, timeout_ms);
return 0;
} }
void ioevent_destroy(IOEventPoller *ioevent) void ioevent_destroy(IOEventPoller *ioevent)
{ {
if (ioevent->events != NULL) { #if IOEVENT_USE_URING
free(ioevent->events); if (ioevent->use_io_uring) {
ioevent->events = NULL; io_uring_queue_exit(&ioevent->ring);
} } else {
#endif
if (ioevent->events != NULL) {
free(ioevent->events);
ioevent->events = NULL;
}
if (ioevent->poll_fd >= 0) { if (ioevent->poll_fd >= 0) {
close(ioevent->poll_fd); close(ioevent->poll_fd);
ioevent->poll_fd = -1; ioevent->poll_fd = -1;
} }
#if IOEVENT_USE_URING
}
#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 #ifdef OS_LINUX
struct epoll_event ev; #if IOEVENT_USE_URING
memset(&ev, 0, sizeof(ev)); if (ioevent->use_io_uring) {
ev.events = e | ioevent->extra_events; struct io_uring_sqe *sqe = io_uring_get_sqe(&ioevent->ring);
ev.data.ptr = data; if (sqe == NULL) {
return epoll_ctl(ioevent->poll_fd, EPOLL_CTL_ADD, fd, &ev); return ENOSPC;
}
io_uring_prep_poll_multishot(sqe, fd, e | ioevent->extra_events);
sqe->user_data = (long)data;
ioevent->submit_count++;
return 0;
} else {
#endif
struct epoll_event ev;
memset(&ev, 0, sizeof(ev));
ev.events = e | ioevent->extra_events;
ev.data.ptr = data;
return epoll_ctl(ioevent->poll_fd, EPOLL_CTL_ADD, fd, &ev);
#if IOEVENT_USE_URING
}
#endif
#elif IOEVENT_USE_KQUEUE #elif IOEVENT_USE_KQUEUE
struct kevent ev[2]; struct kevent ev[2];
int n = 0; int n = 0;
@ -128,15 +176,32 @@ int ioevent_attach(IOEventPoller *ioevent, const int fd, const int e,
#endif #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 #ifdef OS_LINUX
struct epoll_event ev; #if IOEVENT_USE_URING
memset(&ev, 0, sizeof(ev)); if (ioevent->use_io_uring) {
ev.events = e | ioevent->extra_events; struct io_uring_sqe *sqe = io_uring_get_sqe(&ioevent->ring);
ev.data.ptr = data; if (sqe == NULL) {
return epoll_ctl(ioevent->poll_fd, EPOLL_CTL_MOD, fd, &ev); 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;
} else {
#endif
struct epoll_event ev;
memset(&ev, 0, sizeof(ev));
ev.events = e | ioevent->extra_events;
ev.data.ptr = data;
return epoll_ctl(ioevent->poll_fd, EPOLL_CTL_MOD, fd, &ev);
#if IOEVENT_USE_URING
}
#endif
#elif IOEVENT_USE_KQUEUE #elif IOEVENT_USE_KQUEUE
struct kevent ev[2]; struct kevent ev[2];
int result; int result;
@ -171,8 +236,25 @@ int ioevent_modify(IOEventPoller *ioevent, const int fd, const int e,
int ioevent_detach(IOEventPoller *ioevent, const int fd) int ioevent_detach(IOEventPoller *ioevent, const int fd)
{ {
#if IOEVENT_USE_EPOLL #ifdef OS_LINUX
return epoll_ctl(ioevent->poll_fd, EPOLL_CTL_DEL, fd, NULL); #if IOEVENT_USE_URING
if (ioevent->use_io_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;
} else {
#endif
return epoll_ctl(ioevent->poll_fd, EPOLL_CTL_DEL, fd, NULL);
#if IOEVENT_USE_URING
}
#endif
#elif IOEVENT_USE_KQUEUE #elif IOEVENT_USE_KQUEUE
struct kevent ev[1]; struct kevent ev[1];
int r, w; int r, w;
@ -191,15 +273,33 @@ int ioevent_detach(IOEventPoller *ioevent, const int fd)
int ioevent_poll(IOEventPoller *ioevent) int ioevent_poll(IOEventPoller *ioevent)
{ {
#if IOEVENT_USE_EPOLL #ifdef OS_LINUX
return epoll_wait(ioevent->poll_fd, ioevent->events, ioevent->size, ioevent->timeout); #if IOEVENT_USE_URING
if (ioevent->use_io_uring) {
int result;
result = io_uring_wait_cqe_timeout(&ioevent->ring,
&ioevent->cqe, &ioevent->timeout);
if (result < 0) {
errno = -result;
return -1;
}
return 0;
} else {
#endif
return epoll_wait(ioevent->poll_fd, ioevent->events,
ioevent->size, ioevent->timeout_ms);
#if IOEVENT_USE_URING
}
#endif
#elif IOEVENT_USE_KQUEUE #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 #elif IOEVENT_USE_PORT
int result; int result;
int retval; int retval;
unsigned int nget = 1; unsigned int nget = 1;
if((retval = port_getn(ioevent->poll_fd, ioevent->events, if((retval=port_getn(ioevent->poll_fd, ioevent->events,
ioevent->size, &nget, &ioevent->timeout)) == 0) ioevent->size, &nget, &ioevent->timeout)) == 0)
{ {
result = (int)nget; result = (int)nget;
@ -225,4 +325,3 @@ int ioevent_poll(IOEventPoller *ioevent)
#error port me #error port me
#endif #endif
} }

View File

@ -20,17 +20,28 @@
#include <poll.h> #include <poll.h>
#include <sys/time.h> #include <sys/time.h>
#include "_os_define.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 #ifdef OS_LINUX
#include <sys/epoll.h> #include <sys/epoll.h>
#define IOEVENT_EDGE_TRIGGER EPOLLET #define IOEVENT_EDGE_TRIGGER EPOLLET
#endif
#if IOEVENT_USE_EPOLL
#define IOEVENT_READ EPOLLIN #define IOEVENT_READ EPOLLIN
#define IOEVENT_WRITE EPOLLOUT #define IOEVENT_WRITE EPOLLOUT
#define IOEVENT_ERROR (EPOLLERR | EPOLLPRI | EPOLLHUP) #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 #elif IOEVENT_USE_KQUEUE
#include <sys/event.h> #include <sys/event.h>
#include <sys/poll.h> #include <sys/poll.h>
@ -65,18 +76,33 @@ int kqueue_ev_convert(int16_t event, uint16_t flags);
#endif #endif
typedef struct ioevent_puller { typedef struct ioevent_puller {
const char *service_name;
int size; //max events (fd) int size; //max events (fd)
int extra_events; int extra_events;
int poll_fd;
#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
bool use_io_uring;
#endif
int poll_fd;
struct { struct {
int index; int index;
int count; int count;
} iterator; //for deal event loop } iterator; //for deal event loop
#if IOEVENT_USE_EPOLL #ifdef OS_LINUX
struct epoll_event *events; struct epoll_event *events;
int timeout; int timeout_ms; //for epoll
#if IOEVENT_USE_URING
struct io_uring_cqe *cqe;
struct __kernel_timespec timeout;
#endif
bool zero_timeout;
#elif IOEVENT_USE_KQUEUE #elif IOEVENT_USE_KQUEUE
struct kevent *events; struct kevent *events;
struct timespec timeout; struct timespec timeout;
@ -84,9 +110,10 @@ typedef struct ioevent_puller {
port_event_t *events; port_event_t *events;
timespec_t timeout; timespec_t timeout;
#endif #endif
} IOEventPoller; } IOEventPoller;
#if IOEVENT_USE_EPOLL #if OS_LINUX
#define IOEVENT_GET_EVENTS(ioevent, index) \ #define IOEVENT_GET_EVENTS(ioevent, index) \
(ioevent)->events[index].events (ioevent)->events[index].events
#elif IOEVENT_USE_KQUEUE #elif IOEVENT_USE_KQUEUE
@ -99,7 +126,7 @@ typedef struct ioevent_puller {
#error port me #error port me
#endif #endif
#if IOEVENT_USE_EPOLL #ifdef OS_LINUX
#define IOEVENT_GET_DATA(ioevent, index) \ #define IOEVENT_GET_DATA(ioevent, index) \
(ioevent)->events[index].data.ptr (ioevent)->events[index].data.ptr
#elif IOEVENT_USE_KQUEUE #elif IOEVENT_USE_KQUEUE
@ -112,7 +139,7 @@ typedef struct ioevent_puller {
#error port me #error port me
#endif #endif
#if IOEVENT_USE_EPOLL #ifdef OS_LINUX
#define IOEVENT_CLEAR_DATA(ioevent, index) \ #define IOEVENT_CLEAR_DATA(ioevent, index) \
(ioevent)->events[index].data.ptr = NULL (ioevent)->events[index].data.ptr = NULL
#elif IOEVENT_USE_KQUEUE #elif IOEVENT_USE_KQUEUE
@ -129,24 +156,39 @@ typedef struct ioevent_puller {
extern "C" { extern "C" {
#endif #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 bool use_io_uring, const int size, const int timeout_ms,
const int extra_events);
void ioevent_destroy(IOEventPoller *ioevent); 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_detach(IOEventPoller *ioevent, const int fd);
int ioevent_poll(IOEventPoller *ioevent); 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 #if IOEVENT_USE_EPOLL
ioevent->timeout = timeout_ms; ioevent->timeout_ms = timeout_ms;
#else #else
ioevent->timeout.tv_sec = timeout_ms / 1000; #if IOEVENT_USE_URING
ioevent->timeout.tv_nsec = 1000000 * (timeout_ms % 1000); if (!ioevent->use_io_uring) {
ioevent->timeout_ms = timeout_ms;
} else {
#endif
ioevent->timeout.tv_sec = timeout_ms / 1000;
ioevent->timeout.tv_nsec = 1000000 * (timeout_ms % 1000);
#if IOEVENT_USE_URING
}
#endif
#endif
#ifdef OS_LINUX
ioevent->zero_timeout = (timeout_ms == 0);
#endif #endif
} }
@ -156,6 +198,114 @@ static inline int ioevent_poll_ex(IOEventPoller *ioevent, const int timeout_ms)
return ioevent_poll(ioevent); 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 #ifdef __cplusplus
} }
#endif #endif

View File

@ -17,6 +17,76 @@
#include "logger.h" #include "logger.h"
#include "ioevent_loop.h" #include "ioevent_loop.h"
#if IOEVENT_USE_URING
static int ioevent_process_by_uring(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;
}
#endif
static void deal_ioevents(IOEventPoller *ioevent) static void deal_ioevents(IOEventPoller *ioevent)
{ {
int event; int event;
@ -39,35 +109,25 @@ static void deal_ioevents(IOEventPoller *ioevent)
} }
} }
int ioevent_remove(IOEventPoller *ioevent, void *data) static int ioevent_process_by_poll(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) {
pEntry = (IOEventEntry *)IOEVENT_GET_DATA(ioevent, result = errno != 0 ? errno : EINVAL;
ioevent->iterator.index); if (result != EINTR) {
if (pEntry != NULL && (void *)pEntry == data) { logError("file: "__FILE__", line: %d, "
return 0; //do NOT clear current entry "ioevent_poll fail, errno: %d, error info: %s",
} __LINE__, result, STRERROR(result));
return result;
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; return 0;
} }
static void deal_timeouts(FastTimerEntry *head) static void deal_timeouts(FastTimerEntry *head)
@ -82,12 +142,10 @@ static void deal_timeouts(FastTimerEntry *head)
current = entry; current = entry;
entry = entry->next; 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; pEventEntry = (IOEventEntry *)current;
if (pEventEntry != NULL) pEventEntry->callback(pEventEntry->fd, IOEVENT_TIMEOUT, current);
{
pEventEntry->callback(pEventEntry->fd, IOEVENT_TIMEOUT, current);
}
} }
} }
@ -131,10 +189,9 @@ int ioevent_loop(struct nio_thread_data *thread_data,
thread_data->deleted_list = NULL; thread_data->deleted_list = NULL;
last_check_time = g_current_time; last_check_time = g_current_time;
while (*continue_flag) while (*continue_flag) {
{
#ifdef OS_LINUX #ifdef OS_LINUX
if (thread_data->ev_puller.timeout == 0) { if (thread_data->ev_puller.zero_timeout) {
sched_pull = (sched_counter++ & 8) != 0; sched_pull = (sched_counter++ & 8) != 0;
} else { } else {
sched_pull = true; sched_pull = true;
@ -143,43 +200,52 @@ int ioevent_loop(struct nio_thread_data *thread_data,
sched_pull = true; sched_pull = true;
#endif #endif
if (sched_pull) #if IOEVENT_USE_URING
{ if (thread_data->ev_puller.use_io_uring) {
thread_data->ev_puller.iterator.count = ioevent_poll( if (thread_data->ev_puller.submit_count > 0) {
&thread_data->ev_puller); if ((result=ioevent_uring_submit(&thread_data->
if (thread_data->ev_puller.iterator.count > 0) ev_puller)) != 0)
{
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, " \ logError("file: "__FILE__", line: %d, "
"ioevent_poll fail, " \ "io_uring_submit fail, errno: %d, error info: %s",
"errno: %d, error info: %s", \
__LINE__, result, STRERROR(result)); __LINE__, result, STRERROR(result));
return result; return result;
} }
} }
} }
#endif
if (thread_data->busy_polling_callback != NULL) if (sched_pull) {
{ #if IOEVENT_USE_URING
if (thread_data->ev_puller.use_io_uring) {
if ((result=ioevent_process_by_uring(&thread_data->
ev_puller)) != 0)
{
return result;
}
} else {
#endif
if ((result=ioevent_process_by_poll(&thread_data->
ev_puller)) != 0)
{
return result;
}
#if IOEVENT_USE_URING
}
#endif
}
if (thread_data->busy_polling_callback != NULL) {
thread_data->busy_polling_callback(thread_data); thread_data->busy_polling_callback(thread_data);
} }
if (thread_data->deleted_list != NULL) if (thread_data->deleted_list != NULL) {
{ //count = 0;
count = 0; while (thread_data->deleted_list != NULL) {
while (thread_data->deleted_list != NULL)
{
task = thread_data->deleted_list; task = thread_data->deleted_list;
thread_data->deleted_list = task->next; thread_data->deleted_list = task->next;
if (task->polling.in_queue) if (task->polling.in_queue) {
{
fc_list_del_init(&task->polling.dlink); fc_list_del_init(&task->polling.dlink);
task->polling.in_queue = false; task->polling.in_queue = false;
if (fc_list_empty(&task->thread_data->polling_queue)) { if (fc_list_empty(&task->thread_data->polling_queue)) {
@ -188,13 +254,12 @@ int ioevent_loop(struct nio_thread_data *thread_data,
} }
} }
clean_up_callback(task); clean_up_callback(task);
count++; //count++;
} }
//logInfo("cleanup task count: %d", count); //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; last_check_time = g_current_time;
count = fast_timer_timeouts_get( count = fast_timer_timeouts_get(
&thread_data->timer, g_current_time, &head); &thread_data->timer, g_current_time, &head);
@ -204,8 +269,7 @@ int ioevent_loop(struct nio_thread_data *thread_data,
} }
} }
if (thread_data->notify.enabled) if (thread_data->notify.enabled) {
{
int64_t n; int64_t n;
if ((n=__sync_fetch_and_add(&thread_data->notify.counter, 0)) != 0) if ((n=__sync_fetch_and_add(&thread_data->notify.counter, 0)) != 0)
{ {
@ -219,8 +283,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); thread_data->thread_loop_callback(thread_data);
} }
} }
@ -229,22 +292,46 @@ int ioevent_loop(struct nio_thread_data *thread_data,
} }
int ioevent_set(struct fast_task_info *task, struct nio_thread_data *pThread, 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)
{ {
int result; int result;
task->thread_data = pThread; task->thread_data = pThread;
task->event.fd = sock; task->event.fd = sock;
task->event.callback = callback; task->event.callback = callback;
if (ioevent_attach(&pThread->ev_puller, sock, event, task) < 0) #if IOEVENT_USE_URING
{ if (pThread->ev_puller.use_io_uring) {
result = errno != 0 ? errno : ENOENT; if (FC_URING_OP_TYPE(task) == IORING_OP_NOP) {
logError("file: "__FILE__", line: %d, " if ((result=uring_prep_first_recv(task)) != 0) {
"ioevent_attach fail, fd: %d, " logError("file: "__FILE__", line: %d, "
"errno: %d, error info: %s", "uring_prep_recv fail, fd: %d, "
__LINE__, sock, result, STRERROR(result)); "errno: %d, error info: %s",
return result; __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, "
"errno: %d, error info: %s",
__LINE__, sock, result, STRERROR(result));
return result;
}
#if IOEVENT_USE_URING
}
#endif
task->event.timer.expires = g_current_time + timeout; task->event.timer.expires = g_current_time + timeout;
fast_timer_add(&pThread->timer, &task->event.timer); fast_timer_add(&pThread->timer, &task->event.timer);

View File

@ -17,6 +17,14 @@
#define _IOEVENT_LOOP_H #define _IOEVENT_LOOP_H
#include "fast_task_queue.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 #ifdef __cplusplus
extern "C" { extern "C" {
@ -26,11 +34,9 @@ int ioevent_loop(struct nio_thread_data *thread_data,
IOEventCallback recv_notify_callback, TaskCleanUpCallback IOEventCallback recv_notify_callback, TaskCleanUpCallback
clean_up_callback, volatile bool *continue_flag); 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);
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); int ioevent_reset(struct fast_task_info *task, int new_fd, short event);
@ -75,9 +81,170 @@ static inline int ioevent_notify_thread(struct nio_thread_data *thread_data)
return 0; 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 #ifdef __cplusplus
} }
#endif #endif
#endif #endif

View File

@ -46,6 +46,7 @@ int fast_multi_sock_client_init_ex(FastMultiSockClient *client,
fms_client_get_current_time_ms_func get_current_time_ms_func, fms_client_get_current_time_ms_func get_current_time_ms_func,
const int init_recv_buffer_size, const int timeout_ms) const int init_recv_buffer_size, const int timeout_ms)
{ {
const bool use_io_uring = false;
int result; int result;
int new_init_recv_buffer_size; int new_init_recv_buffer_size;
int i; int i;
@ -65,8 +66,8 @@ int fast_multi_sock_client_init_ex(FastMultiSockClient *client,
return EINVAL; return EINVAL;
} }
if ((result=ioevent_init(&client->ioevent, entry_count, if ((result=ioevent_init(&client->ioevent, "client",
timeout_ms, 0)) != 0) use_io_uring, entry_count, timeout_ms, 0)) != 0)
{ {
logError("file: "__FILE__", line: %d, " logError("file: "__FILE__", line: %d, "
"ioevent_init fail, errno: %d, error info: %s", "ioevent_init fail, errno: %d, error info: %s",
@ -316,9 +317,13 @@ static int fast_multi_sock_client_do_recv(FastMultiSockClient *client,
static int fast_multi_sock_client_deal_io(FastMultiSockClient *client) static int fast_multi_sock_client_deal_io(FastMultiSockClient *client)
{ {
int result; int result;
int event;
int count; int count;
#if IOEVENT_USE_URING
unsigned head;
#else
int event;
int index; int index;
#endif
int remain_timeout; int remain_timeout;
FastMultiSockEntry *entry; FastMultiSockEntry *entry;
char formatted_ip[FORMATTED_IP_SIZE]; char formatted_ip[FORMATTED_IP_SIZE];
@ -330,6 +335,37 @@ static int fast_multi_sock_client_deal_io(FastMultiSockClient *client)
break; 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); count = ioevent_poll_ex(&client->ioevent, remain_timeout);
//logInfo("poll count: %d\n", count); //logInfo("poll count: %d\n", count);
for (index=0; index<count; index++) { for (index=0; index<count; index++) {
@ -354,6 +390,7 @@ static int fast_multi_sock_client_deal_io(FastMultiSockClient *client)
fast_multi_sock_client_finish(client, entry, result); fast_multi_sock_client_finish(client, entry, result);
} }
} }
#endif
} }
/* /*

View File

@ -1636,16 +1636,29 @@ static inline int fc_ltostr_ex(int64_t n, char *buff, const int padding_len)
return len; return len;
} }
#if defined(OS_LINUX) && defined(__GNUC__) && __GNUC__ >= 8
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wstringop-overflow"
#endif
fill_len = padding_len - len; fill_len = padding_len - len;
memmove(buff + fill_len, buff, len + 1); memmove(buff + fill_len, buff, len + 1);
memset(buff, '0', fill_len); memset(buff, '0', fill_len);
#if defined(OS_LINUX) && defined(__GNUC__) && __GNUC__ >= 8
#pragma GCC diagnostic pop
#endif
return padding_len; return padding_len;
} }
static inline int fc_ltostr(int64_t n, char *buff) static inline int fc_ltostr(int64_t n, char *buff)
{ {
const int padding_len = 0; int len;
return fc_ltostr_ex(n, buff, padding_len);
len = fc_itoa(n, buff);
*(buff + len) = '\0';
return len;
} }
static inline size_t fc_strlcpy(char *dest, const char *src, const size_t size) static inline size_t fc_strlcpy(char *dest, const char *src, const size_t size)

View File

@ -1,8 +1,8 @@
.SUFFIXES: .c .o .SUFFIXES: .c .o
COMPILE = $(CC) -g -O3 -Wall -D_FILE_OFFSET_BITS=64 -D_GNU_SOURCE -g -DDEBUG_FLAG 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 \ 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 \ test_id_generator test_ini_parser test_char_convert test_char_convert_loader \