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