GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-23 18:37:45 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1558 from fengzeroz/main

Browse Source 
fix(event): lock event loop
This commit is contained in:
fengzeroz
2023-09-22 17:36:11 +08:00
committed by GitHub
parent 4cc7aaae63 0bdb3ecc5e
commit 555e77fe5b
1 changed files with 84 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/event/event_linux.c
+84 -69
View File
@@ -38,43 +38,70 @@
struct neu_event_timer {
int fd;
void * event_data;
struct event_data * event_data;
struct itimerspec value;
neu_event_timer_type_e type;
nng_mtx *mtx;
bool stop;
};
struct neu_event_io {
int fd;
void *event_data;
int fd;
struct event_data *event_data;
};
struct event_data {
enum {
TIMER,
IO,
TIMER = 0,
IO = 1,
} type;
union {
neu_event_io_callback io;
neu_event_timer_callback timer;
} callback;
union {
struct neu_event_io * io;
struct neu_event_timer *timer;
neu_event_io_t io;
neu_event_timer_t timer;
} ctx;
void *usr_data;
int fd;
int index;
bool use;
};
struct neu_events {
int epoll_fd;
pthread_t thread;
bool stop;
pthread_mutex_t mtx;
int n_event;
struct event_data event_datas[256];
};
static int get_free_event(neu_events_t *events)
{
int ret = -1;
pthread_mutex_lock(&events->mtx);
for (int i = 0; i < 256; i++) {
if (events->event_datas[i].use == false) {
events->event_datas[i].use = true;
events->event_datas[i].index = i;
ret = i;
break;
}
}
pthread_mutex_unlock(&events->mtx);
return ret;
}
static void free_event(neu_events_t *events, int index)
{
pthread_mutex_lock(&events->mtx);
events->event_datas[index].use = false;
events->event_datas[index].index = 0;
pthread_mutex_unlock(&events->mtx);
}
static void *event_loop(void *arg)
{
neu_events_t *events = (neu_events_t *) arg;
@@ -106,22 +133,17 @@ static void *event_loop(void *arg)
if ((event.events & EPOLLIN) == EPOLLIN) {
uint64_t t;
nng_mtx_lock(data->ctx.timer->mtx);
ssize_t size = read(data->fd, &t, sizeof(t));
(void) size;
if (!data->ctx.timer->stop) {
if (data->ctx.timer->type == NEU_EVENT_TIMER_BLOCK) {
epoll_ctl(epoll_fd, EPOLL_CTL_DEL, data->fd, NULL);
ret = data->callback.timer(data->usr_data);
timerfd_settime(data->fd, 0, &data->ctx.timer->value,
NULL);
epoll_ctl(epoll_fd, EPOLL_CTL_ADD, data->fd, &event);
} else {
ret = data->callback.timer(data->usr_data);
}
if (data->ctx.timer.type == NEU_EVENT_TIMER_BLOCK) {
epoll_ctl(epoll_fd, EPOLL_CTL_DEL, data->fd, NULL);
ret = data->callback.timer(data->usr_data);
timerfd_settime(data->fd, 0, &data->ctx.timer.value, NULL);
epoll_ctl(epoll_fd, EPOLL_CTL_ADD, data->fd, &event);
} else {
ret = data->callback.timer(data->usr_data);
}
nng_mtx_unlock(data->ctx.timer->mtx);
}
break;
@@ -155,6 +177,8 @@ neu_events_t *neu_event_new(void)
events->epoll_fd = epoll_create(1);
events->stop = false;
events->n_event = 0;
pthread_mutex_init(&events->mtx, NULL);
pthread_create(&events->thread, NULL, event_loop, events);
@@ -167,6 +191,7 @@ int neu_event_close(neu_events_t *events)
close(events->epoll_fd);
pthread_join(events->thread, NULL);
pthread_mutex_destroy(&events->mtx);
free(events);
return 0;
@@ -183,24 +208,28 @@ neu_event_timer_t *neu_event_add_timer(neu_events_t * events,
.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_t *timer_ctx = calloc(1, sizeof(neu_event_timer_t));
int index = get_free_event(events);
assert(index >= 0);
neu_event_timer_t *timer_ctx = &events->event_datas[index].ctx.timer;
timer_ctx->event_data = &events->event_datas[index];
struct epoll_event event = {
.events = EPOLLIN,
.data.ptr = timer_ctx->event_data,
};
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->event_data->type = TIMER;
timer_ctx->event_data->fd = timer_fd;
timer_ctx->event_data->usr_data = timer.usr_data;
timer_ctx->event_data->callback.timer = timer.cb;
timer_ctx->event_data->ctx.timer = events->event_datas[index].ctx.timer;
timer_ctx->value = value;
timer_ctx->fd = timer_fd;
timer_ctx->event_data = data;
timer_ctx->stop = false;
timer_ctx->type = timer.type;
nng_mtx_alloc(&timer_ctx->mtx);
timer_ctx->value = value;
timer_ctx->fd = timer_fd;
timer_ctx->type = timer.type;
ret = epoll_ctl(events->epoll_fd, EPOLL_CTL_ADD, timer_fd, &event);
@@ -219,37 +248,31 @@ int neu_event_del_timer(neu_events_t *events, neu_event_timer_t *timer)
zlog_notice(neuron, "del timer: %d from epoll: %d", timer->fd,
events->epoll_fd);
timer->stop = true;
nng_mtx_lock(timer->mtx);
close(timer->fd);
epoll_ctl(events->epoll_fd, EPOLL_CTL_DEL, timer->fd, NULL);
free(timer->event_data);
nng_mtx_unlock(timer->mtx);
nng_mtx_free(timer->mtx);
free(timer);
close(timer->fd);
free_event(events, timer->event_data->index);
return 0;
}
neu_event_io_t *neu_event_add_io(neu_events_t *events, neu_event_io_param_t io)
{
int ret = 0;
neu_event_io_t * io_ctx = calloc(1, sizeof(neu_event_io_t));
struct event_data *data = calloc(1, sizeof(struct event_data));
struct epoll_event event = { .events =
EPOLLIN | EPOLLERR | EPOLLHUP | EPOLLRDHUP,
.data.ptr = data };
int ret = 0;
int index = get_free_event(events);
neu_event_io_t *io_ctx = &events->event_datas[index].ctx.io;
io_ctx->event_data = &events->event_datas[index];
struct epoll_event event = {
.events = EPOLLIN | EPOLLERR | EPOLLHUP | EPOLLRDHUP,
.data.ptr = io_ctx->event_data,
};
data->type = IO;
data->fd = io.fd;
data->usr_data = io.usr_data;
data->callback.io = io.cb;
data->ctx.io = io_ctx;
io_ctx->event_data->type = IO;
io_ctx->event_data->fd = io.fd;
io_ctx->event_data->usr_data = io.usr_data;
io_ctx->event_data->callback.io = io.cb;
io_ctx->event_data->ctx.io = events->event_datas[index].ctx.io;
io_ctx->fd = io.fd;
io_ctx->event_data = data;
io_ctx->fd = io.fd;
ret = epoll_ctl(events->epoll_fd, EPOLL_CTL_ADD, io.fd, &event);
@@ -265,19 +288,11 @@ int neu_event_del_io(neu_events_t *events, neu_event_io_t *io)
return 0;
}
struct event_data *data = (struct event_data *) io->event_data;
zlog_notice(neuron, "del io: %d from epoll: %d", io->fd, events->epoll_fd);
epoll_ctl(events->epoll_fd, EPOLL_CTL_DEL, io->fd, NULL);
switch (data->type) {
case TIMER:
free(data->ctx.timer);
break;
case IO:
free(data->ctx.io);
break;
}
free(data);
close(io->fd);
free_event(events, io->event_data->index);
return 0;
}