set use_io_uring explicitly

master
YuQing 2025-11-04 15:34:25 +08:00
parent 96c896b09a
commit e8a9967801
6 changed files with 198 additions and 132 deletions

View File

@ -1,4 +1,7 @@
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

View File

@ -46,36 +46,42 @@ int kqueue_ev_convert(int16_t event, uint16_t flags)
#endif
int ioevent_init(IOEventPoller *ioevent, const char *service_name,
const int size, const int timeout_ms, const int extra_events)
const bool use_io_uring, const int size, const int timeout_ms,
const int extra_events)
{
#if IOEVENT_USE_URING
int result;
#else
int bytes;
#ifdef OS_LINUX
int result;
#endif
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);
if (ioevent->poll_fd < 0) {
return errno != 0 ? errno : ENOMEM;
#ifdef OS_LINUX
#if IOEVENT_USE_URING
ioevent->use_io_uring = use_io_uring;
if (use_io_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;
} 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
}
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;
#endif
#elif IOEVENT_USE_KQUEUE
ioevent->poll_fd = kqueue();
if (ioevent->poll_fd < 0) {
@ -93,12 +99,14 @@ int ioevent_init(IOEventPoller *ioevent, const char *service_name,
#endif
#if IOEVENT_USE_URING
#else
if (ioevent->events == NULL) {
close(ioevent->poll_fd);
ioevent->poll_fd = -1;
return ENOMEM;
if (!ioevent->use_io_uring) {
#endif
if (ioevent->events == NULL) {
close(ioevent->poll_fd);
ioevent->poll_fd = -1;
return ENOMEM;
}
#if IOEVENT_USE_URING
}
#endif
@ -109,38 +117,49 @@ int ioevent_init(IOEventPoller *ioevent, const char *service_name,
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;
}
if (ioevent->use_io_uring) {
io_uring_queue_exit(&ioevent->ring);
} else {
#endif
if (ioevent->events != NULL) {
free(ioevent->events);
ioevent->events = NULL;
}
if (ioevent->poll_fd >= 0) {
close(ioevent->poll_fd);
ioevent->poll_fd = -1;
}
if (ioevent->poll_fd >= 0) {
close(ioevent->poll_fd);
ioevent->poll_fd = -1;
}
#if IOEVENT_USE_URING
}
#endif
}
int ioevent_attach(IOEventPoller *ioevent, const int fd,
const int e, void *data)
{
#if IOEVENT_USE_EPOLL
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);
#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;
#ifdef OS_LINUX
#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_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
struct kevent ev[2];
int n = 0;
@ -162,22 +181,29 @@ int ioevent_attach(IOEventPoller *ioevent, const int fd,
int ioevent_modify(IOEventPoller *ioevent, const int fd,
const int e, void *data)
{
#if IOEVENT_USE_EPOLL
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);
#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;
#ifdef OS_LINUX
#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_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
struct kevent ev[2];
int result;
@ -212,18 +238,25 @@ int ioevent_modify(IOEventPoller *ioevent, const int fd,
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;
#ifdef OS_LINUX
#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
struct kevent ev[1];
int r, w;
@ -242,18 +275,25 @@ 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);
#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;
#ifdef OS_LINUX
#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
return kevent(ioevent->poll_fd, NULL, 0, ioevent->events,
ioevent->size, &ioevent->timeout);

View File

@ -25,10 +25,12 @@
#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>
#define IOEVENT_EDGE_TRIGGER EPOLLET
#endif
#if IOEVENT_USE_EPOLL
#define IOEVENT_READ EPOLLIN
#define IOEVENT_WRITE EPOLLOUT
#define IOEVENT_ERROR (EPOLLERR | EPOLLPRI | EPOLLHUP)
@ -77,25 +79,30 @@ 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
bool use_io_uring;
#endif
int poll_fd;
struct {
int index;
int count;
} iterator; //for deal event loop
#endif
#if IOEVENT_USE_EPOLL
#ifdef OS_LINUX
struct epoll_event *events;
int timeout; //in milliseconds
#elif IOEVENT_USE_URING
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
struct kevent *events;
struct timespec timeout;
@ -104,17 +111,11 @@ typedef struct ioevent_puller {
timespec_t timeout;
#endif
#ifdef OS_LINUX
bool zero_timeout;
#endif
} IOEventPoller;
#if IOEVENT_USE_EPOLL
#if OS_LINUX
#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)
@ -125,11 +126,9 @@ typedef struct ioevent_puller {
#error port me
#endif
#if IOEVENT_USE_EPOLL
#ifdef OS_LINUX
#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
@ -140,11 +139,9 @@ typedef struct ioevent_puller {
#error port me
#endif
#if IOEVENT_USE_EPOLL
#ifdef OS_LINUX
#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
@ -160,7 +157,8 @@ extern "C" {
#endif
int ioevent_init(IOEventPoller *ioevent, const char *service_name,
const int size, 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);
int ioevent_attach(IOEventPoller *ioevent, const int fd,
@ -174,16 +172,24 @@ static inline void ioevent_set_timeout(IOEventPoller *ioevent,
const int timeout_ms)
{
#if IOEVENT_USE_EPOLL
ioevent->timeout = timeout_ms;
ioevent->timeout_ms = timeout_ms;
#else
ioevent->timeout.tv_sec = timeout_ms / 1000;
ioevent->timeout.tv_nsec = 1000000 * (timeout_ms % 1000);
#if IOEVENT_USE_URING
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
}
static inline int ioevent_poll_ex(IOEventPoller *ioevent, const int timeout_ms)

View File

@ -18,7 +18,7 @@
#include "ioevent_loop.h"
#if IOEVENT_USE_URING
static int ioevent_process(IOEventPoller *ioevent)
static int ioevent_process_by_uring(IOEventPoller *ioevent)
{
int result;
unsigned head;
@ -85,7 +85,7 @@ static int ioevent_process(IOEventPoller *ioevent)
return 0;
}
#else
#endif
static void deal_ioevents(IOEventPoller *ioevent)
{
@ -109,7 +109,7 @@ static void deal_ioevents(IOEventPoller *ioevent)
}
}
static int ioevent_process(IOEventPoller *ioevent)
static int ioevent_process_by_poll(IOEventPoller *ioevent)
{
int result;
@ -130,8 +130,6 @@ static int ioevent_process(IOEventPoller *ioevent)
return 0;
}
#endif
static void deal_timeouts(FastTimerEntry *head)
{
FastTimerEntry *entry;
@ -206,20 +204,38 @@ int ioevent_loop(struct nio_thread_data *thread_data,
#endif
#if IOEVENT_USE_URING
if (thread_data->ev_puller.submit_count > 0) {
if ((result=ioevent_uring_submit(&thread_data->ev_puller)) != 0) {
logError("file: "__FILE__", line: %d, "
"io_uring_submit fail, errno: %d, error info: %s",
__LINE__, result, STRERROR(result));
return result;
if (thread_data->ev_puller.use_io_uring) {
if (thread_data->ev_puller.submit_count > 0) {
if ((result=ioevent_uring_submit(&thread_data->
ev_puller)) != 0)
{
logError("file: "__FILE__", line: %d, "
"io_uring_submit 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 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) {
@ -280,7 +296,7 @@ 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, const bool use_iouring)
const int timeout)
{
int result;
@ -288,7 +304,7 @@ int ioevent_set(struct fast_task_info *task, struct nio_thread_data *pThread,
task->event.fd = sock;
task->event.callback = callback;
#if IOEVENT_USE_URING
if (use_iouring) {
if (pThread->ev_puller.use_io_uring) {
if (FC_URING_OP_TYPE(task) == IORING_OP_NOP) {
if ((result=uring_prep_first_recv(task)) != 0) {
logError("file: "__FILE__", line: %d, "

View File

@ -36,7 +36,7 @@ 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, const bool use_iouring);
const int timeout);
int ioevent_reset(struct fast_task_info *task, int new_fd, short event);

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,
const int init_recv_buffer_size, const int timeout_ms)
{
const bool use_io_uring = false;
int result;
int new_init_recv_buffer_size;
int i;
@ -66,7 +67,7 @@ int fast_multi_sock_client_init_ex(FastMultiSockClient *client,
}
if ((result=ioevent_init(&client->ioevent, "client",
entry_count, timeout_ms, 0)) != 0)
use_io_uring, entry_count, timeout_ms, 0)) != 0)
{
logError("file: "__FILE__", line: %d, "
"ioevent_init fail, errno: %d, error info: %s",