Compare commits

..

43 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
YuQing ec8e47f831 tests/test_fast_buffer.c support all types 2025-09-12 14:08:46 +08:00
YuQing d9d6255621 test_fast_buffer.c support int2hex 2025-09-11 12:07:45 +08:00
YuQing 2f75958a4a test_fast_buffer.c support -t option 2025-09-11 10:05:25 +08:00
YuQing a4cae13e07 add function fc_ftoa 2025-09-10 15:06:17 +08:00
YuQing f136821c0d getIpaddrByNameEx: IPv4 has priority over IPv6 2025-09-10 15:05:15 +08:00
YuQing b97f23ced2 bytes_to_human_str support round off 2025-09-03 15:23:27 +08:00
YuQing 84a1f90a9a bytes to human readalbe string more gracefully 2025-08-29 13:25:58 +08:00
YuQing 3f19715e45 add function bytes_to_human_str 2025-08-29 11:36:10 +08:00
YuQing ce4c5e23d4 upgrade version to 1.0.79 2025-08-24 22:58:33 +08:00
YuQing d59da03d60 use size_t instead int to avoid compile warning 2025-08-21 20:52:35 +08:00
YuQing 8e51f4de3e logger.h export function log_it_ex3 2025-08-20 17:54:29 +08:00
vazmin 0afae48142 gh actions: upgrade to 1.0.78-1 2025-08-16 16:31:05 +00:00
28 changed files with 1369 additions and 319 deletions

1
.gitignore vendored
View File

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

19
HISTORY
View File

@ -1,4 +1,23 @@
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
* getIpaddrByNameEx: IPv4 has priority over IPv6
* shared_func.[hc]: add function fc_ftoa
Version 1.79 2025-08-29
* logger.h export function log_it_ex3
* shared_func.[hc]: add function bytes_to_human_str
Version 1.78 2025-08-07
* getIpaddrByName: normalize ip addr when input addr is IPv4 or IPv6
* add files: spinlock.[hc]

13
INSTALL
View File

@ -6,11 +6,20 @@ Please visit the libfastcommon Home Page for more detail.
English language: https://github.com/happyfish100/libfastcommon
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,
# github address: https://github.com/happyfish100/libfastcommon.git
# gitee address: https://gitee.com/fastdfs100/libfastcommon.git
# the command lines as:
git clone https://github.com/happyfish100/libfastcommon.git
cd libfastcommon; git checkout V1.0.72
./make.sh clean && ./make.sh && ./make.sh install
cd libfastcommon; git checkout V1.0.83
./make.sh clean && ./make.sh && sudo ./make.sh install

24
debian/changelog vendored
View File

@ -1,3 +1,27 @@
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
* upgrade to 1.0.78-1
-- YuQing <384681@qq.com> Sat, 16 Aug 2025 16:31:05 +0000
libfastcommon (1.0.77-1) unstable; urgency=medium
* upgrade to 1.0.77-1

View File

@ -3,7 +3,7 @@
%define CommitVersion %(echo $COMMIT_VERSION)
Name: libfastcommon
Version: 1.0.78
Version: 1.0.83
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.4
Requires: liburing >= 2.4
%endif
%description
c common functions library extracted from my open source projects FastDFS.
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
if [ "$uname" = "Linux" ]; then
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
LIBS="$LIBS -lrt"
fi
@ -261,3 +275,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

View File

@ -112,13 +112,21 @@ extern int pthread_mutexattr_settype(pthread_mutexattr_t *attr, int kind);
#define SYNC_LOG_BUFF_DEF_INTERVAL 10
#define TIME_NONE -1
#define FC_BYTES_ONE_KB ( 1 << 10)
#define FC_BYTES_ONE_MB ( 1 << 20)
#define FC_BYTES_ONE_GB ( 1 << 30)
#define FC_BYTES_ONE_TB (1LL << 40)
#define FC_BYTES_ONE_PB (1LL << 50)
#define FC_BYTES_ONE_EB (1LL << 60)
#if defined(IOV_MAX) && IOV_MAX > 256
#define FC_IOV_BATCH_SIZE 256
#else
#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)
@ -447,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);

View File

@ -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_is_connected},
conn_pool_disconnect_server_cb,
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)

View File

@ -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

View File

@ -115,7 +115,7 @@ static inline int fast_buffer_append_char(FastBuffer *buffer, const char ch)
return 0;
}
static inline int fast_buffer_append_int(FastBuffer *buffer, const int n)
static inline int fast_buffer_append_int32(FastBuffer *buffer, const int n)
{
int result;
@ -143,6 +143,9 @@ static inline int fast_buffer_append_int64(FastBuffer *buffer, const int64_t n)
return 0;
}
#define fast_buffer_append_int(buffer, n) \
fast_buffer_append_int64(buffer, n)
int fast_buffer_append_file(FastBuffer *buffer, const char *filename);
static inline int fast_buffer_append_string(FastBuffer *buffer, const char *str)

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,
const bool double_buffers, const int max_connections,
const int alloc_task_once, const int min_buff_size,
const int max_buff_size, const int padding_size,
const int arg_size, TaskInitCallback init_callback,
void *init_arg)
const bool double_buffers, const bool need_shrink,
const int max_connections, const int alloc_task_once,
const int min_buff_size, const int max_buff_size,
const int padding_size, const int arg_size,
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
_realloc_buffer(task->send.ptr, task->free_queue->min_buff_size, false);
if (task->free_queue->need_shrink && task->send.
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;
}
}

View File

@ -37,21 +37,27 @@ struct nio_thread_data;
struct fast_task_info;
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 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 +125,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 {
@ -135,7 +148,6 @@ struct fast_task_info
struct fc_list_head dlink; //for polling queue
} polling; //for RDMA busy polling
TaskContinueCallback continue_callback; //for continue stage
TaskFinishCallback finish_callback;
struct nio_thread_data *thread_data;
struct sf_network_handler *handler; //network handler for libserverframe nio
struct fast_task_info *next; //for free queue and deleted list
@ -153,6 +165,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 +177,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 int alloc_task_once, const int min_buff_size,
const int max_buff_size, const int padding_size,
const int arg_size, TaskInitCallback init_callback,
void *init_arg);
const bool double_buffers, const bool need_shrink,
const int max_connections, const int alloc_task_once,
const int min_buff_size, const int max_buff_size,
const int padding_size, const int arg_size,
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 int min_buff_size, const int max_buff_size,
const int arg_size)
const bool need_shrink, const int max_connections,
const int alloc_task_once, const int min_buff_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,
alloc_task_once, min_buff_size, max_buff_size, padding_size,
arg_size, NULL, NULL);
return free_queue_init_ex2(queue, name, double_buffers, need_shrink,
max_connections, alloc_task_once, min_buff_size, max_buff_size,
padding_size, arg_size, NULL, NULL);
}
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 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,
alloc_task_once, min_buff_size, max_buff_size, arg_size);
return free_queue_init_ex(queue, name, double_buffers,
need_shrink, max_connections, alloc_task_once,
min_buff_size, max_buff_size, arg_size);
}
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) {
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

View File

@ -45,71 +45,119 @@ int kqueue_ev_convert(int16_t event, uint16_t flags)
}
#endif
int ioevent_init(IOEventPoller *ioevent, const int size,
const int timeout_ms, const int extra_events)
int ioevent_init(IOEventPoller *ioevent, const char *service_name,
const bool use_io_uring, const int size, const int timeout_ms,
const int extra_events)
{
int bytes;
int bytes;
ioevent->size = size;
ioevent->extra_events = extra_events;
ioevent->iterator.index = 0;
ioevent->iterator.count = 0;
ioevent->iterator.index = 0;
ioevent->iterator.count = 0;
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;
}
bytes = sizeof(struct epoll_event) * size;
ioevent->events = (struct epoll_event *)fc_malloc(bytes);
#ifdef OS_LINUX
#if IOEVENT_USE_URING
ioevent->use_io_uring = use_io_uring;
if (use_io_uring) {
int result;
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
}
#endif
#elif IOEVENT_USE_KQUEUE
ioevent->poll_fd = kqueue();
if (ioevent->poll_fd < 0) {
return errno != 0 ? errno : ENOMEM;
}
bytes = sizeof(struct kevent) * size;
ioevent->events = (struct kevent *)fc_malloc(bytes);
ioevent->poll_fd = kqueue();
if (ioevent->poll_fd < 0) {
return errno != 0 ? errno : ENOMEM;
}
bytes = sizeof(struct kevent) * size;
ioevent->events = (struct kevent *)fc_malloc(bytes);
#elif IOEVENT_USE_PORT
ioevent->poll_fd = port_create();
if (ioevent->poll_fd < 0) {
return errno != 0 ? errno : ENOMEM;
}
bytes = sizeof(port_event_t) * size;
ioevent->events = (port_event_t *)fc_malloc(bytes);
ioevent->poll_fd = port_create();
if (ioevent->poll_fd < 0) {
return errno != 0 ? errno : ENOMEM;
}
bytes = sizeof(port_event_t) * size;
ioevent->events = (port_event_t *)fc_malloc(bytes);
#endif
if (ioevent->events == NULL) {
close(ioevent->poll_fd);
ioevent->poll_fd = -1;
return ENOMEM;
}
ioevent_set_timeout(ioevent, timeout_ms);
#if IOEVENT_USE_URING
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
return 0;
ioevent_set_timeout(ioevent, timeout_ms);
return 0;
}
void ioevent_destroy(IOEventPoller *ioevent)
{
if (ioevent->events != NULL) {
free(ioevent->events);
ioevent->events = NULL;
}
#if IOEVENT_USE_URING
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)
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);
#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;
@ -128,15 +176,32 @@ int ioevent_attach(IOEventPoller *ioevent, const int fd, const int e,
#endif
}
int ioevent_modify(IOEventPoller *ioevent, const int fd, const int e,
void *data)
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);
#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;
@ -171,8 +236,25 @@ int ioevent_modify(IOEventPoller *ioevent, const int fd, const int e,
int ioevent_detach(IOEventPoller *ioevent, const int fd)
{
#if IOEVENT_USE_EPOLL
return epoll_ctl(ioevent->poll_fd, EPOLL_CTL_DEL, fd, NULL);
#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;
@ -191,15 +273,33 @@ 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);
#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);
return kevent(ioevent->poll_fd, NULL, 0, ioevent->events,
ioevent->size, &ioevent->timeout);
#elif IOEVENT_USE_PORT
int result;
int retval;
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)
{
result = (int)nget;
@ -225,4 +325,3 @@ int ioevent_poll(IOEventPoller *ioevent)
#error port me
#endif
}

View File

@ -20,17 +20,28 @@
#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
#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)
#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 +76,33 @@ 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;
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 {
int index;
int count;
} iterator; //for deal event loop
#if IOEVENT_USE_EPOLL
#ifdef OS_LINUX
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
struct kevent *events;
struct timespec timeout;
@ -84,9 +110,10 @@ typedef struct ioevent_puller {
port_event_t *events;
timespec_t timeout;
#endif
} IOEventPoller;
#if IOEVENT_USE_EPOLL
#if OS_LINUX
#define IOEVENT_GET_EVENTS(ioevent, index) \
(ioevent)->events[index].events
#elif IOEVENT_USE_KQUEUE
@ -99,7 +126,7 @@ 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_KQUEUE
@ -112,7 +139,7 @@ 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_KQUEUE
@ -129,24 +156,39 @@ typedef struct ioevent_puller {
extern "C" {
#endif
int ioevent_init(IOEventPoller *ioevent, const int size,
const int timeout_ms, const int extra_events);
int ioevent_init(IOEventPoller *ioevent, const char *service_name,
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, const int e,
void *data);
int ioevent_modify(IOEventPoller *ioevent, const int fd, const int e,
void *data);
int ioevent_attach(IOEventPoller *ioevent, const int fd,
const int e, void *data);
int ioevent_modify(IOEventPoller *ioevent, const int fd,
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;
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
}
@ -156,6 +198,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

View File

@ -17,6 +17,76 @@
#include "logger.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)
{
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 index;
int result;
if (ioevent->iterator.index >= ioevent->iterator.count)
{
return ENOENT;
ioevent->iterator.count = ioevent_poll(ioevent);
if (ioevent->iterator.count > 0) {
deal_ioevents(ioevent);
}
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;
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;
}
}
return ENOENT;
return 0;
}
static void deal_timeouts(FastTimerEntry *head)
@ -82,12 +142,10 @@ 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)
{
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;
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 +200,52 @@ int ioevent_loop(struct nio_thread_data *thread_data,
sched_pull = true;
#endif
if (sched_pull)
{
thread_data->ev_puller.iterator.count = ioevent_poll(
&thread_data->ev_puller);
if (thread_data->ev_puller.iterator.count > 0)
{
deal_ioevents(&thread_data->ev_puller);
}
else if (thread_data->ev_puller.iterator.count < 0)
{
result = errno != 0 ? errno : EINVAL;
if (result != EINTR)
#if IOEVENT_USE_URING
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, " \
"ioevent_poll fail, " \
"errno: %d, error info: %s", \
logError("file: "__FILE__", line: %d, "
"io_uring_submit fail, errno: %d, error info: %s",
__LINE__, result, STRERROR(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);
}
if (thread_data->deleted_list != NULL)
{
count = 0;
while (thread_data->deleted_list != NULL)
{
if (thread_data->deleted_list != NULL) {
//count = 0;
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)) {
@ -188,13 +254,12 @@ int ioevent_loop(struct nio_thread_data *thread_data,
}
}
clean_up_callback(task);
count++;
//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;
count = fast_timer_timeouts_get(
&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;
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);
}
}
@ -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 sock, short event, IOEventCallback callback, const int timeout)
int sock, short event, IOEventCallback callback,
const int timeout)
{
int result;
task->thread_data = pThread;
task->event.fd = sock;
task->event.callback = callback;
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
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, "
"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, "
"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;
fast_timer_add(&pThread->timer, &task->event.timer);

View File

@ -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,11 +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_remove(IOEventPoller *ioevent, void *data);
int ioevent_set(struct fast_task_info *pTask, struct nio_thread_data *pThread,
int sock, short event, IOEventCallback callback, const int timeout);
int ioevent_set(struct fast_task_info *task, 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);
@ -75,9 +81,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

View File

@ -112,7 +112,7 @@ int log_init_ex(LogContext *pContext)
}
pContext->pcurrent_buff = pContext->log_buff;
if ((result=init_pthread_lock(&(pContext->log_thread_lock))) != 0)
if ((result=init_pthread_lock(&(pContext->lock))) != 0)
{
return result;
}
@ -136,8 +136,8 @@ static int log_print_header(LogContext *pContext)
if (pContext->current_size < 0)
{
result = errno != 0 ? errno : EACCES;
fprintf(stderr, "lseek file \"%s\" fail, " \
"errno: %d, error info: %s\n", \
fprintf(stderr, "lseek file \"%s\" fail, "
"errno: %d, error info: %s\n",
pContext->log_filename, result, STRERROR(result));
}
else {
@ -296,9 +296,9 @@ void log_set_header_callback(LogContext *pContext, LogHeaderCallback header_call
{
int64_t current_size;
pthread_mutex_lock(&(pContext->log_thread_lock));
pthread_mutex_lock(&(pContext->lock));
current_size = pContext->current_size;
pthread_mutex_unlock(&(pContext->log_thread_lock));
pthread_mutex_unlock(&(pContext->lock));
if (current_size == 0)
{
log_print_header(pContext);
@ -347,7 +347,7 @@ void log_destroy_ex(LogContext *pContext)
close(pContext->log_fd);
pContext->log_fd = STDERR_FILENO;
pthread_mutex_destroy(&pContext->log_thread_lock);
pthread_mutex_destroy(&pContext->lock);
}
if (pContext->log_buff != NULL)
@ -918,19 +918,19 @@ static int log_fsync(LogContext *pContext, const bool bNeedLock)
{
if (bNeedLock)
{
pthread_mutex_lock(&(pContext->log_thread_lock));
pthread_mutex_lock(&(pContext->lock));
}
result = log_check_rotate(pContext);
if (bNeedLock)
{
pthread_mutex_unlock(&(pContext->log_thread_lock));
pthread_mutex_unlock(&(pContext->lock));
}
return result;
}
}
if (bNeedLock && ((lock_res=pthread_mutex_lock( \
&(pContext->log_thread_lock))) != 0))
&(pContext->lock))) != 0))
{
fprintf(stderr, "file: "__FILE__", line: %d, " \
"call pthread_mutex_lock fail, " \
@ -966,7 +966,7 @@ static int log_fsync(LogContext *pContext, const bool bNeedLock)
}
if (bNeedLock && ((lock_res=pthread_mutex_unlock( \
&(pContext->log_thread_lock))) != 0))
&(pContext->lock))) != 0))
{
fprintf(stderr, "file: "__FILE__", line: %d, " \
"call pthread_mutex_unlock fail, " \
@ -977,8 +977,8 @@ static int log_fsync(LogContext *pContext, const bool bNeedLock)
return result;
}
static void doLogEx(LogContext *pContext, struct timeval *tv, \
const char *caption, const char *text, const int text_len, \
void log_it_ex3(LogContext *pContext, struct timeval *tv,
const char *caption, const char *text, const int text_len,
const bool bNeedSync, const bool bNeedLock)
{
struct tm tm;
@ -1003,7 +1003,7 @@ static void doLogEx(LogContext *pContext, struct timeval *tv, \
}
}
if (bNeedLock && (result=pthread_mutex_lock(&pContext->log_thread_lock)) != 0)
if (bNeedLock && (result=pthread_mutex_lock(&pContext->lock)) != 0)
{
fprintf(stderr, "file: "__FILE__", line: %d, " \
"call pthread_mutex_lock fail, " \
@ -1018,12 +1018,12 @@ static void doLogEx(LogContext *pContext, struct timeval *tv, \
__LINE__, LOG_BUFF_SIZE, text_len + 64);
if (bNeedLock)
{
pthread_mutex_unlock(&(pContext->log_thread_lock));
pthread_mutex_unlock(&(pContext->lock));
}
return;
}
if ((pContext->pcurrent_buff - pContext->log_buff) + text_len + 64 \
if ((pContext->pcurrent_buff - pContext->log_buff) + text_len + 64
> LOG_BUFF_SIZE)
{
log_fsync(pContext, false);
@ -1067,7 +1067,7 @@ static void doLogEx(LogContext *pContext, struct timeval *tv, \
log_fsync(pContext, false);
}
if (bNeedLock && (result=pthread_mutex_unlock(&(pContext->log_thread_lock))) != 0)
if (bNeedLock && (result=pthread_mutex_unlock(&(pContext->lock))) != 0)
{
fprintf(stderr, "file: "__FILE__", line: %d, " \
"call pthread_mutex_unlock fail, " \
@ -1076,8 +1076,8 @@ static void doLogEx(LogContext *pContext, struct timeval *tv, \
}
}
void log_it_ex2(LogContext *pContext, const char *caption, \
const char *text, const int text_len, \
void log_it_ex2(LogContext *pContext, const char *caption,
const char *text, const int text_len,
const bool bNeedSync, const bool bNeedLock)
{
struct timeval tv;
@ -1092,10 +1092,10 @@ void log_it_ex2(LogContext *pContext, const char *caption, \
gettimeofday(&tv, NULL);
}
doLogEx(pContext, &tv, caption, text, text_len, bNeedSync, bNeedLock);
log_it_ex3(pContext, &tv, caption, text, text_len, bNeedSync, bNeedLock);
}
void log_it_ex1(LogContext *pContext, const int priority, \
void log_it_ex1(LogContext *pContext, const int priority,
const char *text, const int text_len)
{
bool bNeedSync;
@ -1291,7 +1291,7 @@ void logAccess(LogContext *pContext, struct timeval *tvStart, \
{
len = sizeof(text) - 1;
}
doLogEx(pContext, tvStart, NULL, text, len, false, true);
log_it_ex3(pContext, tvStart, NULL, text, len, false, true);
}
const char *log_get_level_caption_ex(LogContext *pContext)

View File

@ -62,7 +62,7 @@ typedef struct log_context
char *pcurrent_buff;
/* mutext lock */
pthread_mutex_t log_thread_lock;
pthread_mutex_t lock;
/*
rotate the log when the log file exceeds this parameter
@ -343,7 +343,7 @@ void log_it_ex(LogContext *pContext, const int priority, \
* text_len: text string length (bytes)
* return: none
*/
void log_it_ex1(LogContext *pContext, const int priority, \
void log_it_ex1(LogContext *pContext, const int priority,
const char *text, const int text_len);
/** log to file
@ -355,10 +355,13 @@ void log_it_ex1(LogContext *pContext, const int priority, \
* bNeedSync: if sync to file immediatelly
* return: none
*/
void log_it_ex2(LogContext *pContext, const char *caption, \
const char *text, const int text_len, \
void log_it_ex2(LogContext *pContext, const char *caption,
const char *text, const int text_len,
const bool bNeedSync, const bool bNeedLock);
void log_it_ex3(LogContext *pContext, struct timeval *tv,
const char *caption, const char *text, const int text_len,
const bool bNeedSync, const bool bNeedLock);
/** sync log buffer to log file
* parameters:

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;
@ -65,8 +66,8 @@ int fast_multi_sock_client_init_ex(FastMultiSockClient *client,
return EINVAL;
}
if ((result=ioevent_init(&client->ioevent, entry_count,
timeout_ms, 0)) != 0)
if ((result=ioevent_init(&client->ioevent, "client",
use_io_uring, entry_count, timeout_ms, 0)) != 0)
{
logError("file: "__FILE__", line: %d, "
"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)
{
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 +335,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 +390,7 @@ static int fast_multi_sock_client_deal_io(FastMultiSockClient *client)
fast_multi_sock_client_finish(client, entry, result);
}
}
#endif
}
/*

View File

@ -3251,6 +3251,110 @@ const char *long2str(const int64_t n, char *buff, const bool thousands_separator
return buff;
}
static int format_bytes_string(const int64_t input_bytes,
const int64_t unit_bytes, char *buff)
{
int64_t remain;
int n;
int fragment;
char *p;
n = input_bytes / unit_bytes;
remain = input_bytes - (n * unit_bytes);
if (n < 10)
{
fragment = (remain * 10LL + unit_bytes / 2) / unit_bytes;
if (fragment == 10)
{
++n;
fragment = 0;
}
p = buff + fc_itoa(n, buff);
if (n < 10)
{
*p++ = '.';
p += fc_itoa(fragment, p);
}
}
else
{
if (remain >= unit_bytes / 2)
{
++n;
}
p = buff + fc_itoa(n, buff);
}
return p - buff;
}
const char *bytes_to_human_str(const int64_t bytes, char *buff)
{
char *p;
if (bytes < FC_BYTES_ONE_TB)
{
if (bytes < FC_BYTES_ONE_MB)
{
if (bytes < FC_BYTES_ONE_KB)
{
p = buff + fc_itoa(bytes, buff);
*p++ = ' ';
*p++ = ' '; //for alignment
}
else
{
p = buff + format_bytes_string(bytes, FC_BYTES_ONE_KB, buff);
*p++ = ' ';
*p++ = 'K';
}
}
else
{
if (bytes < FC_BYTES_ONE_GB)
{
p = buff + format_bytes_string(bytes, FC_BYTES_ONE_MB, buff);
*p++ = ' ';
*p++ = 'M';
}
else
{
p = buff + format_bytes_string(bytes, FC_BYTES_ONE_GB, buff);
*p++ = ' ';
*p++ = 'G';
}
}
}
else
{
if (bytes < FC_BYTES_ONE_EB)
{
if (bytes < FC_BYTES_ONE_PB)
{
p = buff + format_bytes_string(bytes, FC_BYTES_ONE_TB, buff);
*p++ = ' ';
*p++ = 'T';
}
else
{
p = buff + format_bytes_string(bytes, FC_BYTES_ONE_PB, buff);
*p++ = ' ';
*p++ = 'P';
}
}
else
{
p = buff + format_bytes_string(bytes, FC_BYTES_ONE_EB, buff);
*p++ = ' ';
*p++ = 'E';
}
}
*p++ = 'B';
*p = '\0';
return buff;
}
bool starts_with(const char *str, const char *needle)
{
int str_len;
@ -4302,6 +4406,119 @@ int fc_itoa(int64_t n, char *buff)
return (start - buff) + len;
}
int fc_ftoa(double d, const int scale, char *buff)
{
int len;
int i;
bool positive;
int64_t n;
double fraction;
char *p;
positive = (d >= 0.00);
switch (scale)
{
case 0:
d += (positive ? 5e-1 : -5e-1);
break;
case 1:
d += (positive ? 5e-2 : -5e-2);
break;
case 2:
d += (positive ? 5e-3 : -5e-3);
break;
case 3:
d += (positive ? 5e-4 : -5e-4);
break;
case 4:
d += (positive ? 5e-5 : -5e-5);
break;
case 5:
d += (positive ? 5e-6 : -5e-6);
break;
case 6:
d += (positive ? 5e-7 : -5e-7);
break;
case 7:
d += (positive ? 5e-8 : -5e-8);
break;
case 8:
d += (positive ? 5e-9 : -5e-9);
break;
case 9:
d += (positive ? 5e-10 : -5e-10);
break;
case 10:
d += (positive ? 5e-11 : -5e-11);
break;
case 11:
d += (positive ? 5e-12 : -5e-12);
break;
case 12:
d += (positive ? 5e-13 : -5e-13);
break;
case 13:
d += (positive ? 5e-14 : -5e-14);
break;
case 14:
d += (positive ? 5e-15 : -5e-15);
break;
case 15:
d += (positive ? 5e-16 : -5e-16);
break;
case 16:
d += (positive ? 5e-17 : -5e-17);
break;
default:
d += (positive ? 5e-18 : -5e-18);
break;
}
n = (int64_t)d;
if (n > -10 && n < 10)
{
if (positive)
{
*buff = '0' + n;
len = 1;
}
else
{
*buff = '-';
*(buff+1) = '0' + (-1) * n;
len = 2;
}
}
else
{
len = fc_itoa(n, buff);
}
if (scale <= 0)
{
return len;
}
p = buff + len;
*p++ = '.';
if (positive)
{
fraction = d - (double)n;
}
else
{
fraction = (double)n - d;
}
for (i=0; i<scale; i++)
{
fraction *= 10;
n = (int)fraction;
fraction -= n;
*p++ = '0' + n;
}
return p - buff;
}
int fc_compare_int64_ptr(const int64_t *n1, const int64_t *n2)
{
return fc_compare_int64(*n1, *n2);

View File

@ -1191,6 +1191,8 @@ static inline const char *long_to_comma_str(const int64_t n, char *buff)
return long2str(n, buff, true);
}
const char *bytes_to_human_str(const int64_t bytes, char *buff);
/** if the string starts with the needle string
* parameters:
* str: the string to detect
@ -1601,12 +1603,28 @@ bool fc_path_contains(const string_t *path, const string_t *needle,
/** itoa output as decimal number
* parameters:
* n: the number to convert
* buff: store the converted string
* return: string length
* n: the integer number to convert
* buff: store the converted string, NOT null-terminated
* return: converted string length
*/
int fc_itoa(int64_t n, char *buff);
/** ftoa output as decimal number
* parameters:
* d: the double number to convert
* scale: number of decimal places (round off)
* buff: store the converted string, NOT null-terminated
* return: converted string length
*/
int fc_ftoa(double d, const int scale, char *buff);
/** output as decimal number
* parameters:
* n: the integer number to convert
* buff: store the converted string, null-terminated
* padding_len: padding length (padding with charactor '0')
* return: converted string length
*/
static inline int fc_ltostr_ex(int64_t n, char *buff, const int padding_len)
{
int len;
@ -1618,21 +1636,34 @@ static inline int fc_ltostr_ex(int64_t n, char *buff, const int padding_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;
memmove(buff + fill_len, buff, len + 1);
memset(buff, '0', fill_len);
#if defined(OS_LINUX) && defined(__GNUC__) && __GNUC__ >= 8
#pragma GCC diagnostic pop
#endif
return padding_len;
}
static inline int fc_ltostr(int64_t n, char *buff)
{
const int padding_len = 0;
return fc_ltostr_ex(n, buff, padding_len);
int 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)
{
int len;
size_t len;
len = strlen(src);
if (len < size) {

View File

@ -1238,18 +1238,18 @@ in_addr_64_t getIpaddrByNameEx(const char *name, char *buff,
return addr4.s_addr;
}
*af = AF_UNSPEC;
memset(&hints, 0, sizeof hints);
hints.ai_family = AF_UNSPEC; // 支持IPv4和IPv6
hints.ai_family = AF_UNSPEC;
if (getaddrinfo(name, NULL, &hints, &res) != 0)
{
*af = AF_UNSPEC;
return INADDR_NONE;
}
ip_addr = INADDR_NONE;
for (p = res; p != NULL; p = p->ai_next)
{
*af = p->ai_family;
if (p->ai_family == AF_INET) // 处理IPv4地址
if (p->ai_family == AF_INET) //IPv4 address
{
struct sockaddr_in *ipv4 = (struct sockaddr_in *)p->ai_addr;
if (buff != NULL)
@ -1260,11 +1260,11 @@ in_addr_64_t getIpaddrByNameEx(const char *name, char *buff,
}
}
*af = p->ai_family;
ip_addr = ipv4->sin_addr.s_addr;
freeaddrinfo(res);
return ip_addr;
break;
}
else if (p->ai_family == AF_INET6) // 处理IPv6地址
else if (p->ai_family == AF_INET6) //IPv6 address
{
struct sockaddr_in6 *ipv6 = (struct sockaddr_in6 *)p->ai_addr;
if (buff != NULL)
@ -1281,15 +1281,14 @@ in_addr_64_t getIpaddrByNameEx(const char *name, char *buff,
}
}
*af = p->ai_family;
ip_addr = *((in_addr_64_t *)((char *)&ipv6->sin6_addr + 8));
freeaddrinfo(res);
return ip_addr;
continue;
}
}
freeaddrinfo(res);
*af = AF_UNSPEC;
return INADDR_NONE;
return ip_addr;
}
int getIpaddrsByName(const char *name,

View File

@ -29,6 +29,7 @@
#include <netinet/tcp.h>
#include <netdb.h>
#include "common_define.h"
#include "shared_func.h"
#define FC_NET_TYPE_NONE 0
#define FC_NET_TYPE_OUTER 1 //extranet IP
@ -53,16 +54,6 @@
#define FAST_WRITE_BUFF_SIZE (256 * 1024)
#define FC_FORMAT_IP_ADDRESS(old_ip_addr, new_ip_addr) \
char new_ip_addr[FORMATTED_IP_SIZE]; \
do { \
if (is_ipv6_addr(old_ip_addr)) { \
sprintf(new_ip_addr, "[%s]", old_ip_addr); \
} else { \
strcpy(new_ip_addr, old_ip_addr); \
} \
} while (0)
typedef struct fast_if_config {
char name[IF_NAMESIZE]; //if name
char mac[64];
@ -719,7 +710,16 @@ static inline const char *format_ip_address(const char *ip, char *buff)
{
if (is_ipv6_addr(ip))
{
sprintf(buff, "[%s]", ip);
int ip_len;
char *p;
ip_len = strlen(ip);
p = buff;
*p++ = '[';
memcpy(p, ip, ip_len);
p += ip_len;
*p++ = ']';
*p = '\0';
}
else
{
@ -732,14 +732,26 @@ static inline const char *format_ip_address(const char *ip, char *buff)
static inline const char *format_ip_port(const char *ip,
const int port, char *buff)
{
if (is_ipv6_addr(ip))
int ip_len;
bool is_ipv6;
char *p;
is_ipv6 = is_ipv6_addr(ip);
ip_len = strlen(ip);
p = buff;
if (is_ipv6)
{
sprintf(buff, "[%s]:%u", ip, port);
*p++ = '[';
}
else
memcpy(p, ip, ip_len);
p += ip_len;
if (is_ipv6)
{
sprintf(buff, "%s:%u", ip, port);
*p++ = ']';
}
*p++ = ':';
p += fc_itoa(port, p);
*p = '\0';
return buff;
}

View File

@ -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
LIB_PATH = -lfastcommon -lpthread
INC_PATH = $(INCS)
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 \

View File

@ -26,6 +26,14 @@
#include "fastcommon/fast_buffer.h"
#include "fastcommon/sched_thread.h"
typedef enum {
TEST_TYPE_NONE = 0,
TEST_TYPE_ITOA,
TEST_TYPE_FTOA,
TEST_TYPE_INT2HEX,
TEST_TYPE_APPEND
} TestType;
typedef enum {
DA_SLICE_TYPE_FILE = 'F', /* in file slice */
DA_SLICE_TYPE_CACHE = 'C', /* in memory cache */
@ -152,20 +160,41 @@ static inline int cache_binlog_filename_by_append(
return p - full_filename;
}
static void usage(const char *program)
{
fprintf(stderr, "Usage: %s [-t {itoa | ftoa | int2hex | append | all}]\n",
program);
}
int main(int argc, char *argv[])
{
const bool binary_mode = true;
const bool check_capacity = false;
const bool have_extra_field = false;
const int LOOP = 10 * 1000 * 1000;
const char *data_path = "/opt/fastcfs/fdir/data";
const char *subdir_name = "binlog";
const uint32_t subdirs = 256;
int result;
TestType test_type = TEST_TYPE_ITOA;
TestType type_start;
TestType type_last;
uint64_t id = 123456;
double d = 123.456;
int ch;
int i;
int len;
int64_t start_time_us;
int append_time_us;
int convert_time_us;
int sprintf_time_us;
double ratio;
FastBuffer buffer;
DATrunkSpaceLogRecord record;
char full_filename1[PATH_MAX];
char full_filename2[PATH_MAX];
char buff[32] = {0};
char *caption;
log_init();
g_current_time = time(NULL);
@ -175,53 +204,127 @@ int main(int argc, char *argv[])
return result;
}
memset(&record, 0, sizeof(record));
record.op_type = 'C';
record.slice_type = DA_SLICE_TYPE_FILE;
record.storage.version = 1111;
record.oid = 9007211709265131LL;
record.fid = 0;
record.storage.trunk_id = 61;
record.storage.length = 62;
record.storage.offset = 12345;
record.storage.size = 64;
const char *data_path = "/opt/fastcfs/fdir/data";
const char *subdir_name = "binlog";
const uint32_t subdirs = 256;
uint64_t id = 123456;
char full_filename1[PATH_MAX];
char full_filename2[PATH_MAX];
start_time_us = get_current_time_us();
for (i=0; i<LOOP; i++) {
cache_binlog_filename_by_sprintf(data_path, subdir_name,
subdirs, ++id, full_filename1, sizeof(full_filename1));
fast_buffer_reset(&buffer);
log_pack_by_sprintf(&record, &buffer, have_extra_field);
}
sprintf_time_us = (get_current_time_us() - start_time_us);
start_time_us = get_current_time_us();
for (i=0; i<LOOP; i++) {
cache_binlog_filename_by_append(data_path, subdir_name,
subdirs, ++id, full_filename2, sizeof(full_filename2));
fast_buffer_reset(&buffer);
log_pack_by_append(&record, &buffer, have_extra_field);
}
append_time_us = (get_current_time_us() - start_time_us);
if (append_time_us > 0) {
ratio = (double)sprintf_time_us / (double)append_time_us;
} else {
ratio = 1.0;
type_start = type_last = TEST_TYPE_ITOA;
while ((ch=getopt(argc, argv, "ht:")) != -1) {
switch (ch) {
case 'h':
usage(argv[0]);
return 0;
case 't':
if (strcasecmp(optarg, "itoa") == 0) {
type_start = type_last = TEST_TYPE_ITOA;
} else if (strcasecmp(optarg, "ftoa") == 0) {
type_start = type_last = TEST_TYPE_FTOA;
} else if (strcasecmp(optarg, "int2hex") == 0) {
type_start = type_last = TEST_TYPE_INT2HEX;
} else if (strcasecmp(optarg, "append") == 0) {
type_start = type_last = TEST_TYPE_APPEND;
} else if (strcasecmp(optarg, "all") == 0) {
type_start = TEST_TYPE_ITOA;
type_last = TEST_TYPE_APPEND;
} else {
fprintf(stderr, "invalid type: %s\n", optarg);
return EINVAL;
}
break;
default:
usage(argv[0]);
return EINVAL;
}
}
printf("sprintf time: %d ms, append time: %d ms, "
"sprintf time / append time: %d%%\n",
sprintf_time_us / 1000, append_time_us / 1000,
(int)(ratio * 100.00));
for (test_type=type_start; test_type<=type_last; test_type++) {
if (test_type == TEST_TYPE_APPEND) {
memset(&record, 0, sizeof(record));
record.op_type = 'C';
record.slice_type = DA_SLICE_TYPE_FILE;
record.storage.version = 1111;
record.oid = 9007211709265131LL;
record.fid = 0;
record.storage.trunk_id = 61;
record.storage.length = 62;
record.storage.offset = 12345;
record.storage.size = 64;
}
start_time_us = get_current_time_us();
for (i=0; i<LOOP; i++) {
switch (test_type) {
case TEST_TYPE_APPEND:
cache_binlog_filename_by_sprintf(data_path, subdir_name,
subdirs, ++id, full_filename1, sizeof(full_filename1));
fast_buffer_reset(&buffer);
log_pack_by_sprintf(&record, &buffer, have_extra_field);
break;
case TEST_TYPE_ITOA:
sprintf(buff, "%"PRId64, id);
break;
case TEST_TYPE_FTOA:
sprintf(buff, "%.2f", d);
break;
case TEST_TYPE_INT2HEX:
sprintf(buff, "%x", (int)id);
break;
default:
break;
}
}
sprintf_time_us = (get_current_time_us() - start_time_us);
start_time_us = get_current_time_us();
for (i=0; i<LOOP; i++) {
switch (test_type) {
case TEST_TYPE_APPEND:
cache_binlog_filename_by_append(data_path, subdir_name,
subdirs, ++id, full_filename2, sizeof(full_filename2));
fast_buffer_reset(&buffer);
log_pack_by_append(&record, &buffer, have_extra_field);
break;
case TEST_TYPE_ITOA:
len = fc_itoa(id, buff);
*(buff + len) = '\0';
break;
case TEST_TYPE_FTOA:
len = fc_ftoa(d, 2, buff);
*(buff + len) = '\0';
break;
case TEST_TYPE_INT2HEX:
int2hex(id, buff, 0);
break;
default:
break;
}
}
convert_time_us = (get_current_time_us() - start_time_us);
if (convert_time_us > 0) {
ratio = (double)sprintf_time_us / (double)convert_time_us;
} else {
ratio = 1.0;
}
switch (test_type) {
case TEST_TYPE_ITOA:
caption = "itoa";
break;
case TEST_TYPE_FTOA:
caption = "ftoa";
break;
case TEST_TYPE_INT2HEX:
caption = "int2hex";
break;
case TEST_TYPE_APPEND:
caption = "append";
break;
default:
caption = "unkown";
break;
}
printf("sprintf time: %d ms, %s time: %d ms, "
"sprintf time / %s time: %d%%\n",
sprintf_time_us / 1000, caption, convert_time_us / 1000,
caption, (int)(ratio * 100.00));
}
fast_buffer_destroy(&buffer);
return 0;

View File

@ -32,6 +32,7 @@ int main(int argc, char *argv[])
int fd;
int result;
int n;
int i;
char buf[1024];
if (argc < 2) {
@ -69,7 +70,7 @@ int main(int argc, char *argv[])
return errno != 0 ? errno : EIO;
}
for (int i=0; i<5; i++) {
for (i=0; i<5; i++) {
if (lseek(fd, 0, SEEK_SET) < 0) {
logError("file: "__FILE__", line: %d, " \
"lseek file %s fail, " \

View File

@ -22,6 +22,7 @@
#include <sys/types.h>
#include <sys/time.h>
#include <sys/stat.h>
#include <sys/file.h>
#include "fastcommon/logger.h"
int main(int argc, char *argv[])
@ -35,6 +36,15 @@ int main(int argc, char *argv[])
log_take_over_stderr();
log_take_over_stdout();
log_set_compress_log_flags(LOG_COMPRESS_FLAGS_ENABLED | LOG_COMPRESS_FLAGS_NEW_THREAD);
log_set_filename("/opt/fastcfs/fuse/test.log");
if (flock(g_log_context.log_fd, LOCK_EX) != 0) {
logError("flock fail");
}
flock(g_log_context.log_fd, LOCK_UN);
printf("sizeof(LogContext): %d, time_precision: %d, compress_log_flags: %d, "
"use_file_write_lock: %d\n", (int)sizeof(LogContext),
@ -46,7 +56,7 @@ int main(int argc, char *argv[])
"by log_it_ex, timestamp: %d", (int)time(NULL));
len = sprintf(buff, "this is by log_it_ex1, "
"timestamp: %d", (int)time(NULL));
"timestamp: %ld", (long)time(NULL));
log_it_ex1(&g_log_context, LOG_INFO, buff, len);
return 0;