mirror of
https://github.com/emqx/neuron.git
synced 2026-10-18 13:41:22 +08:00
add asynchronous io and timer
asyn io is not implemented use epoll/timerfd on linux use kqueue on unix
This commit is contained in:
fengzero
@@ -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
|
||||
|
||||
@@ -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 <stdint.h>
|
||||
|
||||
#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
|
||||
@@ -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 <errno.h>
|
||||
#include <pthread.h>
|
||||
#include <stdbool.h>
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <sys/queue.h>
|
||||
|
||||
#include "event/event.h"
|
||||
#include "neu_log.h"
|
||||
|
||||
#ifdef NEU_PLATFORM_LINUX
|
||||
#include <sys/epoll.h>
|
||||
#include <sys/timerfd.h>
|
||||
|
||||
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
|
||||
@@ -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 <errno.h>
|
||||
#include <pthread.h>
|
||||
#include <stdbool.h>
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <sys/queue.h>
|
||||
|
||||
#include "event/event.h"
|
||||
#include "neu_log.h"
|
||||
|
||||
#ifdef NEU_PLATFORM_DRAWIN
|
||||
|
||||
#include <sys/event.h>
|
||||
|
||||
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
|
||||
Reference in New Issue
Block a user