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 void backend_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         free_conn(conn);
 193 }
 194
 195 void del_listener(struct io_listener *l)
 196 {
 197         del_fd(&l->fd);
 198 }
 199
 200 static void set_plan(struct io_conn *conn, struct io_plan plan)
 201 {
 202         conn->plan = plan;
 203         backend_plan_changed(conn);
 204 }
 205
 206 static void accept_conn(struct io_listener *l)
 207 {
 208         int fd = accept(l->fd.fd, NULL, NULL);
 209
 210         /* FIXME: What to do here? */
 211         if (fd < 0)
 212                 return;
 213         l->init(fd, l->arg);
 214 }
 215
 216 /* It's OK to miss some, as long as we make progress. */
 217 static bool finish_conns(struct io_conn **ready)
 218 {
 219         unsigned int i;
 220
 221         for (i = 0; !io_loop_return && i < num_fds; i++) {
 222                 struct io_conn *c, *duplex;
 223
 224                 if (!num_closing)
 225                         break;
 226
 227                 if (fds[i]->listener)
 228                         continue;
 229                 c = (void *)fds[i];
 230                 for (duplex = c->duplex; c; c = duplex, duplex = NULL) {
 231                         if (!c->plan.next) {
 232                                 if (doing_debug_on(c) && ready) {
 233                                         *ready = c;
 234                                         return true;
 235                                 }
 236                                 backend_del_conn(c);
 237                                 i--;
 238                         }
 239                 }
 240         }
 241         return false;
 242 }
 243
 244 void backend_add_timeout(struct io_conn *conn, struct timespec duration)
 245 {
 246         if (!timeouts.base)
 247                 timers_init(&timeouts, time_now());
 248         timer_add(&timeouts, &conn->timeout->timer,
 249                   time_add(time_now(), duration));
 250         conn->timeout->conn = conn;
 251 }
 252
 253 void backend_del_timeout(struct io_conn *conn)
 254 {
 255         assert(conn->timeout->conn == conn);
 256         timer_del(&timeouts, &conn->timeout->timer);
 257         conn->timeout->conn = NULL;
 258 }
 259
 260 /* This is the main loop. */
 261 void *do_io_loop(struct io_conn **ready)
 262 {
 263         void *ret;
 264
 265         io_loop_enter();
 266
 267         while (!io_loop_return) {
 268                 int i, r, timeout = INT_MAX;
 269                 struct timespec now;
 270                 bool some_timeouts = false;
 271
 272                 if (timeouts.base) {
 273                         struct timespec first;
 274                         struct list_head expired;
 275                         struct io_timeout *t;
 276
 277                         now = time_now();
 278
 279                         /* Call functions for expired timers. */
 280                         timers_expire(&timeouts, now, &expired);
 281                         while ((t = list_pop(&expired, struct io_timeout, timer.list))) {
 282                                 struct io_conn *conn = t->conn;
 283                                 /* Clear, in case timer re-adds */
 284                                 t->conn = NULL;
 285                                 set_current(conn);
 286                                 set_plan(conn, t->next(conn, t->next_arg));
 287                                 some_timeouts = true;
 288                         }
 289
 290                         /* Now figure out how long to wait for the next one. */
 291                         if (timer_earliest(&timeouts, &first)) {
 292                                 uint64_t f = time_to_msec(time_sub(first, now));
 293                                 if (f < INT_MAX)
 294                                         timeout = f;
 295                         }
 296                 }
 297
 298                 if (num_closing) {
 299                         /* If this finishes a debugging con, return now. */
 300                         if (finish_conns(ready))
 301                                 return NULL;
 302                         /* Could have started/finished more. */
 303                         continue;
 304                 }
 305
 306                 /* debug can recurse on io_loop; anything can change. */
 307                 if (doing_debug() && some_timeouts)
 308                         continue;
 309
 310                 if (num_fds == 0)
 311                         break;
 312
 313                 /* You can't tell them all to go to sleep! */
 314                 assert(num_waiting);
 315
 316                 r = poll(pollfds, num_fds, timeout);
 317                 if (r < 0)
 318                         break;
 319
 320                 for (i = 0; i < num_fds && !io_loop_return; i++) {
 321                         struct io_conn *c = (void *)fds[i];
 322                         int events = pollfds[i].revents;
 323
 324                         if (r == 0)
 325                                 break;
 326
 327                         if (fds[i]->listener) {
 328                                 if (events & POLLIN) {
 329                                         accept_conn((void *)c);
 330                                         r--;
 331                                 }
 332                         } else if (events & (POLLIN|POLLOUT)) {
 333                                 r--;
 334                                 if (c->duplex) {
 335                                         int mask = c->duplex->plan.pollflag;
 336                                         if (events & mask) {
 337                                                 if (doing_debug_on(c->duplex)
 338                                                         && ready) {
 339                                                         *ready = c->duplex;
 340                                                         return NULL;
 341                                                 }
 342                                                 io_ready(c->duplex);
 343                                                 events &= ~mask;
 344                                                 /* debug can recurse;
 345                                                  * anything can change. */
 346                                                 if (doing_debug())
 347                                                         break;
 348                                                 if (!(events&(POLLIN|POLLOUT)))
 349                                                         continue;
 350                                         }
 351                                 }
 352                                 if (doing_debug_on(c) && ready) {
 353                                         *ready = c;
 354                                         return NULL;
 355                                 }
 356                                 io_ready(c);
 357                                 /* debug can recurse; anything can change. */
 358                                 if (doing_debug())
 359                                         break;
 360                         } else if (events & (POLLHUP|POLLNVAL|POLLERR)) {
 361                                 r--;
 362                                 set_current(c);
 363                                 errno = EBADF;
 364                                 set_plan(c, io_close());
 365                                 if (c->duplex) {
 366                                         set_current(c->duplex);
 367                                         set_plan(c->duplex, io_close());
 368                                 }
 369                         }
 370                 }
 371         }
 372
 373         while (num_closing && !io_loop_return) {
 374                 if (finish_conns(ready))
 375                         return NULL;
 376         }
 377
 378         ret = io_loop_return;
 379         io_loop_return = NULL;
 380
 381         io_loop_exit();
 382         return ret;
 383 }
 384
 385 void *io_loop(void)
 386 {
 387         return do_io_loop(NULL);
 388 }