42 #include <sys/epoll.h>
45 #define EPOLL_EVENT_INIT(events, fd) \
47 (((events) & IO_EVENT_IN) ? (EPOLLIN | EPOLLRDHUP) : 0) \
48 | (((events) & IO_EVENT_PRI) ? EPOLLPRI : 0) \
49 | (((events) & IO_EVENT_OUT) ? EPOLLOUT : 0) \
55 #ifndef LELY_IO_EPOLL_MAXEVENTS
56 #define LELY_IO_EPOLL_MAXEVENTS \
57 MAX((LELY_VLA_SIZE_MAX / sizeof(struct epoll_event)), 1)
70 static const struct io_svc_vtbl io_poll_svc_vtbl = {
71 &io_poll_svc_notify_fork,
76 static void *io_poll_poll_self(
const ev_poll_t *poll);
77 static int io_poll_poll_wait(
ev_poll_t *poll,
int timeout);
78 static int io_poll_poll_kill(
ev_poll_t *poll,
void *thr);
93 struct sigaction oact;
106 static int io_poll_open(
io_poll_t *poll);
107 static int io_poll_close(
io_poll_t *poll);
109 static void io_poll_process(
112 static int io_fd_cmp(
const void *p1,
const void *p2);
117 static void sig_ign(
int signo);
126 io_poll_free(
void *ptr)
145 poll->poll_vptr = &io_poll_poll_vtbl;
149 struct sigaction act;
150 act.sa_handler = &sig_ign;
151 sigemptyset(&act.sa_mask);
153 if (sigaction(poll->signo, &act, &poll->oact) == -1) {
155 goto error_sigaction;
161 sigaddset(&set, poll->signo);
163 if (sigprocmask(SIG_BLOCK, &set, &poll->oset) == -1) {
166 if ((errsv = pthread_sigmask(SIG_BLOCK, &set, &poll->oset))) {
173 if ((errsv = pthread_mutex_init(&poll->
mtx, NULL)))
180 if (io_poll_open(poll) == -1) {
192 pthread_mutex_destroy(&poll->
mtx);
196 sigprocmask(SIG_SETMASK, &poll->oset, NULL);
198 pthread_sigmask(SIG_SETMASK, &poll->oset, NULL);
201 sigaction(poll->signo, &poll->oact, NULL);
217 pthread_mutex_destroy(&poll->
mtx);
223 sigaddset(&set, poll->signo);
225 while (sigtimedwait(&set, &(siginfo_t){ 0 }, &(
struct timespec){ 0, 0 })
231 sigprocmask(SIG_SETMASK, &poll->oset, NULL);
233 pthread_sigmask(SIG_SETMASK, &poll->oset, NULL);
235 sigaction(poll->signo, &poll->oact, NULL);
249 io_poll_t *tmp = io_poll_init(poll, ctx, signo);
283 return &poll->poll_vptr;
299 int epfd = poll->
epfd;
301 if (fd == -1 || fd == epfd) {
310 events &= IO_EVENT_MASK;
315 pthread_mutex_lock(&poll->
mtx);
319 if (node && node != &watch->_node) {
325 struct epoll_event event = EPOLL_EVENT_INIT(events, fd);
326 if (node && events != watch->_events) {
327 if (epoll_ctl(epfd, EPOLL_CTL_MOD, fd, &event) == -1) {
329 epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL);
337 if (epoll_ctl(epfd, EPOLL_CTL_ADD, fd, &event) == -1) {
348 watch->_events = events;
350 epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL);
361 pthread_mutex_unlock(&poll->
mtx);
378 if (io_poll_close(poll) == -1 && !result) {
383 if (io_poll_open(poll) == -1 && !result) {
388 int epfd = poll->
epfd;
392 int events = watch->_events;
394 struct epoll_event event = EPOLL_EVENT_INIT(events, fd);
396 if (epoll_ctl(epfd, EPOLL_CTL_ADD, fd, &event) == -1) {
423 thread = pthread_self();
424 thr.thread = &thread;
431 io_poll_poll_wait(
ev_poll_t *poll_,
int timeout)
433 io_poll_t *poll = io_poll_from_poll(poll_);
434 void *thr_ = io_poll_poll_self(poll_);
443 struct epoll_event events[LELY_IO_EPOLL_MAXEVENTS];
446 pthread_mutex_lock(&poll->
mtx);
460 pthread_mutex_unlock(&poll->
mtx);
463 int nevents = epoll_pwait(poll->
epfd, events,
464 LELY_IO_EPOLL_MAXEVENTS, timeout, &set);
465 if (nevents == -1 && errno == EINTR) {
472 sigaddset(&set, poll->signo);
474 pthread_mutex_lock(&poll->
mtx);
485 pthread_mutex_lock(&poll->
mtx);
490 for (
int i = 0; i < nevents; i++) {
492 if (events[i].events & (EPOLLIN | EPOLLRDHUP))
494 if (events[i].events & EPOLLPRI)
496 if (events[i].events & EPOLLOUT)
498 if (events[i].events & EPOLLERR)
500 if (events[i].events & EPOLLHUP)
502 int fd = events[i].data.fd;
504 pthread_mutex_lock(&poll->
mtx);
509 io_poll_watch_from_node(node);
510 io_poll_process(poll, revents, watch);
514 pthread_mutex_unlock(&poll->
mtx);
519 pthread_mutex_lock(&poll->
mtx);
522 stopped = nevents != LELY_IO_EPOLL_MAXEVENTS;
526 pthread_mutex_unlock(&poll->
mtx);
534 io_poll_poll_kill(
ev_poll_t *poll_,
void *thr_)
537 io_poll_t *poll = io_poll_from_poll(poll_);
542 if (thr_ == io_poll_poll_self(poll_))
546 pthread_mutex_lock(&poll->
mtx);
548 int stopped = thr->stopped;
552 pthread_mutex_unlock(&poll->
mtx);
554 int errsv = pthread_kill(*thr->thread, poll->signo);
565 io_poll_from_svc(
const struct io_svc *svc)
583 if (io_poll_close(poll) == -1)
586 return (poll->
epfd = epoll_create1(EPOLL_CLOEXEC)) == -1 ? -1 : 0;
594 int epfd = poll->
epfd;
606 assert(poll->nwatch);
608 assert(watch->_events);
615 pthread_mutex_unlock(&poll->
mtx);
617 watch->
func(watch, revents);
619 pthread_mutex_lock(&poll->
mtx);
625 io_fd_cmp(
const void *p1,
const void *p2)
628 int fd1 = *(
const int *)p1;
630 int fd2 = *(
const int *)p2;
632 return (fd2 < fd1) - (fd1 < fd2);
636 io_poll_watch_from_node(
struct rbnode *node)