From: Rusty Russell Date: Fri, 15 Nov 2013 05:23:31 +0000 (+1030) Subject: io: io_close_other() X-Git-Url: http://git.ozlabs.org/?p=ccan;a=commitdiff_plain;h=95b59482c1bb18b7904ea60149eff4809dd28d80;hp=f508ed408b230c21341778370ab04def8e7aba3a io: io_close_other() And add test for that which also tests duplex case. Signed-off-by: Rusty Russell --- diff --git a/ccan/io/io.c b/ccan/io/io.c index a58b3f01..faf8b87b 100644 --- a/ccan/io/io.c +++ b/ccan/io/io.c @@ -435,6 +435,10 @@ void io_wake_(struct io_conn *conn, struct io_plan plan) void io_ready(struct io_conn *conn) { + /* Beware io_close_other! */ + if (!conn->plan.next) + return; + set_current(conn); switch (conn->plan.io(conn->fd.fd, &conn->plan)) { case -1: /* Failure means a new plan: close up. */ @@ -474,6 +478,12 @@ struct io_plan io_close_cb(struct io_conn *conn, void *arg) return io_close(); } +void io_close_other(struct io_conn *conn) +{ + conn->plan = io_close_(); + backend_plan_changed(conn); +} + /* Exit the loop, returning this (non-NULL) arg. */ struct io_plan io_break_(void *ret, struct io_plan plan) { diff --git a/ccan/io/io.h b/ccan/io/io.h index 4468cee3..558a8769 100644 --- a/ccan/io/io.h +++ b/ccan/io/io.h @@ -495,6 +495,24 @@ struct io_plan io_close_(void); */ struct io_plan io_close_cb(struct io_conn *, void *unused); +/** + * io_close_other - close different connection next time around the I/O loop. + * @conn: the connection to close. + * + * This is used to force a different connection to close: no more I/O will + * happen on @conn, even if it's pending. + * + * It's a bug to use this on the current connection! + * + * Example: + * static void stop_connection(struct io_conn *conn) + * { + * printf("forcing stop on connection\n"); + * io_close_other(conn); + * } + */ +void io_close_other(struct io_conn *conn); + /** * io_loop - process fds until all closed on io_break. * diff --git a/ccan/io/test/run-16-duplex-test.c b/ccan/io/test/run-16-duplex-test.c new file mode 100644 index 00000000..8a1a04f0 --- /dev/null +++ b/ccan/io/test/run-16-duplex-test.c @@ -0,0 +1,142 @@ +/* Tests when the last connection is a duplex, and poll.c moves it over + * deleted fd. */ +#include +/* Include the C files directly. */ +#include +#include +#include +#include +#include + +#ifndef PORT +#define PORT "65016" +#endif + +struct data { + struct io_listener *l; + struct io_conn *writer; + int state; + char buf[4]; + char wbuf[32]; +}; + +static void finish_ok(struct io_conn *conn, struct data *d) +{ + d->state++; +} + +static struct io_plan write_done(struct io_conn *conn, struct data *d) +{ + d->state++; + return io_idle(); +} + +static struct io_plan read_done(struct io_conn *conn, struct data *d) +{ + d->state++; + io_close_other(d->writer); + return io_close(); +} + +static void init_conn(int fd, struct data *d) +{ + struct io_conn *conn; + + ok1(d->state == 0); + d->state++; + + memset(d->wbuf, 7, sizeof(d->wbuf)); + + conn = io_new_conn(fd, io_read(d->buf, sizeof(d->buf), read_done, d)); + io_set_finish(conn, finish_ok, d); + d->writer = io_duplex(conn, io_write(d->wbuf, sizeof(d->wbuf), write_done, d)); + ok1(d->writer); + io_set_finish(d->writer, finish_ok, d); + + io_close_listener(d->l); +} + +static int make_listen_fd(const char *port, struct addrinfo **info) +{ + int fd, on = 1; + struct addrinfo *addrinfo, hints; + + memset(&hints, 0, sizeof(hints)); + hints.ai_family = AF_UNSPEC; + hints.ai_socktype = SOCK_STREAM; + hints.ai_flags = AI_PASSIVE; + hints.ai_protocol = 0; + + if (getaddrinfo(NULL, port, &hints, &addrinfo) != 0) + return -1; + + fd = socket(addrinfo->ai_family, addrinfo->ai_socktype, + addrinfo->ai_protocol); + if (fd < 0) + return -1; + + setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on)); + if (bind(fd, addrinfo->ai_addr, addrinfo->ai_addrlen) != 0) { + close(fd); + return -1; + } + if (listen(fd, 1) != 0) { + close(fd); + return -1; + } + *info = addrinfo; + return fd; +} + +int main(void) +{ + struct data *d = malloc(sizeof(*d)); + struct addrinfo *addrinfo; + int fd, status; + + /* This is how many tests you plan to run */ + plan_tests(10); + d->state = 0; + fd = make_listen_fd(PORT, &addrinfo); + ok1(fd >= 0); + d->l = io_new_listener(fd, init_conn, d); + ok1(d->l); + fflush(stdout); + if (!fork()) { + int i; + char buf[32]; + + io_close_listener(d->l); + free(d); + fd = socket(addrinfo->ai_family, addrinfo->ai_socktype, + addrinfo->ai_protocol); + if (fd < 0) + exit(1); + if (connect(fd, addrinfo->ai_addr, addrinfo->ai_addrlen) != 0) + exit(2); + signal(SIGPIPE, SIG_IGN); + for (i = 0; i < 32; i++) { + if (read(fd, buf+i, 1) != 1) + break; + } + for (i = 0; i < strlen("hellothere"); i++) { + if (write(fd, "hellothere" + i, 1) != 1) + break; + } + close(fd); + freeaddrinfo(addrinfo); + exit(0); + } + freeaddrinfo(addrinfo); + ok1(io_loop() == NULL); + ok1(d->state == 5); + ok1(memcmp(d->buf, "hellothere", sizeof(d->buf)) == 0); + free(d); + + ok1(wait(&status)); + ok1(WIFEXITED(status)); + ok1(WEXITSTATUS(status) == 0); + + /* This exits depending on whether all tests passed */ + return exit_status(); +}