From 66b74c690ea0ee44678e5d2f3f5d3c4e509ea2f0 Mon Sep 17 00:00:00 2001 From: fengzero Date: Mon, 29 Nov 2021 05:32:49 +0000 Subject: [PATCH] add asynchronous io and timer asyn io is not implemented use epoll/timerfd on linux use kqueue on unix --- CMakeLists.txt | 5 + include/event/event.h | 71 ++++++++++++ src/event/event_linux.c | 250 ++++++++++++++++++++++++++++++++++++++++ src/event/event_unix.c | 162 ++++++++++++++++++++++++++ 4 files changed, 488 insertions(+) create mode 100644 include/event/event.h create mode 100644 src/event/event_linux.c create mode 100644 src/event/event_unix.c diff --git a/CMakeLists.txt b/CMakeLists.txt index 058cb835..fa3df0bc 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -65,6 +65,9 @@ find_package(nng CONFIG REQUIRED) find_package(Threads) add_definitions(-DPROJECT_ROOT_DIR="${CMAKE_CURRENT_SOURCE_DIR}") +if(NOT CMAKE_SYSTEM_NAME) + set(CMAKE_SYSTEM_NAME "Linux") +endif() if(CMAKE_SYSTEM_NAME MATCHES "Linux") add_definitions(-DNEU_PLATFORM_LINUX) @@ -121,6 +124,8 @@ set(NEURON_BASE_SOURCES src/base/neu_errcodes.c src/connection/tcp_client.c src/connection/tcp_server.c + src/event/event_linux.c + src/event/event_unix.c src/utils/atomic_data.c src/utils/mem_alloc.c src/utils/idhash.c diff --git a/include/event/event.h b/include/event/event.h new file mode 100644 index 00000000..a22b7dee --- /dev/null +++ b/include/event/event.h @@ -0,0 +1,71 @@ +/** + * NEURON IIoT System for Industry 4.0 + * Copyright (C) 2020-2021 EMQ Technologies Co., Ltd All rights reserved. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 3 of the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with this program; if not, write to the Free Software Foundation, + * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. + **/ + +#ifndef NEURON_EVENT_H +#define NEURON_EVENT_H + +#include + +#ifdef __cplusplus +extern "C" { +#endif + +typedef struct neu_event_ctx neu_event_ctx_t; + +neu_event_ctx_t *neu_event_new(void); +int neu_event_close(neu_event_ctx_t *events); + +typedef struct neu_event_timer_ctx neu_event_timer_ctx_t; +typedef int (*neu_event_timer_callback)(void *usr_data); + +typedef struct neu_event_timer { + int64_t second; + int64_t millisecond; + void * usr_data; + neu_event_timer_callback cb; +} neu_event_timer_t; + +neu_event_timer_ctx_t *neu_event_add_timer(neu_event_ctx_t * events, + neu_event_timer_t timer); +int neu_event_del_timer(neu_event_ctx_t *events, neu_event_timer_ctx_t *timer); + +enum neu_event_io_type { + NEU_EVENT_IO_READ = 0x1, + NEU_EVENT_IO_PEER_CLOSE = 0x2, +}; +typedef struct neu_event_io_ctx neu_event_io_ctx_t; +typedef int (*neu_event_io_callback)(enum neu_event_io_type type, int fd, + void *usr_data); + +typedef struct neu_event_io { + int type; + int fd; + void * usr_data; + neu_event_io_callback cb; +} neu_event_io_t; + +neu_event_io_ctx_t *neu_event_add_io(neu_event_ctx_t *events, + neu_event_io_t io); +int neu_event_del_io(neu_event_ctx_t *events, neu_event_io_ctx_t *io); + +#ifdef __cplusplus +} +#endif + +#endif diff --git a/src/event/event_linux.c b/src/event/event_linux.c new file mode 100644 index 00000000..88534fad --- /dev/null +++ b/src/event/event_linux.c @@ -0,0 +1,250 @@ +/** + * NEURON IIoT System for Industry 4.0 + * Copyright (C) 2020-2021 EMQ Technologies Co., Ltd All rights reserved. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 3 of the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with this program; if not, write to the Free Software Foundation, + * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. + **/ +#include +#include +#include +#include +#include +#include +#include + +#include "event/event.h" +#include "neu_log.h" + +#ifdef NEU_PLATFORM_LINUX +#include +#include + +struct neu_event_timer_ctx { + int fd; + void *event_data; +}; + +struct neu_event_io_ctx { +}; + +struct event_data { + enum { + TIMER, + IO, + } type; + union { + neu_event_io_callback io; + neu_event_timer_callback timer; + } callback; + union { + struct neu_event_io_ctx * io; + struct neu_event_timer_ctx *timer; + } ctx; + + void *usr_data; + int fd; + + TAILQ_ENTRY(event_data) node; +}; + +struct neu_event_ctx { + int epoll_fd; + bool running; + pthread_t thread; + pthread_mutex_t mtx; + + TAILQ_HEAD(, event_data) datas; +}; + +static void *event_loop(void *arg) +{ + neu_event_ctx_t *events = (neu_event_ctx_t *) arg; + bool running = events->running; + + while (running) { + struct epoll_event event = { 0 }; + struct event_data *data = NULL; + + int ret = epoll_wait(events->epoll_fd, &event, 1, 1000); + if (ret == 0) { + continue; + } + if (ret == -1) { + log_error("epoll wait error: %d(%s), fd: %d", errno, + strerror(errno), events->epoll_fd); + continue; + } + + data = (struct event_data *) event.data.ptr; + + switch (data->type) { + case TIMER: + if ((event.events & EPOLLIN) == EPOLLIN) { + uint64_t t; + + read(data->fd, &t, sizeof(t)); + + ret = data->callback.timer(data->usr_data); + log_debug("timer trigger: %d, ret: %d", data->fd, ret); + } + break; + case IO: + if ((event.events & EPOLLIN) == EPOLLIN) { + data->callback.io(NEU_EVENT_IO_READ, data->fd, data->usr_data); + } + if ((event.events & EPOLLHUP) == EPOLLHUP || + (event.events & EPOLLRDHUP) == EPOLLRDHUP) { + data->callback.io(NEU_EVENT_IO_PEER_CLOSE, data->fd, + data->usr_data); + } + break; + } + + pthread_mutex_lock(&events->mtx); + running = events->running; + pthread_mutex_unlock(&events->mtx); + } + + return NULL; +}; + +neu_event_ctx_t *neu_event_new(void) +{ + neu_event_ctx_t *events = calloc(1, sizeof(struct neu_event_ctx)); + + events->epoll_fd = epoll_create(1); + events->running = true; + + TAILQ_INIT(&events->datas); + pthread_mutex_init(&events->mtx, NULL); + + pthread_create(&events->thread, NULL, event_loop, events); + + return events; +}; + +int neu_event_close(neu_event_ctx_t *events) +{ + struct event_data *data = NULL; + + pthread_mutex_lock(&events->mtx); + events->running = false; + + data = TAILQ_FIRST(&events->datas); + while (data != NULL) { + epoll_ctl(events->epoll_fd, EPOLL_CTL_DEL, data->fd, NULL); + TAILQ_REMOVE(&events->datas, data, node); + close(data->fd); + switch (data->type) { + case TIMER: + free(data->ctx.timer); + break; + case IO: + free(data->ctx.io); + break; + } + free(data); + + data = TAILQ_FIRST(&events->datas); + } + + pthread_mutex_unlock(&events->mtx); + + close(events->epoll_fd); + pthread_mutex_destroy(&events->mtx); + + free(events); + return 0; +} + +neu_event_timer_ctx_t *neu_event_add_timer(neu_event_ctx_t * events, + neu_event_timer_t timer) +{ + int ret = 0; + int timer_fd = timerfd_create(CLOCK_MONOTONIC, 0); + struct itimerspec value = { + .it_value.tv_sec = timer.second, + .it_value.tv_nsec = timer.millisecond * 1000 * 1000, + .it_interval.tv_sec = timer.second, + .it_interval.tv_nsec = timer.millisecond * 1000 * 1000, + }; + struct event_data * data = calloc(1, sizeof(struct event_data)); + struct epoll_event event = { .events = EPOLLIN, .data.ptr = data }; + neu_event_timer_ctx_t *timer_ctx = calloc(1, sizeof(neu_event_timer_ctx_t)); + + timerfd_settime(timer_fd, 0, &value, NULL); + + data->type = TIMER; + data->fd = timer_fd; + data->usr_data = timer.usr_data; + data->callback.timer = timer.cb; + data->ctx.timer = timer_ctx; + + timer_ctx->fd = timer_fd; + timer_ctx->event_data = data; + + pthread_mutex_lock(&events->mtx); + TAILQ_INSERT_TAIL(&events->datas, data, node); + pthread_mutex_unlock(&events->mtx); + + ret = epoll_ctl(events->epoll_fd, EPOLL_CTL_ADD, timer_fd, &event); + + log_info("add timer, second: %ld, millisecond: %ld, timer: %d in epoll %d, " + "ret: %d", + timer.second, timer.millisecond, timer_fd, events->epoll_fd, ret); + + return timer_ctx; +} + +int neu_event_del_timer(neu_event_ctx_t *events, neu_event_timer_ctx_t *timer) +{ + struct event_data *data = NULL; + log_info("del timer: %d from epoll: %d", timer->fd, events->epoll_fd); + + epoll_ctl(events->epoll_fd, EPOLL_CTL_DEL, timer->fd, NULL); + + pthread_mutex_lock(&events->mtx); + TAILQ_FOREACH(data, &events->datas, node) + { + if (data->fd == timer->fd) { + TAILQ_REMOVE(&events->datas, data, node); + close(timer->fd); + + free(timer->event_data); + free(timer); + break; + } + } + + pthread_mutex_unlock(&events->mtx); + + return 0; +} + +neu_event_io_ctx_t *neu_event_add_io(neu_event_ctx_t *events, neu_event_io_t io) +{ + (void) events; + (void) io; + return NULL; +} + +int neu_event_del_io(neu_event_ctx_t *events, neu_event_io_ctx_t *io) +{ + (void) events; + (void) io; + return 0; +} + +#endif \ No newline at end of file diff --git a/src/event/event_unix.c b/src/event/event_unix.c new file mode 100644 index 00000000..43599cf8 --- /dev/null +++ b/src/event/event_unix.c @@ -0,0 +1,162 @@ +/** + * NEURON IIoT System for Industry 4.0 + * Copyright (C) 2020-2021 EMQ Technologies Co., Ltd All rights reserved. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 3 of the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with this program; if not, write to the Free Software Foundation, + * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. + **/ +#include +#include +#include +#include +#include +#include +#include + +#include "event/event.h" +#include "neu_log.h" + +#ifdef NEU_PLATFORM_DRAWIN + +#include + +struct neu_event_timer_ctx { + int id; + void * usr_data; + neu_event_timer_callback timer; +}; + +struct neu_event_io_ctx { +}; + +struct neu_event_ctx { + int kq; + bool running; + pthread_t thread; + pthread_mutex_t mtx; + int32_t timer_id; +}; + +static void *event_loop(void *arg) +{ + neu_event_ctx_t *events = (neu_event_ctx_t *) arg; + bool running = events->running; + + while (running) { + struct kevent event = { 0 }; + struct timespec timeout = { .tv_sec = 1, .tv_nsec = 0 }; + int ret = kevent(events->kq, NULL, 0, &event, 1, &timeout); + + if (ret == 0) { + continue; + } + + if (ret == -1) { + log_error("kevent wait error: %d(%s), fd: %d", errno, + strerror(errno), events->kq); + continue; + } + + if (event.filter == EVFILT_TIMER) { + neu_event_timer_ctx_t *ctx = (neu_event_timer_ctx_t *) event.udata; + + ret = ctx->timer(ctx->usr_data); + log_debug("timer trigger: %d, ret: %d", ctx->id, ret); + } + + pthread_mutex_lock(&events->mtx); + running = events->running; + pthread_mutex_unlock(&events->mtx); + } + + return NULL; +} + +neu_event_ctx_t *neu_event_new(void) +{ + neu_event_ctx_t *events = calloc(1, sizeof(neu_event_ctx_t)); + + pthread_mutex_init(&events->mtx, NULL); + + events->running = true; + events->kq = kqueue(); + events->timer_id = 1; + + pthread_create(&events->thread, NULL, event_loop, events); + + return events; +}; + +int neu_event_close(neu_event_ctx_t *events) +{ + pthread_mutex_lock(&events->mtx); + events->running = false; + pthread_mutex_unlock(&events->mtx); + + pthread_join(events->thread, NULL); + pthread_mutex_destroy(&events->mtx); + + free(events); + return 0; +} + +neu_event_timer_ctx_t *neu_event_add_timer(neu_event_ctx_t * events, + neu_event_timer_t timer) +{ + struct kevent ke = { 0 }; + neu_event_timer_ctx_t *ctx = calloc(1, sizeof(neu_event_timer_ctx_t)); + int ret = 0; + + ctx->id = events->timer_id++; + ctx->usr_data = timer.usr_data; + ctx->timer = timer.cb; + + EV_SET(&ke, ctx->id, EVFILT_TIMER, EV_ADD | EV_ENABLE, 0, + timer.second * 1000 + timer.millisecond, ctx); + + ret = kevent(events->kq, &ke, 1, NULL, 0, NULL); + + log_info("add timer, second: %ld, millisecond: %ld, timer: %d in epoll %d, " + "ret: %d", + timer.second, timer.millisecond, ctx->id, events->kq, ret); + return ctx; +} + +int neu_event_del_timer(neu_event_ctx_t *events, neu_event_timer_ctx_t *timer) +{ + struct kevent ke = { 0 }; + + EV_SET(&ke, timer->id, EVFILT_TIMER, EV_DELETE, 0, 0, timer); + + kevent(events->kq, &ke, 1, NULL, 0, NULL); + + free(timer); + return 0; +} + +neu_event_io_ctx_t *neu_event_add_io(neu_event_ctx_t *events, neu_event_io_t io) +{ + (void) events; + (void) io; + return NULL; +} + +int neu_event_del_io(neu_event_ctx_t *events, neu_event_io_ctx_t *io) +{ + (void) events; + (void) io; + return 0; +} + +#endif