X-Git-Url: https://git.ozlabs.org/?a=blobdiff_plain;f=ccan%2Fio%2Fpoll.c;h=f9221d1ff6d91fc398563def5d24d33ec626bdcf;hb=57d9d1be33905691ec756b14b066181ca6850ced;hp=7ceb8a7eec8bc6fd3f3de65c0c22e27a2e20e43e;hpb=02388bfc242abbd33c7ba64b2c32d425f5359b67;p=ccan diff --git a/ccan/io/poll.c b/ccan/io/poll.c index 7ceb8a7e..f9221d1f 100644 --- a/ccan/io/poll.c +++ b/ccan/io/poll.c @@ -1,4 +1,4 @@ -/* Licensed under BSD-MIT - see LICENSE file for details */ +/* Licensed under LGPLv2.1+ - see LICENSE file for details */ #include "io.h" #include "backend.h" #include @@ -6,10 +6,12 @@ #include #include #include +#include static size_t num_fds = 0, max_fds = 0, num_next = 0, num_finished = 0, num_waiting = 0; static struct pollfd *pollfds = NULL; static struct fd **fds = NULL; +static struct timers timeouts; static bool add_fd(struct fd *fd, short events) { @@ -35,6 +37,9 @@ static bool add_fd(struct fd *fd, short events) fds[num_fds] = fd; fd->backend_info = num_fds; num_fds++; + if (events) + num_waiting++; + return true; } @@ -44,6 +49,8 @@ static void del_fd(struct fd *fd) assert(n != -1); assert(n < num_fds); + if (pollfds[n].events) + num_waiting--; if (n != num_fds - 1) { /* Move last one over us. */ pollfds[n] = pollfds[num_fds-1]; @@ -67,38 +74,68 @@ bool add_listener(struct io_listener *l) { if (!add_fd(&l->fd, POLLIN)) return false; - num_waiting++; return true; } +static void adjust_counts(enum io_state state) +{ + if (state == IO_NEXT) + num_next++; + else if (state == IO_FINISHED) + num_finished++; +} + +static void update_pollevents(struct io_conn *conn) +{ + struct pollfd *pfd = &pollfds[conn->fd.backend_info]; + + if (pfd->events) + num_waiting--; + + pfd->events = conn->plan.pollflag; + if (conn->duplex) { + int mask = conn->duplex->plan.pollflag; + /* You can't *both* read/write. */ + assert(!mask || pfd->events != mask); + pfd->events |= mask; + } + if (pfd->events) + num_waiting++; + + adjust_counts(conn->plan.state); +} + bool add_conn(struct io_conn *c) { - if (!add_fd(&c->fd, 0)) + if (!add_fd(&c->fd, c->plan.pollflag)) return false; - num_next++; + adjust_counts(c->plan.state); return true; } bool add_duplex(struct io_conn *c) { c->fd.backend_info = c->duplex->fd.backend_info; - num_next++; + update_pollevents(c); return true; } static void del_conn(struct io_conn *conn) { - if (conn->fd.finish) - conn->fd.finish(conn, conn->fd.finish_arg); + if (conn->finish) + conn->finish(conn, conn->finish_arg); + if (timeout_active(conn)) + backend_del_timeout(conn); + free(conn->timeout); if (conn->duplex) { /* In case fds[] pointed to the other one. */ fds[conn->fd.backend_info] = &conn->duplex->fd; conn->duplex->duplex = NULL; } else del_fd(&conn->fd); - if (conn->state == FINISHED) + if (conn->plan.state == IO_FINISHED) num_finished--; - else if (conn->state == NEXT) + else if (conn->plan.state == IO_NEXT) num_next--; } @@ -107,59 +144,25 @@ void del_listener(struct io_listener *l) del_fd(&l->fd); } -static int pollmask(enum io_state state) +static void backend_set_state(struct io_conn *conn, struct io_plan plan) { - switch (state) { - case READ: - case READPART: - return POLLIN; - case WRITE: - case WRITEPART: - return POLLOUT; - default: - return 0; - } + conn->plan = plan; + update_pollevents(conn); } -void backend_set_state(struct io_conn *conn, struct io_op *op) +void backend_wakeup(struct io_conn *conn) { - enum io_state state = from_ioop(op); - struct pollfd *pfd = &pollfds[conn->fd.backend_info]; - - if (pfd->events) - num_waiting--; - - pfd->events = pollmask(state); - if (conn->duplex) { - int mask = pollmask(conn->duplex->state); - /* You can't *both* read/write. */ - assert(!mask || pfd->events != mask); - pfd->events |= mask; - } - if (pfd->events) - num_waiting++; - - if (state == NEXT) - num_next++; - else if (state == FINISHED) - num_finished++; - - conn->state = state; + update_pollevents(conn); } static void accept_conn(struct io_listener *l) { - struct io_conn *c; int fd = accept(l->fd.fd, NULL, NULL); /* FIXME: What to do here? */ if (fd < 0) return; - c = io_new_conn(fd, l->fd.next, l->fd.finish, l->fd.next_arg); - if (!c) { - close(fd); - return; - } + l->init(fd, l->arg); } /* It's OK to miss some, as long as we make progress. */ @@ -178,14 +181,12 @@ static void finish_and_next(bool finished_only) continue; c = (void *)fds[i]; for (duplex = c->duplex; c; c = duplex, duplex = NULL) { - if (c->state == FINISHED) { + if (c->plan.state == IO_FINISHED) { del_conn(c); free(c); i--; - } else if (!finished_only && c->state == NEXT) { - backend_set_state(c, - c->fd.next(c, - c->fd.next_arg)); + } else if (!finished_only && c->plan.state == IO_NEXT) { + backend_set_state(c, c->plan.next(c, c->plan.next_arg)); num_next--; } } @@ -197,13 +198,54 @@ static void ready(struct io_conn *c) backend_set_state(c, do_ready(c)); } +void backend_add_timeout(struct io_conn *conn, struct timespec duration) +{ + if (!timeouts.base) + timers_init(&timeouts, time_now()); + timer_add(&timeouts, &conn->timeout->timer, + time_add(time_now(), duration)); + conn->timeout->conn = conn; +} + +void backend_del_timeout(struct io_conn *conn) +{ + assert(conn->timeout->conn == conn); + timer_del(&timeouts, &conn->timeout->timer); + conn->timeout->conn = NULL; +} + /* This is the main loop. */ void *io_loop(void) { void *ret; while (!io_loop_return) { - int i, r; + int i, r, timeout = INT_MAX; + struct timespec now; + + if (timeouts.base) { + struct timespec first; + struct list_head expired; + struct io_timeout *t; + + now = time_now(); + + /* Call functions for expired timers. */ + timers_expire(&timeouts, now, &expired); + while ((t = list_pop(&expired, struct io_timeout, timer.list))) { + struct io_conn *conn = t->conn; + /* Clear, in case timer re-adds */ + t->conn = NULL; + backend_set_state(conn, t->next(conn, t->next_arg)); + } + + /* Now figure out how long to wait for the next one. */ + if (timer_earliest(&timeouts, &first)) { + uint64_t f = time_to_msec(time_sub(first, now)); + if (f < INT_MAX) + timeout = f; + } + } if (num_finished || num_next) { finish_and_next(false); @@ -217,7 +259,7 @@ void *io_loop(void) /* You can't tell them all to go to sleep! */ assert(num_waiting); - r = poll(pollfds, num_fds, -1); + r = poll(pollfds, num_fds, timeout); if (r < 0) break; @@ -225,12 +267,18 @@ void *io_loop(void) struct io_conn *c = (void *)fds[i]; int events = pollfds[i].revents; + if (r == 0) + break; + if (fds[i]->listener) { - if (events & POLLIN) + if (events & POLLIN) { accept_conn((void *)c); + r--; + } } else if (events & (POLLIN|POLLOUT)) { + r--; if (c->duplex) { - int mask = pollmask(c->duplex->state); + int mask = c->duplex->plan.pollflag; if (events & mask) { ready(c->duplex); events &= ~mask; @@ -240,13 +288,13 @@ void *io_loop(void) } ready(c); } else if (events & POLLHUP) { + r--; backend_set_state(c, io_close(c, NULL)); if (c->duplex) backend_set_state(c->duplex, io_close(c->duplex, NULL)); } - } }