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