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 <unistd.h>
  12 #include <fcntl.h>
  13 #include <ccan/container_of/container_of.h>
  14
  15 void *io_loop_return;
  16
  17 struct io_plan io_conn_freed;
  18
  19 struct io_listener *io_new_listener_(const tal_t *ctx, int fd,
  20                                      struct io_plan *(*init)(struct io_conn *,
  21                                                              void *),
  22                                      void *arg)
  23 {
  24         struct io_listener *l = tal(ctx, struct io_listener);
  25         if (!l)
  26                 return NULL;
  27
  28         l->fd.listener = true;
  29         l->fd.fd = fd;
  30         l->init = init;
  31         l->arg = arg;
  32         l->ctx = ctx;
  33         if (!add_listener(l))
  34                 return tal_free(l);
  35         return l;
  36 }
  37
  38 void io_close_listener(struct io_listener *l)
  39 {
  40         tal_free(l);
  41 }
  42
  43 static struct io_plan *io_never_called(struct io_conn *conn, void *arg)
  44 {
  45         abort();
  46 }
  47
  48 /* Returns false if conn was freed. */
  49 static bool next_plan(struct io_conn *conn, struct io_plan *plan)
  50 {
  51         struct io_plan *(*next)(struct io_conn *, void *arg);
  52
  53         next = plan->next;
  54
  55         plan->status = IO_UNSET;
  56         plan->io = NULL;
  57         plan->next = io_never_called;
  58
  59         plan = next(conn, plan->next_arg);
  60
  61         if (plan == &io_conn_freed)
  62                 return false;
  63
  64         assert(plan == &conn->plan[plan->dir]);
  65         assert(conn->plan[IO_IN].status != IO_UNSET
  66                || conn->plan[IO_OUT].status != IO_UNSET);
  67
  68         backend_new_plan(conn);
  69         return true;
  70 }
  71
  72 bool io_fd_block(int fd, bool block)
  73 {
  74         int flags = fcntl(fd, F_GETFL);
  75
  76         if (flags == -1)
  77                 return false;
  78
  79         if (block)
  80                 flags &= ~O_NONBLOCK;
  81         else
  82                 flags |= O_NONBLOCK;
  83
  84         return fcntl(fd, F_SETFL, flags) != -1;
  85 }
  86
  87 struct io_conn *io_new_conn_(const tal_t *ctx, int fd,
  88                              struct io_plan *(*init)(struct io_conn *, void *),
  89                              void *arg)
  90 {
  91         struct io_conn *conn = tal(ctx, struct io_conn);
  92
  93         if (!conn)
  94                 return NULL;
  95
  96         conn->fd.listener = false;
  97         conn->fd.fd = fd;
  98         conn->finish = NULL;
  99         conn->finish_arg = NULL;
 100         list_node_init(&conn->always);
 101
 102         if (!add_conn(conn))
 103                 return tal_free(conn);
 104
 105         /* Keep our I/O async. */
 106         io_fd_block(fd, false);
 107
 108         /* So we can get back from plan -> conn later */
 109         conn->plan[IO_OUT].dir = IO_OUT;
 110         conn->plan[IO_IN].dir = IO_IN;
 111
 112         /* We start with out doing nothing, and in doing our init. */
 113         conn->plan[IO_OUT].status = IO_UNSET;
 114
 115         conn->plan[IO_IN].next = init;
 116         conn->plan[IO_IN].next_arg = arg;
 117         if (!next_plan(conn, &conn->plan[IO_IN]))
 118                 return NULL;
 119
 120         return conn;
 121 }
 122
 123 void io_set_finish_(struct io_conn *conn,
 124                     void (*finish)(struct io_conn *, void *),
 125                     void *arg)
 126 {
 127         conn->finish = finish;
 128         conn->finish_arg = arg;
 129 }
 130
 131 struct io_plan_arg *io_plan_arg(struct io_conn *conn, enum io_direction dir)
 132 {
 133         assert(conn->plan[dir].status == IO_UNSET);
 134
 135         conn->plan[dir].status = IO_POLLING_NOTSTARTED;
 136         return &conn->plan[dir].arg;
 137 }
 138
 139 static struct io_plan *set_always(struct io_conn *conn,
 140                                   enum io_direction dir,
 141                                   struct io_plan *(*next)(struct io_conn *,
 142                                                           void *),
 143                                   void *arg)
 144 {
 145         struct io_plan *plan = &conn->plan[dir];
 146
 147         plan->status = IO_ALWAYS;
 148         backend_new_always(conn);
 149         return io_set_plan(conn, dir, NULL, next, arg);
 150 }
 151
 152 static struct io_plan *io_always_dir(struct io_conn *conn,
 153                                      enum io_direction dir,
 154                                      struct io_plan *(*next)(struct io_conn *,
 155                                                              void *),
 156                                      void *arg)
 157 {
 158         return set_always(conn, dir, next, arg);
 159 }
 160
 161 struct io_plan *io_always_(struct io_conn *conn,
 162                            struct io_plan *(*next)(struct io_conn *, void *),
 163                            void *arg)
 164 {
 165         return io_always_dir(conn, IO_IN, next, arg);
 166 }
 167
 168 struct io_plan *io_out_always_(struct io_conn *conn,
 169                                struct io_plan *(*next)(struct io_conn *,
 170                                                        void *),
 171                                void *arg)
 172 {
 173         return io_always_dir(conn, IO_OUT, next, arg);
 174 }
 175
 176 static int do_write(int fd, struct io_plan_arg *arg)
 177 {
 178         ssize_t ret = write(fd, arg->u1.cp, arg->u2.s);
 179         if (ret < 0)
 180                 return -1;
 181
 182         arg->u1.cp += ret;
 183         arg->u2.s -= ret;
 184         return arg->u2.s == 0;
 185 }
 186
 187 /* Queue some data to be written. */
 188 struct io_plan *io_write_(struct io_conn *conn, const void *data, size_t len,
 189                           struct io_plan *(*next)(struct io_conn *, void *),
 190                           void *next_arg)
 191 {
 192         struct io_plan_arg *arg = io_plan_arg(conn, IO_OUT);
 193
 194         if (len == 0)
 195                 return set_always(conn, IO_OUT, next, next_arg);
 196
 197         arg->u1.const_vp = data;
 198         arg->u2.s = len;
 199
 200         return io_set_plan(conn, IO_OUT, do_write, next, next_arg);
 201 }
 202
 203 static int do_read(int fd, struct io_plan_arg *arg)
 204 {
 205         ssize_t ret = read(fd, arg->u1.cp, arg->u2.s);
 206         if (ret <= 0) {
 207                 /* Errno isn't set if we hit EOF, so set it to distinct value */
 208                 if (ret == 0)
 209                         errno = 0;
 210                 return -1;
 211         }
 212
 213         arg->u1.cp += ret;
 214         arg->u2.s -= ret;
 215         return arg->u2.s == 0;
 216 }
 217
 218 /* Queue a request to read into a buffer. */
 219 struct io_plan *io_read_(struct io_conn *conn,
 220                          void *data, size_t len,
 221                          struct io_plan *(*next)(struct io_conn *, void *),
 222                          void *next_arg)
 223 {
 224         struct io_plan_arg *arg = io_plan_arg(conn, IO_IN);
 225
 226         if (len == 0)
 227                 return set_always(conn, IO_IN, next, next_arg);
 228
 229         arg->u1.cp = data;
 230         arg->u2.s = len;
 231
 232         return io_set_plan(conn, IO_IN, do_read, next, next_arg);
 233 }
 234
 235 static int do_read_partial(int fd, struct io_plan_arg *arg)
 236 {
 237         ssize_t ret = read(fd, arg->u1.cp, *(size_t *)arg->u2.vp);
 238         if (ret <= 0) {
 239                 /* Errno isn't set if we hit EOF, so set it to distinct value */
 240                 if (ret == 0)
 241                         errno = 0;
 242                 return -1;
 243         }
 244
 245         *(size_t *)arg->u2.vp = ret;
 246         return 1;
 247 }
 248
 249 /* Queue a partial request to read into a buffer. */
 250 struct io_plan *io_read_partial_(struct io_conn *conn,
 251                                  void *data, size_t maxlen, size_t *len,
 252                                  struct io_plan *(*next)(struct io_conn *,
 253                                                          void *),
 254                                  void *next_arg)
 255 {
 256         struct io_plan_arg *arg = io_plan_arg(conn, IO_IN);
 257
 258         if (maxlen == 0)
 259                 return set_always(conn, IO_IN, next, next_arg);
 260
 261         arg->u1.cp = data;
 262         /* We store the max len in here temporarily. */
 263         *len = maxlen;
 264         arg->u2.vp = len;
 265
 266         return io_set_plan(conn, IO_IN, do_read_partial, next, next_arg);
 267 }
 268
 269 static int do_write_partial(int fd, struct io_plan_arg *arg)
 270 {
 271         ssize_t ret = write(fd, arg->u1.cp, *(size_t *)arg->u2.vp);
 272         if (ret < 0)
 273                 return -1;
 274
 275         *(size_t *)arg->u2.vp = ret;
 276         return 1;
 277 }
 278
 279 /* Queue a partial write request. */
 280 struct io_plan *io_write_partial_(struct io_conn *conn,
 281                                   const void *data, size_t maxlen, size_t *len,
 282                                   struct io_plan *(*next)(struct io_conn *,
 283                                                           void*),
 284                                   void *next_arg)
 285 {
 286         struct io_plan_arg *arg = io_plan_arg(conn, IO_OUT);
 287
 288         if (maxlen == 0)
 289                 return set_always(conn, IO_OUT, next, next_arg);
 290
 291         arg->u1.const_vp = data;
 292         /* We store the max len in here temporarily. */
 293         *len = maxlen;
 294         arg->u2.vp = len;
 295
 296         return io_set_plan(conn, IO_OUT, do_write_partial, next, next_arg);
 297 }
 298
 299 static int do_connect(int fd, struct io_plan_arg *arg)
 300 {
 301         int err, ret;
 302         socklen_t len = sizeof(err);
 303
 304         /* Has async connect finished? */
 305         ret = getsockopt(fd, SOL_SOCKET, SO_ERROR, &err, &len);
 306         if (ret < 0)
 307                 return -1;
 308
 309         if (err == 0) {
 310                 return 1;
 311         } else if (err == EINPROGRESS)
 312                 return 0;
 313
 314         errno = err;
 315         return -1;
 316 }
 317
 318 struct io_plan *io_connect_(struct io_conn *conn, const struct addrinfo *addr,
 319                             struct io_plan *(*next)(struct io_conn *, void *),
 320                             void *next_arg)
 321 {
 322         int fd = io_conn_fd(conn);
 323
 324         /* We don't actually need the arg, but we need it polling. */
 325         io_plan_arg(conn, IO_OUT);
 326
 327         /* Note that io_new_conn() will make fd O_NONBLOCK */
 328
 329         /* Immediate connect can happen. */
 330         if (connect(fd, addr->ai_addr, addr->ai_addrlen) == 0)
 331                 return set_always(conn, IO_OUT, next, next_arg);
 332
 333         if (errno != EINPROGRESS)
 334                 return io_close(conn);
 335
 336         return io_set_plan(conn, IO_OUT, do_connect, next, next_arg);
 337 }
 338
 339 static struct io_plan *io_wait_dir(struct io_conn *conn,
 340                                    const void *wait,
 341                                    enum io_direction dir,
 342                                    struct io_plan *(*next)(struct io_conn *,
 343                                                            void *),
 344                                    void *next_arg)
 345 {
 346         struct io_plan_arg *arg = io_plan_arg(conn, dir);
 347         arg->u1.const_vp = wait;
 348
 349         conn->plan[dir].status = IO_WAITING;
 350
 351         return io_set_plan(conn, dir, NULL, next, next_arg);
 352 }
 353
 354 struct io_plan *io_wait_(struct io_conn *conn,
 355                          const void *wait,
 356                          struct io_plan *(*next)(struct io_conn *, void *),
 357                          void *next_arg)
 358 {
 359         return io_wait_dir(conn, wait, IO_IN, next, next_arg);
 360 }
 361
 362 struct io_plan *io_out_wait_(struct io_conn *conn,
 363                              const void *wait,
 364                              struct io_plan *(*next)(struct io_conn *, void *),
 365                              void *next_arg)
 366 {
 367         return io_wait_dir(conn, wait, IO_OUT, next, next_arg);
 368 }
 369
 370 void io_wake(const void *wait)
 371 {
 372         backend_wake(wait);
 373 }
 374
 375 /* Returns false if this should not be touched (eg. freed). */
 376 static bool do_plan(struct io_conn *conn, struct io_plan *plan,
 377                     bool idle_on_epipe)
 378 {
 379         /* We shouldn't have polled for this event if this wasn't true! */
 380         assert(plan->status == IO_POLLING_NOTSTARTED
 381                || plan->status == IO_POLLING_STARTED);
 382
 383         switch (plan->io(conn->fd.fd, &plan->arg)) {
 384         case -1:
 385                 if (errno == EPIPE && idle_on_epipe) {
 386                         plan->status = IO_UNSET;
 387                         backend_new_plan(conn);
 388                         return false;
 389                 }
 390                 io_close(conn);
 391                 return false;
 392         case 0:
 393                 plan->status = IO_POLLING_STARTED;
 394                 return true;
 395         case 1:
 396                 return next_plan(conn, plan);
 397         default:
 398                 /* IO should only return -1, 0 or 1 */
 399                 abort();
 400         }
 401 }
 402
 403 void io_ready(struct io_conn *conn, int pollflags)
 404 {
 405         if (pollflags & POLLIN)
 406                 if (!do_plan(conn, &conn->plan[IO_IN], false))
 407                         return;
 408
 409         if (pollflags & POLLOUT)
 410                 /* If we're writing to a closed pipe, we need to wait for
 411                  * read to fail if we're duplex: we want to drain it! */
 412                 do_plan(conn, &conn->plan[IO_OUT],
 413                         conn->plan[IO_IN].status == IO_POLLING_NOTSTARTED
 414                         || conn->plan[IO_IN].status == IO_POLLING_STARTED);
 415 }
 416
 417 void io_do_always(struct io_conn *conn)
 418 {
 419         /* There's a corner case where the in next_plan wakes up the
 420          * out, placing it in IO_ALWAYS and we end up processing it immediately,
 421          * only to leave it in the always list.
 422          *
 423          * Yet we can't just process one, in case they are both supposed
 424          * to be done, so grab state beforehand.
 425          */
 426         bool always_out = (conn->plan[IO_OUT].status == IO_ALWAYS);
 427
 428         if (conn->plan[IO_IN].status == IO_ALWAYS)
 429                 if (!next_plan(conn, &conn->plan[IO_IN]))
 430                         return;
 431
 432         if (always_out) {
 433                 /* You can't *unalways* a conn (except by freeing, in which
 434                  * case next_plan() returned false */
 435                 assert(conn->plan[IO_OUT].status == IO_ALWAYS);
 436                 next_plan(conn, &conn->plan[IO_OUT]);
 437         }
 438 }
 439
 440 void io_do_wakeup(struct io_conn *conn, enum io_direction dir)
 441 {
 442         struct io_plan *plan = &conn->plan[dir];
 443
 444         assert(plan->status == IO_WAITING);
 445
 446         set_always(conn, dir, plan->next, plan->next_arg);
 447 }
 448
 449 /* Close the connection, we're done. */
 450 struct io_plan *io_close(struct io_conn *conn)
 451 {
 452         tal_free(conn);
 453         return &io_conn_freed;
 454 }
 455
 456 struct io_plan *io_close_cb(struct io_conn *conn, void *next_arg)
 457 {
 458         return io_close(conn);
 459 }
 460
 461 struct io_plan *io_close_taken_fd(struct io_conn *conn)
 462 {
 463         io_fd_block(conn->fd.fd, true);
 464
 465         cleanup_conn_without_close(conn);
 466         return io_close(conn);
 467 }
 468
 469 /* Exit the loop, returning this (non-NULL) arg. */
 470 void io_break(const void *ret)
 471 {
 472         assert(ret);
 473         io_loop_return = (void *)ret;
 474 }
 475
 476 struct io_plan *io_never(struct io_conn *conn, void *unused)
 477 {
 478         return io_always(conn, io_never_called, NULL);
 479 }
 480
 481 int io_conn_fd(const struct io_conn *conn)
 482 {
 483         return conn->fd.fd;
 484 }
 485
 486 struct io_plan *io_duplex(struct io_conn *conn,
 487                           struct io_plan *in_plan, struct io_plan *out_plan)
 488 {
 489         assert(conn == container_of(in_plan, struct io_conn, plan[IO_IN]));
 490         /* in_plan must be conn->plan[IO_IN], out_plan must be [IO_OUT] */
 491         assert(out_plan == in_plan + 1);
 492         return in_plan;
 493 }
 494
 495 struct io_plan *io_halfclose(struct io_conn *conn)
 496 {
 497         /* Both unset?  OK. */
 498         if (conn->plan[IO_IN].status == IO_UNSET
 499             && conn->plan[IO_OUT].status == IO_UNSET)
 500                 return io_close(conn);
 501
 502         /* We leave this unset then. */
 503         if (conn->plan[IO_IN].status == IO_UNSET)
 504                 return &conn->plan[IO_IN];
 505         else
 506                 return &conn->plan[IO_OUT];
 507 }
 508
 509 struct io_plan *io_set_plan(struct io_conn *conn, enum io_direction dir,
 510                             int (*io)(int fd, struct io_plan_arg *arg),
 511                             struct io_plan *(*next)(struct io_conn *, void *),
 512                             void *next_arg)
 513 {
 514         struct io_plan *plan = &conn->plan[dir];
 515
 516         plan->io = io;
 517         plan->next = next;
 518         plan->next_arg = next_arg;
 519         assert(next != NULL);
 520
 521         return plan;
 522 }
 523
 524 bool io_plan_in_started(const struct io_conn *conn)
 525 {
 526         return conn->plan[IO_IN].status == IO_POLLING_STARTED;
 527 }
 528
 529 bool io_plan_out_started(const struct io_conn *conn)
 530 {
 531         return conn->plan[IO_OUT].status == IO_POLLING_STARTED;
 532 }
 533
 534 bool io_flush_sync(struct io_conn *conn)
 535 {
 536         struct io_plan *plan = &conn->plan[IO_OUT];
 537         bool ok;
 538
 539         /* Not writing?  Nothing to do. */
 540         if (plan->status != IO_POLLING_STARTED
 541             && plan->status != IO_POLLING_NOTSTARTED)
 542                 return true;
 543
 544         /* Synchronous please. */
 545         io_fd_block(io_conn_fd(conn), true);
 546
 547 again:
 548         switch (plan->io(conn->fd.fd, &plan->arg)) {
 549         case -1:
 550                 ok = false;
 551                 break;
 552         /* Incomplete, try again. */
 553         case 0:
 554                 plan->status = IO_POLLING_STARTED;
 555                 goto again;
 556         case 1:
 557                 ok = true;
 558                 /* In case they come back. */
 559                 set_always(conn, IO_OUT, plan->next, plan->next_arg);
 560                 break;
 561         default:
 562                 /* IO should only return -1, 0 or 1 */
 563                 abort();
 564         }
 565
 566         io_fd_block(io_conn_fd(conn), false);
 567         return ok;
 568 }