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
  13 void *io_loop_return;
  14
  15 #ifdef DEBUG
  16 /* Set to skip the next plan. */
  17 bool io_plan_nodebug;
  18 /* The current connection to apply plan to. */
  19 struct io_conn *current;
  20 /* User-defined function to select which connection(s) to debug. */
  21 bool (*io_debug_conn)(struct io_conn *conn);
  22 /* Set when we wake up an connection we are debugging. */
  23 bool io_debug_wakeup;
  24
  25 struct io_plan io_debug(struct io_plan plan)
  26 {
  27         struct io_conn *ready = NULL;
  28
  29         if (io_plan_nodebug) {
  30                 io_plan_nodebug = false;
  31                 return plan;
  32         }
  33
  34         if (!current || !doing_debug_on(current)) {
  35                 if (!io_debug_wakeup)
  36                         return plan;
  37         }
  38
  39         io_debug_wakeup = false;
  40         current->plan = plan;
  41         backend_plan_changed(current);
  42
  43         /* Call back into the loop immediately. */
  44         io_loop_return = do_io_loop(&ready);
  45
  46         if (ready) {
  47                 set_current(ready);
  48                 if (!ready->plan.next) {
  49                         /* Call finish function immediately. */
  50                         if (ready->finish) {
  51                                 errno = ready->plan.u.close.saved_errno;
  52                                 ready->finish(ready, ready->finish_arg);
  53                                 ready->finish = NULL;
  54                         }
  55                         backend_del_conn(ready);
  56                 } else {
  57                         /* Calls back in itself, via io_debug_io(). */
  58                         if (ready->plan.io(ready->fd.fd, &ready->plan) != 2)
  59                                 abort();
  60                 }
  61                 set_current(NULL);
  62         }
  63
  64         /* Return a do-nothing plan, so backend_plan_changed in
  65          * io_ready doesn't do anything (it's already been called). */
  66         return io_idle_();
  67 }
  68
  69 int io_debug_io(int ret)
  70 {
  71         /* Cache it for debugging; current changes. */
  72         struct io_conn *conn = current;
  73         int saved_errno = errno;
  74
  75         if (!doing_debug_on(conn))
  76                 return ret;
  77
  78         /* These will all go linearly through the io_debug() path above. */
  79         switch (ret) {
  80         case -1:
  81                 /* This will call io_debug above. */
  82                 errno = saved_errno;
  83                 io_close();
  84                 break;
  85         case 0: /* Keep going with plan. */
  86                 io_debug(conn->plan);
  87                 break;
  88         case 1: /* Done: get next plan. */
  89                 if (timeout_active(conn))
  90                         backend_del_timeout(conn);
  91                 conn->plan.next(conn, conn->plan.next_arg);
  92                 break;
  93         default:
  94                 abort();
  95         }
  96
  97         /* Normally-invalid value, used for sanity check. */
  98         return 2;
  99 }
 100
 101 static void debug_io_wake(struct io_conn *conn)
 102 {
 103         /* We want linear if we wake a debugged connection, too. */
 104         if (io_debug_conn && io_debug_conn(conn))
 105                 io_debug_wakeup = true;
 106 }
 107
 108 /* Counterpart to io_plan_no_debug(), called in macros in io.h */
 109 static void io_plan_debug_again(void)
 110 {
 111         io_plan_nodebug = false;
 112 }
 113 #else
 114 static void debug_io_wake(struct io_conn *conn)
 115 {
 116 }
 117 static void io_plan_debug_again(void)
 118 {
 119 }
 120 #endif
 121
 122 struct io_listener *io_new_listener_(int fd,
 123                                      void (*init)(int fd, void *arg),
 124                                      void *arg)
 125 {
 126         struct io_listener *l = malloc(sizeof(*l));
 127
 128         if (!l)
 129                 return NULL;
 130
 131         l->fd.listener = true;
 132         l->fd.fd = fd;
 133         l->init = init;
 134         l->arg = arg;
 135         if (!add_listener(l)) {
 136                 free(l);
 137                 return NULL;
 138         }
 139         return l;
 140 }
 141
 142 void io_close_listener(struct io_listener *l)
 143 {
 144         close(l->fd.fd);
 145         del_listener(l);
 146         free(l);
 147 }
 148
 149 struct io_conn *io_new_conn_(int fd, struct io_plan plan)
 150 {
 151         struct io_conn *conn = malloc(sizeof(*conn));
 152
 153         io_plan_debug_again();
 154
 155         if (!conn)
 156                 return NULL;
 157
 158         conn->fd.listener = false;
 159         conn->fd.fd = fd;
 160         conn->plan = plan;
 161         conn->finish = NULL;
 162         conn->finish_arg = NULL;
 163         conn->duplex = NULL;
 164         conn->timeout = NULL;
 165         if (!add_conn(conn)) {
 166                 free(conn);
 167                 return NULL;
 168         }
 169         return conn;
 170 }
 171
 172 void io_set_finish_(struct io_conn *conn,
 173                     void (*finish)(struct io_conn *, void *),
 174                     void *arg)
 175 {
 176         conn->finish = finish;
 177         conn->finish_arg = arg;
 178 }
 179
 180 struct io_conn *io_duplex_(struct io_conn *old, struct io_plan plan)
 181 {
 182         struct io_conn *conn;
 183
 184         io_plan_debug_again();
 185
 186         assert(!old->duplex);
 187
 188         conn = malloc(sizeof(*conn));
 189         if (!conn)
 190                 return NULL;
 191
 192         conn->fd.listener = false;
 193         conn->fd.fd = old->fd.fd;
 194         conn->plan = plan;
 195         conn->duplex = old;
 196         conn->finish = NULL;
 197         conn->finish_arg = NULL;
 198         conn->timeout = NULL;
 199         if (!add_duplex(conn)) {
 200                 free(conn);
 201                 return NULL;
 202         }
 203         old->duplex = conn;
 204         return conn;
 205 }
 206
 207 bool io_timeout_(struct io_conn *conn, struct timespec ts,
 208                  struct io_plan (*cb)(struct io_conn *, void *), void *arg)
 209 {
 210         assert(cb);
 211
 212         if (!conn->timeout) {
 213                 conn->timeout = malloc(sizeof(*conn->timeout));
 214                 if (!conn->timeout)
 215                         return false;
 216         } else
 217                 assert(!timeout_active(conn));
 218
 219         conn->timeout->next = cb;
 220         conn->timeout->next_arg = arg;
 221         backend_add_timeout(conn, ts);
 222         return true;
 223 }
 224
 225 /* Returns true if we're finished. */
 226 static int do_write(int fd, struct io_plan *plan)
 227 {
 228         ssize_t ret = write(fd, plan->u.write.buf, plan->u.write.len);
 229         if (ret < 0)
 230                 return io_debug_io(-1);
 231
 232         plan->u.write.buf += ret;
 233         plan->u.write.len -= ret;
 234         return io_debug_io(plan->u.write.len == 0);
 235 }
 236
 237 /* Queue some data to be written. */
 238 struct io_plan io_write_(const void *data, size_t len,
 239                          struct io_plan (*cb)(struct io_conn *, void *),
 240                          void *arg)
 241 {
 242         struct io_plan plan;
 243
 244         assert(cb);
 245         plan.u.write.buf = data;
 246         plan.u.write.len = len;
 247         plan.io = do_write;
 248         plan.next = cb;
 249         plan.next_arg = arg;
 250         plan.pollflag = POLLOUT;
 251
 252         return plan;
 253 }
 254
 255 static int do_read(int fd, struct io_plan *plan)
 256 {
 257         ssize_t ret = read(fd, plan->u.read.buf, plan->u.read.len);
 258         if (ret <= 0)
 259                 return io_debug_io(-1);
 260
 261         plan->u.read.buf += ret;
 262         plan->u.read.len -= ret;
 263         return io_debug_io(plan->u.read.len == 0);
 264 }
 265
 266 /* Queue a request to read into a buffer. */
 267 struct io_plan io_read_(void *data, size_t len,
 268                         struct io_plan (*cb)(struct io_conn *, void *),
 269                         void *arg)
 270 {
 271         struct io_plan plan;
 272
 273         assert(cb);
 274         plan.u.read.buf = data;
 275         plan.u.read.len = len;
 276         plan.io = do_read;
 277         plan.next = cb;
 278         plan.next_arg = arg;
 279         plan.pollflag = POLLIN;
 280
 281         return plan;
 282 }
 283
 284 static int do_read_partial(int fd, struct io_plan *plan)
 285 {
 286         ssize_t ret = read(fd, plan->u.readpart.buf, *plan->u.readpart.lenp);
 287         if (ret <= 0)
 288                 return io_debug_io(-1);
 289
 290         *plan->u.readpart.lenp = ret;
 291         return io_debug_io(1);
 292 }
 293
 294 /* Queue a partial request to read into a buffer. */
 295 struct io_plan io_read_partial_(void *data, size_t *len,
 296                                 struct io_plan (*cb)(struct io_conn *, void *),
 297                                 void *arg)
 298 {
 299         struct io_plan plan;
 300
 301         assert(cb);
 302         plan.u.readpart.buf = data;
 303         plan.u.readpart.lenp = len;
 304         plan.io = do_read_partial;
 305         plan.next = cb;
 306         plan.next_arg = arg;
 307         plan.pollflag = POLLIN;
 308
 309         return plan;
 310 }
 311
 312 static int do_write_partial(int fd, struct io_plan *plan)
 313 {
 314         ssize_t ret = write(fd, plan->u.writepart.buf, *plan->u.writepart.lenp);
 315         if (ret < 0)
 316                 return io_debug_io(-1);
 317
 318         *plan->u.writepart.lenp = ret;
 319         return io_debug_io(1);
 320 }
 321
 322 /* Queue a partial write request. */
 323 struct io_plan io_write_partial_(const void *data, size_t *len,
 324                                  struct io_plan (*cb)(struct io_conn*, void *),
 325                                  void *arg)
 326 {
 327         struct io_plan plan;
 328
 329         assert(cb);
 330         plan.u.writepart.buf = data;
 331         plan.u.writepart.lenp = len;
 332         plan.io = do_write_partial;
 333         plan.next = cb;
 334         plan.next_arg = arg;
 335         plan.pollflag = POLLOUT;
 336
 337         return plan;
 338 }
 339
 340 struct io_plan io_idle_(void)
 341 {
 342         struct io_plan plan;
 343
 344         plan.pollflag = 0;
 345         plan.io = NULL;
 346         /* Never called (overridden by io_wake), but NULL means closing */
 347         plan.next = (void *)io_idle_;
 348
 349         return plan;
 350 }
 351
 352 void io_wake_(struct io_conn *conn, struct io_plan plan)
 353
 354 {
 355         io_plan_debug_again();
 356
 357         /* It might be closing, but we haven't called its finish() yet. */
 358         if (!conn->plan.next)
 359                 return;
 360         /* It was idle, right? */
 361         assert(!conn->plan.io);
 362         conn->plan = plan;
 363         backend_plan_changed(conn);
 364
 365         debug_io_wake(conn);
 366 }
 367
 368 void io_ready(struct io_conn *conn)
 369 {
 370         set_current(conn);
 371         switch (conn->plan.io(conn->fd.fd, &conn->plan)) {
 372         case -1: /* Failure means a new plan: close up. */
 373                 conn->plan = io_close();
 374                 backend_plan_changed(conn);
 375                 break;
 376         case 0: /* Keep going with plan. */
 377                 break;
 378         case 1: /* Done: get next plan. */
 379                 if (timeout_active(conn))
 380                         backend_del_timeout(conn);
 381                 conn->plan = conn->plan.next(conn, conn->plan.next_arg);
 382                 backend_plan_changed(conn);
 383         }
 384         set_current(NULL);
 385 }
 386
 387 /* Close the connection, we're done. */
 388 struct io_plan io_close_(void)
 389 {
 390         struct io_plan plan;
 391
 392         plan.pollflag = 0;
 393         /* This means we're closing. */
 394         plan.next = NULL;
 395         plan.u.close.saved_errno = errno;
 396
 397         return plan;
 398 }
 399
 400 struct io_plan io_close_cb(struct io_conn *conn, void *arg)
 401 {
 402         return io_close();
 403 }
 404
 405 /* Exit the loop, returning this (non-NULL) arg. */
 406 struct io_plan io_break_(void *ret, struct io_plan plan)
 407 {
 408         io_plan_debug_again();
 409
 410         assert(ret);
 411         io_loop_return = ret;
 412
 413         return plan;
 414 }