lightningd/subd: support multiple fds sent at once in normal messages.

Rather than returning SUBD_NEED_FD, callback returns how many fds it needs.

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

lightningd/gossip_control.c

@@ -104,8 +104,7 @@ static void peer_ready(struct subd *gossip, const u8 *msg, int fd)
 	peer_set_condition(peer, "Exchanging gossip");
 }
 
-static enum subd_msg_ret gossip_msg(struct subd *gossip,
-				    const u8 *msg, int fd)
+static size_t gossip_msg(struct subd *gossip, const u8 *msg, const int *fds)
 {
 	enum gossip_wire_type t = fromwire_peektype(msg);
 
@@ -128,18 +127,17 @@ static enum subd_msg_ret gossip_msg(struct subd *gossip,
 		peer_bad_message(gossip, msg);
 		break;
 	case WIRE_GOSSIPSTATUS_PEER_NONGOSSIP:
-		if (fd == -1)
-			return SUBD_NEED_FD;
-		peer_nongossip(gossip, msg, fd);
+		if (tal_count(fds) != 1)
+			return 1;
+		peer_nongossip(gossip, msg, fds[0]);
 		break;
 	case WIRE_GOSSIPSTATUS_PEER_READY:
-		if (fd == -1) {
-			return SUBD_NEED_FD;
-		}
-		peer_ready(gossip, msg, fd);
+		if (tal_count(fds) != 1)
+			return 1;
+		peer_ready(gossip, msg, fds[0]);
 		break;
 	}
-	return SUBD_COMPLETE;
+	return 0;
 }
 
 void gossip_init(struct lightningd *ld)

lightningd/hsm_control.c

@@ -38,7 +38,7 @@ static void hsm_finished(struct subd *hsm, int status)
 	errx(1, "HSM failed (signal %u), exiting.", WTERMSIG(status));
 }
 
-static enum subd_msg_ret hsm_msg(struct subd *hsm, const u8 *msg, int fd)
+static size_t hsm_msg(struct subd *hsm, const u8 *msg, const int *fds)
 {
 	enum hsm_wire_type t = fromwire_peektype(msg);
 	u8 *badmsg;
@@ -79,7 +79,7 @@ static enum subd_msg_ret hsm_msg(struct subd *hsm, const u8 *msg, int fd)
 	case WIRE_HSMCTL_SIGN_FUNDING_REPLY:
 		errx(1, "HSM gave invalid message %s", hsm_wire_type_name(t));
 	}
-	return SUBD_COMPLETE;
+	return 0;
 }
 
 void hsm_init(struct lightningd *ld, bool newdir)

lightningd/peer_control.c

@@ -604,9 +604,9 @@ static bool opening_got_hsm_funding_sig(struct subd *hsm, const u8 *resp,
 	return true;
 }
 
-static enum subd_msg_ret update_channel_status(struct subd *sd,
-					       const u8 *msg,
-					       int unused)
+static size_t update_channel_status(struct subd *sd,
+				    const u8 *msg,
+				    const int *unused)
 {
 	enum channel_wire_type t = fromwire_peektype(msg);
 
@@ -629,7 +629,7 @@ static enum subd_msg_ret update_channel_status(struct subd *sd,
 		break;
 	}
 
-	return SUBD_COMPLETE;
+	return 0;
 }
 
 /* opening is done, start lightningd_channel for peer. */

lightningd/subd.c

@@ -47,8 +47,7 @@ struct subd_req {
 	bool (*replycb)(struct subd *, const u8 *, const int *, void *);
 	void *replycb_data;
 
-	size_t num_fds_read;
-	int *fds_in;
+	size_t num_reply_fds;
 };
 
 static void free_subd_req(struct subd_req *sr)
@@ -66,8 +65,7 @@ static void add_req(struct subd *sd, int type, size_t num_fds_in,
 	sr->reply_type = type + SUBD_REPLY_OFFSET;
 	sr->replycb = replycb;
 	sr->replycb_data = replycb_data;
-	sr->fds_in = num_fds_in ? tal_arr(sr, int, num_fds_in) : NULL;
-	sr->num_fds_read = 0;
+	sr->num_reply_fds = num_fds_in;
 	assert(strends(sd->msgname(sr->reply_type), "_REPLY"));
 
 	/* Keep in FIFO order: we sent in order, so replies will be too. */
@@ -185,26 +183,46 @@ static struct io_plan *sd_msg_reply(struct io_conn *conn, struct subd *sd,
 {
 	int type = fromwire_peektype(sd->msg_in);
 	bool keep_open;
-	size_t i;
 
 	log_info(sd->log, "REPLY %s with %zu fds",
-		 sd->msgname(type), tal_count(sr->fds_in));
-
-	/* Don't trust subd to set it blocking. */
-	for (i = 0; i < tal_count(sr->fds_in); i++)
-		set_blocking(sr->fds_in[i], true);
+		 sd->msgname(type), tal_count(sd->fds_in));
 
 	/* If not stolen, we'll free this below. */
 	tal_steal(sr, sd->msg_in);
-	keep_open = sr->replycb(sd, sd->msg_in, sr->fds_in, sr->replycb_data);
+	keep_open = sr->replycb(sd, sd->msg_in, sd->fds_in, sr->replycb_data);
 	tal_free(sr);
 
 	if (!keep_open)
 		return io_close(conn);
 
+	/* Free any fd array. */
+	sd->fds_in = tal_free(sd->fds_in);
 	return io_read_wire(conn, sd, &sd->msg_in, sd_msg_read, sd);
 }
 
+static struct io_plan *read_fds(struct io_conn *conn, struct subd *sd)
+{
+	if (sd->num_fds_in_read == tal_count(sd->fds_in)) {
+		size_t i;
+
+		/* Don't trust subd to set it blocking. */
+		for (i = 0; i < tal_count(sd->fds_in); i++)
+			set_blocking(sd->fds_in[i], true);
+		return sd_msg_read(conn, sd);
+	}
+	return io_recv_fd(conn, &sd->fds_in[sd->num_fds_in_read++],
+			  read_fds, sd);
+}
+
+static struct io_plan *sd_collect_fds(struct io_conn *conn, struct subd *sd,
+				      size_t num_fds)
+{
+	assert(!sd->fds_in);
+	sd->fds_in = tal_arr(sd, int, num_fds);
+	sd->num_fds_in_read = 0;
+	return read_fds(conn, sd);
+}
+
 static struct io_plan *sd_msg_read(struct io_conn *conn, struct subd *sd)
 {
 	int type = fromwire_peektype(sd->msg_in);
@@ -221,11 +239,10 @@ static struct io_plan *sd_msg_read(struct io_conn *conn, struct subd *sd)
 	/* First, check for replies. */
 	sr = get_req(sd, type);
 	if (sr) {
-		/* If we need (another) fd, read it and call us again. */
-		if (sr->num_fds_read < tal_count(sr->fds_in)) {
-			return io_recv_fd(conn, &sr->fds_in[sr->num_fds_read++],
-					  sd_msg_read, sd);
-		}
+		if (sr->num_reply_fds && sd->fds_in == NULL)
+			return sd_collect_fds(conn, sd, sr->num_reply_fds);
+
+		assert(sr->num_reply_fds == tal_count(sd->fds_in));
 		return sd_msg_reply(conn, sd, sr);
 	}
 
@@ -244,30 +261,21 @@ static struct io_plan *sd_msg_read(struct io_conn *conn, struct subd *sd)
 			    sd->msgname(type), str_len, str);
 	else {
 		log_info(sd->log, "UPDATE %s", sd->msgname(type));
+
 		if (sd->msgcb) {
-			enum subd_msg_ret r;
-
-			/* If received from subd, set blocking. */
-			if (sd->fd_in != -1)
-				set_blocking(sd->fd_in, true);
-			r = sd->msgcb(sd, sd->msg_in, sd->fd_in);
-			switch (r) {
-			case SUBD_NEED_FD:
+			size_t i = sd->msgcb(sd, sd->msg_in, sd->fds_in);
+			if (i != 0) {
+				/* Don't ask for fds twice! */
+				assert(!sd->fds_in);
 				/* Don't free msg_in: we go around again. */
 				tal_steal(sd, sd->msg_in);
 				tal_free(tmpctx);
-				return io_recv_fd(conn, &sd->fd_in,
-						  sd_msg_read, sd);
-			case SUBD_COMPLETE:
-				break;
-			default:
-				fatal("Unknown msgcb return for %s:%s: %u",
-				      sd->name, sd->msgname(type), r);
+				return sd_collect_fds(conn, sd, i);
 			}
 		}
 	}
 	sd->msg_in = NULL;
-	sd->fd_in = -1;
+	sd->fds_in = tal_free(sd->fds_in);
 	tal_free(tmpctx);
 	return io_read_wire(conn, sd, &sd->msg_in, sd_msg_read, sd);
 }
@@ -321,8 +329,8 @@ struct subd *new_subd(const tal_t *ctx,
 				const char *name,
 				struct peer *peer,
 				const char *(*msgname)(int msgtype),
-				enum subd_msg_ret (*msgcb)
-				(struct subd *, const u8 *, int fd),
+				size_t (*msgcb)(struct subd *, const u8 *,
+						const int *fds),
 				void (*finished)(struct subd *, int),
 				...)
 {
@@ -347,7 +355,7 @@ struct subd *new_subd(const tal_t *ctx,
 	sd->finished = finished;
 	sd->msgname = msgname;
 	sd->msgcb = msgcb;
-	sd->fd_in = -1;
+	sd->fds_in = NULL;
 	msg_queue_init(&sd->outq, sd);
 	tal_add_destructor(sd, destroy_subd);
 	list_head_init(&sd->reqs);

lightningd/subd.h

@@ -9,11 +9,6 @@
 
 struct io_conn;
 
-enum subd_msg_ret {
-	SUBD_NEED_FD,
-	SUBD_COMPLETE
-};
-
 /* By convention, replies are requests + 100 */
 #define SUBD_REPLY_OFFSET 100
 
@@ -35,14 +30,16 @@ struct subd {
 	struct log *log;
 
 	/* Callback when non-reply message comes in. */
-	enum subd_msg_ret (*msgcb)(struct subd *, const u8 *, int);
+	size_t (*msgcb)(struct subd *, const u8 *, const int *);
 	const char *(*msgname)(int msgtype);
 	void (*finished)(struct subd *sd, int status);
 
 	/* Buffer for input. */
 	u8 *msg_in;
-	/* While we're reading an fd in. */
-	int fd_in;
+
+	/* While we're reading fds in. */
+	size_t num_fds_in_read;
+	int *fds_in;
 
 	/* Messages queue up here. */
 	struct msg_queue outq;
@@ -62,9 +59,8 @@ struct subd {
  * @finished: function to call when it's finished (with exit status).
  * @...: the fds to hand as fd 3, 4... terminated with -1.
  *
- * @msgcb is called with fd == -1 when a message is received; if it
- * returns SUBD_NEED_FD, we read an fd from the daemon and call it
- * again with that as the third arg.
+ * @msgcb gets called with @fds set to NULL: if it returns a positive number,
+ * that many @fds are received before calling again.
  *
  * If this succeeds subd owns @peer.
  */
@@ -73,8 +69,7 @@ struct subd *new_subd(const tal_t *ctx,
 		      const char *name,
 		      struct peer *peer,
 		      const char *(*msgname)(int msgtype),
-		      enum subd_msg_ret (*msgcb)
-		      (struct subd *, const u8 *, int fd),
+		      size_t (*msgcb)(struct subd *, const u8 *, const int *fds),
 		      void (*finished)(struct subd *, int), ...);
 
 /**