X-Git-Url: http://git.ozlabs.org/?p=ccan;a=blobdiff_plain;f=ccan%2Fio%2Fio.c;h=12df06d82cb6474fed825cd381f7f225fc6213b1;hp=bdcbf9e26b8f9f580fea08c2c0913d31efe9dcc7;hb=HEAD;hpb=fedf515165bfafaf4fb98252ecda1abe050c8da5 diff --git a/ccan/io/io.c b/ccan/io/io.c index bdcbf9e2..58ae19bd 100644 --- a/ccan/io/io.c +++ b/ccan/io/io.c @@ -15,6 +15,7 @@ void *io_loop_return; struct io_plan io_conn_freed; +static bool io_extended_errors; struct io_listener *io_new_listener_(const tal_t *ctx, int fd, struct io_plan *(*init)(struct io_conn *, @@ -61,10 +62,7 @@ static bool next_plan(struct io_conn *conn, struct io_plan *plan) if (plan == &io_conn_freed) return false; - /* It should have set a plan inside this conn (or duplex) */ - assert(plan == &conn->plan[IO_IN] - || plan == &conn->plan[IO_OUT] - || plan == &conn->plan[2]); + assert(plan == &conn->plan[plan->dir]); assert(conn->plan[IO_IN].status != IO_UNSET || conn->plan[IO_OUT].status != IO_UNSET); @@ -100,7 +98,6 @@ struct io_conn *io_new_conn_(const tal_t *ctx, int fd, conn->fd.fd = fd; conn->finish = NULL; conn->finish_arg = NULL; - list_node_init(&conn->always); if (!add_conn(conn)) return tal_free(conn); @@ -108,6 +105,10 @@ struct io_conn *io_new_conn_(const tal_t *ctx, int fd, /* Keep our I/O async. */ io_fd_block(fd, false); + /* So we can get back from plan -> conn later */ + conn->plan[IO_OUT].dir = IO_OUT; + conn->plan[IO_IN].dir = IO_IN; + /* We start with out doing nothing, and in doing our init. */ conn->plan[IO_OUT].status = IO_UNSET; @@ -119,6 +120,16 @@ struct io_conn *io_new_conn_(const tal_t *ctx, int fd, return conn; } +bool io_conn_exclusive(struct io_conn *conn, bool exclusive) +{ + return backend_set_exclusive(&conn->plan[IO_IN], exclusive); +} + +bool io_conn_out_exclusive(struct io_conn *conn, bool exclusive) +{ + return backend_set_exclusive(&conn->plan[IO_OUT], exclusive); +} + void io_set_finish_(struct io_conn *conn, void (*finish)(struct io_conn *, void *), void *arg) @@ -144,7 +155,9 @@ static struct io_plan *set_always(struct io_conn *conn, struct io_plan *plan = &conn->plan[dir]; plan->status = IO_ALWAYS; - backend_new_always(conn); + /* Only happens on OOM, and only with non-default tal_backend. */ + if (!backend_new_always(plan)) + return NULL; return io_set_plan(conn, dir, NULL, next, arg); } @@ -371,9 +384,19 @@ void io_wake(const void *wait) backend_wake(wait); } -/* Returns false if this should not be touched (eg. freed). */ -static bool do_plan(struct io_conn *conn, struct io_plan *plan, - bool idle_on_epipe) +enum plan_result { + /* Destroyed, do not touch */ + FREED, + /* Worked, call again. */ + KEEP_GOING, + /* Failed with EAGAIN or did partial. */ + EXHAUSTED, + /* No longer interested in read (or write) */ + UNINTERESTED +}; + +static enum plan_result do_plan(struct io_conn *conn, struct io_plan *plan, + bool idle_on_epipe) { /* We shouldn't have polled for this event if this wasn't true! */ assert(plan->status == IO_POLLING_NOTSTARTED @@ -381,18 +404,26 @@ static bool do_plan(struct io_conn *conn, struct io_plan *plan, switch (plan->io(conn->fd.fd, &plan->arg)) { case -1: + /* This is expected, as we call optimistically! */ + if (errno == EAGAIN) + return EXHAUSTED; if (errno == EPIPE && idle_on_epipe) { plan->status = IO_UNSET; backend_new_plan(conn); - return false; + return UNINTERESTED; } io_close(conn); - return false; + return FREED; case 0: plan->status = IO_POLLING_STARTED; - return true; + /* If it started but didn't finish, don't call again. */ + return EXHAUSTED; case 1: - return next_plan(conn, plan); + if (!next_plan(conn, plan)) + return FREED; + if (plan->status == IO_POLLING_NOTSTARTED) + return KEEP_GOING; + return UNINTERESTED; default: /* IO should only return -1, 0 or 1 */ abort(); @@ -401,41 +432,55 @@ static bool do_plan(struct io_conn *conn, struct io_plan *plan, void io_ready(struct io_conn *conn, int pollflags) { - if (pollflags & POLLIN) - if (!do_plan(conn, &conn->plan[IO_IN], false)) - return; - - if (pollflags & POLLOUT) - /* If we're writing to a closed pipe, we need to wait for - * read to fail if we're duplex: we want to drain it! */ - do_plan(conn, &conn->plan[IO_OUT], - conn->plan[IO_IN].status == IO_POLLING_NOTSTARTED - || conn->plan[IO_IN].status == IO_POLLING_STARTED); -} - -void io_do_always(struct io_conn *conn) -{ - /* There's a corner case where the in next_plan wakes up the - * out, placing it in IO_ALWAYS and we end up processing it immediately, - * only to leave it in the always list. - * - * Yet we can't just process one, in case they are both supposed - * to be done, so grab state beforehand. - */ - bool always_out = (conn->plan[IO_OUT].status == IO_ALWAYS); - - if (conn->plan[IO_IN].status == IO_ALWAYS) - if (!next_plan(conn, &conn->plan[IO_IN])) - return; - - if (always_out) { - /* You can't *unalways* a conn (except by freeing, in which - * case next_plan() returned false */ - assert(conn->plan[IO_OUT].status == IO_ALWAYS); - next_plan(conn, &conn->plan[IO_OUT]); + enum plan_result res; + + if (pollflags & POLLIN) { + for (;;) { + res = do_plan(conn, &conn->plan[IO_IN], false); + switch (res) { + case FREED: + return; + case EXHAUSTED: + case UNINTERESTED: + goto try_write; + case KEEP_GOING: + continue; + } + abort(); + } + } + +try_write: + if (pollflags & POLLOUT) { + for (;;) { + /* If we're writing to a closed pipe, we need to wait for + * read to fail if we're duplex: we want to drain it! */ + res = do_plan(conn, &conn->plan[IO_OUT], + conn->plan[IO_IN].status == IO_POLLING_NOTSTARTED + || conn->plan[IO_IN].status == IO_POLLING_STARTED); + switch (res) { + case FREED: + case EXHAUSTED: + case UNINTERESTED: + return; + case KEEP_GOING: + continue; + } + abort(); + } } } +void io_do_always(struct io_plan *plan) +{ + struct io_conn *conn; + + assert(plan->status == IO_ALWAYS); + conn = container_of(plan, struct io_conn, plan[plan->dir]); + + next_plan(conn, plan); +} + void io_do_wakeup(struct io_conn *conn, enum io_direction dir) { struct io_plan *plan = &conn->plan[dir]; @@ -488,7 +533,7 @@ struct io_plan *io_duplex(struct io_conn *conn, assert(conn == container_of(in_plan, struct io_conn, plan[IO_IN])); /* in_plan must be conn->plan[IO_IN], out_plan must be [IO_OUT] */ assert(out_plan == in_plan + 1); - return out_plan + 1; + return in_plan; } struct io_plan *io_halfclose(struct io_conn *conn) @@ -530,6 +575,18 @@ bool io_plan_out_started(const struct io_conn *conn) return conn->plan[IO_OUT].status == IO_POLLING_STARTED; } +/* Despite being a TCP expert, I missed the full extent of this + * problem. The legendary ZmnSCPxj implemented it (with the URL + * pointing to the explanation), and I imitate that here. */ +struct io_plan *io_sock_shutdown(struct io_conn *conn) +{ + if (shutdown(io_conn_fd(conn), SHUT_WR) != 0) + return io_close(conn); + + /* And leave unset .*/ + return &conn->plan[IO_IN]; +} + bool io_flush_sync(struct io_conn *conn) { struct io_plan *plan = &conn->plan[IO_OUT]; @@ -565,3 +622,13 @@ again: io_fd_block(io_conn_fd(conn), false); return ok; } + +void io_set_extended_errors(bool state) +{ + io_extended_errors = state; +} + +bool io_get_extended_errors(void) +{ + return io_extended_errors; +}