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