lightningd: use jsonrpc_io for reading JSON commands.

This is more efficient if we have lots of incoming commands, too.

Signed-off-by: Rusty Russell <rusty@rustcorp.com.au>

Author: rustyrussell

lightningd/jsonrpc.c

@@ -25,6 +25,7 @@
 #include <common/codex32.h>
 #include <common/json_command.h>
 #include <common/json_filter.h>
+#include <common/jsonrpc_io.h>
 #include <common/memleak.h>
 #include <common/timeout.h>
 #include <common/trace.h>
@@ -78,18 +79,8 @@ struct json_connection {
 	/* Logging for this json connection. */
 	struct logger *log;
 
-	/* The buffer (required to interpret tokens). */
-	char *buffer;
-
-	/* Internal state: */
-	/* How much is already filled. */
-	size_t used;
-	/* How much has just been filled. */
-	size_t len_read;
-
-	/* JSON parsing state. */
-	jsmn_parser input_parser;
-	jsmntok_t *input_toks;
+	/* The buffer and state reading in the JSON commands */
+	struct jsonrpc_io *json_in;
 
 	/* Local deprecated support? */
 	bool deprecated_ok;
@@ -1210,93 +1201,61 @@ static struct io_plan *stream_out_complete(struct io_conn *conn,
 static struct io_plan *read_json(struct io_conn *conn,
 				 struct json_connection *jcon)
 {
-	bool complete;
 	bool in_transaction = false;
 	struct timemono start_time = time_mono();
+	size_t len_read;
+	const jsmntok_t *toks;
+	const char *buffer, *error;
 
-	if (jcon->len_read)
-		log_io(jcon->log, LOG_IO_IN, NULL, "",
-		       jcon->buffer + jcon->used, jcon->len_read);
-
-	/* Resize larger if we're full. */
-	jcon->used += jcon->len_read;
-	if (jcon->used == tal_count(jcon->buffer))
-		tal_resize(&jcon->buffer, jcon->used * 2);
+	buffer = jsonrpc_newly_read(jcon->json_in, &len_read);
+	if (len_read)
+		log_io(jcon->log, LOG_IO_IN, NULL, "", buffer, len_read);
 
 	/* We wait for pending output to be consumed, to avoid DoS */
 	if (tal_count(jcon->js_arr) != 0) {
-		jcon->len_read = 0;
 		return io_wait(conn, conn, read_json, jcon);
 	}
 
 again:
-	if (!json_parse_input(&jcon->input_parser, &jcon->input_toks,
-			      jcon->buffer, jcon->used,
-			      &complete)) {
-		json_command_malformed(
-		    jcon, "null",
-		    tal_fmt(tmpctx, "Invalid token in json input: '%s'",
-			    tal_hexstr(tmpctx, jcon->buffer, jcon->used)));
+	error = jsonrpc_io_parse(tmpctx, jcon->json_in, &toks, &buffer);
+	if (error) {
+		json_command_malformed(jcon, "null", error);
 		if (in_transaction)
 			db_commit_transaction(jcon->ld->wallet->db);
 		return io_halfclose(conn);
 	}
 
-	if (!complete)
-		goto read_more;
-
-	/* Empty buffer? (eg. just whitespace). */
-	if (tal_count(jcon->input_toks) == 1) {
-		jcon->used = 0;
-
-		/* Reset parser. */
-		jsmn_init(&jcon->input_parser);
-		toks_reset(jcon->input_toks);
+	if (!toks)
 		goto read_more;
-	}
 
 	if (!in_transaction) {
 		db_begin_transaction(jcon->ld->wallet->db);
 		in_transaction = true;
 	}
-	parse_request(jcon, jcon->buffer, jcon->input_toks);
-
-	/* Remove first {}. */
-	memmove(jcon->buffer, jcon->buffer + jcon->input_toks[0].end,
-		tal_count(jcon->buffer) - jcon->input_toks[0].end);
-	jcon->used -= jcon->input_toks[0].end;
-
-	/* Reset parser. */
-	jsmn_init(&jcon->input_parser);
-	toks_reset(jcon->input_toks);
+	parse_request(jcon, buffer, toks);
+	jsonrpc_io_parse_done(jcon->json_in);
 
-	/* Do we have more already read? */
-	if (jcon->used) {
-		if (!jcon->db_batching) {
+	if (!jcon->db_batching) {
+		db_commit_transaction(jcon->ld->wallet->db);
+		in_transaction = false;
+	} else {
+		/* FIXME: io_always() should interleave with
+		 * real IO, and then we should rotate order we
+		 * service fds in, to avoid starvation. */
+		if (time_greater(timemono_between(time_mono(),
+						  start_time),
+				 time_from_msec(250))) {
 			db_commit_transaction(jcon->ld->wallet->db);
-			in_transaction = false;
-		} else {
-			/* FIXME: io_always() should interleave with
-			 * real IO, and then we should rotate order we
-			 * service fds in, to avoid starvation. */
-			if (time_greater(timemono_between(time_mono(),
-							  start_time),
-					 time_from_msec(250))) {
-				db_commit_transaction(jcon->ld->wallet->db);
-				/* Call us back, as if we read nothing new */
-				jcon->len_read = 0;
-				return io_always(conn, read_json, jcon);
-			}
+			/* Call us back, as if we read nothing new */
+			return io_always(conn, read_json, jcon);
 		}
-		goto again;
 	}
+	goto again;
 
 read_more:
 	if (in_transaction)
 		db_commit_transaction(jcon->ld->wallet->db);
-	return io_read_partial(conn, jcon->buffer + jcon->used,
-			       tal_count(jcon->buffer) - jcon->used,
-			       &jcon->len_read, read_json, jcon);
+	return jsonrpc_io_read(conn, jcon->json_in, read_json, jcon);
 }
 
 static struct io_plan *jcon_connected(struct io_conn *conn,
@@ -1308,12 +1267,8 @@ static struct io_plan *jcon_connected(struct io_conn *conn,
 	jcon = notleak(tal(conn, struct json_connection));
 	jcon->conn = conn;
 	jcon->ld = ld;
-	jcon->used = 0;
-	jcon->buffer = tal_arr(jcon, char, 64);
 	jcon->js_arr = tal_arr(jcon, struct json_stream *, 0);
-	jcon->len_read = 0;
-	jsmn_init(&jcon->input_parser);
-	jcon->input_toks = toks_alloc(jcon);
+	jcon->json_in = jsonrpc_io_new(jcon);
 	jcon->notifications_enabled = false;
 	jcon->db_batching = false;
 	jcon->deprecated_ok = ld->deprecated_ok;

tests/test_misc.py

@@ -951,11 +951,14 @@ def test_malformed_rpc(node_factory):
     obj, _ = l1.rpc._readobj(sock, b'')
     assert obj['error']['code'] == -32600
 
-    # Complete crap
+    # Complete crap (this makes it hang up!)
     sock.sendall(b'[]')
     obj, _ = l1.rpc._readobj(sock, b'')
     assert obj['error']['code'] == -32600
 
+    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+    sock.connect(l1.rpc.socket_path)
+
     # Bad ID
     sock.sendall(b'{"id":{}, "jsonrpc":"2.0","method":"getinfo","params":[]}')
     obj, _ = l1.rpc._readobj(sock, b'')