libaggregate: add timeout arg to aggregate_push call

Problem: The aggregate_push_json() convenience function call doesn't
allow the local aggregate timeout to be set.

Add the missing prameter and update the one caller.

src/cmd/builtin/hwloc.c

@@ -741,7 +741,7 @@ static int aggregate_topo_summary (flux_t *h, const char *key, const char *xml)
     if (!(o = topo_tojson (topo)))
         log_err_exit ("Failed to convert topology to JSON");
 
-    if (!(f = aggregator_push_json (h, fwd_count, key, o))
+    if (!(f = aggregator_push_json (h, fwd_count, 1., key, o))
         || (flux_future_get (f, NULL) < 0))
         log_err_exit ("aggregator_push_json");
 

src/common/libaggregate/aggregate.c

@@ -162,8 +162,8 @@ int aggregate_unpack_to_kvs (flux_future_t *f, const char *path)
     return (rc);
 }
 
-flux_future_t *aggregator_push_json (flux_t *h, int fwd_count,
-                                            const char *key, json_t *o)
+flux_future_t *aggregator_push_json (flux_t *h, int fwd_count, double timeout,
+                                     const char *key, json_t *o)
 {
     uint32_t size;
     uint32_t rank;
@@ -176,10 +176,22 @@ flux_future_t *aggregator_push_json (flux_t *h, int fwd_count,
         || (n >= sizeof (rankstr)))
         return NULL;
 
-    return flux_rpc_pack (h, "aggregator.push", FLUX_NODEID_ANY, 0,
+    if (timeout >= 0.)
+        return flux_rpc_pack (h, "aggregator.push", FLUX_NODEID_ANY, 0,
+                             "{s:s,s:i,s:i,s:f,s:{s:o}}",
+                             "key", key,
+                             "total", size,
+                             "fwd_count", fwd_count,
+                             "timeout" , timeout,
+                             "entries", rankstr, o);
+    else
+        return flux_rpc_pack (h, "aggregator.push", FLUX_NODEID_ANY, 0,
                              "{s:s,s:i,s:i,s:{s:o}}",
                              "key", key,
                              "total", size,
                              "fwd_count", fwd_count,
                              "entries", rankstr, o);
 }
+
+/* vi: ts=4 sw=4 expandtab
+ */

src/common/libaggregate/aggregate.h

@@ -10,9 +10,11 @@
 
 /*
  *  Push single json object `o` to local aggregator module via RPC.
- *   Steals the reference to `o`.
+ *   Steals the reference to `o`. If fwd_count > 0, then set local
+ *   forward count in aggregator message. If `t` is non-negative,
+ *   then set local forward timeout to this value.
  */
-flux_future_t *aggregator_push_json (flux_t *h, int fwd_count,
+flux_future_t *aggregator_push_json (flux_t *h, int fwd_count, double t,
 		                     const char *key, json_t *o);
 
 /*  Fulfill future when aggregate at `key` is "complete", i.e.