libflux: add msg_route_sendto()

Problem: iteratively sending the same message to multiple peers,
as occurs in the broker, is not as lightweight as it could be.

Moreover, this code path has shown up in a hot spot in #6806.

Add a private helper function to libflux that avoids a malloc
at each iteration.

Author: garlick

src/common/libflux/message_route.c

@@ -36,6 +36,7 @@ static struct route_id *route_id_create (const char *id, unsigned int id_len)
     struct route_id *r;
     if (!(r = calloc (1, sizeof (*r) + id_len + 1)))
         return NULL;
+    r->id = (char *)(r + 1);
     if (id && id_len) {
         memcpy (r->id, id, id_len);
         list_node_init (&(r->route_id_node));
@@ -93,6 +94,36 @@ int msg_route_delete_last (flux_msg_t *msg)
     return 0;
 }
 
+/* This function was added to streamline a critical path in the broker that
+ * iteratively sends the same message to multiple peers.  It is equivalent to
+ *   flux_msg_route_push (msg, id, strlen (id))
+ *   cb (msg, arg)
+ *   flux_msg_route_delete_last (msg)
+ * but borrows the copy of id rather than copying it each time.
+ */
+int msg_route_sendto (const flux_msg_t *cmsg,
+                      const char *id,
+                      msg_route_send_f cb,
+                      void *arg)
+{
+    flux_msg_t *msg = (flux_msg_t *)cmsg; // drop const since net effect is nil
+    struct route_id r;
+    int rc;
+
+    list_node_init (&r.route_id_node);
+    r.id = (char *)id;
+
+    list_add (&msg->routes, &r.route_id_node);
+    msg->routes_len++;
+
+    rc = cb (msg, arg);
+
+    list_pop (&msg->routes, struct route_id, route_id_node);
+    msg->routes_len--;
+
+    return rc;
+}
+
 /*
  * vi:tabstop=4 shiftwidth=4 expandtab
  */

src/common/libflux/message_route.h

@@ -11,9 +11,11 @@
 #ifndef _FLUX_CORE_MESSAGE_ROUTE_H
 #define _FLUX_CORE_MESSAGE_ROUTE_H
 
+#include "ccan/list/list.h"
+
 struct route_id {
     struct list_node route_id_node;
-    char id[0];                 /* variable length id stored at end of struct */
+    char *id;                   /* variable length id stored at end of struct */
 };
 
 int msg_route_push (flux_msg_t *msg,
@@ -28,6 +30,13 @@ void msg_route_clear (flux_msg_t *msg);
 
 int msg_route_delete_last (flux_msg_t *msg);
 
+typedef int (*msg_route_send_f)(const flux_msg_t *msg, void *arg);
+
+int msg_route_sendto (const flux_msg_t *msg,
+                      const char *id,
+                      msg_route_send_f cb,
+                      void *arg);
+
 #endif /* !_FLUX_CORE_MESSAGE_ROUTE_H */
 
 /*