git.ozlabs.org Git - ccan/blob - ccan/io/poll.c

   1 /* Licensed under LGPLv2.1+ - see LICENSE file for details */
   2 #include "io.h"
   3 #include "backend.h"
   4 #include <assert.h>
   5 #include <poll.h>
   6 #include <stdlib.h>
   7 #include <sys/types.h>
   8 #include <sys/socket.h>
   9 #include <limits.h>
  10 #include <errno.h>
  11
  12 static size_t num_fds = 0, max_fds = 0, num_closing = 0, num_waiting = 0;
  13 static struct pollfd *pollfds = NULL;
  14 static struct fd **fds = NULL;
  15 static struct timers timeouts;
  16 #ifdef DEBUG
  17 static unsigned int io_loop_level;
  18 static struct io_conn *free_later;
  19 static void io_loop_enter(void)
  20 {
  21         io_loop_level++;
  22 }
  23 static void io_loop_exit(void)
  24 {
  25         io_loop_level--;
  26         if (io_loop_level == 0) {
  27                 /* Delayed free. */
  28                 while (free_later) {
  29                         struct io_conn *c = free_later;
  30                         free_later = c->finish_arg;
  31                         free(c);
  32                 }
  33         }
  34 }
  35 static void free_conn(struct io_conn *conn)
  36 {
  37         /* Only free on final exit: chain via finish. */
  38         if (io_loop_level > 1) {
  39                 struct io_conn *c;
  40                 for (c = free_later; c; c = c->finish_arg)
  41                         assert(c != conn);
  42                 conn->finish_arg = free_later;
  43                 free_later = conn;
  44         } else
  45                 free(conn);
  46 }
  47 #else
  48 static void io_loop_enter(void)
  49 {
  50 }
  51 static void io_loop_exit(void)
  52 {
  53 }
  54 static void free_conn(struct io_conn *conn)
  55 {
  56         free(conn);
  57 }
  58 #endif
  59
  60 static bool add_fd(struct fd *fd, short events)
  61 {
  62         if (num_fds + 1 > max_fds) {
  63                 struct pollfd *newpollfds;
  64                 struct fd **newfds;
  65                 size_t num = max_fds ? max_fds * 2 : 8;
  66
  67                 newpollfds = realloc(pollfds, sizeof(*newpollfds) * num);
  68                 if (!newpollfds)
  69                         return false;
  70                 pollfds = newpollfds;
  71                 newfds = realloc(fds, sizeof(*newfds) * num);
  72                 if (!newfds)
  73                         return false;
  74                 fds = newfds;
  75                 max_fds = num;
  76         }
  77
  78         pollfds[num_fds].events = events;
  79         /* In case it's idle. */
  80         if (!events)
  81                 pollfds[num_fds].fd = -fd->fd;
  82         else
  83                 pollfds[num_fds].fd = fd->fd;
  84         pollfds[num_fds].revents = 0; /* In case we're iterating now */
  85         fds[num_fds] = fd;
  86         fd->backend_info = num_fds;
  87         num_fds++;
  88         if (events)
  89                 num_waiting++;
  90
  91         return true;
  92 }
  93
  94 static void del_fd(struct fd *fd)
  95 {
  96         size_t n = fd->backend_info;
  97
  98         assert(n != -1);
  99         assert(n < num_fds);
 100         if (pollfds[n].events)
 101                 num_waiting--;
 102         if (n != num_fds - 1) {
 103                 /* Move last one over us. */
 104                 pollfds[n] = pollfds[num_fds-1];
 105                 fds[n] = fds[num_fds-1];
 106                 assert(fds[n]->backend_info == num_fds-1);
 107                 fds[n]->backend_info = n;
 108         } else if (num_fds == 1) {
 109                 /* Free everything when no more fds. */
 110                 free(pollfds);
 111                 free(fds);
 112                 pollfds = NULL;
 113                 fds = NULL;
 114                 max_fds = 0;
 115         }
 116         num_fds--;
 117         fd->backend_info = -1;
 118         close(fd->fd);
 119 }
 120
 121 bool add_listener(struct io_listener *l)
 122 {
 123         if (!add_fd(&l->fd, POLLIN))
 124                 return false;
 125         return true;
 126 }
 127
 128 void backend_plan_changed(struct io_conn *conn)
 129 {
 130         struct pollfd *pfd;
 131
 132         /* This can happen with debugging and delayed free... */
 133         if (conn->fd.backend_info == -1)
 134                 return;
 135
 136         pfd = &pollfds[conn->fd.backend_info];
 137
 138         if (pfd->events)
 139                 num_waiting--;
 140
 141         pfd->events = conn->plan.pollflag;
 142         if (conn->duplex) {
 143                 int mask = conn->duplex->plan.pollflag;
 144                 /* You can't *both* read/write. */
 145                 assert(!mask || pfd->events != mask);
 146                 pfd->events |= mask;
 147         }
 148         if (pfd->events) {
 149                 num_waiting++;
 150                 pfd->fd = conn->fd.fd;
 151         } else
 152                 pfd->fd = -conn->fd.fd;
 153
 154         if (!conn->plan.next)
 155                 num_closing++;
 156 }
 157
 158 bool add_conn(struct io_conn *c)
 159 {
 160         if (!add_fd(&c->fd, c->plan.pollflag))
 161                 return false;
 162         /* Immediate close is allowed. */
 163         if (!c->plan.next)
 164                 num_closing++;
 165         return true;
 166 }
 167
 168 bool add_duplex(struct io_conn *c)
 169 {
 170         c->fd.backend_info = c->duplex->fd.backend_info;
 171         backend_plan_changed(c);
 172         return true;
 173 }
 174
 175 static void del_conn(struct io_conn *conn)
 176 {
 177         if (conn->finish) {
 178                 errno = conn->plan.u.close.saved_errno;
 179                 conn->finish(conn, conn->finish_arg);
 180         }
 181         if (timeout_active(conn))
 182                 backend_del_timeout(conn);
 183         free(conn->timeout);
 184         if (conn->duplex) {
 185                 /* In case fds[] pointed to the other one. */
 186                 fds[conn->fd.backend_info] = &conn->duplex->fd;
 187                 conn->duplex->duplex = NULL;
 188                 conn->fd.backend_info = -1;
 189         } else
 190                 del_fd(&conn->fd);
 191         num_closing--;
 192 }
 193
 194 void del_listener(struct io_listener *l)
 195 {
 196         del_fd(&l->fd);
 197 }
 198
 199 static void set_plan(struct io_conn *conn, struct io_plan plan)
 200 {
 201         conn->plan = plan;
 202         backend_plan_changed(conn);
 203 }
 204
 205 static void accept_conn(struct io_listener *l)
 206 {
 207         int fd = accept(l->fd.fd, NULL, NULL);
 208
 209         /* FIXME: What to do here? */
 210         if (fd < 0)
 211                 return;
 212         l->init(fd, l->arg);
 213 }
 214
 215 /* It's OK to miss some, as long as we make progress. */
 216 static void finish_conns(void)
 217 {
 218         unsigned int i;
 219
 220         for (i = 0; !io_loop_return && i < num_fds; i++) {
 221                 struct io_conn *c, *duplex;
 222
 223                 if (!num_closing)
 224                         break;
 225
 226                 if (fds[i]->listener)
 227                         continue;
 228                 c = (void *)fds[i];
 229                 for (duplex = c->duplex; c; c = duplex, duplex = NULL) {
 230                         if (!c->plan.next) {
 231                                 del_conn(c);
 232                                 free_conn(c);
 233                                 i--;
 234                         }
 235                 }
 236         }
 237 }
 238
 239 void backend_add_timeout(struct io_conn *conn, struct timespec duration)
 240 {
 241         if (!timeouts.base)
 242                 timers_init(&timeouts, time_now());
 243         timer_add(&timeouts, &conn->timeout->timer,
 244                   time_add(time_now(), duration));
 245         conn->timeout->conn = conn;
 246 }
 247
 248 void backend_del_timeout(struct io_conn *conn)
 249 {
 250         assert(conn->timeout->conn == conn);
 251         timer_del(&timeouts, &conn->timeout->timer);
 252         conn->timeout->conn = NULL;
 253 }
 254
 255 /* This is the main loop. */
 256 void *io_loop(void)
 257 {
 258         void *ret;
 259
 260         io_loop_enter();
 261
 262         while (!io_loop_return) {
 263                 int i, r, timeout = INT_MAX;
 264                 struct timespec now;
 265                 bool some_timeouts = false;
 266
 267                 if (timeouts.base) {
 268                         struct timespec first;
 269                         struct list_head expired;
 270                         struct io_timeout *t;
 271
 272                         now = time_now();
 273
 274                         /* Call functions for expired timers. */
 275                         timers_expire(&timeouts, now, &expired);
 276                         while ((t = list_pop(&expired, struct io_timeout, timer.list))) {
 277                                 struct io_conn *conn = t->conn;
 278                                 /* Clear, in case timer re-adds */
 279                                 t->conn = NULL;
 280                                 set_current(conn);
 281                                 set_plan(conn, t->next(conn, t->next_arg));
 282                                 some_timeouts = true;
 283                         }
 284
 285                         /* Now figure out how long to wait for the next one. */
 286                         if (timer_earliest(&timeouts, &first)) {
 287                                 uint64_t f = time_to_msec(time_sub(first, now));
 288                                 if (f < INT_MAX)
 289                                         timeout = f;
 290                         }
 291                 }
 292
 293                 if (num_closing) {
 294                         finish_conns();
 295                         /* Could have started/finished more. */
 296                         continue;
 297                 }
 298
 299                 /* debug can recurse on io_loop; anything can change. */
 300                 if (doing_debug() && some_timeouts)
 301                         continue;
 302
 303                 if (num_fds == 0)
 304                         break;
 305
 306                 /* You can't tell them all to go to sleep! */
 307                 assert(num_waiting);
 308
 309                 r = poll(pollfds, num_fds, timeout);
 310                 if (r < 0)
 311                         break;
 312
 313                 for (i = 0; i < num_fds && !io_loop_return; i++) {
 314                         struct io_conn *c = (void *)fds[i];
 315                         int events = pollfds[i].revents;
 316
 317                         if (r == 0)
 318                                 break;
 319
 320                         if (fds[i]->listener) {
 321                                 if (events & POLLIN) {
 322                                         accept_conn((void *)c);
 323                                         r--;
 324                                 }
 325                         } else if (events & (POLLIN|POLLOUT)) {
 326                                 r--;
 327                                 if (c->duplex) {
 328                                         int mask = c->duplex->plan.pollflag;
 329                                         if (events & mask) {
 330                                                 io_ready(c->duplex);
 331                                                 events &= ~mask;
 332                                                 /* debug can recurse;
 333                                                  * anything can change. */
 334                                                 if (doing_debug())
 335                                                         break;
 336                                                 if (!(events&(POLLIN|POLLOUT)))
 337                                                         continue;
 338                                         }
 339                                 }
 340                                 io_ready(c);
 341                                 /* debug can recurse; anything can change. */
 342                                 if (doing_debug())
 343                                         break;
 344                         } else if (events & (POLLHUP|POLLNVAL|POLLERR)) {
 345                                 r--;
 346                                 set_current(c);
 347                                 errno = EBADF;
 348                                 set_plan(c, io_close());
 349                                 if (c->duplex) {
 350                                         set_current(c->duplex);
 351                                         set_plan(c->duplex, io_close());
 352                                 }
 353                         }
 354                 }
 355         }
 356
 357         while (num_closing)
 358                 finish_conns();
 359
 360         ret = io_loop_return;
 361         io_loop_return = NULL;
 362
 363         io_loop_exit();
 364         return ret;
 365 }