wip: launch jobs to instances

Signed-off-by: vsoch <vsoch@users.noreply.github.com>

Author: vsoch

README.md

@@ -9,8 +9,9 @@
 
 This library includes a cluster agnostic language to setup a job (one unit of work in a jobspec).
 It is a transformational layer, or a simple language that converts steps needed to prepare a job
-for a specific clusters scheduler. We are currently prototyping off of the Flux JobSpec, and intent
-to derive some variant between that and something more. It is JobSpec... the next generation! 🚀️
+for a specific clusters scheduler. It also provides a means to submit to a hierarchy.
+We are currently prototyping off of the Flux JobSpec, and intend to derive some variant
+between that and something more. It is JobSpec... the next generation! 🚀️
 
 ⭐️ [Documentation](https://compspec.github.io/jobspec) ⭐️
 

docs/docs/commands.md

@@ -37,6 +37,29 @@ If run from python, the function "satisfied" would return False and the broker c
 is expected to be in JSON and our jobspec files are in yaml, so we can throw them into the same examples directory without issue.
 
 
+### Submit to a Hierarchy
+
+#### 1. Start Flux
+
+Start up the development environment to find yourself in a container with flux. Start a test instance:
+
+```bash
+flux start --test-size=4
+```
+
+Note that we have 4 faux nodes and 40 faux cores.
+
+```bash
+flux resource list
+```
+```console
+     STATE NNODES   NCORES    NGPUS NODELIST
+      free      4       40        0 194c2b9f4f3c,194c2b9f4f3c,194c2b9f4f3c,194c2b9f4f3c
+ allocated      0        0        0
+      down      0        0        0
+```
+
+
 ### Run
 
 #### 1. Start Flux

examples/launch/hierarchy.yaml

@@ -0,0 +1,44 @@
+version: 2
+
+# The entrypoint is where to start the hierarchy.
+entrypoint: level-1
+
+# Groups are templates for a level in the hierarchy.
+groups:
+  # Each of these are intermediate groups - they only exist to launch children.
+  - name: level-1
+    resources: level1
+    launch:
+      - group: level-2
+        count: 2  # Launch 2 instances of level-2
+
+  - name: level-2
+    resources: level2
+    launch:
+      - group: level-3
+        count: 2  # Each level-2 instance launches 2 instances of level-3
+
+  - name: level-3
+    resources: level3
+    launch:
+      - group: level-4
+        count: 2  # Each level-3 instance launches 2 instances of level-4
+
+  # This is a leaf group. No launch section == start a broker!
+  - name: level-4
+    resources: level4
+
+# These are nested resources. If we have 8 at the top, we have to split that up
+resources:
+  level1: # Consumed by the single level-1 broker
+    count: 8
+    type: core
+  level2: # Consumed by each of the two level-2 brokers
+    count: 4
+    type: core
+  level3: # Consumed by each of the four level-3 brokers
+    count: 2
+    type: core
+  level4: # Consumed by each of the eight level-4 leaf brokers
+    count: 1
+    type: core

examples/launch/throughput.py

@@ -0,0 +1,61 @@
+#!/usr/bin/env python3
+
+import time
+import argparse
+
+from jobspec.transformer.flux import FluxHierarchy
+from jobspec.logger import LogColors
+
+def parse_args():
+    parser = argparse.ArgumentParser(description="Run job throughput test on a Flux Hierarchy")
+    parser.add_argument("config", help="Path to the FluxHierarchy YAML configuration file")
+    parser.add_argument("-n", "--njobs", type=int, metavar="N", help="Total number of jobs to run", default=100)
+    parser.add_argument("-t", "--runtime", help="Simulated runtime of each job (default=1ms)", default="0.001s")
+    parser.add_argument("-x", "--exec", help="Do not simulate, actually run jobs", action="store_true")
+    parser.add_argument("-o", "--setopt", action="append", help="Set shell option OPT or OPT=VAL", metavar="OPT")
+    parser.add_argument("--setattr", action="append", help="Set job attribute ATTR=VAL", metavar="ATTR=VAL")
+    parser.add_argument("command", nargs=argparse.REMAINDER, default=["true"])
+    return parser.parse_args()
+
+def main():
+    args = parse_args()
+
+    # Instantiate, build, and connect to the Flux Hierarchy!
+    hierarchy = FluxHierarchy(args.config)
+    hierarchy.start(interactive=False)
+
+    # Default to true if not set.
+    if not args.command:
+        args.command = ["true"]
+
+    # Run the throughput test using the specialized 'throughput' method
+    time0 = time.time()
+    jobs = hierarchy.throughput(args.command, args.njobs)
+    
+    if not jobs:
+        print(f"{LogColors.RED}No jobs were tracked. Cannot calculate throughput.{LogColors.ENDC}")
+        return
+
+    # 3. Analyze and print results (logic is unchanged)
+    first = jobs[min(jobs.keys(), key=lambda x: jobs[x].get("submit", type("o", (), {"timestamp": float('inf')})()).timestamp)]
+    last = jobs[max(jobs.keys(), key=lambda x: jobs[x].get("clean", type("o", (), {"timestamp": float('-inf')})()).timestamp)]
+    lastsubmit = jobs[max(jobs.keys(), key=lambda x: jobs[x]["t_submit"])]
+    
+    submit_time = lastsubmit["t_submit"] - time0
+    sjps = args.njobs / submit_time if submit_time > 0 else float('inf')    
+    script_runtime = time.time() - time0
+
+    job_runtime = last["clean"].timestamp - first["submit"].timestamp
+    jps = args.njobs / job_runtime if job_runtime > 0 else float('inf')
+    jpsb = args.njobs / script_runtime if script_runtime > 0 else float('inf')
+
+    print(f"\n--- Throughput Results ---")
+    print(f"number of jobs: {args.njobs} (on {len(hierarchy.handles)} workers)")
+    print(f"   submit time: {submit_time:<6.3f}s ({sjps:5.1f} job/s)")
+    print(f"script runtime: {script_runtime:<6.3f}s")
+    print(f"   job runtime: {job_runtime:<6.3f}s")
+    print(f"    throughput: {jps:<.1f} job/s (script: {jpsb:5.1f} job/s)")
+
+
+if __name__ == "__main__":
+    main()

jobspec/cli/__init__.py

@@ -46,6 +46,15 @@ def get_parser():
     )
     subparsers.add_parser("version", description="show software version")
 
+    # start an instane hierarchy
+    start = subparsers.add_parser(
+        "start",
+        formatter_class=argparse.RawTextHelpFormatter,
+        description="start an instance hierarchy",
+    )
+    start.add_argument("config", help="configuration file for hierarchy")
+    start.add_argument("-o", "--out", help="output directory for hierarchy assets", default=None)
+
     # Maybe this warrants a better name, but this seems to be what we'd want to do -
     # run a jobspec
     run = subparsers.add_parser(
@@ -125,6 +134,8 @@ def help(return_code=0):
         from .run import main
     elif args.command == "satisfy":
         from .satisfy import main
+    elif args.command == "start":
+        from .start import main
     else:
         help(1)
     main(args, extra)

jobspec/cli/start.py

@@ -0,0 +1,13 @@
+#!/usr/bin/env python
+
+from jobspec.transformer.flux import FluxHierarchy
+
+
+def main(args, _):
+    """
+    Start a set of nested instances to submit jobs to.
+    """
+    hier = FluxHierarchy(args.config, outdir=args.out)
+    # Note that this returns the uri lookup, if we wanted
+    # to use it somewhere.
+    hier.start()

jobspec/runner.py

@@ -57,15 +57,21 @@ def run(self, filename):
         """
         # Load the jobspec
         jobspec = self.load_jobspec(filename)
+        print(jobspec.data)
 
         # Get validated transformation steps
         # These will depend on the transformer logic
         steps = self.parse(jobspec)
+        print(jobspec.data)
         self.announce()
 
         # Run each step to submit the job, and that's it.
         for step in steps:
             step_runner.run(self.name, step)
+        print("POST RUN")
+        import IPython
+
+        IPython.embed()
 
     def load_jobspec(self, filename):
         """

jobspec/transformer/flux/__init__.py

@@ -1 +1,5 @@
+from .hierarchy import FluxHierarchy
 from .workload import FluxWorkload as Transformer
+
+assert Transformer
+assert FluxHierarchy