cmd: support addition of node 0 brokers in flux-batch/alloc

Problem: For subinstances of Flux with large node counts, it may be
useful to launch an extra set of brokers on node 0 of the allocation
to assist with handling message distribution, but there's no easy
way for users to request that in flux-alloc and flux-batch.

Add a new `--add-brokers=N` option to `flux alloc` and `flux batch`
which requests that `N` extra brokers be started on node 0 of the
allocation.  The option is hidden and undocumented for now in case
a better solution is implemented in the future. The `--add-brokers`
option is only available when a number of nodes is explicitly requested
with `-N, --nodes`.

The `--add-brokers` option adjusts the job's taskmap to force extra
brokers onto the first allocated node and automatically excludes the
extra ranks to ensure their resources are not available for scheduling.
The jobspec is also updated to reflect the updated task count.

Author: grondo

src/bindings/python/flux/cli/base.py

@@ -1661,6 +1661,9 @@ class BatchAllocCmd(MiniCmd):
     def __init__(self, prog, usage=None, description=None, exclude_io=True):
         self.t0 = None
         super().__init__(prog, usage, description, exclude_io)
+        self.parser.add_argument(
+            "--add-brokers", default=0, type=int, help=argparse.SUPPRESS
+        )
         self.parser.add_argument(
             "--conf",
             metavar="CONF",
@@ -1739,6 +1742,35 @@ def init_common(self, args):
             args.broker_opts = args.broker_opts or []
             args.broker_opts.append("-Scontent.dump=" + args.dump)
 
+        if args.add_brokers > 0:
+            if not args.nodes:
+                raise ValueError(
+                    "--add-brokers may only be specified with -N, --nnodes"
+                )
+            nbrokers = args.add_brokers
+            nnodes = args.nodes
+
+            # Force update taskmap with extra ranks on nodeid 0:
+            args.taskmap = f"manual:[[0,1,{1+nbrokers},1],[1,{nnodes-1},1,1]]"
+
+            # Exclude the additional brokers via configuration. However,
+            # don't throw away any ranks already excluded bythe user.
+            # Note: raises an exception if user excluded by hostname (unlikely)
+            exclude = IDset(args.conf.get("resource.exclude", default="")).set(
+                1, nbrokers
+            )
+            args.conf.update(f'resource.exclude="{exclude}"')
+
     def update_jobspec_common(self, args, jobspec):
         """Common jobspec update code for batch/alloc"""
-        pass
+        # If args.add_brokers is being used, update jobspec task count
+        # to accurately reflect the updated task count.
+        if args.add_brokers > 0:
+            # Note: args.nodes required with add_brokers already checked above
+            total_tasks = args.nodes + args.add_brokers
+
+            # Overwrite task count with new total_tasks:
+            jobspec.tasks[0]["count"] = {"total": total_tasks}
+
+            # remove per-resource shell option which is no longer necessary:
+            del jobspec.attributes["system"]["shell"]["options"]["per-resource"]