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

   1 /* Licensed under BSD-MIT - 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
  10 static size_t num_fds = 0, max_fds = 0, num_next = 0, num_finished = 0, num_waiting = 0;
  11 static struct pollfd *pollfds = NULL;
  12 static struct fd **fds = NULL;
  13
  14 static bool add_fd(struct fd *fd, short events)
  15 {
  16         if (num_fds + 1 > max_fds) {
  17                 struct pollfd *newpollfds;
  18                 struct fd **newfds;
  19                 size_t num = max_fds ? max_fds * 2 : 8;
  20
  21                 newpollfds = realloc(pollfds, sizeof(*newpollfds) * num);
  22                 if (!newpollfds)
  23                         return false;
  24                 pollfds = newpollfds;
  25                 newfds = realloc(fds, sizeof(*newfds) * num);
  26                 if (!newfds)
  27                         return false;
  28                 fds = newfds;
  29                 max_fds = num;
  30         }
  31
  32         pollfds[num_fds].fd = fd->fd;
  33         pollfds[num_fds].events = events;
  34         pollfds[num_fds].revents = 0; /* In case we're iterating now */
  35         fds[num_fds] = fd;
  36         fd->backend_info = num_fds;
  37         num_fds++;
  38         return true;
  39 }
  40
  41 static void del_fd(struct fd *fd)
  42 {
  43         size_t n = fd->backend_info;
  44
  45         assert(n != -1);
  46         assert(n < num_fds);
  47         if (n != num_fds - 1) {
  48                 /* Move last one over us. */
  49                 pollfds[n] = pollfds[num_fds-1];
  50                 fds[n] = fds[num_fds-1];
  51                 assert(fds[n]->backend_info == num_fds-1);
  52                 fds[n]->backend_info = n;
  53         } else if (num_fds == 1) {
  54                 /* Free everything when no more fds. */
  55                 free(pollfds);
  56                 free(fds);
  57                 pollfds = NULL;
  58                 fds = NULL;
  59                 max_fds = 0;
  60         }
  61         num_fds--;
  62         fd->backend_info = -1;
  63         close(fd->fd);
  64 }
  65
  66 bool add_listener(struct io_listener *l)
  67 {
  68         if (!add_fd(&l->fd, POLLIN))
  69                 return false;
  70         num_waiting++;
  71         return true;
  72 }
  73
  74 bool add_conn(struct io_conn *c)
  75 {
  76         if (!add_fd(&c->fd, 0))
  77                 return false;
  78         num_next++;
  79         return true;
  80 }
  81
  82 bool add_duplex(struct io_conn *c)
  83 {
  84         c->fd.backend_info = c->duplex->fd.backend_info;
  85         num_next++;
  86         return true;
  87 }
  88
  89 static void del_conn(struct io_conn *conn)
  90 {
  91         if (conn->fd.finish)
  92                 conn->fd.finish(conn, conn->fd.finish_arg);
  93         if (conn->duplex) {
  94                 /* In case fds[] pointed to the other one. */
  95                 fds[conn->fd.backend_info] = &conn->duplex->fd;
  96                 conn->duplex->duplex = NULL;
  97         } else
  98                 del_fd(&conn->fd);
  99         if (conn->state == FINISHED)
 100                 num_finished--;
 101         else if (conn->state == NEXT)
 102                 num_next--;
 103 }
 104
 105 void del_listener(struct io_listener *l)
 106 {
 107         del_fd(&l->fd);
 108 }
 109
 110 static int pollmask(enum io_state state)
 111 {
 112         switch (state) {
 113         case READ:
 114         case READPART:
 115                 return POLLIN;
 116         case WRITE:
 117         case WRITEPART:
 118                 return POLLOUT;
 119         default:
 120                 return 0;
 121         }
 122 }
 123
 124 void backend_set_state(struct io_conn *conn, struct io_op *op)
 125 {
 126         enum io_state state = from_ioop(op);
 127         struct pollfd *pfd = &pollfds[conn->fd.backend_info];
 128
 129         if (pfd->events)
 130                 num_waiting--;
 131
 132         pfd->events = pollmask(state);
 133         if (conn->duplex) {
 134                 int mask = pollmask(conn->duplex->state);
 135                 /* You can't *both* read/write. */
 136                 assert(!mask || pfd->events != mask);
 137                 pfd->events |= mask;
 138         }
 139         if (pfd->events)
 140                 num_waiting++;
 141
 142         if (state == NEXT)
 143                 num_next++;
 144         else if (state == FINISHED)
 145                 num_finished++;
 146
 147         conn->state = state;
 148 }
 149
 150 static void accept_conn(struct io_listener *l)
 151 {
 152         struct io_conn *c;
 153         int fd = accept(l->fd.fd, NULL, NULL);
 154
 155         /* FIXME: What to do here? */
 156         if (fd < 0)
 157                 return;
 158         c = io_new_conn(fd, l->fd.next, l->fd.finish, l->fd.next_arg);
 159         if (!c) {
 160                 close(fd);
 161                 return;
 162         }
 163 }
 164
 165 /* It's OK to miss some, as long as we make progress. */
 166 static void finish_and_next(bool finished_only)
 167 {
 168         unsigned int i;
 169
 170         for (i = 0; !io_loop_return && i < num_fds; i++) {
 171                 struct io_conn *c, *duplex;
 172
 173                 if (!num_finished) {
 174                         if (finished_only || num_next == 0)
 175                                 break;
 176                 }
 177                 if (fds[i]->listener)
 178                         continue;
 179                 c = (void *)fds[i];
 180                 for (duplex = c->duplex; c; c = duplex, duplex = NULL) {
 181                         if (c->state == FINISHED) {
 182                                 del_conn(c);
 183                                 free(c);
 184                                 i--;
 185                         } else if (!finished_only && c->state == NEXT) {
 186                                 backend_set_state(c,
 187                                                   c->fd.next(c,
 188                                                              c->fd.next_arg));
 189                                 num_next--;
 190                         }
 191                 }
 192         }
 193 }
 194
 195 static void ready(struct io_conn *c)
 196 {
 197         backend_set_state(c, do_ready(c));
 198 }
 199
 200 /* This is the main loop. */
 201 void *io_loop(void)
 202 {
 203         void *ret;
 204
 205         while (!io_loop_return) {
 206                 int i, r;
 207
 208                 if (num_finished || num_next) {
 209                         finish_and_next(false);
 210                         /* Could have started/finished more. */
 211                         continue;
 212                 }
 213
 214                 if (num_fds == 0)
 215                         break;
 216
 217                 /* You can't tell them all to go to sleep! */
 218                 assert(num_waiting);
 219
 220                 r = poll(pollfds, num_fds, -1);
 221                 if (r < 0)
 222                         break;
 223
 224                 for (i = 0; i < num_fds && !io_loop_return; i++) {
 225                         struct io_conn *c = (void *)fds[i];
 226                         int events = pollfds[i].revents;
 227
 228                         if (fds[i]->listener) {
 229                                 if (events & POLLIN)
 230                                         accept_conn((void *)c);
 231                         } else if (events & (POLLIN|POLLOUT)) {
 232                                 if (c->duplex) {
 233                                         int mask = pollmask(c->duplex->state);
 234                                         if (events & mask) {
 235                                                 ready(c->duplex);
 236                                                 events &= ~mask;
 237                                                 if (!(events&(POLLIN|POLLOUT)))
 238                                                         continue;
 239                                         }
 240                                 }
 241                                 ready(c);
 242                         } else if (events & POLLHUP) {
 243                                 backend_set_state(c, io_close(c, NULL));
 244                                 if (c->duplex)
 245                                         backend_set_state(c->duplex,
 246                                                           io_close(c->duplex,
 247                                                                    NULL));
 248                         }
 249
 250                 }
 251         }
 252
 253         while (num_finished)
 254                 finish_and_next(true);
 255
 256         ret = io_loop_return;
 257         io_loop_return = NULL;
 258         return ret;
 259 }