X-Git-Url: https://git.ozlabs.org/?a=blobdiff_plain;f=ccan%2Fio%2Fio.c;h=b74520444353761451529af0d0467522293f2686;hb=490b63852f281f0d72eb6f6dfa5e0dce36bcbe0d;hp=ddcd502f367d3afdbbb3562b91f883be13072ece;hpb=c321a1d02755a77ae790d6e8976b2ff15edba89d;p=ccan diff --git a/ccan/io/io.c b/ccan/io/io.c index ddcd502f..b7452044 100644 --- a/ccan/io/io.c +++ b/ccan/io/io.c @@ -9,9 +9,15 @@ #include #include #include +#include +#include void *io_loop_return; +struct io_alloc io_alloc = { + malloc, realloc, free +}; + #ifdef DEBUG /* Set to skip the next plan. */ bool io_plan_nodebug; @@ -40,6 +46,12 @@ struct io_plan io_debug(struct io_plan plan) current->plan = plan; backend_plan_changed(current); + /* If it closed, close duplex. */ + if (!current->plan.next && current->duplex) { + current->duplex->plan = io_close_(); + backend_plan_changed(current->duplex); + } + /* Call back into the loop immediately. */ io_loop_return = do_io_loop(&ready); @@ -48,7 +60,7 @@ struct io_plan io_debug(struct io_plan plan) if (!ready->plan.next) { /* Call finish function immediately. */ if (ready->finish) { - errno = ready->plan.u.close.saved_errno; + errno = ready->plan.u1.s; ready->finish(ready, ready->finish_arg); ready->finish = NULL; } @@ -123,7 +135,7 @@ struct io_listener *io_new_listener_(int fd, void (*init)(int fd, void *arg), void *arg) { - struct io_listener *l = malloc(sizeof(*l)); + struct io_listener *l = io_alloc.alloc(sizeof(*l)); if (!l) return NULL; @@ -133,7 +145,7 @@ struct io_listener *io_new_listener_(int fd, l->init = init; l->arg = arg; if (!add_listener(l)) { - free(l); + io_alloc.free(l); return NULL; } return l; @@ -143,12 +155,12 @@ void io_close_listener(struct io_listener *l) { close(l->fd.fd); del_listener(l); - free(l); + io_alloc.free(l); } struct io_conn *io_new_conn_(int fd, struct io_plan plan) { - struct io_conn *conn = malloc(sizeof(*conn)); + struct io_conn *conn = io_alloc.alloc(sizeof(*conn)); io_plan_debug_again(); @@ -163,7 +175,7 @@ struct io_conn *io_new_conn_(int fd, struct io_plan plan) conn->duplex = NULL; conn->timeout = NULL; if (!add_conn(conn)) { - free(conn); + io_alloc.free(conn); return NULL; } return conn; @@ -185,7 +197,7 @@ struct io_conn *io_duplex_(struct io_conn *old, struct io_plan plan) assert(!old->duplex); - conn = malloc(sizeof(*conn)); + conn = io_alloc.alloc(sizeof(*conn)); if (!conn) return NULL; @@ -197,7 +209,7 @@ struct io_conn *io_duplex_(struct io_conn *old, struct io_plan plan) conn->finish_arg = NULL; conn->timeout = NULL; if (!add_duplex(conn)) { - free(conn); + io_alloc.free(conn); return NULL; } old->duplex = conn; @@ -210,7 +222,7 @@ bool io_timeout_(struct io_conn *conn, struct timespec ts, assert(cb); if (!conn->timeout) { - conn->timeout = malloc(sizeof(*conn->timeout)); + conn->timeout = io_alloc.alloc(sizeof(*conn->timeout)); if (!conn->timeout) return false; } else @@ -225,13 +237,13 @@ bool io_timeout_(struct io_conn *conn, struct timespec ts, /* Returns true if we're finished. */ static int do_write(int fd, struct io_plan *plan) { - ssize_t ret = write(fd, plan->u.write.buf, plan->u.write.len); + ssize_t ret = write(fd, plan->u1.cp, plan->u2.s); if (ret < 0) return io_debug_io(-1); - plan->u.write.buf += ret; - plan->u.write.len -= ret; - return io_debug_io(plan->u.write.len == 0); + plan->u1.cp += ret; + plan->u2.s -= ret; + return io_debug_io(plan->u2.s == 0); } /* Queue some data to be written. */ @@ -242,8 +254,8 @@ struct io_plan io_write_(const void *data, size_t len, struct io_plan plan; assert(cb); - plan.u.write.buf = data; - plan.u.write.len = len; + plan.u1.const_vp = data; + plan.u2.s = len; plan.io = do_write; plan.next = cb; plan.next_arg = arg; @@ -254,13 +266,13 @@ struct io_plan io_write_(const void *data, size_t len, static int do_read(int fd, struct io_plan *plan) { - ssize_t ret = read(fd, plan->u.read.buf, plan->u.read.len); + ssize_t ret = read(fd, plan->u1.cp, plan->u2.s); if (ret <= 0) return io_debug_io(-1); - plan->u.read.buf += ret; - plan->u.read.len -= ret; - return io_debug_io(plan->u.read.len == 0); + plan->u1.cp += ret; + plan->u2.s -= ret; + return io_debug_io(plan->u2.s == 0); } /* Queue a request to read into a buffer. */ @@ -271,8 +283,8 @@ struct io_plan io_read_(void *data, size_t len, struct io_plan plan; assert(cb); - plan.u.read.buf = data; - plan.u.read.len = len; + plan.u1.cp = data; + plan.u2.s = len; plan.io = do_read; plan.next = cb; plan.next_arg = arg; @@ -283,11 +295,11 @@ struct io_plan io_read_(void *data, size_t len, static int do_read_partial(int fd, struct io_plan *plan) { - ssize_t ret = read(fd, plan->u.readpart.buf, *plan->u.readpart.lenp); + ssize_t ret = read(fd, plan->u1.cp, *(size_t *)plan->u2.vp); if (ret <= 0) return io_debug_io(-1); - *plan->u.readpart.lenp = ret; + *(size_t *)plan->u2.vp = ret; return io_debug_io(1); } @@ -299,8 +311,8 @@ struct io_plan io_read_partial_(void *data, size_t *len, struct io_plan plan; assert(cb); - plan.u.readpart.buf = data; - plan.u.readpart.lenp = len; + plan.u1.cp = data; + plan.u2.vp = len; plan.io = do_read_partial; plan.next = cb; plan.next_arg = arg; @@ -311,11 +323,11 @@ struct io_plan io_read_partial_(void *data, size_t *len, static int do_write_partial(int fd, struct io_plan *plan) { - ssize_t ret = write(fd, plan->u.writepart.buf, *plan->u.writepart.lenp); + ssize_t ret = write(fd, plan->u1.cp, *(size_t *)plan->u2.vp); if (ret < 0) return io_debug_io(-1); - *plan->u.writepart.lenp = ret; + *(size_t *)plan->u2.vp = ret; return io_debug_io(1); } @@ -327,8 +339,8 @@ struct io_plan io_write_partial_(const void *data, size_t *len, struct io_plan plan; assert(cb); - plan.u.writepart.buf = data; - plan.u.writepart.lenp = len; + plan.u1.const_vp = data; + plan.u2.vp = len; plan.io = do_write_partial; plan.next = cb; plan.next_arg = arg; @@ -337,6 +349,59 @@ struct io_plan io_write_partial_(const void *data, size_t *len, return plan; } +static int already_connected(int fd, struct io_plan *plan) +{ + return io_debug_io(1); +} + +static int do_connect(int fd, struct io_plan *plan) +{ + int err, ret; + socklen_t len = sizeof(err); + + /* Has async connect finished? */ + ret = getsockopt(fd, SOL_SOCKET, SO_ERROR, &err, &len); + if (ret < 0) + return -1; + + if (err == 0) { + /* Restore blocking if it was initially. */ + fcntl(fd, F_SETFD, plan->u1.s); + return 1; + } + return 0; +} + +struct io_plan io_connect_(int fd, const struct addrinfo *addr, + struct io_plan (*cb)(struct io_conn*, void *), + void *arg) +{ + struct io_plan plan; + + assert(cb); + + plan.next = cb; + plan.next_arg = arg; + + /* Save old flags, set nonblock if not already. */ + plan.u1.s = fcntl(fd, F_GETFD); + fcntl(fd, F_SETFD, plan.u1.s | O_NONBLOCK); + + /* Immediate connect can happen. */ + if (connect(fd, addr->ai_addr, addr->ai_addrlen) == 0) { + /* Dummy will be called immediately. */ + plan.pollflag = POLLOUT; + plan.io = already_connected; + } else { + if (errno != EINPROGRESS) + return io_close_(); + + plan.pollflag = POLLIN; + plan.io = do_connect; + } + return plan; +} + struct io_plan io_idle_(void) { struct io_plan plan; @@ -382,6 +447,14 @@ void io_ready(struct io_conn *conn) backend_plan_changed(conn); } set_current(NULL); + + /* If it closed, close duplex. */ + if (!conn->plan.next && conn->duplex) { + set_current(conn->duplex); + conn->duplex->plan = io_close(); + backend_plan_changed(conn->duplex); + set_current(NULL); + } } /* Close the connection, we're done. */ @@ -392,7 +465,7 @@ struct io_plan io_close_(void) plan.pollflag = 0; /* This means we're closing. */ plan.next = NULL; - plan.u.close.saved_errno = errno; + plan.u1.s = errno; return plan; } @@ -412,3 +485,17 @@ struct io_plan io_break_(void *ret, struct io_plan plan) return plan; } + +int io_conn_fd(const struct io_conn *conn) +{ + return conn->fd.fd; +} + +void io_set_alloc(void *(*allocfn)(size_t size), + void *(*reallocfn)(void *ptr, size_t size), + void (*freefn)(void *ptr)) +{ + io_alloc.alloc = allocfn; + io_alloc.realloc = reallocfn; + io_alloc.free = freefn; +}