json: Only start parsing once we're sure we have a message

PR #3957 improved performance considerably, however we still look over the
entire message for the message separator. If instead we just look in the
incrementally read data, we remove the quadratic behavior for large messages.

This is safe since we then loop over the messages which would drain any
message separator from the buffer before we attempt the next read.

lightningd/plugin.c

@@ -491,30 +491,41 @@ static struct io_plan *plugin_read_json(struct io_conn *conn,
 					struct plugin *plugin)
 {
 	bool success;
+	bool have_full;
+	int offset = plugin->used >= 1 ? 1 : 0;
 
 	log_io(plugin->log, LOG_IO_IN, NULL, "",
 	       plugin->buffer + plugin->used, plugin->len_read);
 
+	/* Peek into the buffer increment we just read (including a tiny
+	 * overlap if we're not at the start) to see if we have a '\n\n'
+	 * separator. If not we can skip the JSON attempt. */
+	have_full = memmem(plugin->buffer + plugin->used - offset,
+			   plugin->len_read + offset, "\n\n", 2);
+
 	plugin->used += plugin->len_read;
 	if (plugin->used == tal_count(plugin->buffer))
 		tal_resize(&plugin->buffer, plugin->used * 2);
 
 	/* Read and process all messages from the connection */
-	do {
-		bool destroyed;
-		const char *err;
-		err = plugin_read_json_one(plugin, &success, &destroyed);
-
-		/* If it's destroyed, conn is already freed! */
-		if (destroyed)
-			return io_close(NULL);
-
-		if (err) {
-			plugin_kill(plugin, err);
-			/* plugin_kill frees plugin */
-			return io_close(NULL);
-		}
-	} while (success);
+	if (have_full) {
+		do {
+			bool destroyed;
+			const char *err;
+			err =
+			    plugin_read_json_one(plugin, &success, &destroyed);
+
+			/* If it's destroyed, conn is already freed! */
+			if (destroyed)
+				return io_close(NULL);
+
+			if (err) {
+				plugin_kill(plugin, err);
+				/* plugin_kill frees plugin */
+				return io_close(NULL);
+			}
+		} while (success);
+	}
 
 	/* Now read more from the connection */
 	return io_read_partial(plugin->stdout_conn,