daemon_conn: make it a tal object, typesafe callbacks.

It means an extra allocation at startup, but it means we can hide the definition,
and use standard patterns (new_daemon_conn and typesafe callbacks).

Signed-off-by: Rusty Russell <[email protected]>

Author: rustyrussell

common/daemon_conn.c

@@ -5,12 +5,40 @@
 #include <wire/wire_io.h>
 #include <wire/wire_sync.h>
 
+struct daemon_conn {
+	/* Last message we received */
+	u8 *msg_in;
+
+	/* Queue of outgoing messages */
+	struct msg_queue out;
+
+	/* Underlying connection: we're freed if it closes, and vice versa */
+	struct io_conn *conn;
+
+	/* Callback for incoming messages */
+	struct io_plan *(*recv)(struct io_conn *conn, const u8 *, void *);
+
+	/* Called whenever we've cleared the msg_out queue. If it returns
+	 * true, it has added packets to msg_out queue. */
+	bool (*outq_empty)(void *);
+
+	/* Arg for both callbacks. */
+	void *arg;
+};
+
+static struct io_plan *handle_read(struct io_conn *conn,
+				      struct daemon_conn *dc)
+{
+	return dc->recv(conn, dc->msg_in, dc->arg);
+}
+
 struct io_plan *daemon_conn_read_next(struct io_conn *conn,
 				      struct daemon_conn *dc)
 {
-	dc->msg_in = tal_free(dc->msg_in);
-	return io_read_wire(conn, dc->ctx, &dc->msg_in, dc->daemon_conn_recv,
-			    dc);
+	/* FIXME: We could use disposable parent instead, and recv() could
+	 * tal_steal() it?  If they did that now, we'd free it here. */
+	tal_free(dc->msg_in);
+	return io_read_wire(conn, dc, &dc->msg_in, handle_read, dc);
 }
 
 static struct io_plan *daemon_conn_write_next(struct io_conn *conn,
@@ -29,8 +57,8 @@ static struct io_plan *daemon_conn_write_next(struct io_conn *conn,
 		}
 		return io_write_wire(conn, take(msg), daemon_conn_write_next,
 				     dc);
-	} else if (dc->msg_queue_cleared_cb) {
-		if (dc->msg_queue_cleared_cb(conn, dc))
+	} else if (dc->outq_empty) {
+		if (dc->outq_empty(dc->arg))
 			goto again;
 	}
 	return msg_queue_wait(conn, &dc->out, daemon_conn_write_next, dc);
@@ -70,26 +98,36 @@ static struct io_plan *daemon_conn_start(struct io_conn *conn,
 					 struct daemon_conn *dc)
 {
 	return io_duplex(conn, daemon_conn_read_next(conn, dc),
-			 daemon_conn_write_next(conn, dc));
+			 /* Could call daemon_conn_write_next, but we don't
+			  * want it to call empty_cb just yet! */
+			 msg_queue_wait(conn, dc->out,
+					daemon_conn_write_next, dc));
 }
 
-void daemon_conn_init(tal_t *ctx, struct daemon_conn *dc, int fd,
-		      struct io_plan *(*daemon_conn_recv)(struct io_conn *,
-							  struct daemon_conn *))
+static void destroy_dc_from_conn(struct io_conn *conn, struct daemon_conn *dc)
 {
-	dc->daemon_conn_recv = daemon_conn_recv;
-
-	dc->ctx = ctx;
-	dc->msg_in = NULL;
-	msg_queue_init(&dc->out, dc->ctx);
-	dc->msg_queue_cleared_cb = NULL;
-	dc->conn = io_new_conn(ctx, fd, daemon_conn_start, dc);
+	/* Harmless free loop if conn is being destroyed because dc freed */
+	tal_free(dc);
 }
 
-void daemon_conn_clear(struct daemon_conn *dc)
+struct daemon_conn *daemon_conn_new_(const tal_t *ctx, int fd,
+				     struct io_plan *(*recv)(struct io_conn *,
+							     const u8 *,
+							     void *),
+				     bool (*outq_empty)(void *),
+				     void *arg)
 {
-	io_set_finish(dc->conn, NULL, NULL);
-	io_close(dc->conn);
+	struct daemon_conn *dc = tal(NULL, struct daemon_conn);
+
+	dc->recv = recv;
+	dc->outq_empty = outq_empty;
+	dc->arg = arg;
+	dc->msg_in = NULL;
+	msg_queue_init(&dc->out, dc);
+
+	dc->conn = io_new_conn(dc, fd, daemon_conn_start, dc);
+	tal_add_destructor2(dc->conn, destroy_dc_from_conn, dc);
+	return dc;
 }
 
 void daemon_conn_send(struct daemon_conn *dc, const u8 *msg)

common/daemon_conn.h

@@ -6,49 +6,30 @@
 #include <ccan/short_types/short_types.h>
 #include <common/msg_queue.h>
 
-struct daemon_conn {
-	/* Context to tallocate all things from, possibly the
-	 * container of this connection. */
-	tal_t *ctx;
-
-	/* Last message we received */
-	u8 *msg_in;
-
-	/* Queue of outgoing messages */
-	struct msg_queue out;
-
-	/* Underlying connection */
-	struct io_conn *conn;
-
-	/* Callback for incoming messages */
-	struct io_plan *(*daemon_conn_recv)(struct io_conn *conn,
-					    struct daemon_conn *);
-
-	/* Called whenever we've cleared the msg_out queue. If it returns
-	 * true, it has added packets to msg_out queue. */
-	bool (*msg_queue_cleared_cb)(struct io_conn *, struct daemon_conn *);
-};
-
 /**
- * daemon_conn_init - Initialize a new daemon connection
+ * daemon_conn_new - Allocate a new daemon connection
  *
- * @ctx: context to allocate from
- * @dc: daemon_conn to initialize
+ * @ctx: context to allocate the daemon_conn's conn from
  * @fd: socket file descriptor to wrap
- * @daemon_conn_recv: callback function to be called upon receiving a message
- */
-void daemon_conn_init(tal_t *ctx, struct daemon_conn *dc, int fd,
-		      struct io_plan *(*daemon_conn_recv)(
-			  struct io_conn *, struct daemon_conn *));
-
-/**
- * daemon_conn_clear - discard a daemon conn without triggering finish.
- * @dc: the daemon_conn to clean up.
- *
- * This is used by gossipd when a peer is handed back, and we no longer
- * want to deal with it via daemon_conn.  @dc must not be used after this!
+ * @recv: callback function to be called upon receiving a message
+ * @outq_empty: callback function to be called when queue is empty: returns
+ *     true if it added something to the queue.  Can be NULL.
  */
-void daemon_conn_clear(struct daemon_conn *dc);
+#define daemon_conn_new(ctx, fd, recv, outq_empty, arg)		       \
+	daemon_conn_new_((ctx), (fd),				       \
+			 typesafe_cb_preargs(struct io_plan *, void *, \
+					     (recv), (arg), 	       \
+					     struct io_conn *,		\
+					     const u8 *),		\
+			 typesafe_cb(bool, void *,  (outq_empty), (arg)), \
+			 arg)
+
+struct daemon_conn *daemon_conn_new_(const tal_t *ctx, int fd,
+				     struct io_plan *(*recv)(struct io_conn *,
+							     const u8 *,
+							     void *),
+				     bool (*outq_empty)(void *),
+				     void *arg);
 
 /**
  * daemon_conn_send - Enqueue an outgoing message to be sent

connectd/connectd.c

@@ -121,7 +121,7 @@ struct daemon {
 	struct list_head connecting;
 
 	/* Connection to main daemon. */
-	struct daemon_conn master;
+	struct daemon_conn *master;
 
 	/* Local and global features to offer to peers. */
 	u8 *localfeatures, *globalfeatures;
@@ -372,7 +372,7 @@ static struct io_plan *peer_reconnected(struct io_conn *conn,
 
 	/* Tell master to kill it: will send peer_disconnect */
 	msg = towire_connect_reconnected(NULL, id);
-	daemon_conn_send(&daemon->master, take(msg));
+	daemon_conn_send(daemon->master, take(msg));
 
 	/* Save arguments for next time. */
 	r = tal(daemon, struct peer_reconnected);
@@ -425,10 +425,10 @@ struct io_plan *peer_connected(struct io_conn *conn,
 	/*~ daemon_conn is a message queue for inter-daemon communication: we
 	 * queue up the `connect_peer_connected` message to tell lightningd
 	 * we have connected, and give the the peer and gossip fds. */
-	daemon_conn_send(&daemon->master, peer_connected_msg);
+	daemon_conn_send(daemon->master, peer_connected_msg);
 	/* io_conn_fd() extracts the fd from ccan/io's io_conn */
-	daemon_conn_send_fd(&daemon->master, io_conn_fd(conn));
-	daemon_conn_send_fd(&daemon->master, gossip_fd);
+	daemon_conn_send_fd(daemon->master, io_conn_fd(conn));
+	daemon_conn_send_fd(daemon->master, gossip_fd);
 
 	/*~ Finally, we add it to the set of pubkeys: tal_dup will handle
 	 * take() args for us, by simply tal_steal()ing it. */
@@ -550,7 +550,7 @@ static void PRINTF_FMT(5,6)
 	 * asking. */
 	msg = towire_connectctl_connect_failed(NULL, id, err, wait_seconds,
 					       addrhint);
-	daemon_conn_send(&daemon->master, take(msg));
+	daemon_conn_send(daemon->master, take(msg));
 
 	status_trace("Failed connected out for %s: %s",
 		     type_to_string(tmpctx, struct pubkey, id),
@@ -1080,9 +1080,9 @@ static struct wireaddr_internal *setup_listeners(const tal_t *ctx,
 /*~ Parse the incoming connect init message from lightningd ("master") and
  * assign config variables to the daemon; it should be the first message we
  * get. */
-static struct io_plan *connect_init(struct daemon_conn *master,
-				   struct daemon *daemon,
-				   const u8 *msg)
+static struct io_plan *connect_init(struct io_conn *conn,
+				    struct daemon *daemon,
+				    const u8 *msg)
 {
 	struct wireaddr *proxyaddr;
 	struct wireaddr_internal *binding;
@@ -1127,19 +1127,19 @@ static struct io_plan *connect_init(struct daemon_conn *master,
 				  &announcable);
 
 	/* Tell it we're ready, handing it the addresses we have. */
-	daemon_conn_send(&daemon->master,
+	daemon_conn_send(daemon->master,
 			 take(towire_connectctl_init_reply(NULL,
 							   binding,
 							   announcable)));
 
 	/* Read the next message. */
-	return daemon_conn_read_next(master->conn, master);
+	return daemon_conn_read_next(conn, daemon->master);
 }
 
 /*~ lightningd tells us to go! */
-static struct io_plan *connect_activate(struct daemon_conn *master,
-				       struct daemon *daemon,
-				       const u8 *msg)
+static struct io_plan *connect_activate(struct io_conn *conn,
+					struct daemon *daemon,
+					const u8 *msg)
 {
 	bool do_listen;
 
@@ -1166,9 +1166,9 @@ static struct io_plan *connect_activate(struct daemon_conn *master,
 	daemon->listen_fds = tal_free(daemon->listen_fds);
 
 	/* OK, we're ready! */
-	daemon_conn_send(&daemon->master,
+	daemon_conn_send(daemon->master,
 			 take(towire_connectctl_activate_reply(NULL)));
-	return daemon_conn_read_next(master->conn, master);
+	return daemon_conn_read_next(conn, daemon->master);
 }
 
 /*~ This is where we'd put a BOLT #10 reference, but it doesn't exist :( */
@@ -1334,7 +1334,7 @@ static struct io_plan *connect_to_peer(struct io_conn *conn,
 		master_badmsg(WIRE_CONNECTCTL_CONNECT_TO_PEER, msg);
 
 	try_connect_peer(daemon, &id, seconds_waited, addrhint);
-	return daemon_conn_read_next(conn, &daemon->master);
+	return daemon_conn_read_next(conn, daemon->master);
 }
 
 /* lightningd tells us a peer has disconnected. */
@@ -1362,28 +1362,29 @@ static struct io_plan *peer_disconnected(struct io_conn *conn,
 	tal_free(key);
 
 	/* Read the next message from lightningd. */
-	return daemon_conn_read_next(conn, &daemon->master);
+	return daemon_conn_read_next(conn, daemon->master);
 }
 
-static struct io_plan *recv_req(struct io_conn *conn, struct daemon_conn *master)
+static struct io_plan *recv_req(struct io_conn *conn,
+				const u8 *msg,
+				struct daemon *daemon)
 {
-	struct daemon *daemon = container_of(master, struct daemon, master);
-	enum connect_wire_type t = fromwire_peektype(master->msg_in);
+	enum connect_wire_type t = fromwire_peektype(msg);
 
 	/* Demux requests from lightningd: we expect INIT then ACTIVATE, then
 	 * connect requests and disconnected messages. */
 	switch (t) {
 	case WIRE_CONNECTCTL_INIT:
-		return connect_init(master, daemon, master->msg_in);
+		return connect_init(conn, daemon, msg);
 
 	case WIRE_CONNECTCTL_ACTIVATE:
-		return connect_activate(master, daemon, master->msg_in);
+		return connect_activate(conn, daemon, msg);
 
 	case WIRE_CONNECTCTL_CONNECT_TO_PEER:
-		return connect_to_peer(conn, daemon, master->msg_in);
+		return connect_to_peer(conn, daemon, msg);
 
 	case WIRE_CONNECTCTL_PEER_DISCONNECTED:
-		return peer_disconnected(conn, daemon, master->msg_in);
+		return peer_disconnected(conn, daemon, msg);
 
 	/* We send these, we don't receive them */
 	case WIRE_CONNECTCTL_INIT_REPLY:
@@ -1396,7 +1397,7 @@ static struct io_plan *recv_req(struct io_conn *conn, struct daemon_conn *master
 
 	/* Master shouldn't give bad requests. */
 	status_failed(STATUS_FAIL_MASTER_IO, "%i: %s",
-		      t, tal_hex(tmpctx, master->msg_in));
+		      t, tal_hex(tmpctx, msg));
 }
 
 /*~ Helper for handshake.c: we ask `hsmd` to do the ECDH to get the shared
@@ -1424,8 +1425,7 @@ bool hsm_do_ecdh(struct secret *ss, const struct pubkey *point)
  *
  * The C++ method of omitting unused parameter names is *much* neater, and I
  * hope we'll eventually see it in a C standard. */
-static void master_gone(struct io_conn *unused UNUSED,
-			struct daemon *daemon UNUSED)
+static void master_gone(struct daemon_conn *master UNUSED)
 {
 	/* Can't tell master, it's gone. */
 	exit(2);
@@ -1446,12 +1446,13 @@ int main(int argc, char *argv[])
 	list_head_init(&daemon->connecting);
 	daemon->listen_fds = tal_arr(daemon, struct listen_fd, 0);
 	/* stdin == control */
-	daemon_conn_init(daemon, &daemon->master, STDIN_FILENO, recv_req);
-	io_set_finish(daemon->master.conn, master_gone, daemon);
+	daemon->master = daemon_conn_new(daemon, STDIN_FILENO, recv_req, NULL,
+					 daemon);
+	tal_add_destructor(daemon->master, master_gone);
 
 	/* This tells the status_* subsystem to use this connection to send
 	 * our status_ and failed messages. */
-	status_setup_async(&daemon->master);
+	status_setup_async(daemon->master);
 
 	/* Should never exit. */
 	io_loop(NULL, NULL);