X-Git-Url: http://git.ozlabs.org/?p=ccan;a=blobdiff_plain;f=ccan%2Fio%2Fio.c;h=12b2489023d94077d5562db876ab9004caae3dd4;hp=0e716fced062c594d42955603e3fcfd2af67d3bb;hb=d00c9d1be5a56fbf990882f9a6da5704b3f1ea75;hpb=897626152d12d7fd13a8feb36989eb5c8c1f3485 diff --git a/ccan/io/io.c b/ccan/io/io.c index 0e716fce..12b24890 100644 --- a/ccan/io/io.c +++ b/ccan/io/io.c @@ -14,6 +14,8 @@ void *io_loop_return; +struct io_plan io_conn_freed; + struct io_listener *io_new_listener_(const tal_t *ctx, int fd, struct io_plan *(*init)(struct io_conn *, void *), @@ -35,8 +37,6 @@ struct io_listener *io_new_listener_(const tal_t *ctx, int fd, void io_close_listener(struct io_listener *l) { - close(l->fd.fd); - del_listener(l); tal_free(l); } @@ -45,7 +45,8 @@ static struct io_plan *io_never_called(struct io_conn *conn, void *arg) abort(); } -static void next_plan(struct io_conn *conn, struct io_plan *plan) +/* Returns false if conn was freed. */ +static bool next_plan(struct io_conn *conn, struct io_plan *plan) { struct io_plan *(*next)(struct io_conn *, void *arg); @@ -57,6 +58,9 @@ static void next_plan(struct io_conn *conn, struct io_plan *plan) plan = next(conn, plan->next_arg); + 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] @@ -65,6 +69,7 @@ static void next_plan(struct io_conn *conn, struct io_plan *plan) || conn->plan[IO_OUT].status != IO_UNSET); backend_new_plan(conn); + return true; } static void set_blocking(int fd, bool block) @@ -92,8 +97,7 @@ struct io_conn *io_new_conn_(const tal_t *ctx, int fd, conn->fd.fd = fd; conn->finish = NULL; conn->finish_arg = NULL; - conn->list = NULL; - conn->debug = false; + list_node_init(&conn->always); if (!add_conn(conn)) return tal_free(conn); @@ -106,7 +110,8 @@ struct io_conn *io_new_conn_(const tal_t *ctx, int fd, conn->plan[IO_IN].next = init; conn->plan[IO_IN].next_arg = arg; - next_plan(conn, &conn->plan[IO_IN]); + if (!next_plan(conn, &conn->plan[IO_IN])) + return NULL; return conn; } @@ -355,24 +360,20 @@ void io_wake(const void *wait) backend_wake(wait); } -static int do_plan(struct io_conn *conn, struct io_plan *plan) +/* Returns false if this has been freed. */ +static bool do_plan(struct io_conn *conn, struct io_plan *plan) { - /* Someone else might have called io_close() on us. */ - if (plan->status == IO_CLOSING) - return -1; - /* We shouldn't have polled for this event if this wasn't true! */ assert(plan->status == IO_POLLING); switch (plan->io(conn->fd.fd, &plan->arg)) { case -1: io_close(conn); - return -1; + return false; case 0: - return 0; + return true; case 1: - next_plan(conn, plan); - return 1; + return next_plan(conn, plan); default: /* IO should only return -1, 0 or 1 */ abort(); @@ -382,7 +383,8 @@ static int do_plan(struct io_conn *conn, struct io_plan *plan) void io_ready(struct io_conn *conn, int pollflags) { if (pollflags & POLLIN) - do_plan(conn, &conn->plan[IO_IN]); + if (!do_plan(conn, &conn->plan[IO_IN])) + return; if (pollflags & POLLOUT) do_plan(conn, &conn->plan[IO_OUT]); @@ -390,11 +392,25 @@ void io_ready(struct io_conn *conn, int pollflags) 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) - next_plan(conn, &conn->plan[IO_IN]); + if (!next_plan(conn, &conn->plan[IO_IN])) + return; - if (conn->plan[IO_OUT].status == IO_ALWAYS) + 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]); + } } void io_do_wakeup(struct io_conn *conn, enum io_direction dir) @@ -409,15 +425,8 @@ void io_do_wakeup(struct io_conn *conn, enum io_direction dir) /* Close the connection, we're done. */ struct io_plan *io_close(struct io_conn *conn) { - /* Already closing? Don't close twice. */ - if (conn->plan[IO_IN].status == IO_CLOSING) - return &conn->plan[IO_IN]; - - conn->plan[IO_IN].status = conn->plan[IO_OUT].status = IO_CLOSING; - conn->plan[IO_IN].arg.u1.s = errno; - backend_new_closing(conn); - - return io_set_plan(conn, IO_IN, NULL, NULL, NULL); + tal_free(conn); + return &io_conn_freed; } struct io_plan *io_close_cb(struct io_conn *conn, void *next_arg) @@ -425,6 +434,14 @@ struct io_plan *io_close_cb(struct io_conn *conn, void *next_arg) return io_close(conn); } +struct io_plan *io_close_taken_fd(struct io_conn *conn) +{ + set_blocking(conn->fd.fd, true); + + cleanup_conn_without_close(conn); + return io_close(conn); +} + /* Exit the loop, returning this (non-NULL) arg. */ void io_break(const void *ret) { @@ -442,43 +459,17 @@ int io_conn_fd(const struct io_conn *conn) return conn->fd.fd; } -void io_duplex_prepare(struct io_conn *conn) -{ - assert(conn->plan[IO_IN].status == IO_UNSET); - assert(conn->plan[IO_OUT].status == IO_UNSET); - - /* We can't sync debug until we've set both: io_wait() and io_always - * can't handle it. */ - conn->debug_saved = conn->debug; - io_set_debug(conn, false); -} - -struct io_plan *io_duplex_(struct io_plan *in_plan, struct io_plan *out_plan) +struct io_plan *io_duplex(struct io_conn *conn, + struct io_plan *in_plan, struct io_plan *out_plan) { - 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); - - /* Restore debug. */ - conn = container_of(in_plan, struct io_conn, plan[IO_IN]); - io_set_debug(conn, conn->debug_saved); - - /* Now set the plans again, to invoke sync debug. */ - io_set_plan(conn, IO_OUT, - out_plan->io, out_plan->next, out_plan->next_arg); - io_set_plan(conn, IO_IN, - in_plan->io, in_plan->next, in_plan->next_arg); - return out_plan + 1; } struct io_plan *io_halfclose(struct io_conn *conn) { - /* Already closing? Don't close twice. */ - if (conn->plan[IO_IN].status == IO_CLOSING) - return &conn->plan[IO_IN]; - /* Both unset? OK. */ if (conn->plan[IO_IN].status == IO_UNSET && conn->plan[IO_OUT].status == IO_UNSET) @@ -501,45 +492,41 @@ struct io_plan *io_set_plan(struct io_conn *conn, enum io_direction dir, plan->io = io; plan->next = next; plan->next_arg = next_arg; - assert(plan->status == IO_CLOSING || next != NULL); - - if (!conn->debug) - return plan; - - if (io_loop_return) { - io_debug_complete(conn); - return plan; - } - - switch (plan->status) { - case IO_POLLING: - while (do_plan(conn, plan) == 0); - break; - /* Shouldn't happen, since you said you did plan! */ - case IO_UNSET: - abort(); - case IO_ALWAYS: - /* If other one is ALWAYS, leave in list! */ - if (conn->plan[!dir].status != IO_ALWAYS) - remove_from_always(conn); - next_plan(conn, plan); - break; - case IO_WAITING: - case IO_CLOSING: - io_debug_complete(conn); - } + assert(next != NULL); return plan; } -void io_set_debug(struct io_conn *conn, bool debug) +bool io_flush_sync(struct io_conn *conn) { - conn->debug = debug; + struct io_plan *plan = &conn->plan[IO_OUT]; + bool ok; - /* Debugging means fds must block. */ - set_blocking(io_conn_fd(conn), debug); -} + /* Not writing? Nothing to do. */ + if (plan->status != IO_POLLING) + return true; -void io_debug_complete(struct io_conn *conn) -{ + /* Synchronous please. */ + set_blocking(io_conn_fd(conn), true); + +again: + switch (plan->io(conn->fd.fd, &plan->arg)) { + case -1: + ok = false; + break; + /* Incomplete, try again. */ + case 0: + goto again; + case 1: + ok = true; + /* In case they come back. */ + set_always(conn, IO_OUT, plan->next, plan->next_arg); + break; + default: + /* IO should only return -1, 0 or 1 */ + abort(); + } + + set_blocking(io_conn_fd(conn), false); + return ok; }