Big nodes, big loads and lots of (overdue) optimizations

ccan/README

@@ -1,3 +1,3 @@
 CCAN imported from http://ccodearchive.net.
 
-CCAN version: init-2605-gc47bf0d9
+CCAN version: init-2606-g5f219f03

ccan/ccan/io/poll.c

@@ -373,6 +373,8 @@ static void restore_pollfds(void)
 void *io_loop(struct timers *timers, struct timer **expired)
 {
 	void *ret;
+	/* This ensures we don't always service lower fds first */
+	static int fairness_counter;
 
 	/* if timers is NULL, expired must be.  If not, not. */
 	assert(!timers == !expired);
@@ -384,17 +386,12 @@ void *io_loop(struct timers *timers, struct timer **expired)
 	while (!io_loop_return) {
 		int i, r, ms_timeout = -1;
 
-		if (handle_always()) {
-			/* Could have started/finished more. */
-			continue;
-		}
-
 		/* Everything closed? */
 		if (num_fds == 0)
 			break;
 
 		/* You can't tell them all to go to sleep! */
-		assert(num_waiting);
+		assert(num_waiting || num_always);
 
 		if (timers) {
 			struct timemono now, first;
@@ -417,6 +414,10 @@ void *io_loop(struct timers *timers, struct timer **expired)
 			}
 		}
 
+		/* Don't wait if we have always requests pending! */
+		if (num_always != 0)
+			ms_timeout = 0;
+
 		/* We do this temporarily, assuming exclusive is unusual */
 		exclude_pollfds();
 		r = pollfn(pollfds, num_fds, ms_timeout);
@@ -430,15 +431,29 @@ void *io_loop(struct timers *timers, struct timer **expired)
 			break;
 		}
 
-		for (i = 0; i < num_fds && !io_loop_return; i++) {
-			struct io_conn *c = (void *)fds[i];
-			int events = pollfds[i].revents;
+		fairness_counter++;
+		for (size_t rotation = 0; rotation < num_fds && !io_loop_return; rotation++) {
+			struct io_conn *c;
+			int events;
+
+			i = (rotation + fairness_counter) % num_fds;
+			c = (void *)fds[i];
 
 			/* Clear so we don't get confused if exclusive next time */
+			events = pollfds[i].revents;
 			pollfds[i].revents = 0;
 
-			if (r == 0)
+			/* Timeout? */
+			if (r == 0) {
+				handle_always();
 				break;
+			}
+
+			/* We interleave always before the first fd */
+			if (i == 0 && handle_always()) {
+				/* Could have started/finished more. */
+				break;
+			}
 
 			if (fds[i]->listener) {
 				struct io_listener *l = (void *)fds[i];

common/json_parse_simple.c

@@ -486,24 +486,24 @@ bool json_parse_input(jsmn_parser *parser,
 
 again:
 	ret = jsmn_parse(parser, input, len, *toks, tal_count(*toks) - 1);
-
-	switch (ret) {
-	case JSMN_ERROR_INVAL:
+	if (ret == JSMN_ERROR_INVAL)
 		return false;
-	case JSMN_ERROR_NOMEM:
-		tal_resize(toks, tal_count(*toks) * 2);
-		goto again;
-	}
 
 	/* Check whether we read at least one full root element, i.e., root
 	 * element has its end set. */
 	if ((*toks)[0].type == JSMN_UNDEFINED || (*toks)[0].end == -1) {
+		/* If it ran out of tokens, provide more. */
+		if (ret == JSMN_ERROR_NOMEM) {
+			tal_resize(toks, tal_count(*toks) * 2);
+			goto again;
+		}
+		/* Otherwise, must be incomplete */
 		*complete = false;
 		return true;
 	}
 
 	/* If we read a partial element at the end of the stream we'll get a
-	 * ret=JSMN_ERROR_PART, but due to the previous check we know we read at
+	 * errro, but due to the previous check we know we read at
 	 * least one full element, so count tokens that are part of this root
 	 * element. */
 	ret = json_next(*toks) - *toks;

common/json_stream.h

@@ -9,6 +9,7 @@
 # include <external/jsmn/jsmn.h>
 
 #include <bitcoin/short_channel_id.h>
+#include <ccan/list/list.h>
 #include <ccan/time/time.h>
 #include <common/amount.h>
 #include <common/jsonrpc_errors.h>
@@ -36,6 +37,8 @@ struct wireaddr;
 struct wireaddr_internal;
 
 struct json_stream {
+	struct list_node list;
+
 	struct json_out *jout;
 
 	/* Who is writing to this buffer now; NULL if nobody is. */

common/jsonrpc_io.c

@@ -8,7 +8,7 @@
 #include <errno.h>
 #include <unistd.h>
 
-#define READ_CHUNKSIZE 64
+#define READ_CHUNKSIZE (1024*1024)
 
 struct jsonrpc_io {
 	MEMBUF(char) membuf;
@@ -22,13 +22,14 @@ struct jsonrpc_io {
 struct jsonrpc_io *jsonrpc_io_new(const tal_t *ctx)
 {
 	struct jsonrpc_io *json_in;
+	const size_t bufsize = READ_CHUNKSIZE * 2;
 
 	json_in = tal(ctx, struct jsonrpc_io);
 	json_in->bytes_read = 0;
 
 	membuf_init(&json_in->membuf,
-		    tal_arr(json_in, char, READ_CHUNKSIZE),
-		    READ_CHUNKSIZE, membuf_tal_resize);
+		    tal_arr(json_in, char, bufsize),
+		    bufsize, membuf_tal_resize);
 	json_in->toks = toks_alloc(json_in);
 	jsmn_init(&json_in->parser);