X-Git-Url: http://git.ozlabs.org/?a=blobdiff_plain;f=ccan%2Fio%2Fio.c;h=c725850fd810eeb3e5a14c1ad0e6a64b924a8b6f;hb=bafb520171bacdc19bcf2f75bb224f36563d7851;hp=84c2ccad0c2c0ac032201ab6fb8bce44a05f1a23;hpb=1fe2db9cbe84812e1465db36f538a222e9984c93;p=ccan diff --git a/ccan/io/io.c b/ccan/io/io.c index 84c2ccad..c725850f 100644 --- a/ccan/io/io.c +++ b/ccan/io/io.c @@ -8,12 +8,13 @@ #include #include #include +#include void *io_loop_return; struct io_listener *io_new_listener_(int fd, - struct io_op *(*start)(struct io_conn *, - void *arg), + struct io_plan *(*start)(struct io_conn *, + void *arg), void (*finish)(struct io_conn *, void *), void *arg) { @@ -24,9 +25,9 @@ struct io_listener *io_new_listener_(int fd, l->fd.listener = true; l->fd.fd = fd; - l->fd.next = start; - l->fd.finish = finish; - l->fd.finish_arg = l->fd.next_arg = arg; + l->next = start; + l->finish = finish; + l->conn_arg = arg; if (!add_listener(l)) { free(l); return NULL; @@ -42,7 +43,7 @@ void io_close_listener(struct io_listener *l) } struct io_conn *io_new_conn_(int fd, - struct io_op *(*start)(struct io_conn *, void *), + struct io_plan *(*start)(struct io_conn *, void *), void (*finish)(struct io_conn *, void *), void *arg) { @@ -53,10 +54,11 @@ struct io_conn *io_new_conn_(int fd, conn->fd.listener = false; conn->fd.fd = fd; - conn->fd.next = start; - conn->fd.finish = finish; - conn->fd.finish_arg = conn->fd.next_arg = arg; - conn->state = NEXT; + conn->next = start; + conn->finish = finish; + conn->finish_arg = conn->next_arg = arg; + conn->pollflag = 0; + conn->state = IO_NEXT; conn->duplex = NULL; conn->timeout = NULL; if (!add_conn(conn)) { @@ -67,7 +69,7 @@ struct io_conn *io_new_conn_(int fd, } struct io_conn *io_duplex_(struct io_conn *old, - struct io_op *(*start)(struct io_conn *, void *), + struct io_plan *(*start)(struct io_conn *, void *), void (*finish)(struct io_conn *, void *), void *arg) { @@ -81,10 +83,11 @@ struct io_conn *io_duplex_(struct io_conn *old, conn->fd.listener = false; conn->fd.fd = old->fd.fd; - conn->fd.next = start; - conn->fd.finish = finish; - conn->fd.finish_arg = conn->fd.next_arg = arg; - conn->state = NEXT; + conn->next = start; + conn->finish = finish; + conn->finish_arg = conn->next_arg = arg; + conn->pollflag = 0; + conn->state = IO_NEXT; conn->duplex = old; conn->timeout = NULL; if (!add_duplex(conn)) { @@ -95,34 +98,13 @@ struct io_conn *io_duplex_(struct io_conn *old, return conn; } -/* Convenient token which only we can produce. */ -static inline struct io_next *to_ionext(struct io_conn *conn) +static inline struct io_plan *to_ioplan(enum io_state state) { - return (struct io_next *)conn; -} - -static inline struct io_op *to_ioop(enum io_state state) -{ - return (struct io_op *)(long)state; -} - -static inline struct io_conn *from_ionext(struct io_next *next) -{ - return (struct io_conn *)next; -} - -struct io_next *io_next_(struct io_conn *conn, - struct io_op *(*next)(struct io_conn *, void *), - void *arg) -{ - conn->fd.next = next; - conn->fd.next_arg = arg; - - return to_ionext(conn); + return (struct io_plan *)(long)state; } bool io_timeout_(struct io_conn *conn, struct timespec ts, - struct io_op *(*next)(struct io_conn *, void *), void *arg) + struct io_plan *(*cb)(struct io_conn *, void *), void *arg) { if (!conn->timeout) { conn->timeout = malloc(sizeof(*conn->timeout)); @@ -131,132 +113,172 @@ bool io_timeout_(struct io_conn *conn, struct timespec ts, } else assert(!timeout_active(conn)); - conn->timeout->next = next; + conn->timeout->next = cb; conn->timeout->next_arg = arg; backend_add_timeout(conn, ts); return true; } +static enum io_result do_write(struct io_conn *conn) +{ + ssize_t ret = write(conn->fd.fd, conn->u.write.buf, conn->u.write.len); + if (ret < 0) + return RESULT_CLOSE; + + conn->u.write.buf += ret; + conn->u.write.len -= ret; + if (conn->u.write.len == 0) + return RESULT_FINISHED; + else + return RESULT_AGAIN; +} + /* Queue some data to be written. */ -struct io_op *io_write(const void *data, size_t len, struct io_next *next) +struct io_plan *io_write_(struct io_conn *conn, const void *data, size_t len, + struct io_plan *(*cb)(struct io_conn *, void *), + void *arg) { - struct io_conn *conn = from_ionext(next); conn->u.write.buf = data; conn->u.write.len = len; - return to_ioop(WRITE); + conn->io = do_write; + conn->next = cb; + conn->next_arg = arg; + conn->pollflag = POLLOUT; + return to_ioplan(IO_IO); +} + +static enum io_result do_read(struct io_conn *conn) +{ + ssize_t ret = read(conn->fd.fd, conn->u.read.buf, conn->u.read.len); + if (ret <= 0) + return RESULT_CLOSE; + conn->u.read.buf += ret; + conn->u.read.len -= ret; + if (conn->u.read.len == 0) + return RESULT_FINISHED; + else + return RESULT_AGAIN; } /* Queue a request to read into a buffer. */ -struct io_op *io_read(void *data, size_t len, struct io_next *next) +struct io_plan *io_read_(struct io_conn *conn, void *data, size_t len, + struct io_plan *(*cb)(struct io_conn *, void *), + void *arg) { - struct io_conn *conn = from_ionext(next); conn->u.read.buf = data; conn->u.read.len = len; - return to_ioop(READ); + conn->io = do_read; + conn->next = cb; + conn->next_arg = arg; + conn->pollflag = POLLIN; + return to_ioplan(IO_IO); +} + +static enum io_result do_read_partial(struct io_conn *conn) +{ + ssize_t ret = read(conn->fd.fd, conn->u.readpart.buf, + *conn->u.readpart.lenp); + if (ret <= 0) + return RESULT_CLOSE; + *conn->u.readpart.lenp = ret; + return RESULT_FINISHED; } /* Queue a partial request to read into a buffer. */ -struct io_op *io_read_partial(void *data, size_t *len, struct io_next *next) +struct io_plan *io_read_partial_(struct io_conn *conn, void *data, size_t *len, + struct io_plan *(*cb)(struct io_conn *, void *), + void *arg) { - struct io_conn *conn = from_ionext(next); conn->u.readpart.buf = data; conn->u.readpart.lenp = len; - return to_ioop(READPART); + conn->io = do_read_partial; + conn->next = cb; + conn->next_arg = arg; + conn->pollflag = POLLIN; + return to_ioplan(IO_IO); +} + +static enum io_result do_write_partial(struct io_conn *conn) +{ + ssize_t ret = write(conn->fd.fd, conn->u.writepart.buf, + *conn->u.writepart.lenp); + if (ret < 0) + return RESULT_CLOSE; + *conn->u.writepart.lenp = ret; + return RESULT_FINISHED; } /* Queue a partial write request. */ -struct io_op *io_write_partial(const void *data, size_t *len, struct io_next *next) +struct io_plan *io_write_partial_(struct io_conn *conn, + const void *data, size_t *len, + struct io_plan *(*cb)(struct io_conn*, void *), + void *arg) { - struct io_conn *conn = from_ionext(next); conn->u.writepart.buf = data; conn->u.writepart.lenp = len; - return to_ioop(WRITEPART); + conn->io = do_write_partial; + conn->next = cb; + conn->next_arg = arg; + conn->pollflag = POLLOUT; + return to_ioplan(IO_IO); } -struct io_op *io_idle(struct io_conn *conn) +struct io_plan *io_idle(struct io_conn *conn) { - return to_ioop(IDLE); + conn->pollflag = 0; + return to_ioplan(IO_IDLE); } void io_wake_(struct io_conn *conn, - struct io_op *(*next)(struct io_conn *, void *), void *arg) + struct io_plan *(*fn)(struct io_conn *, void *), void *arg) { /* It might have finished, but we haven't called its finish() yet. */ - if (conn->state == FINISHED) + if (conn->state == IO_FINISHED) return; - assert(conn->state == IDLE); - conn->fd.next = next; - conn->fd.next_arg = arg; - backend_set_state(conn, to_ioop(NEXT)); + assert(conn->state == IO_IDLE); + conn->next = fn; + conn->next_arg = arg; + backend_set_state(conn, to_ioplan(IO_NEXT)); } -static struct io_op *do_next(struct io_conn *conn) +static struct io_plan *do_next(struct io_conn *conn) { if (timeout_active(conn)) backend_del_timeout(conn); - return conn->fd.next(conn, conn->fd.next_arg); + return conn->next(conn, conn->next_arg); } -struct io_op *do_ready(struct io_conn *conn) +struct io_plan *do_ready(struct io_conn *conn) { - ssize_t ret; - bool finished; - - switch (conn->state) { - case WRITE: - ret = write(conn->fd.fd, conn->u.write.buf, conn->u.write.len); - if (ret < 0) - return io_close(conn, NULL); - conn->u.write.buf += ret; - conn->u.write.len -= ret; - finished = (conn->u.write.len == 0); - break; - case WRITEPART: - ret = write(conn->fd.fd, conn->u.writepart.buf, - *conn->u.writepart.lenp); - if (ret < 0) - return io_close(conn, NULL); - *conn->u.writepart.lenp = ret; - finished = true; - break; - case READ: - ret = read(conn->fd.fd, conn->u.read.buf, conn->u.read.len); - if (ret <= 0) - return io_close(conn, NULL); - conn->u.read.buf += ret; - conn->u.read.len -= ret; - finished = (conn->u.read.len == 0); - break; - case READPART: - ret = read(conn->fd.fd, conn->u.readpart.buf, - *conn->u.readpart.lenp); - if (ret <= 0) - return io_close(conn, NULL); - *conn->u.readpart.lenp = ret; - finished = true; - break; + assert(conn->state == IO_IO); + switch (conn->io(conn)) { + case RESULT_CLOSE: + return io_close(conn, NULL); + case RESULT_FINISHED: + return do_next(conn); + case RESULT_AGAIN: + return to_ioplan(conn->state); default: - /* Shouldn't happen. */ abort(); } - - if (finished) - return do_next(conn); - return to_ioop(conn->state); } /* Useful next functions. */ /* Close the connection, we're done. */ -struct io_op *io_close(struct io_conn *conn, void *arg) +struct io_plan *io_close(struct io_conn *conn, void *arg) { - return to_ioop(FINISHED); + return to_ioplan(IO_FINISHED); } /* Exit the loop, returning this (non-NULL) arg. */ -struct io_op *io_break(void *arg, struct io_next *next) +struct io_plan *io_break_(struct io_conn *conn, void *ret, + struct io_plan *(*fn)(struct io_conn *, void *), + void *arg) { - io_loop_return = arg; + io_loop_return = ret; + conn->next = fn; + conn->next_arg = arg; - return to_ioop(NEXT); + return to_ioplan(IO_NEXT); }