Merge pull request #6811 from garlick/mcast_lightweight

broker: avoid malloc in event distribution critical path

Author: mergify[bot]

src/broker/overlay.c

@@ -36,6 +36,7 @@
 #include "src/common/libutil/monotime.h"
 #include "src/common/libutil/errprintf.h"
 #include "src/common/librouter/rpc_track.h"
+#include "src/common/libflux/message_route.h" // for msg_route_sendto()
 #include "ccan/str/str.h"
 
 #include "overlay.h"
@@ -851,17 +852,11 @@ static int overlay_sendmsg_child (struct overlay *ov, const flux_msg_t *msg)
     return rc;
 }
 
-/* Push child->uuid onto the message, then pop it off again after sending.
- */
-static int overlay_mcast_child_one (struct overlay *ov,
-                                    flux_msg_t *msg,
-                                    struct child *child)
+// callback for msg_route_sendto()
+static int overlay_mcast_send (const flux_msg_t *msg, void *arg)
 {
-    if (flux_msg_route_push (msg, child->uuid) < 0)
-        return -1;
-    int rc = overlay_sendmsg_child (ov, msg);
-    (void)flux_msg_route_delete_last (msg);
-    return rc;
+    struct overlay *ov = arg;
+    return overlay_sendmsg_child (ov, msg);
 }
 
 static void overlay_mcast_child (struct overlay *ov, flux_msg_t *msg)
@@ -873,7 +868,10 @@ static void overlay_mcast_child (struct overlay *ov, flux_msg_t *msg)
 
     foreach_overlay_child (ov, child) {
         if (subtree_is_online (child->status)) {
-            if (overlay_mcast_child_one (ov, msg, child) < 0) {
+            if (msg_route_sendto (msg,
+                                  child->uuid,
+                                  overlay_mcast_send,
+                                  ov) < 0) {
                 if (errno != EHOSTUNREACH) {
                     flux_log_error (ov->h,
                                     "mcast error to child rank %lu",

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 */
 
 /*