Make Json rpc more efficient

common/Makefile

@@ -62,6 +62,7 @@ COMMON_SRC_NOGEN :=				\
 	common/json_parse.c			\
 	common/json_parse_simple.c		\
 	common/json_stream.c			\
+	common/jsonrpc_io.c			\
 	common/key_derive.c			\
 	common/keyset.c				\
 	common/lease_rates.c			\

common/json_parse.c

@@ -676,3 +676,13 @@ json_tok_channel_id(const char *buffer, const jsmntok_t *tok,
 	return hex_decode(buffer + tok->start, tok->end - tok->start,
 			  cid, sizeof(*cid));
 }
+
+void json_dup_contents(const tal_t *ctx,
+		       const char *buffer,
+		       const jsmntok_t *tok,
+		       const char **new_buffer,
+		       const jsmntok_t **new_toks)
+{
+	*new_buffer = tal_dup_arr(ctx, char, buffer, tok->end, 0);
+	*new_toks = tal_dup_arr(ctx, jsmntok_t, tok, json_next(tok) - tok, 0);
+}

common/json_parse.h

@@ -137,6 +137,13 @@ const char *json_scan(const tal_t *ctx,
 		      const char *guide,
 		      ...);
 
+/* Duplicate the tok(s) and buffer required (don't assume they're tal objects!) */
+void json_dup_contents(const tal_t *ctx,
+		       const char *buffer,
+		       const jsmntok_t *tok,
+		       const char **new_buffer,
+		       const jsmntok_t **new_toks);
+
 /* eg. JSON_SCAN(json_to_bool, &boolvar) */
 #define JSON_SCAN(fmt, var)						\
 	json_scan,							\

common/jsonrpc_io.c

@@ -0,0 +1,163 @@
+#include "config.h"
+
+#include <ccan/io/io.h>
+#include <ccan/membuf/membuf.h>
+#include <ccan/tal/str/str.h>
+#include <common/jsonrpc_io.h>
+#include <common/utils.h>
+#include <errno.h>
+#include <unistd.h>
+
+#define READ_CHUNKSIZE 64
+
+struct jsonrpc_io {
+	MEMBUF(char) membuf;
+	jsmn_parser parser;
+	jsmntok_t *toks;
+
+	/* Amount of unparsed JSON from previous reads */
+	size_t bytes_unparsed;
+	/* Amount just read by io_read_partial */
+	size_t bytes_read;
+};
+
+struct jsonrpc_io *jsonrpc_io_new(const tal_t *ctx)
+{
+	struct jsonrpc_io *json_in;
+
+	json_in = tal(ctx, struct jsonrpc_io);
+	json_in->bytes_unparsed = 0;
+	json_in->bytes_read = 0;
+
+	membuf_init(&json_in->membuf,
+		    tal_arr(json_in, char, READ_CHUNKSIZE),
+		    READ_CHUNKSIZE, membuf_tal_resize);
+	json_in->toks = toks_alloc(json_in);
+	jsmn_init(&json_in->parser);
+
+	return json_in;
+}
+
+/* Empty new bytes read into our unparsed buffer */
+static void add_newly_read(struct jsonrpc_io *json_in)
+{
+	/* Now added it to our ubparsed buffer */
+	membuf_added(&json_in->membuf, json_in->bytes_read);
+	json_in->bytes_unparsed += json_in->bytes_read;
+	json_in->bytes_read = 0;
+}
+
+const char *jsonrpc_newly_read(struct jsonrpc_io *json_in,
+			       size_t *len)
+{
+	const char *ret;
+
+	ret = membuf_elems(&json_in->membuf) + json_in->bytes_unparsed;
+	*len = json_in->bytes_read;
+
+	add_newly_read(json_in);
+	return ret;
+}
+
+const char *jsonrpc_io_parse(const tal_t *ctx,
+			     struct jsonrpc_io *json_in,
+			     const jsmntok_t **toks,
+			     const char **buf)
+{
+	bool complete;
+
+	/* If we're read any more, add that */
+	add_newly_read(json_in);
+	*toks = NULL;
+	*buf = NULL;
+
+	/* Our JSON parser is pretty good at incremental parsing, but
+	 * `getrawblock` gives a giant 2MB token, which forces it to re-parse
+	 * every time until we have all of it. However, we can't complete a
+	 * JSON object without a '}', so we do a cheaper check here.
+	 */
+	if (!memchr(membuf_elems(&json_in->membuf), '}',
+		    membuf_num_elems(&json_in->membuf)))
+		return NULL;
+
+	if (!json_parse_input(&json_in->parser, &json_in->toks,
+			      membuf_elems(&json_in->membuf),
+			      membuf_num_elems(&json_in->membuf),
+			      &complete)) {
+		return tal_fmt(ctx, "Failed to parse RPC JSON response '%.*s'",
+			       (int)membuf_num_elems(&json_in->membuf),
+			       membuf_elems(&json_in->membuf));
+	}
+
+	if (!complete)
+		return NULL;
+
+	/* Must have jsonrpc to be valid! */
+	if (!json_get_member(membuf_elems(&json_in->membuf),
+			     json_in->toks,
+			     "jsonrpc")) {
+		return tal_fmt(ctx,
+			       "JSON-RPC message does not contain \"jsonrpc\" field: '%.*s'",
+			       (int)membuf_num_elems(&json_in->membuf),
+			       membuf_elems(&json_in->membuf));
+	}
+
+	*toks = json_in->toks;
+	*buf = membuf_elems(&json_in->membuf);
+	return NULL;
+}
+
+void jsonrpc_io_parse_done(struct jsonrpc_io *json_in)
+{
+	size_t bytes_parsed = json_in->toks[0].end;
+	json_in->bytes_unparsed -= bytes_parsed;
+	membuf_consume(&json_in->membuf, bytes_parsed);
+
+	jsmn_init(&json_in->parser);
+	toks_reset(json_in->toks);
+}
+
+struct io_plan *jsonrpc_io_read_(struct io_conn *conn,
+				 struct jsonrpc_io *json_in,
+				 struct io_plan *(*next)(struct io_conn *,
+							 void *),
+				 void *arg)
+{
+	/* Make sure there's more room */
+	membuf_prepare_space(&json_in->membuf, READ_CHUNKSIZE);
+
+	/* Try to read more. */
+	json_in->bytes_read = 0;
+	return io_read_partial(conn,
+			       membuf_elems(&json_in->membuf)
+			       + json_in->bytes_unparsed,
+			       membuf_num_elems(&json_in->membuf)
+			       - json_in->bytes_unparsed
+			       + membuf_num_space(&json_in->membuf),
+			       &json_in->bytes_read,
+			       next, arg);
+}
+
+bool jsonrpc_sync_read(struct jsonrpc_io *json_in, int infd)
+{
+	int r;
+
+	/* Make sure there's more room */
+	membuf_prepare_space(&json_in->membuf, READ_CHUNKSIZE);
+
+	/* Try to read more. */
+	r = read(infd,
+		 membuf_elems(&json_in->membuf)
+		 + json_in->bytes_unparsed,
+		 membuf_num_elems(&json_in->membuf)
+		 - json_in->bytes_unparsed
+		 + membuf_num_space(&json_in->membuf));
+	if (r < 0)
+		return false;
+	if (r == 0) {
+		errno = 0;
+		return false;
+	}
+	json_in->bytes_read = r;
+	return true;
+}

common/jsonrpc_io.h

@@ -0,0 +1,80 @@
+/* Low-level helper library for C plugins using ccan/io and jsonrpc socket. */
+#ifndef LIGHTNING_COMMON_JSONRPC_IO_H
+#define LIGHTNING_COMMON_JSONRPC_IO_H
+#include "config.h"
+#include <ccan/tal/tal.h>
+#include <ccan/typesafe_cb/typesafe_cb.h>
+#include <common/json_parse_simple.h>
+
+struct io_conn;
+struct plugin;
+
+/**
+ * jsonrpc_io_new: allocate a fresh jsonrpc_io
+ */
+struct jsonrpc_io *jsonrpc_io_new(const tal_t *ctx);
+
+
+/**
+ * jsonrpc_io_read: set io_plan for reading more into buffer.
+ * @conn: the io_conn to read.
+ * @json_in: the jsonrpc_io.
+ * @next: the callback once a read is done.
+ * @arg: the argument for @next (typesafe).
+ */
+struct io_plan *jsonrpc_io_read_(struct io_conn *conn,
+				 struct jsonrpc_io *json_in,
+				 struct io_plan *(*next)(struct io_conn *,
+							 void *),
+				 void *arg);
+#define jsonrpc_io_read(ctx, json_in, next, arg)			\
+	jsonrpc_io_read_((ctx), (json_in),				\
+			 typesafe_cb_preargs(struct io_plan *, void *,	\
+					     (next), (arg),		\
+					     struct io_conn *),		\
+			 (arg))
+
+/**
+ * jsonrpc_newly_read: how much did we read into the buffer?
+ *
+ * Returns the buffer and sets *len to the bytes just read.  After
+ * that it will return *len == 0.
+ */
+const char *jsonrpc_newly_read(struct jsonrpc_io *json_in,
+			       size_t *len);
+
+/**
+ * jsonrpc_sync_read: read from fd into buffer.
+ * @json_in: buffer to read into.
+ * @infd: file descriptort to read.
+ *
+ * Returns false on error or EOF; for EOF errno will be 0.
+ */
+bool jsonrpc_sync_read(struct jsonrpc_io *json_in, int infd);
+
+/**
+ * jsonrpc_io_parse: try to parse more of the buffer.
+ * @ctx: context to allocate error message off.
+ * @json_in: json_in after jsonrpc_io_read.
+ * @toks: returned non-NULL if there's a whole valid json object.
+ * @buf: returned non-NULL as above.
+ *
+ * On error, a message is returned.  On incomplete, *@toks and *@buf
+ * are NULL.  Usually you call this, the use the result and call
+ * jsonrpc_io_parse_done(), then call it again.
+ */
+const char *jsonrpc_io_parse(const tal_t *ctx,
+			     struct jsonrpc_io *json_in,
+			     const jsmntok_t **toks,
+			     const char **buf);
+
+/**
+ * jsonrpc_io_parse_done: call aftr using toks from jsonrpc_io_parse.
+ * @json_in: json_in after jsonrpc_io_parse.
+ *
+ * You must call this if jsonrpc_io_parse() sets *toks non-NULL
+ * (i.e. complete, and no error).
+ */
+void jsonrpc_io_parse_done(struct jsonrpc_io *json_in);
+
+#endif /* LIGHTNING_COMMON_JSONRPC_IO_H */

common/msg_queue.c

@@ -36,21 +36,11 @@ static void destroy_msg_queue(struct msg_queue *q)
 	}
 }
 
-/* Realloc helper for tal membufs */
-static void *membuf_tal_realloc(struct membuf *mb, void *rawelems,
-				size_t newsize)
-{
-	char *p = rawelems;
-
-	tal_resize(&p, newsize);
-	return p;
-}
-
 struct msg_queue *msg_queue_new(const tal_t *ctx, bool fd_passing)
 {
 	struct msg_queue *q = tal(ctx, struct msg_queue);
 	q->fd_passing = fd_passing;
-	membuf_init(&q->mb, tal_arr(q, const u8 *, 0), 0, membuf_tal_realloc);
+	membuf_init(&q->mb, tal_arr(q, const u8 *, 0), 0, membuf_tal_resize);
 
 	if (q->fd_passing)
 		tal_add_destructor(q, destroy_msg_queue);

common/test/Makefile

@@ -130,4 +130,6 @@ common/test/run-shutdown_scriptpubkey: wire/towire.o wire/fromwire.o
 
 common/test/run-wireaddr: wire/towire.o wire/fromwire.o
 
+common/test/run-jsonrpc_io: common/json_parse_simple.o
+
 check-units: $(COMMON_TEST_PROGRAMS:%=unittest/%)