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

   1 /* Licensed under LGPLv2.1+ - see LICENSE file for details */
   2 #include "io.h"
   3 #include "backend.h"
   4 #include <sys/types.h>
   5 #include <sys/socket.h>
   6 #include <netdb.h>
   7 #include <string.h>
   8 #include <errno.h>
   9 #include <stdlib.h>
  10 #include <assert.h>
  11 #include <poll.h>
  12 #include <unistd.h>
  13 #include <fcntl.h>
  14
  15 void *io_loop_return;
  16
  17 struct io_alloc io_alloc = {
  18         malloc, realloc, free
  19 };
  20
  21 #ifdef DEBUG
  22 /* Set to skip the next plan. */
  23 bool io_plan_nodebug;
  24 /* The current connection to apply plan to. */
  25 struct io_conn *current;
  26 /* User-defined function to select which connection(s) to debug. */
  27 bool (*io_debug_conn)(struct io_conn *conn);
  28 /* Set when we wake up an connection we are debugging. */
  29 bool io_debug_wakeup;
  30
  31 struct io_plan io_debug(struct io_plan plan)
  32 {
  33         struct io_conn *ready = NULL;
  34
  35         if (io_plan_nodebug) {
  36                 io_plan_nodebug = false;
  37                 return plan;
  38         }
  39
  40         if (!current || !doing_debug_on(current)) {
  41                 if (!io_debug_wakeup)
  42                         return plan;
  43         }
  44
  45         io_debug_wakeup = false;
  46         current->plan = plan;
  47         backend_plan_changed(current);
  48
  49         /* Call back into the loop immediately. */
  50         io_loop_return = do_io_loop(&ready);
  51
  52         if (ready) {
  53                 set_current(ready);
  54                 if (!ready->plan.next) {
  55                         /* Call finish function immediately. */
  56                         if (ready->finish) {
  57                                 errno = ready->plan.u1.s;
  58                                 ready->finish(ready, ready->finish_arg);
  59                                 ready->finish = NULL;
  60                         }
  61                         backend_del_conn(ready);
  62                 } else {
  63                         /* Calls back in itself, via io_debug_io(). */
  64                         if (ready->plan.io(ready->fd.fd, &ready->plan) != 2)
  65                                 abort();
  66                 }
  67                 set_current(NULL);
  68         }
  69
  70         /* Return a do-nothing plan, so backend_plan_changed in
  71          * io_ready doesn't do anything (it's already been called). */
  72         return io_idle_();
  73 }
  74
  75 int io_debug_io(int ret)
  76 {
  77         /* Cache it for debugging; current changes. */
  78         struct io_conn *conn = current;
  79         int saved_errno = errno;
  80
  81         if (!doing_debug_on(conn))
  82                 return ret;
  83
  84         /* These will all go linearly through the io_debug() path above. */
  85         switch (ret) {
  86         case -1:
  87                 /* This will call io_debug above. */
  88                 errno = saved_errno;
  89                 io_close();
  90                 break;
  91         case 0: /* Keep going with plan. */
  92                 io_debug(conn->plan);
  93                 break;
  94         case 1: /* Done: get next plan. */
  95                 if (timeout_active(conn))
  96                         backend_del_timeout(conn);
  97                 /* In case they call io_duplex, clear our poll flags so
  98                  * both sides don't seem to be both doing read or write
  99                  * (See assert(!mask || pfd->events != mask) in poll.c) */
 100                 conn->plan.pollflag = 0;
 101                 conn->plan.next(conn, conn->plan.next_arg);
 102                 break;
 103         default:
 104                 abort();
 105         }
 106
 107         /* Normally-invalid value, used for sanity check. */
 108         return 2;
 109 }
 110
 111 static void debug_io_wake(struct io_conn *conn)
 112 {
 113         /* We want linear if we wake a debugged connection, too. */
 114         if (io_debug_conn && io_debug_conn(conn))
 115                 io_debug_wakeup = true;
 116 }
 117
 118 /* Counterpart to io_plan_no_debug(), called in macros in io.h */
 119 static void io_plan_debug_again(void)
 120 {
 121         io_plan_nodebug = false;
 122 }
 123 #else
 124 static void debug_io_wake(struct io_conn *conn)
 125 {
 126 }
 127 static void io_plan_debug_again(void)
 128 {
 129 }
 130 #endif
 131
 132 struct io_listener *io_new_listener_(int fd,
 133                                      void (*init)(int fd, void *arg),
 134                                      void *arg)
 135 {
 136         struct io_listener *l = io_alloc.alloc(sizeof(*l));
 137
 138         if (!l)
 139                 return NULL;
 140
 141         l->fd.listener = true;
 142         l->fd.fd = fd;
 143         l->init = init;
 144         l->arg = arg;
 145         if (!add_listener(l)) {
 146                 io_alloc.free(l);
 147                 return NULL;
 148         }
 149         return l;
 150 }
 151
 152 void io_close_listener(struct io_listener *l)
 153 {
 154         close(l->fd.fd);
 155         del_listener(l);
 156         io_alloc.free(l);
 157 }
 158
 159 struct io_conn *io_new_conn_(int fd, struct io_plan plan)
 160 {
 161         struct io_conn *conn = io_alloc.alloc(sizeof(*conn));
 162
 163         io_plan_debug_again();
 164
 165         if (!conn)
 166                 return NULL;
 167
 168         conn->fd.listener = false;
 169         conn->fd.fd = fd;
 170         conn->plan = plan;
 171         conn->finish = NULL;
 172         conn->finish_arg = NULL;
 173         conn->duplex = NULL;
 174         conn->timeout = NULL;
 175         if (!add_conn(conn)) {
 176                 io_alloc.free(conn);
 177                 return NULL;
 178         }
 179         return conn;
 180 }
 181
 182 void io_set_finish_(struct io_conn *conn,
 183                     void (*finish)(struct io_conn *, void *),
 184                     void *arg)
 185 {
 186         conn->finish = finish;
 187         conn->finish_arg = arg;
 188 }
 189
 190 struct io_conn *io_duplex_(struct io_conn *old, struct io_plan plan)
 191 {
 192         struct io_conn *conn;
 193
 194         io_plan_debug_again();
 195
 196         assert(!old->duplex);
 197
 198         conn = io_alloc.alloc(sizeof(*conn));
 199         if (!conn)
 200                 return NULL;
 201
 202         conn->fd.listener = false;
 203         conn->fd.fd = old->fd.fd;
 204         conn->plan = plan;
 205         conn->duplex = old;
 206         conn->finish = NULL;
 207         conn->finish_arg = NULL;
 208         conn->timeout = NULL;
 209         if (!add_duplex(conn)) {
 210                 io_alloc.free(conn);
 211                 return NULL;
 212         }
 213         old->duplex = conn;
 214         return conn;
 215 }
 216
 217 bool io_timeout_(struct io_conn *conn, struct timespec ts,
 218                  struct io_plan (*cb)(struct io_conn *, void *), void *arg)
 219 {
 220         assert(cb);
 221
 222         if (!conn->timeout) {
 223                 conn->timeout = io_alloc.alloc(sizeof(*conn->timeout));
 224                 if (!conn->timeout)
 225                         return false;
 226         } else
 227                 assert(!timeout_active(conn));
 228
 229         conn->timeout->next = cb;
 230         conn->timeout->next_arg = arg;
 231         backend_add_timeout(conn, ts);
 232         return true;
 233 }
 234
 235 /* Returns true if we're finished. */
 236 static int do_write(int fd, struct io_plan *plan)
 237 {
 238         ssize_t ret = write(fd, plan->u1.cp, plan->u2.s);
 239         if (ret < 0)
 240                 return io_debug_io(-1);
 241
 242         plan->u1.cp += ret;
 243         plan->u2.s -= ret;
 244         return io_debug_io(plan->u2.s == 0);
 245 }
 246
 247 /* Queue some data to be written. */
 248 struct io_plan io_write_(const void *data, size_t len,
 249                          struct io_plan (*cb)(struct io_conn *, void *),
 250                          void *arg)
 251 {
 252         struct io_plan plan;
 253
 254         assert(cb);
 255         plan.u1.const_vp = data;
 256         plan.u2.s = len;
 257         plan.io = do_write;
 258         plan.next = cb;
 259         plan.next_arg = arg;
 260         plan.pollflag = POLLOUT;
 261
 262         return plan;
 263 }
 264
 265 static int do_read(int fd, struct io_plan *plan)
 266 {
 267         ssize_t ret = read(fd, plan->u1.cp, plan->u2.s);
 268         if (ret <= 0)
 269                 return io_debug_io(-1);
 270
 271         plan->u1.cp += ret;
 272         plan->u2.s -= ret;
 273         return io_debug_io(plan->u2.s == 0);
 274 }
 275
 276 /* Queue a request to read into a buffer. */
 277 struct io_plan io_read_(void *data, size_t len,
 278                         struct io_plan (*cb)(struct io_conn *, void *),
 279                         void *arg)
 280 {
 281         struct io_plan plan;
 282
 283         assert(cb);
 284         plan.u1.cp = data;
 285         plan.u2.s = len;
 286         plan.io = do_read;
 287         plan.next = cb;
 288         plan.next_arg = arg;
 289         plan.pollflag = POLLIN;
 290
 291         return plan;
 292 }
 293
 294 static int do_read_partial(int fd, struct io_plan *plan)
 295 {
 296         ssize_t ret = read(fd, plan->u1.cp, *(size_t *)plan->u2.vp);
 297         if (ret <= 0)
 298                 return io_debug_io(-1);
 299
 300         *(size_t *)plan->u2.vp = ret;
 301         return io_debug_io(1);
 302 }
 303
 304 /* Queue a partial request to read into a buffer. */
 305 struct io_plan io_read_partial_(void *data, size_t *len,
 306                                 struct io_plan (*cb)(struct io_conn *, void *),
 307                                 void *arg)
 308 {
 309         struct io_plan plan;
 310
 311         assert(cb);
 312         plan.u1.cp = data;
 313         plan.u2.vp = len;
 314         plan.io = do_read_partial;
 315         plan.next = cb;
 316         plan.next_arg = arg;
 317         plan.pollflag = POLLIN;
 318
 319         return plan;
 320 }
 321
 322 static int do_write_partial(int fd, struct io_plan *plan)
 323 {
 324         ssize_t ret = write(fd, plan->u1.cp, *(size_t *)plan->u2.vp);
 325         if (ret < 0)
 326                 return io_debug_io(-1);
 327
 328         *(size_t *)plan->u2.vp = ret;
 329         return io_debug_io(1);
 330 }
 331
 332 /* Queue a partial write request. */
 333 struct io_plan io_write_partial_(const void *data, size_t *len,
 334                                  struct io_plan (*cb)(struct io_conn*, void *),
 335                                  void *arg)
 336 {
 337         struct io_plan plan;
 338
 339         assert(cb);
 340         plan.u1.const_vp = data;
 341         plan.u2.vp = len;
 342         plan.io = do_write_partial;
 343         plan.next = cb;
 344         plan.next_arg = arg;
 345         plan.pollflag = POLLOUT;
 346
 347         return plan;
 348 }
 349
 350 static int already_connected(int fd, struct io_plan *plan)
 351 {
 352         return io_debug_io(1);
 353 }
 354
 355 static int do_connect(int fd, struct io_plan *plan)
 356 {
 357         int err, ret;
 358         socklen_t len = sizeof(err);
 359
 360         /* Has async connect finished? */
 361         ret = getsockopt(fd, SOL_SOCKET, SO_ERROR, &err, &len);
 362         if (ret < 0)
 363                 return -1;
 364
 365         if (err == 0) {
 366                 /* Restore blocking if it was initially. */
 367                 fcntl(fd, F_SETFD, plan->u1.s);
 368                 return 1;
 369         }
 370         return 0;
 371 }
 372
 373 struct io_plan io_connect_(int fd, const struct addrinfo *addr,
 374                            struct io_plan (*cb)(struct io_conn*, void *),
 375                            void *arg)
 376 {
 377         struct io_plan plan;
 378
 379         assert(cb);
 380
 381         plan.next = cb;
 382         plan.next_arg = arg;
 383
 384         /* Save old flags, set nonblock if not already. */
 385         plan.u1.s = fcntl(fd, F_GETFD);
 386         fcntl(fd, F_SETFD, plan.u1.s | O_NONBLOCK);
 387
 388         /* Immediate connect can happen. */
 389         if (connect(fd, addr->ai_addr, addr->ai_addrlen) == 0) {
 390                 /* Dummy will be called immediately. */
 391                 plan.pollflag = POLLOUT;
 392                 plan.io = already_connected;
 393         } else {
 394                 if (errno != EINPROGRESS)
 395                         return io_close_();
 396
 397                 plan.pollflag = POLLIN;
 398                 plan.io = do_connect;
 399         }
 400         return plan;
 401 }
 402
 403 struct io_plan io_idle_(void)
 404 {
 405         struct io_plan plan;
 406
 407         plan.pollflag = 0;
 408         plan.io = NULL;
 409         /* Never called (overridden by io_wake), but NULL means closing */
 410         plan.next = (void *)io_idle_;
 411
 412         return plan;
 413 }
 414
 415 bool io_is_idle(const struct io_conn *conn)
 416 {
 417         return conn->plan.io == NULL;
 418 }
 419
 420 void io_wake_(struct io_conn *conn, struct io_plan plan)
 421
 422 {
 423         io_plan_debug_again();
 424
 425         /* It might be closing, but we haven't called its finish() yet. */
 426         if (!conn->plan.next)
 427                 return;
 428         /* It was idle, right? */
 429         assert(!conn->plan.io);
 430         conn->plan = plan;
 431         backend_plan_changed(conn);
 432
 433         debug_io_wake(conn);
 434 }
 435
 436 void io_ready(struct io_conn *conn)
 437 {
 438         set_current(conn);
 439         switch (conn->plan.io(conn->fd.fd, &conn->plan)) {
 440         case -1: /* Failure means a new plan: close up. */
 441                 conn->plan = io_close();
 442                 backend_plan_changed(conn);
 443                 break;
 444         case 0: /* Keep going with plan. */
 445                 break;
 446         case 1: /* Done: get next plan. */
 447                 if (timeout_active(conn))
 448                         backend_del_timeout(conn);
 449                 /* In case they call io_duplex, clear our poll flags so
 450                  * both sides don't seem to be both doing read or write
 451                  * (See assert(!mask || pfd->events != mask) in poll.c) */
 452                 conn->plan.pollflag = 0;
 453                 conn->plan = conn->plan.next(conn, conn->plan.next_arg);
 454                 backend_plan_changed(conn);
 455         }
 456         set_current(NULL);
 457 }
 458
 459 /* Close the connection, we're done. */
 460 struct io_plan io_close_(void)
 461 {
 462         struct io_plan plan;
 463
 464         plan.pollflag = 0;
 465         /* This means we're closing. */
 466         plan.next = NULL;
 467         plan.u1.s = errno;
 468
 469         return plan;
 470 }
 471
 472 struct io_plan io_close_cb(struct io_conn *conn, void *arg)
 473 {
 474         return io_close();
 475 }
 476
 477 /* Exit the loop, returning this (non-NULL) arg. */
 478 struct io_plan io_break_(void *ret, struct io_plan plan)
 479 {
 480         io_plan_debug_again();
 481
 482         assert(ret);
 483         io_loop_return = ret;
 484
 485         return plan;
 486 }
 487
 488 int io_conn_fd(const struct io_conn *conn)
 489 {
 490         return conn->fd.fd;
 491 }
 492
 493 void io_set_alloc(void *(*allocfn)(size_t size),
 494                   void *(*reallocfn)(void *ptr, size_t size),
 495                   void (*freefn)(void *ptr))
 496 {
 497         io_alloc.alloc = allocfn;
 498         io_alloc.realloc = reallocfn;
 499         io_alloc.free = freefn;
 500 }