X-Git-Url: https://git.ozlabs.org/?a=blobdiff_plain;f=ccan%2Fio%2Fio.c;h=ef79e410217a8b76a4751112d8f0ac2e1c992475;hb=a5af4b170139a6e6a0dae7535fb00278ed4639c6;hp=6efc68ee732c15ab99d00104e2f6adb6ef0b5f71;hpb=a2dffefa5ef8d0cf71d99755c4640a8004679b1d;p=ccan diff --git a/ccan/io/io.c b/ccan/io/io.c index 6efc68ee..ef79e410 100644 --- a/ccan/io/io.c +++ b/ccan/io/io.c @@ -1,4 +1,4 @@ -/* Licensed under BSD-MIT - see LICENSE file for details */ +/* Licensed under LGPLv2.1+ - see LICENSE file for details */ #include "io.h" #include "backend.h" #include @@ -8,13 +8,12 @@ #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), - void (*finish)(struct io_conn *, void *), + void (*init)(int fd, void *arg), void *arg) { struct io_listener *l = malloc(sizeof(*l)); @@ -24,9 +23,8 @@ 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->init = init; + l->arg = arg; if (!add_listener(l)) { free(l); return NULL; @@ -42,7 +40,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 plan, void (*finish)(struct io_conn *, void *), void *arg) { @@ -53,11 +51,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->plan = plan; + conn->finish = finish; + conn->finish_arg = arg; conn->duplex = NULL; + conn->timeout = NULL; if (!add_conn(conn)) { free(conn); return NULL; @@ -66,9 +64,9 @@ 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 *), - void (*finish)(struct io_conn *, void *), - void *arg) + struct io_plan plan, + void (*finish)(struct io_conn *, void *), + void *arg) { struct io_conn *conn; @@ -80,11 +78,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->plan = plan; conn->duplex = old; + conn->finish = finish; + conn->finish_arg = arg; + conn->timeout = NULL; if (!add_duplex(conn)) { free(conn); return NULL; @@ -93,150 +91,200 @@ 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) +bool io_timeout_(struct io_conn *conn, struct timespec ts, + struct io_plan (*cb)(struct io_conn *, void *), void *arg) { - return (struct io_next *)conn; -} + if (!conn->timeout) { + conn->timeout = malloc(sizeof(*conn->timeout)); + if (!conn->timeout) + return false; + } else + assert(!timeout_active(conn)); -static inline struct io_op *to_ioop(enum io_state state) -{ - return (struct io_op *)(long)state; + conn->timeout->next = cb; + conn->timeout->next_arg = arg; + backend_add_timeout(conn, ts); + return true; } -static inline struct io_conn *from_ionext(struct io_next *next) +static enum io_result do_write(struct io_conn *conn) { - return (struct io_conn *)next; + ssize_t ret = write(conn->fd.fd, conn->plan.u.write.buf, conn->plan.u.write.len); + if (ret < 0) + return RESULT_CLOSE; + + conn->plan.u.write.buf += ret; + conn->plan.u.write.len -= ret; + if (conn->plan.u.write.len == 0) + return RESULT_FINISHED; + else + return RESULT_AGAIN; } -struct io_next *io_next_(struct io_conn *conn, - struct io_op *(*next)(struct io_conn *, void *), +/* Queue some data to be written. */ +struct io_plan io_write_(const void *data, size_t len, + struct io_plan (*cb)(struct io_conn *, void *), void *arg) { - conn->fd.next = next; - conn->fd.next_arg = arg; + struct io_plan plan; - return to_ionext(conn); + plan.u.write.buf = data; + plan.u.write.len = len; + plan.io = do_write; + plan.next = cb; + plan.next_arg = arg; + plan.pollflag = POLLOUT; + plan.state = IO_IO; + return plan; } -/* Queue some data to be written. */ -struct io_op *io_write(const void *data, size_t len, struct io_next *next) +static enum io_result do_read(struct io_conn *conn) { - struct io_conn *conn = from_ionext(next); - conn->u.write.buf = data; - conn->u.write.len = len; - return to_ioop(WRITE); + ssize_t ret = read(conn->fd.fd, conn->plan.u.read.buf, + conn->plan.u.read.len); + if (ret <= 0) + return RESULT_CLOSE; + conn->plan.u.read.buf += ret; + conn->plan.u.read.len -= ret; + if (conn->plan.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_(void *data, size_t len, + struct io_plan (*cb)(struct io_conn *, void *), + void *arg) +{ + struct io_plan plan; + + plan.u.read.buf = data; + plan.u.read.len = len; + plan.io = do_read; + plan.next = cb; + plan.next_arg = arg; + plan.pollflag = POLLIN; + plan.state = IO_IO; + return plan; +} + +static enum io_result do_read_partial(struct io_conn *conn) { - struct io_conn *conn = from_ionext(next); - conn->u.read.buf = data; - conn->u.read.len = len; - return to_ioop(READ); + ssize_t ret = read(conn->fd.fd, conn->plan.u.readpart.buf, + *conn->plan.u.readpart.lenp); + if (ret <= 0) + return RESULT_CLOSE; + *conn->plan.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_(void *data, size_t *len, + struct io_plan (*cb)(struct io_conn *, void *), + void *arg) +{ + struct io_plan plan; + + plan.u.readpart.buf = data; + plan.u.readpart.lenp = len; + plan.io = do_read_partial; + plan.next = cb; + plan.next_arg = arg; + plan.pollflag = POLLIN; + plan.state = IO_IO; + + return plan; +} + +static enum io_result do_write_partial(struct io_conn *conn) { - struct io_conn *conn = from_ionext(next); - conn->u.readpart.buf = data; - conn->u.readpart.lenp = len; - return to_ioop(READPART); + ssize_t ret = write(conn->fd.fd, conn->plan.u.writepart.buf, + *conn->plan.u.writepart.lenp); + if (ret < 0) + return RESULT_CLOSE; + *conn->plan.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_(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); + struct io_plan plan; + + plan.u.writepart.buf = data; + plan.u.writepart.lenp = len; + plan.io = do_write_partial; + plan.next = cb; + plan.next_arg = arg; + plan.pollflag = POLLOUT; + plan.state = IO_IO; + + return plan; } -struct io_op *io_idle(struct io_conn *conn) +struct io_plan io_idle(void) { - return to_ioop(IDLE); + struct io_plan plan; + + plan.pollflag = 0; + plan.state = IO_IO; + plan.io = NULL; + + return plan; } -void io_wake_(struct io_conn *conn, - struct io_op *(*next)(struct io_conn *, void *), void *arg) +void io_wake(struct io_conn *conn, struct io_plan plan) { /* It might have finished, but we haven't called its finish() yet. */ - if (conn->state == FINISHED) + if (conn->plan.state == IO_FINISHED) return; - assert(conn->state == IDLE); - conn->fd.next = next; - conn->fd.next_arg = arg; - backend_set_state(conn, to_ioop(NEXT)); -} - -static struct io_op *do_next(struct io_conn *conn) -{ - return conn->fd.next(conn, conn->fd.next_arg); -} - -struct io_op *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->plan.io); + conn->plan = plan; + backend_wakeup(conn); +} + +static struct io_plan do_next(struct io_conn *conn) +{ + if (timeout_active(conn)) + backend_del_timeout(conn); + return conn->plan.next(conn, conn->plan.next_arg); +} + +struct io_plan do_ready(struct io_conn *conn) +{ + assert(conn->plan.state == IO_IO); + switch (conn->plan.io(conn)) { + case RESULT_CLOSE: + return io_close(conn, NULL); + case RESULT_FINISHED: + return do_next(conn); + case RESULT_AGAIN: + return conn->plan; 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); + struct io_plan plan; + + plan.state = IO_FINISHED; + plan.pollflag = 0; + + return plan; } /* Exit the loop, returning this (non-NULL) arg. */ -struct io_op *io_break(void *arg, struct io_next *next) +struct io_plan io_break(void *ret, struct io_plan plan) { - io_loop_return = arg; + io_loop_return = ret; - return to_ioop(NEXT); + return plan; }