libflux: Support flux_msg_disable_route()

Problem: libflux currently supports flux_msg_enable_route() and
flux_msg_clear_route(), which seem imbalanced.  Hypothethically, there
should be a flux_msg_disable_route().

Solution: Add a new function flux_msg_disable_route() that behaves
identically to flux_msg_clear_route() before this commit.  Change the
behavior of flux_msg_clear_route() to only clear internally stored route
ids, not to disable routing on a message.

Adjust all callers accordingly.  In locations that previously called
flux_msg_clear_route() and flux_msg_enable_route() in series, only
call flux_msg_clear_route().

src/broker/overlay.c

@@ -640,7 +640,7 @@ static void parent_cb (flux_reactor_t *r, flux_watcher_t *w,
         goto handled;
     }
     if (type == FLUX_MSGTYPE_EVENT)
-        flux_msg_clear_route (msg);
+        flux_msg_disable_route (msg);
     ov->recv_cb (msg, OVERLAY_UPSTREAM, ov->recv_arg);
 handled:
     flux_msg_destroy (msg);

src/broker/publisher.c

@@ -131,7 +131,7 @@ int publisher_send (struct publisher *pub, const flux_msg_t *msg)
 
     if (!(cpy = flux_msg_copy (msg, true)))
         return -1;
-    flux_msg_clear_route (cpy);
+    flux_msg_disable_route (cpy);
     if (flux_msg_set_seq (cpy, ++pub->seq) < 0)
         goto error_restore_seq;
     send_event (pub, cpy);

src/common/libflux/message.c

@@ -929,6 +929,14 @@ void flux_msg_enable_route (flux_msg_t *msg)
     (void) flux_msg_set_flags (msg, msg->flags | FLUX_MSGFLAG_ROUTE);
 }
 
+void flux_msg_disable_route (flux_msg_t *msg)
+{
+    if (!msg || (!(msg->flags & FLUX_MSGFLAG_ROUTE)))
+        return;
+    flux_msg_clear_route (msg);
+    (void) flux_msg_set_flags (msg, msg->flags & ~(uint8_t)FLUX_MSGFLAG_ROUTE);
+}
+
 void flux_msg_clear_route (flux_msg_t *msg)
 {
     struct route_id *r;
@@ -938,7 +946,6 @@ void flux_msg_clear_route (flux_msg_t *msg)
         route_id_destroy (r);
     list_head_init (&msg->routes);
     msg->routes_len = 0;
-    (void) flux_msg_set_flags (msg, msg->flags & ~(uint8_t)FLUX_MSGFLAG_ROUTE);
 }
 
 int flux_msg_push_route (flux_msg_t *msg, const char *id)

src/common/libflux/message.h

@@ -307,13 +307,19 @@ const char *flux_msg_typestr (int type);
  * routing.
  */
 
-/* Prepare a message for routing by setting FLUX_MSGFLAG_ROUTE.  This
+/* Enable routes on a message by setting FLUX_MSGFLAG_ROUTE.  This
  * function is a no-op if the flag is already set.
  */
 void flux_msg_enable_route (flux_msg_t *msg);
 
-/* Clear routes from msg and clear FLUX_MSGFLAG_ROUTE flag.  This
- * function is a no-op if the flag is already clear.
+/* Disable routes on a message by clearing the FLUX_MSGFLAG_ROUTE
+ * flag.  In addition, clear all presently stored routes.  This
+ * function is a no-op if the flag is already set.
+ */
+void flux_msg_disable_route (flux_msg_t *msg);
+
+/* Clear all routes stored in a message.  This function is a no-op if
+ * routes are not enabled.
  */
 void flux_msg_clear_route (flux_msg_t *msg);
 

src/common/libflux/test/message.c

@@ -252,6 +252,8 @@ void check_cornercase (void)
 
     lives_ok ({flux_msg_enable_route (NULL);},
         "flux_msg_enable_route msg=NULL doesnt crash");
+    lives_ok ({flux_msg_disable_route (NULL);},
+        "flux_msg_disable_route msg=NULL doesnt crash");
     lives_ok ({flux_msg_clear_route (NULL);},
         "flux_msg_clear_route msg=NULL doesnt crash");
 
@@ -314,6 +316,9 @@ void check_routes (void)
     flux_msg_clear_route (msg);
     ok (flux_msg_frames (msg) == 1,
         "flux_msg_clear_route works, is no-op on msg w/o routes enabled");
+    flux_msg_disable_route (msg);
+    ok (flux_msg_frames (msg) == 1,
+        "flux_msg_disable_route works, is no-op on msg w/o routes enabled");
     flux_msg_enable_route (msg);
     ok (flux_msg_frames (msg) == 2,
         "flux_msg_enable_route works, adds one frame on msg w/ routes enabled");
@@ -372,8 +377,23 @@ void check_routes (void)
         "flux_msg_get_route_count returns 1 on message w/ id1");
 
     flux_msg_clear_route (msg);
-    ok (flux_msg_frames (msg) == 1,
+    ok (flux_msg_get_route_count (msg) == 0,
         "flux_msg_clear_route clear routing frames");
+    ok (flux_msg_frames (msg) == 2,
+        "flux_msg_clear_route did not disable routing frames");
+
+    ok (flux_msg_push_route (msg, "foobar") == 0 && flux_msg_frames (msg) == 3,
+        "flux_msg_push_route works and adds a frame after flux_msg_clear_route()");
+    ok ((flux_msg_get_route_count (msg) == 1),
+        "flux_msg_get_route_count returns 1 on msg w/ id1");
+
+    flux_msg_disable_route (msg);
+    ok (flux_msg_frames (msg) == 1,
+        "flux_msg_disable_route clear routing frames");
+
+    ok (flux_msg_push_route (msg, "boobar") < 0 && errno == EPROTO,
+        "flux_msg_push_route fails with EPROTO after flux_msg_disable_route()");
+
     flux_msg_destroy (msg);
 }
 

src/common/librouter/servhash.c

@@ -68,7 +68,6 @@ static flux_msg_t *request_copy_clear_routes (const flux_msg_t *msg)
     if (!(cpy = flux_msg_copy (msg, true)))
         return NULL;
     flux_msg_clear_route (cpy);
-    flux_msg_enable_route (cpy);
     return cpy;
 }