ccan/io: go linear for debugging.

Debugging an async library is a pain: it's nice to force it into a
linear call chain to try to track problems.

Ugly code, though.
Signed-off-by: Rusty Russell <rusty@rustcorp.com.au>

ccan/io/backend.h

@@ -47,6 +47,26 @@ static inline bool timeout_active(const struct io_conn *conn)
 
 extern void *io_loop_return;
 
+#ifdef DEBUG
+extern struct io_conn *current;
+static inline void set_current(struct io_conn *conn)
+{
+	current = conn;
+}
+static inline bool doing_debug(void)
+{
+	return io_debug != NULL;
+}
+#else
+static inline void set_current(struct io_conn *conn)
+{
+}
+static inline bool doing_debug(void)
+{
+	return false;
+}
+#endif
+
 bool add_listener(struct io_listener *l);
 bool add_conn(struct io_conn *c);
 bool add_duplex(struct io_conn *c);

ccan/io/io.c

@@ -12,6 +12,48 @@
 
 void *io_loop_return;
 
+#ifdef DEBUG
+bool io_plan_for_other;
+struct io_conn *current;
+bool (*io_debug)(struct io_conn *conn);
+bool io_debug_wakeup;
+
+static void debug_io_plan(struct io_plan *plan)
+{
+	if (io_plan_for_other) {
+		io_plan_for_other = false;
+		return;
+	}
+
+	if (!io_debug || !current)
+		return;
+
+	if (!io_debug(current) && !io_debug_wakeup)
+		return;
+
+	io_debug_wakeup = false;
+	current->plan = *plan;
+	backend_plan_changed(current);
+
+	/* Call back into the loop immediately. */
+	io_loop_return = io_loop();
+}
+
+static void debug_io_wake(struct io_conn *conn)
+{
+	/* We want linear if we wake a debugged connection, too. */
+	if (io_debug && io_debug(conn))
+		io_debug_wakeup = true;
+}
+#else
+static void debug_io_plan(struct io_plan *plan)
+{
+}
+static void debug_io_wake(struct io_conn *conn)
+{
+}
+#endif
+
 struct io_listener *io_new_listener_(int fd,
 				     void (*init)(int fd, void *arg),
 				     void *arg)
@@ -138,6 +180,8 @@ struct io_plan io_write_(const void *data, size_t len,
 	plan.next = cb;
 	plan.next_arg = arg;
 	plan.pollflag = POLLOUT;
+
+	debug_io_plan(&plan);
 	return plan;
 }
 
@@ -169,6 +213,8 @@ struct io_plan io_read_(void *data, size_t len,
 	plan.next = cb;
 	plan.next_arg = arg;
 	plan.pollflag = POLLIN;
+
+	debug_io_plan(&plan);
 	return plan;
 }
 
@@ -200,6 +246,7 @@ struct io_plan io_read_partial_(void *data, size_t *len,
 	plan.next_arg = arg;
 	plan.pollflag = POLLIN;
 
+	debug_io_plan(&plan);
 	return plan;
 }
 
@@ -231,6 +278,7 @@ struct io_plan io_write_partial_(const void *data, size_t *len,
 	plan.next_arg = arg;
 	plan.pollflag = POLLOUT;
 
+	debug_io_plan(&plan);
 	return plan;
 }
 
@@ -243,10 +291,11 @@ struct io_plan io_idle(void)
 	/* Never called (overridded by io_wake), but NULL means closing */
 	plan.next = io_close;
 
+	debug_io_plan(&plan);
 	return plan;
 }
 
-void io_wake(struct io_conn *conn, struct io_plan plan)
+void io_wake_(struct io_conn *conn, struct io_plan plan)
 
 {
 	/* It might be closing, but we haven't called its finish() yet. */
@@ -256,15 +305,19 @@ void io_wake(struct io_conn *conn, struct io_plan plan)
 	assert(!conn->plan.io);
 	conn->plan = plan;
 	backend_plan_changed(conn);
+
+	debug_io_wake(conn);
 }
 
 void io_ready(struct io_conn *conn)
 {
 	if (conn->plan.io(conn->fd.fd, &conn->plan)) {
+		set_current(conn);
 		if (timeout_active(conn))
 			backend_del_timeout(conn);
 		conn->plan = conn->plan.next(conn, conn->plan.next_arg);
 		backend_plan_changed(conn);
+		set_current(NULL);
 	}
 }
 
@@ -278,11 +331,12 @@ struct io_plan io_close(struct io_conn *conn, void *arg)
 	/* This means we're closing. */
 	plan.next = NULL;
 
+	debug_io_plan(&plan);
 	return plan;
 }
 
 /* Exit the loop, returning this (non-NULL) arg. */
-struct io_plan io_break(void *ret, struct io_plan plan)
+struct io_plan io_break_(void *ret, struct io_plan plan)
 {
 	assert(ret);
 	io_loop_return = ret;

ccan/io/io.h

@@ -8,6 +8,14 @@
 
 struct io_conn;
 
+#ifdef DEBUG
+extern bool io_plan_for_other;
+extern bool (*io_debug)(struct io_conn *conn);
+#define io_plan_other() ((io_plan_for_other = true))
+#else
+#define io_plan_other() (void)0
+#endif
+
 struct io_state_read {
 	char *buf;
 	size_t len;
@@ -63,10 +71,11 @@ struct io_plan {
  * Returns NULL on error (and sets errno).
  */
 #define io_new_conn(fd, plan, finish, arg)				\
-	io_new_conn_((fd), (plan),					\
-		     typesafe_cb_preargs(void, void *, (finish), (arg),	\
+	(io_plan_other(), io_new_conn_((fd), (plan),			\
+				       typesafe_cb_preargs(void, void *, \
+							   (finish), (arg), \
 							   struct io_conn *), \
-		     (arg))
+				       (arg)))
 struct io_conn *io_new_conn_(int fd,
 			     struct io_plan plan,
 			     void (*finish)(struct io_conn *, void *),
@@ -237,10 +246,11 @@ bool io_timeout_(struct io_conn *conn, struct timespec ts,
  * You must io_close() both of them to close the fd.
  */
 #define io_duplex(conn, plan, finish, arg)				\
-	io_duplex_((conn), (plan),					\
-		   typesafe_cb_preargs(void, void *, (finish), (arg),	\
+	(io_plan_other(), io_duplex_((conn), (plan),			\
+				     typesafe_cb_preargs(void, void *,	\
+							 (finish), (arg), \
 							 struct io_conn *), \
-		   (arg))
+				     (arg)))
 
 struct io_conn *io_duplex_(struct io_conn *conn,
 			   struct io_plan plan,
@@ -254,7 +264,8 @@ struct io_conn *io_duplex_(struct io_conn *conn,
  *
  * This makes @conn do I/O the next time around the io_loop().
  */
-void io_wake(struct io_conn *conn, struct io_plan plan);
+#define io_wake(conn, plan) (io_plan_other(), io_wake_((conn), (plan)))
+void io_wake_(struct io_conn *conn, struct io_plan plan);
 
 /**
  * io_break - return from io_loop()
@@ -267,7 +278,8 @@ void io_wake(struct io_conn *conn, struct io_plan plan);
  *
  * If io_loop() is called again, then @plan will be carried out.
  */
-struct io_plan io_break(void *ret, struct io_plan plan);
+#define io_break(ret, plan) (io_plan_other(), io_break_((ret), (plan)))
+struct io_plan io_break_(void *ret, struct io_plan plan);
 
 /* FIXME: io_recvfrom/io_sendto */
 

ccan/io/poll.c

@@ -12,6 +12,49 @@ static size_t num_fds = 0, max_fds = 0, num_closing = 0, num_waiting = 0;
 static struct pollfd *pollfds = NULL;
 static struct fd **fds = NULL;
 static struct timers timeouts;
+#ifdef DEBUG
+static unsigned int io_loop_level;
+static struct io_conn *free_later;
+static void io_loop_enter(void)
+{
+	io_loop_level++;
+}
+static void io_loop_exit(void)
+{
+	io_loop_level--;
+	if (io_loop_level == 0) {
+		/* Delayed free. */
+		while (free_later) {
+			struct io_conn *c = free_later;
+			free_later = c->finish_arg;
+			free(c);
+		}
+	}
+}
+static void free_conn(struct io_conn *conn)
+{
+	/* Only free on final exit: chain via finish. */
+	if (io_loop_level > 1) {
+		struct io_conn *c;
+		for (c = free_later; c; c = c->finish_arg)
+			assert(c != conn);
+		conn->finish_arg = free_later;
+		free_later = conn;
+	} else
+		free(conn);
+}
+#else
+static void io_loop_enter(void)
+{
+}
+static void io_loop_exit(void)
+{
+}
+static void free_conn(struct io_conn *conn)
+{
+	free(conn);
+}
+#endif
 
 static bool add_fd(struct fd *fd, short events)
 {
@@ -83,7 +126,13 @@ bool add_listener(struct io_listener *l)
 
 void backend_plan_changed(struct io_conn *conn)
 {
-	struct pollfd *pfd = &pollfds[conn->fd.backend_info];
+	struct pollfd *pfd;
+
+	/* This can happen with debugging and delayed free... */
+	if (conn->fd.backend_info == -1)
+		return;
+
+	pfd = &pollfds[conn->fd.backend_info];
 
 	if (pfd->events)
 		num_waiting--;
@@ -133,6 +182,7 @@ static void del_conn(struct io_conn *conn)
 		/* In case fds[] pointed to the other one. */
 		fds[conn->fd.backend_info] = &conn->duplex->fd;
 		conn->duplex->duplex = NULL;
+		conn->fd.backend_info = -1;
 	} else
 		del_fd(&conn->fd);
 	num_closing--;
@@ -176,7 +226,7 @@ static void finish_conns(void)
 		for (duplex = c->duplex; c; c = duplex, duplex = NULL) {
 			if (!c->plan.next) {
 				del_conn(c);
-				free(c);
+				free_conn(c);
 				i--;
 			}
 		}
@@ -204,9 +254,12 @@ void *io_loop(void)
 {
 	void *ret;
 
+	io_loop_enter();
+
 	while (!io_loop_return) {
 		int i, r, timeout = INT_MAX;
 		struct timespec now;
+		bool some_timeouts = false;
 
 		if (timeouts.base) {
 			struct timespec first;
@@ -221,7 +274,9 @@ void *io_loop(void)
 				struct io_conn *conn = t->conn;
 				/* Clear, in case timer re-adds */
 				t->conn = NULL;
+				set_current(conn);
 				set_plan(conn, t->next(conn, t->next_arg));
+				some_timeouts = true;
 			}
 
 			/* Now figure out how long to wait for the next one. */
@@ -238,6 +293,10 @@ void *io_loop(void)
 			continue;
 		}
 
+		/* debug can recurse on io_loop; anything can change. */
+		if (doing_debug() && some_timeouts)
+			continue;
+
 		if (num_fds == 0)
 			break;
 
@@ -267,25 +326,37 @@ void *io_loop(void)
 					if (events & mask) {
 						io_ready(c->duplex);
 						events &= ~mask;
+						/* debug can recurse;
+						 * anything can change. */
+						if (doing_debug())
+							break;
 						if (!(events&(POLLIN|POLLOUT)))
 							continue;
 					}
 				}
 				io_ready(c);
+				/* debug can recurse; anything can change. */
+				if (doing_debug())
+					break;
 			} else if (events & POLLHUP) {
 				r--;
+				set_current(c);
 				set_plan(c, io_close(c, NULL));
-				if (c->duplex)
+				if (c->duplex) {
+					set_current(c->duplex);
 					set_plan(c->duplex,
 						 io_close(c->duplex, NULL));
 				}
 			}
 		}
+	}
 
 	while (num_closing)
 		finish_conns();
 
 	ret = io_loop_return;
 	io_loop_return = NULL;
+
+	io_loop_exit();
 	return ret;
 }

ccan/io/test/run-01-start-finish-DEBUG.c

+#define DEBUG
+#define PORT "64001"
+#define main real_main
+int real_main(void);
+#include "run-01-start-finish.c"
+#undef main
+static bool always_debug(struct io_conn *conn) { return true; }
+int main(void) { io_debug = always_debug; return real_main(); }

ccan/io/test/run-01-start-finish.c

@@ -6,6 +6,10 @@
 #include <sys/wait.h>
 #include <stdio.h>
 
+#ifndef PORT
+#define PORT "65001"
+#endif
+
 static void finish_ok(struct io_conn *conn, int *state)
 {
 	ok1(*state == 1);
@@ -62,7 +66,7 @@ int main(void)
 
 	/* This is how many tests you plan to run */
 	plan_tests(9);
-	fd = make_listen_fd("65001", &addrinfo);
+	fd = make_listen_fd(PORT, &addrinfo);
 	ok1(fd >= 0);
 	l = io_new_listener(fd, init_conn, &state);
 	ok1(l);

ccan/io/test/run-02-read-DEBUG.c

+#define DEBUG
+#define PORT "64002"
+#define main real_main
+int real_main(void);
+#include "run-02-read.c"
+#undef main
+static bool always_debug(struct io_conn *conn) { return true; }
+int main(void) { io_debug = always_debug; return real_main(); }

ccan/io/test/run-02-read.c

@@ -6,6 +6,10 @@
 #include <sys/wait.h>
 #include <stdio.h>
 
+#ifndef PORT
+#define PORT "65002"
+#endif
+
 struct data {
 	int state;
 	char buf[4];
@@ -70,7 +74,7 @@ int main(void)
 	/* This is how many tests you plan to run */
 	plan_tests(10);
 	d->state = 0;
-	fd = make_listen_fd("65002", &addrinfo);
+	fd = make_listen_fd(PORT, &addrinfo);
 	ok1(fd >= 0);
 	l = io_new_listener(fd, init_conn, d);
 	ok1(l);

ccan/io/test/run-03-readpartial-DEBUG.c

+#define DEBUG
+#define PORT "64003"
+#define main real_main
+int real_main(void);
+#include "run-03-readpartial.c"
+#undef main
+static bool always_debug(struct io_conn *conn) { return true; }
+int main(void) { io_debug = always_debug; return real_main(); }

ccan/io/test/run-03-readpartial.c

@@ -6,6 +6,10 @@
 #include <sys/wait.h>
 #include <stdio.h>
 
+#ifndef PORT
+#define PORT "65003"
+#endif
+
 struct data {
 	int state;
 	size_t bytes;
@@ -90,7 +94,7 @@ int main(void)
 	/* This is how many tests you plan to run */
 	plan_tests(22);
 	d->state = 0;
-	fd = make_listen_fd("65003", &addrinfo);
+	fd = make_listen_fd(PORT, &addrinfo);
 	ok1(fd >= 0);
 	l = io_new_listener(fd, init_conn, d);
 	ok1(l);

ccan/io/test/run-04-writepartial-DEBUG.c

+#define DEBUG
+#define PORT "64004"
+#define main real_main
+int real_main(void);
+#include "run-04-writepartial.c"
+#undef main
+static bool always_debug(struct io_conn *conn) { return true; }
+int main(void) { io_debug = always_debug; return real_main(); }

ccan/io/test/run-04-writepartial.c

@@ -6,6 +6,10 @@
 #include <sys/wait.h>
 #include <stdio.h>
 
+#ifndef PORT
+#define PORT "65004"
+#endif
+
 struct data {
 	int state;
 	size_t bytes;
@@ -91,7 +95,7 @@ int main(void)
 	d->bytes = 1024*1024;
 	d->buf = malloc(d->bytes);
 	memset(d->buf, 'a', d->bytes);
-	fd = make_listen_fd("65004", &addrinfo);
+	fd = make_listen_fd(PORT, &addrinfo);
 	ok1(fd >= 0);
 	l = io_new_listener(fd, init_conn, d);
 	ok1(l);

ccan/io/test/run-05-write-DEBUG.c

+#define DEBUG
+#define PORT "64005"
+#define main real_main
+int real_main(void);
+#include "run-05-write.c"
+#undef main
+static bool always_debug(struct io_conn *conn) { return true; }
+int main(void) { io_debug = always_debug; return real_main(); }

ccan/io/test/run-05-write.c

@@ -6,6 +6,10 @@
 #include <sys/wait.h>
 #include <stdio.h>
 
+#ifndef PORT
+#define PORT "65005"
+#endif
+
 struct data {
 	int state;
 	size_t bytes;
@@ -94,7 +98,7 @@ int main(void)
 	d->bytes = 1024*1024;
 	d->buf = malloc(d->bytes);
 	memset(d->buf, 'a', d->bytes);
-	fd = make_listen_fd("65005", &addrinfo);
+	fd = make_listen_fd(PORT, &addrinfo);
 	ok1(fd >= 0);
 	l = io_new_listener(fd, init_conn, d);
 	ok1(l);

ccan/io/test/run-06-idle-DEBUG.c

+#define DEBUG
+#define PORT "64006"
+#define main real_main
+int real_main(void);
+#include "run-06-idle.c"
+#undef main
+static bool always_debug(struct io_conn *conn) { return true; }
+int main(void) { io_debug = always_debug; return real_main(); }

ccan/io/test/run-06-idle.c

@@ -9,6 +9,10 @@
 #include <sys/stat.h>
 #include <fcntl.h>
 
+#ifndef PORT
+#define PORT "65006"
+#endif
+
 static struct io_conn *idler;
 
 struct data {
@@ -98,7 +102,7 @@ int main(void)
 	/* This is how many tests you plan to run */
 	plan_tests(14);
 	d->state = 0;
-	fd = make_listen_fd("65006", &addrinfo);
+	fd = make_listen_fd(PORT, &addrinfo);
 	ok1(fd >= 0);
 	l = io_new_listener(fd, init_conn, d);
 	ok1(l);

ccan/io/test/run-07-break-DEBUG.c

+#define DEBUG
+#define PORT "64007"
+#define main real_main
+int real_main(void);
+#include "run-07-break.c"
+#undef main
+static bool always_debug(struct io_conn *conn) { return true; }
+int main(void) { io_debug = always_debug; return real_main(); }

ccan/io/test/run-07-break.c

@@ -6,6 +6,10 @@
 #include <sys/wait.h>
 #include <stdio.h>
 
+#ifndef PORT
+#define PORT "65007"
+#endif
+
 struct data {
 	int state;
 	char buf[4];
@@ -78,7 +82,7 @@ int main(void)
 	/* This is how many tests you plan to run */
 	plan_tests(13);
 	d->state = 0;
-	fd = make_listen_fd("65007", &addrinfo);
+	fd = make_listen_fd(PORT, &addrinfo);
 	ok1(fd >= 0);
 	l = io_new_listener(fd, init_conn, d);
 	ok1(l);

ccan/io/test/run-08-hangup-on-idle-DEBUG.c

+#define DEBUG
+#define main real_main
+int real_main(void);
+#include "run-08-hangup-on-idle.c"
+#undef main
+static bool always_debug(struct io_conn *conn) { return true; }
+int main(void) { io_debug = always_debug; return real_main(); }

ccan/io/test/run-08-read-after-hangup-DEBUG.c

+#define DEBUG
+#define main real_main
+int real_main(void);
+#include "run-08-read-after-hangup.c"
+#undef main
+static bool always_debug(struct io_conn *conn) { return true; }
+int main(void) { io_debug = always_debug; return real_main(); }

ccan/io/test/run-10-many-DEBUG.c

+#define DEBUG
+#define PORT "64010"
+#define main real_main
+int real_main(void);
+#include "run-10-many.c"
+#undef main
+/* We stack overflow if we debug all of them! */
+static bool debug_one(struct io_conn *conn)
+{
+	return conn == buf[1].reader;
+}
+int main(void) { io_debug = debug_one; return real_main(); }

ccan/io/test/run-12-bidir-DEBUG.c

+#define DEBUG
+#define PORT "64012"
+#define main real_main
+int real_main(void);
+#include "run-12-bidir.c"
+#undef main
+static bool always_debug(struct io_conn *conn) { return true; }
+int main(void) { io_debug = always_debug; return real_main(); }

ccan/io/test/run-12-bidir.c

@@ -6,6 +6,10 @@
 #include <sys/wait.h>
 #include <stdio.h>
 
+#ifndef PORT
+#define PORT "65012"
+#endif
+
 struct data {
 	struct io_listener *l;
 	int state;
@@ -82,7 +86,7 @@ int main(void)
 	/* This is how many tests you plan to run */
 	plan_tests(10);
 	d->state = 0;
-	fd = make_listen_fd("65012", &addrinfo);
+	fd = make_listen_fd(PORT, &addrinfo);
 	ok1(fd >= 0);
 	d->l = io_new_listener(fd, init_conn, d);
 	ok1(d->l);

ccan/io/test/run-13-all-idle-DEBUG.c

+#define DEBUG
+#define PORT "64013"
+#define main real_main
+int real_main(void);
+#include "run-13-all-idle.c"
+#undef main
+static bool always_debug(struct io_conn *conn) { return true; }
+int main(void) { io_debug = always_debug; return real_main(); }

ccan/io/test/run-15-timeout-DEBUG.c

+#define DEBUG
+#define PORT "64015"
+#define main real_main
+int real_main(void);
+#include "run-15-timeout.c"
+#undef main
+static bool always_debug(struct io_conn *conn) { return true; }
+int main(void) { io_debug = always_debug; return real_main(); }

ccan/io/test/run-15-timeout.c

@@ -7,6 +7,10 @@
 #include <stdio.h>
 #include <unistd.h>
 
+#ifndef PORT
+#define PORT "65015"
+#endif
+
 struct data {
 	int state;
 	int timeout_usec;
@@ -93,7 +97,7 @@ int main(void)
 	d->state = 0;
 	d->timed_out = false;
 	d->timeout_usec = 100000;
-	fd = make_listen_fd("65002", &addrinfo);
+	fd = make_listen_fd(PORT, &addrinfo);
 	ok1(fd >= 0);
 	l = io_new_listener(fd, init_conn, d);
 	ok1(l);