X-Git-Url: http://git.ozlabs.org/?a=blobdiff_plain;f=ccan%2Fio%2Fio.c;h=cd5557d16af9a5d7bf2e3003febfb89237ff7603;hb=8f116a81b39b55031212abb0be2ad088cbf9fd87;hp=6d1c0522d131c8e15809b8deb2e84ba752710302;hpb=43185ec669720903fa86aac5f2d1b1ee62e3a12b;p=ccan diff --git a/ccan/io/io.c b/ccan/io/io.c index 6d1c0522..cd5557d1 100644 --- a/ccan/io/io.c +++ b/ccan/io/io.c @@ -67,6 +67,18 @@ static void next_plan(struct io_conn *conn, struct io_plan *plan) backend_new_plan(conn); } +static void set_blocking(int fd, bool block) +{ + int flags = fcntl(fd, F_GETFL); + + if (block) + flags &= ~O_NONBLOCK; + else + flags |= O_NONBLOCK; + + fcntl(fd, F_SETFL, flags); +} + struct io_conn *io_new_conn_(const tal_t *ctx, int fd, struct io_plan *(*init)(struct io_conn *, void *), void *arg) @@ -80,12 +92,16 @@ 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; + list_node_init(&conn->always); + list_node_init(&conn->closing); conn->debug = false; if (!add_conn(conn)) return tal_free(conn); + /* Keep our I/O async. */ + set_blocking(fd, false); + /* We start with out doing nothing, and in doing our init. */ conn->plan[IO_OUT].status = IO_UNSET; @@ -125,16 +141,28 @@ static struct io_plan *set_always(struct io_conn *conn, return io_set_plan(conn, dir, NULL, next, arg); } +static struct io_plan *io_always_dir(struct io_conn *conn, + enum io_direction dir, + struct io_plan *(*next)(struct io_conn *, + void *), + void *arg) +{ + return set_always(conn, dir, next, arg); +} + struct io_plan *io_always_(struct io_conn *conn, struct io_plan *(*next)(struct io_conn *, void *), void *arg) { - /* If we're duplex, we want this on the current plan. Otherwise, - * doesn't matter. */ - if (conn->plan[IO_IN].status == IO_UNSET) - return set_always(conn, IO_IN, next, arg); - else - return set_always(conn, IO_OUT, next, arg); + return io_always_dir(conn, IO_IN, next, arg); +} + +struct io_plan *io_out_always_(struct io_conn *conn, + struct io_plan *(*next)(struct io_conn *, + void *), + void *arg) +{ + return io_always_dir(conn, IO_OUT, next, arg); } static int do_write(int fd, struct io_plan_arg *arg) @@ -263,8 +291,6 @@ static int do_connect(int fd, struct io_plan_arg *arg) return -1; if (err == 0) { - /* Restore blocking if it was initially. */ - fcntl(fd, F_SETFL, arg->u1.s); return 1; } else if (err == EINPROGRESS) return 0; @@ -277,39 +303,31 @@ struct io_plan *io_connect_(struct io_conn *conn, const struct addrinfo *addr, struct io_plan *(*next)(struct io_conn *, void *), void *next_arg) { - struct io_plan_arg *arg = io_plan_arg(conn, IO_IN); int fd = io_conn_fd(conn); - /* Save old flags, set nonblock if not already. */ - arg->u1.s = fcntl(fd, F_GETFL); - fcntl(fd, F_SETFL, arg->u1.s | O_NONBLOCK); + /* We don't actually need the arg, but we need it polling. */ + io_plan_arg(conn, IO_OUT); + + /* Note that io_new_conn() will make fd O_NONBLOCK */ /* Immediate connect can happen. */ if (connect(fd, addr->ai_addr, addr->ai_addrlen) == 0) - return set_always(conn, IO_IN, next, next_arg); + return set_always(conn, IO_OUT, next, next_arg); if (errno != EINPROGRESS) return io_close(conn); - return io_set_plan(conn, IO_IN, do_connect, next, next_arg); + return io_set_plan(conn, IO_OUT, do_connect, next, next_arg); } -struct io_plan *io_wait_(struct io_conn *conn, - const void *wait, - struct io_plan *(*next)(struct io_conn *, void *), - void *next_arg) +static struct io_plan *io_wait_dir(struct io_conn *conn, + const void *wait, + enum io_direction dir, + struct io_plan *(*next)(struct io_conn *, + void *), + void *next_arg) { - enum io_direction dir; - struct io_plan_arg *arg; - - /* If we're duplex, we want this on the current plan. Otherwise, - * doesn't matter. */ - if (conn->plan[IO_IN].status == IO_UNSET) - dir = IO_IN; - else - dir = IO_OUT; - - arg = io_plan_arg(conn, dir); + struct io_plan_arg *arg = io_plan_arg(conn, dir); arg->u1.const_vp = wait; conn->plan[dir].status = IO_WAITING; @@ -317,6 +335,22 @@ struct io_plan *io_wait_(struct io_conn *conn, return io_set_plan(conn, dir, NULL, next, next_arg); } +struct io_plan *io_wait_(struct io_conn *conn, + const void *wait, + struct io_plan *(*next)(struct io_conn *, void *), + void *next_arg) +{ + return io_wait_dir(conn, wait, IO_IN, next, next_arg); +} + +struct io_plan *io_out_wait_(struct io_conn *conn, + const void *wait, + struct io_plan *(*next)(struct io_conn *, void *), + void *next_arg) +{ + return io_wait_dir(conn, wait, IO_OUT, next, next_arg); +} + void io_wake(const void *wait) { backend_wake(wait); @@ -364,10 +398,13 @@ void io_do_always(struct io_conn *conn) next_plan(conn, &conn->plan[IO_OUT]); } -void io_do_wakeup(struct io_conn *conn, struct io_plan *plan) +void io_do_wakeup(struct io_conn *conn, enum io_direction dir) { + struct io_plan *plan = &conn->plan[dir]; + assert(plan->status == IO_WAITING); - next_plan(conn, plan); + + set_always(conn, dir, plan->next, plan->next_arg); } /* Close the connection, we're done. */ @@ -414,7 +451,7 @@ void io_duplex_prepare(struct io_conn *conn) /* We can't sync debug until we've set both: io_wait() and io_always * can't handle it. */ conn->debug_saved = conn->debug; - conn->debug = false; + io_set_debug(conn, false); } struct io_plan *io_duplex_(struct io_plan *in_plan, struct io_plan *out_plan) @@ -426,7 +463,7 @@ struct io_plan *io_duplex_(struct io_plan *in_plan, struct io_plan *out_plan) /* Restore debug. */ conn = container_of(in_plan, struct io_conn, plan[IO_IN]); - conn->debug = conn->debug_saved; + io_set_debug(conn, conn->debug_saved); /* Now set the plans again, to invoke sync debug. */ io_set_plan(conn, IO_OUT, @@ -437,6 +474,24 @@ struct io_plan *io_duplex_(struct io_plan *in_plan, struct io_plan *out_plan) 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) + return io_close(conn); + + /* We leave this unset then. */ + if (conn->plan[IO_IN].status == IO_UNSET) + return &conn->plan[IO_IN]; + else + return &conn->plan[IO_OUT]; +} + struct io_plan *io_set_plan(struct io_conn *conn, enum io_direction dir, int (*io)(int fd, struct io_plan_arg *arg), struct io_plan *(*next)(struct io_conn *, void *), @@ -481,6 +536,9 @@ struct io_plan *io_set_plan(struct io_conn *conn, enum io_direction dir, void io_set_debug(struct io_conn *conn, bool debug) { conn->debug = debug; + + /* Debugging means fds must block. */ + set_blocking(io_conn_fd(conn), debug); } void io_debug_complete(struct io_conn *conn)