rustyrussell · Aug 5, 2014 · Aug 5, 2014 · Aug 5, 2014 · Aug 5, 2014 · Aug 5, 2014
diff --git a/blockfile.c b/blockfile.c
@@ -60,11 +60,50 @@ static bool load_tx_in_block(struct state *state,
 	return e == PROTOCOL_ECODE_NONE;
 }
 
+struct load_state {
+	struct state *state;
+	off_t processed;
+	struct protocol_net_hdr *pkt;
+};
+
+static struct io_plan *load_packet(struct io_conn *conn, struct load_state *ls)
+{
+	switch (le32_to_cpu(ls->pkt->type)) {
+	case PROTOCOL_PKT_BLOCK:
+		if (!load_block(ls->state, ls->pkt)) {
+			log_unusual(ls->state->log,
+				    "blockfile partial block");
+			return io_close(conn);
+		}
+		break;
+	case PROTOCOL_PKT_TX_IN_BLOCK:
+		if (!load_tx_in_block(ls->state, (const void *)ls->pkt)) {
+			log_unusual(ls->state->log,
+				    "blockfile partial transaction");
+			return io_close(conn);
+		}
+		break;
+	default:
+		log_unusual(ls->state->log, "blockfile unknown type %u",
+			    le32_to_cpu(ls->pkt->type));
+		return io_close(conn);
+	}
+
+	ls->processed = lseek(io_conn_fd(conn), 0, SEEK_CUR);
+	return io_read_packet(conn, &ls->pkt, load_packet, ls);
+}
+
+static struct io_plan *setup_load_conn(struct io_conn *conn,
+				       struct load_state *ls)
+{
+	return io_read_packet(conn, &ls->pkt, load_packet, ls);
+}
+
 void load_blocks(struct state *state)
 {
-	off_t off = 0;
-	struct stat st;
 	int fd;
+	struct load_state ls;
+	off_t len;
 
 	fd = open("blockfile", O_RDWR|O_CREAT, 0600);
 	if (fd < 0)
@@ -73,55 +112,26 @@ void load_blocks(struct state *state)
 	/* Prevent us saving blocks as we're reading them. */
 	state->blockfd = -1;
 
-	for (;;) {
-		struct protocol_net_hdr *pkt;
-		struct io_plan plan;
-		int ret;
-
-		plan = io_read_packet_(&pkt, (void *)1, NULL);
-		while ((ret = plan.io(fd, &plan)) != 1) {
-			if (ret == -1) {
-				/* Did we do a partial read? */
-				if (lseek(fd, 0, SEEK_CUR) != off) {
-					log_unusual(state->log,
-						    "blockfile partial read");
-					goto truncate;
-				}
-				goto out;
-			}
-		}
+	ls.state = state;
+	ls.processed = 0;
+	io_new_conn(state, fd, setup_load_conn, &ls);
 
-		switch (le32_to_cpu(pkt->type)) {
-		case PROTOCOL_PKT_BLOCK:
-			if (!load_block(state, pkt)) {
-				log_unusual(state->log,
-					    "blockfile partial block");
-				goto truncate;
-			}
-			break;
-		case PROTOCOL_PKT_TX_IN_BLOCK:
-			if (!load_tx_in_block(state, (const void *)pkt)) {
-				log_unusual(state->log,
-					    "blockfile partial transaction");
-				goto truncate;
-			}
-			break;
-		default:
-			log_unusual(state->log, "blockfile unknown type %u",
-				    le32_to_cpu(pkt->type));
-			goto truncate;
-		}
-		off += le32_to_cpu(pkt->len);
-	}
+	/* When it reads 0 bytes, it will close, so dup fd. */
+	fd = dup(fd);
 
-truncate:
-	fstat(fd, &st);
-	log_unusual(state->log, "Truncating blockfile from %llu to %llu",
-		    (long long)st.st_size, (long long)off);
-	ftruncate(fd, off);
-	lseek(fd, SEEK_SET, off);
+	/* Process them all. */
+	io_loop(NULL, NULL);
+
+	len = lseek(fd, 0, SEEK_END);
+	/* If we didn't process the entire file, truncate it. */
+	if (len != ls.processed) {
+		log_unusual(state->log,
+			    "Truncating blockfile from %llu to %llu",
+			    (long long)len, (long long)ls.processed);
+		ftruncate(fd, ls.processed);
+		lseek(fd, SEEK_SET, ls.processed);
+	}
 
-out:
 	/* Now we can save more. */
 	state->blockfd = fd;
 

diff --git a/ccan/README b/ccan/README
@@ -1,3 +1,3 @@
 CCAN imported from http://ccodearchive.net.
 
-CCAN version: init-1886-gc2b2d22
+CCAN version: init-1898-g0fbc790
diff --git a/ccan/ccan/io/_info b/ccan/ccan/io/_info
@@ -10,12 +10,6 @@
  * (eg. read, write).  It is also possible to write custom I/O
  * plans.
  *
- * When compiled with DEBUG, control flow is changed so that rather
- * than returning to the main io_loop(), plans are executed sequentially
- * providing a backtrace showing what has occurred on that connection.
- * Which connection(s) do this depends on the user-specified io_debug
- * function.
- *
  * Example:
  * // Given tr A-Z a-z outputs tr a-z a-z
  * #include <ccan/io/io.h>
@@ -41,7 +35,7 @@
  * 	io_wake(b);
  * }
  *
- * static struct io_plan read_in(struct io_conn *c, struct buffer *b)
+ * static struct io_plan *read_in(struct io_conn *c, struct buffer *b)
  * {
  * 	// Add what we just read.
  * 	b->end += b->rlen;
@@ -55,35 +49,33 @@
  * 	if (b->start == b->end)
  * 		b->start = b->end = 0;
  *
- * 	// Read in some of the rest.
- * 	b->rlen = sizeof(b->buf) - b->end;
- *
  * 	// No room?  Wait for writer
- * 	if (b->rlen == 0)
- * 		return io_wait(b, read_in, b);
+ * 	if (b->end == sizeof(b->buf))
+ * 		return io_wait(c, b, read_in, b);
  *
- * 	return io_read_partial(b->buf + b->end, &b->rlen, read_in, b);
+ * 	return io_read_partial(c, b->buf + b->end, sizeof(b->buf) - b->end,
+ *			       &b->rlen, read_in, b);
  * }
  *
- * static struct io_plan write_out(struct io_conn *c, struct buffer *b)
+ * static struct io_plan *write_out(struct io_conn *c, struct buffer *b)
  * {
  * 	// Remove what we just wrote.
  * 	b->start += b->wlen;
  * 	assert(b->start <= sizeof(b->buf));
  *
- * 	// If we wrote somthing, wake writer.
+ * 	// If we wrote something, wake writer.
  * 	if (b->wlen != 0)
  * 		io_wake(b);
  *
- * 	b->wlen = b->end - b->start;
  * 	// Nothing to write?  Wait for reader.
- * 	if (b->wlen == 0) {
+ * 	if (b->end == b->start) {
  * 		if (b->finished)
- * 			return io_close();
- * 		return io_wait(b, write_out, b);
+ * 			return io_close(c);
+ * 		return io_wait(c, b, write_out, b);
  * 	}
  *
- * 	return io_write_partial(b->buf + b->start, &b->wlen, write_out, b);
+ * 	return io_write_partial(c, b->buf + b->start, b->end - b->start,
+ *				&b->wlen, write_out, b);
  * }
  *
  * // Feed a program our stdin, gather its stdout, print that at end.
@@ -117,16 +109,16 @@
  *
  * 	// Read from stdin, write to child.
  * 	memset(&to, 0, sizeof(to));
- * 	reader = io_new_conn(STDIN_FILENO, read_in(NULL, &to));
+ * 	reader = io_new_conn(NULL, STDIN_FILENO, read_in, &to);
  * 	io_set_finish(reader, finish, &to);
- * 	io_new_conn(tochild[1], write_out(NULL, &to));
+ * 	io_new_conn(NULL, tochild[1], write_out, &to);
  *
  * 	// Read from child, write to stdout.
- * 	reader = io_new_conn(fromchild[0], read_in(NULL, &from));
+ * 	reader = io_new_conn(NULL, fromchild[0], read_in, &from);
  * 	io_set_finish(reader, finish, &from);
- * 	io_new_conn(STDOUT_FILENO, write_out(NULL, &from));
+ * 	io_new_conn(NULL, STDOUT_FILENO, write_out, &from);
  *
- * 	io_loop();
+ * 	io_loop(NULL, NULL);
  * 	wait(&status);
  *
  * 	return WIFEXITED(status) ? WEXITSTATUS(status) : 2;
@@ -141,9 +133,12 @@ int main(int argc, char *argv[])
 		return 1;
 
 	if (strcmp(argv[1], "depends") == 0) {
-		printf("ccan/typesafe_cb\n");
+		printf("ccan/container_of\n");
+		printf("ccan/list\n");
+		printf("ccan/tal\n");
 		printf("ccan/time\n");
 		printf("ccan/timer\n");
+		printf("ccan/typesafe_cb\n");
 		return 0;
 	}
 

diff --git a/ccan/ccan/io/backend.h b/ccan/ccan/io/backend.h
@@ -2,18 +2,8 @@
 #ifndef CCAN_IO_BACKEND_H
 #define CCAN_IO_BACKEND_H
 #include <stdbool.h>
-#include <ccan/timer/timer.h>
 #include <poll.h>
-
-/* A setting for actions to always run (eg. zero-length reads). */
-#define POLLALWAYS (((POLLIN|POLLOUT) + 1) & ~((POLLIN|POLLOUT)))
-
-struct io_alloc {
-	void *(*alloc)(size_t size);
-	void *(*realloc)(void *ptr, size_t size);
-	void (*free)(void *ptr);
-};
-extern struct io_alloc io_alloc;
+#include "io_plan.h"
 
 struct fd {
 	int fd;
@@ -25,77 +15,79 @@ struct fd {
 struct io_listener {
 	struct fd fd;
 
+	const tal_t *ctx;
+
 	/* These are for connections we create. */
-	void (*init)(int fd, void *arg);
+	struct io_plan *(*init)(struct io_conn *conn, void *arg);
 	void *arg;
 };
 
-struct io_timeout {
-	struct timer timer;
-	struct io_conn *conn;
+enum io_plan_status {
+	/* As before calling next function. */
+	IO_UNSET,
+	/* Normal. */
+	IO_POLLING,
+	/* Waiting for io_wake */
+	IO_WAITING,
+	/* Always do this. */
+	IO_ALWAYS,
+	/* Closing (both plans will be the same). */
+	IO_CLOSING
+};
 
-	struct io_plan (*next)(struct io_conn *, void *arg);
+/**
+ * struct io_plan - one half of I/O to do
+ * @status: the status of this plan.
+ * @io: function to call when fd becomes read/writable, returns 0 to be
+ *      called again, 1 if it's finished, and -1 on error (fd will be closed)
+ * @next: the next function which is called if io returns 1.
+ * @next_arg: the argument to @next
+ * @u1, @u2: scratch space for @io.
+ */
+struct io_plan {
+	enum io_plan_status status;
+
+	int (*io)(int fd, struct io_plan_arg *arg);
+
+	struct io_plan *(*next)(struct io_conn *, void *next_arg);
 	void *next_arg;
+
+	struct io_plan_arg arg;
 };
 
 /* One connection per client. */
 struct io_conn {
 	struct fd fd;
+	bool debug;
+	/* For duplex to save. */
+	bool debug_saved;
+
+	/* always or closing list. */
+	struct io_conn *list;
 
 	void (*finish)(struct io_conn *, void *arg);
 	void *finish_arg;
 
-	struct io_conn *duplex;
-	struct io_timeout *timeout;
-
-	struct io_plan plan;
+	struct io_plan plan[2];
 };
 
-static inline bool timeout_active(const struct io_conn *conn)
-{
-	return conn->timeout && conn->timeout->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_on(struct io_conn *conn)
-{
-	return io_debug_conn && io_debug_conn(conn);
-}
-static inline bool doing_debug(void)
-{
-	return io_debug_conn;
-}
-#else
-static inline void set_current(struct io_conn *conn)
-{
-}
-static inline bool doing_debug_on(struct io_conn *conn)
-{
-	return false;
-}
-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);
 void del_listener(struct io_listener *l);
-void backend_plan_changed(struct io_conn *conn);
-void backend_wait_changed(const void *wait);
-void backend_add_timeout(struct io_conn *conn, struct timerel duration);
-void backend_del_timeout(struct io_conn *conn);
+void backend_new_closing(struct io_conn *conn);
+void backend_new_always(struct io_conn *conn);
+void backend_new_plan(struct io_conn *conn);
+void remove_from_always(struct io_conn *conn);
+void backend_plan_done(struct io_conn *conn);
+
+void backend_wake(const void *wait);
 void backend_del_conn(struct io_conn *conn);
 
-void io_ready(struct io_conn *conn);
+void io_ready(struct io_conn *conn, int pollflags);
+void io_do_always(struct io_conn *conn);
+void io_do_wakeup(struct io_conn *conn, struct io_plan *plan);
 void *do_io_loop(struct io_conn **ready);
 #endif /* CCAN_IO_BACKEND_H */