dbus: let a dropped peer outlive its own read loop

A peer can be dropped from inside its own read loop: a handler emits a
signal, the write to that very peer fails, and the drop lands while
link_connection_process() still holds the connection and will touch
its rx buffer on the way out.  Freeing there pulls the ground out from
under it.  Unlink the peer and let the event loop free it once the
stack has unwound.

The work has to be scheduled with a non-zero delay.  A uev timer armed
with zero is a disarmed timer, so the queue would never run and the
connections would leak instead.

Losing a peer is also not a warning.  It is what shutdown looks like
from here, and every reboot said so on the console.

Signed-off-by: Joachim Wiberg <troglobit@gmail.com>

src/dbus.c

@@ -50,6 +50,7 @@
 #include "conf.h"
 #include "log.h"
 #include "private.h"
+#include "schedule.h"
 #include "service.h"
 #include "sig.h"
 #include "sm.h"
@@ -61,10 +62,12 @@
 struct peer {
 	uev_t              watcher;
 	link_connection_t  *conn;
+	int                dead;
 	TAILQ_ENTRY(peer)  link;
 };
 
 static TAILQ_HEAD(, peer) peers = TAILQ_HEAD_INITIALIZER(peers);
+static TAILQ_HEAD(, peer) reapq = TAILQ_HEAD_INITIALIZER(reapq);
 static link_server_t      *server;
 static uev_t              accept_watcher;
 static size_t             peer_count;
@@ -82,23 +85,55 @@ static void link_log_cb(void *userdata, const char *func, const char *msg)
 	logit(LOG_DEBUG, "%s():%s", func, msg);
 }
 
+/* Close and free everything dropped since the last time round the
+ * event loop.  Safe here because no connection's read loop is on the
+ * stack; see peer_drop(). */
+static void peer_reap(void *arg)
+{
+	struct peer *p;
+
+	(void)arg;
+	while ((p = TAILQ_FIRST(&reapq))) {
+		TAILQ_REMOVE(&reapq, p, link);
+		link_connection_close(p->conn);
+		free(p);
+	}
+}
+
+/* Not zero: a uev timer armed with a zero timeout is a disarmed
+ * timer, so the work would never run and the connections would leak.
+ * Any short delay does, the peer is already unlinked and its watcher
+ * stopped, so nothing touches it in the meantime. */
+static struct wq reap_work = { .cb = peer_reap, .delay = 10 };
+
+/*
+ * A peer can be dropped from inside its own read loop: a handler emits
+ * a signal, the write to this very peer fails, and dbus_emit_signal()
+ * lands here while link_connection_process() still holds the
+ * connection and will touch its rx buffer on the way out.  Freeing now
+ * would pull that out from under it, so unlink the peer and let the
+ * event loop free it once the stack has unwound.
+ */
 static void peer_drop(struct peer *p)
 {
-	int was_sysbus = p == sysbus_peer;
+	if (p->dead)
+		return;		/* already on its way out */
+	p->dead = 1;
 
 	uev_io_stop(&p->watcher);
-	link_connection_close(p->conn);
 	TAILQ_REMOVE(&peers, p, link);
 	peer_count--;
-	free(p);
 
 	/* broker gone; the notify paths probe for its return.  Its unique
 	 * names die with it, so nothing we learned about them is safe to
 	 * carry over to whatever takes its place. */
-	if (was_sysbus) {
+	if (p == sysbus_peer) {
 		sysbus_peer = NULL;
 		sender_cache_flush();
 	}
+
+	TAILQ_INSERT_TAIL(&reapq, p, link);
+	schedule_work(&reap_work);
 }
 
 static void peer_cb(uev_t *w, void *arg, int events)
@@ -813,6 +848,19 @@ void dbus_unregister_service(svc_t *svc)
  * letting each connection apply its AddMatch filter.  Short-circuits
  * when no peers are connected so dbus_notify_* callers don't have
  * to inspect that state themselves. */
+/* A peer that has gone away is routine rather than a fault: initctl
+ * calls and exits, and on the way down every peer goes at once, which
+ * is not something an operator watching the shutdown needs to read
+ * about.  Keep the warning for a write that failed for some other
+ * reason, where something really is wrong. */
+static int quiet_drop(int err)
+{
+	if (err == EPIPE || err == ECONNRESET || err == ENOTCONN)
+		return 1;
+
+	return runlevel == 0 || runlevel == 6;
+}
+
 static void dbus_emit_signal(const char *path,
 			     const char *interface,
 			     const char *member,
@@ -824,18 +872,23 @@ static void dbus_emit_signal(const char *path,
 	if (!server || TAILQ_EMPTY(&peers))
 		return;
 	TAILQ_FOREACH_SAFE(p, &peers, link, tmp) {
+		int err;
+
 		if (link_connection_emit_signal(p->conn,
 				path, interface, member,
-				signature, body, body_len) < 0) {
-			/* nothing hit the wire, and same for every peer */
-			if (errno == EMSGSIZE || errno == EINVAL)
-				break;
-			logit(LOG_WARNING, "D-Bus peer fd %d write failed: "
-			      "%s, dropping",
-			      link_connection_get_fd(p->conn),
-			      strerror(errno));
-			peer_drop(p);
-		}
+				signature, body, body_len) >= 0)
+			continue;
+
+		err = errno;
+
+		/* nothing hit the wire, and same for every peer */
+		if (err == EMSGSIZE || err == EINVAL)
+			break;
+
+		logit(quiet_drop(err) ? LOG_DEBUG : LOG_WARNING,
+		      "D-Bus peer fd %d write failed: %s, dropping",
+		      link_connection_get_fd(p->conn), strerror(err));
+		peer_drop(p);
 	}
 }
 
@@ -1601,6 +1654,9 @@ int dbus_exit(void)
 	while ((p = TAILQ_FIRST(&peers)))
 		peer_drop(p);
 
+	/* No read loop is running now, and the timer never will again. */
+	peer_reap(NULL);
+
 	if (server) {
 		link_server_free(server);
 		server = NULL;