diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..fdc6b75 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,64 @@ +# TopoQueue engineering guidance + +## Product invariants + +- Determinism is a product feature. Identical inputs must produce byte-for-byte stable text and JSON output. +- TopoQueue is an educational simulator. Never claim Kubernetes scheduler, Kueue, or production-scheduler compatibility. +- Preserve the behavior and stable output shapes of the existing `schedule` and `compare` commands. +- Use logical integer ticks for simulations. Do not use wall-clock time, sleeps, randomness, or background timers. +- Keep state ownership explicit. Independent policy runs must not share mutable scheduling state. +- Prefer the Go standard library. Do not add a dependency unless the feature cannot reasonably be implemented without it. +- Reject invalid or ambiguous input early, with contextual errors. +- Check arithmetic that can overflow. +- Keep unrelated refactors out of feature commits. + +## Repository map + +- `cmd/topoqueue/`: CLI parsing and command wiring +- `internal/model/`: strict YAML models, loading, cloning, and validation +- `internal/scheduler/`: placement, policies, comparisons, and simulation logic +- `internal/output/`: deterministic text and JSON renderers +- `examples/`: runnable scenarios +- `docs/`: design documents and executable plans + +## Verification + +After each milestone: + +1. Run `gofmt` on changed Go files. +2. Run focused package tests. +3. Run `go test ./...` before committing. + +Before declaring the branch complete, run: + +```sh +make fmt-check +make vet +make test +make build +make demo +make demo-simulate +make benchmark +git diff --check main...HEAD +``` + +Fix failures before proceeding. Do not weaken, skip, or delete tests merely to make a check pass. + +## Git discipline + +- Work only on the current feature branch; never commit directly to `main`. +- Use the repository's existing commit-signing configuration. Do not change Git configuration or disable signing. +- Create coherent, reviewable commits. Target 8–12 commits for a large milestone, but never split changes merely to increase the count. +- Every commit must compile and pass `go test ./...`. +- Do not create empty commits. +- Do not amend or rewrite commits that existed before this feature branch. +- Do not push, force-push, tag, create a release, or open/merge a pull request. +- Finish with a clean worktree. + +## ExecPlan + +For the event-driven simulation milestone, maintain the living plan at: + +`docs/exec-plans/v0.2-event-driven-simulation.md` + +Update its progress, decisions, validation evidence, and known limitations as work proceeds. diff --git a/Makefile b/Makefile index d8d64e8..7c8cf26 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: build test vet fmt-check demo benchmark +.PHONY: build test vet fmt-check demo demo-simulate benchmark BINARY := bin/topoqueue @@ -26,5 +26,12 @@ demo: --jobs examples/jobs.yaml \ --output text +demo-simulate: + go run ./cmd/topoqueue simulate \ + --cluster examples/cluster.yaml \ + --jobs examples/timed-jobs.yaml \ + --policy all \ + --output text + benchmark: go test -run '^$$' -bench . ./internal/scheduler diff --git a/README.md b/README.md index 9e53bf7..ff8a1f3 100644 --- a/README.md +++ b/README.md @@ -5,7 +5,7 @@ TopoQueue is a deterministic Go CLI for exploring how queue policy and topology constraints affect accelerator-job admission. -> TopoQueue is an educational snapshot simulator. It is not a Kubernetes +> TopoQueue is an educational scheduling simulator. It is not a Kubernetes > scheduler and does not implement Kueue semantics. ## Features @@ -18,6 +18,8 @@ topology constraints affect accelerator-job admission. - Human-readable text and stable, indented JSON output. - Concurrent comparison of policies with isolated mutable scheduling state. - Strict YAML decoding that rejects unknown fields. +- Discrete-event workload simulation with logical arrivals, fixed durations, + completion-driven resource release, and terminal lifecycle results. ## Installation @@ -64,6 +66,16 @@ go run ./cmd/topoqueue schedule \ --output json ``` +Simulate both policies over the checked-in timed workload: + +```sh +./bin/topoqueue simulate \ + --cluster examples/cluster.yaml \ + --jobs examples/timed-jobs.yaml \ + --policy all \ + --output text +``` + ## Example comparison This is the output of `make demo` using the checked-in example files: @@ -88,6 +100,26 @@ batch pending - - 0/16 0/4 head_of_line_blocked: head_ The `CPU` and `GPU` columns in job decisions show allocated/requested units. +## Event-driven simulation + +The `simulate` command uses a separate timed-jobs schema with required +`arrivalTick` and `durationTicks` fields. It advances directly between logical +event ticks; it does not sleep or use wall-clock time. At each tick it completes +and releases all finishing jobs, enqueues all arrivals, and then runs one +admission cycle. + +The checked-in example demonstrates the difference over time: + +| Policy | Completed | Unscheduled | Makespan | Total wait | Average wait | Maximum wait | +|---|---:|---:|---:|---:|---:|---:| +| backfill | 3 | 0 | 12 | 7 | 2.33 | 7 | +| strict-fifo | 3 | 0 | 14 | 17 | 5.67 | 10 | + +Backfill starts `batch` at tick 2 while `train-xl` waits for the whole cluster. +Strict FIFO retains `train-xl` at the queue head, so `batch` starts only after +`train-xl` completes. See [Event-driven simulation](docs/event-driven-simulation.md) +for the input contract, event ordering, lifecycle fields, and metric definitions. + ## Algorithm Jobs are considered in YAML order. For each attempted job, TopoQueue computes @@ -106,6 +138,13 @@ head-of-line blocked. Backfill records an unplaceable job and continues trying later jobs. A comparison runs both simulations concurrently, but each policy run is single-threaded and owns its scheduling state. +Timed simulation reuses the same placement algorithm. Arrivals are ordered by +logical tick and input position, and running jobs are ordered by completion tick +and input position. Completion releases the exact recorded per-node allocation. +Pending work is retried only at arrival and completion ticks. When neither can +occur again, remaining jobs become terminally unscheduled with final placement +reasons. + ## Complexity Let `N` be the number of nodes, `D` the number of topology domains, and `J` the @@ -114,17 +153,21 @@ sorting nodes and domain candidates, for a worst-case bound of `O(N log N + D log D)` time and `O(N + D)` temporary space. A full backfill run is `O(J * (N log N + D log D))`; strict FIFO may stop before all `J` jobs. Policy comparison performs two independent runs with the same per-run bounds. +For a timed workload, heap operations add `O(log J)` per admission/completion. +One admission cycle may inspect every pending job, and up to `O(J)` event ticks +can retry pending work, so the deliberately simple worst case is +`O(J^2 * (N log N + D log D) + J log J)` time and `O(N + D + J)` state. ## Limitations -- The input is a static snapshot; there is no API server, watch loop, or - cluster controller. +- Cluster capacity is still an input snapshot; there is no API server, watch + loop, or cluster controller. - Resources are whole, non-negative CPU and GPU units; cluster and per-job totals must fit in a signed 64-bit integer. - Replicas are identical, cannot be split across nodes, and support at most one required topology key per job. -- There is no preemption, priority, fairness, retry, reservation, or job - lifecycle model. +- Timed jobs have fixed successful durations. There are no failures, failure + retries, preemption, priorities, fairness, reservations, or elastic jobs. - TopoQueue does not implement Kubernetes scheduling behavior or exact Kueue semantics. - The simulator is educational and is not intended for production scheduling. @@ -137,6 +180,7 @@ make vet make test make build make demo +make demo-simulate make benchmark ``` @@ -149,6 +193,7 @@ internal/scheduler/ policies, placement, comparison, tests, and benchmark internal/output/ deterministic text and JSON renderers examples/ example cluster and ordered jobs docs/design.md design decisions and deliberate non-goals +docs/event-driven-simulation.md timed input, lifecycle, and metric semantics .github/workflows/ formatting, vet, race-test, and build CI ``` diff --git a/cmd/topoqueue/main.go b/cmd/topoqueue/main.go index 7e67ad9..817cd44 100644 --- a/cmd/topoqueue/main.go +++ b/cmd/topoqueue/main.go @@ -18,6 +18,7 @@ import ( const ( outputText = "text" outputJSON = "json" + policyAll = "all" ) func main() { @@ -48,6 +49,8 @@ func run(ctx context.Context, args []string, stdout, stderr io.Writer) error { return runSchedule(ctx, args[1:], stdout, stderr) case "compare": return runCompare(ctx, args[1:], stdout, stderr) + case "simulate": + return runSimulate(ctx, args[1:], stdout, stderr) case "help", "-h", "--help": return writeRootUsage(stdout) default: @@ -58,6 +61,75 @@ func run(ctx context.Context, args []string, stdout, stderr io.Writer) error { } } +func runSimulate(ctx context.Context, args []string, stdout, stderr io.Writer) error { + flags := flag.NewFlagSet("simulate", flag.ContinueOnError) + flags.SetOutput(stderr) + clusterPath := flags.String("cluster", "", "path to the cluster YAML file") + jobsPath := flags.String("jobs", "", "path to the timed jobs YAML file") + policyName := flags.String("policy", string(scheduler.PolicyStrictFIFO), "queue policy: strict-fifo, backfill, or all") + outputName := flags.String("output", outputText, "output format: text or json") + if err := flags.Parse(args); err != nil { + if errors.Is(err, flag.ErrHelp) { + return nil + } + return fmt.Errorf("parse simulate flags: %w", err) + } + if flags.NArg() != 0 { + return fmt.Errorf("simulate: unexpected positional arguments: %v", flags.Args()) + } + if err := requireInputPaths(*clusterPath, *jobsPath); err != nil { + return fmt.Errorf("simulate: %w", err) + } + if err := validateSimulationPolicy(*policyName); err != nil { + return fmt.Errorf("simulate: %w", err) + } + if err := validateOutputFormat(*outputName); err != nil { + return fmt.Errorf("simulate: %w", err) + } + + cluster, jobs, err := loadSimulationInputs(*clusterPath, *jobsPath) + if err != nil { + return err + } + if *policyName == policyAll { + results, err := scheduler.CompareSimulations(ctx, cluster, jobs) + if err != nil { + return fmt.Errorf("simulate all policies: %w", err) + } + switch *outputName { + case outputText: + if err := output.WriteSimulationComparisonText(stdout, results); err != nil { + return fmt.Errorf("render simulation comparison text output: %w", err) + } + case outputJSON: + if err := output.WriteSimulationComparisonJSON(stdout, results); err != nil { + return fmt.Errorf("render simulation comparison JSON output: %w", err) + } + } + return nil + } + + policy, err := scheduler.ParsePolicy(*policyName) + if err != nil { + return fmt.Errorf("simulate: %w", err) + } + result, err := scheduler.Simulate(ctx, cluster, jobs, policy) + if err != nil { + return fmt.Errorf("simulate policy %q: %w", policy, err) + } + switch *outputName { + case outputText: + if err := output.WriteSimulationText(stdout, result); err != nil { + return fmt.Errorf("render simulation text output: %w", err) + } + case outputJSON: + if err := output.WriteSimulationJSON(stdout, result); err != nil { + return fmt.Errorf("render simulation JSON output: %w", err) + } + } + return nil +} + func runSchedule(ctx context.Context, args []string, stdout, stderr io.Writer) error { flags := flag.NewFlagSet("schedule", flag.ContinueOnError) flags.SetOutput(stderr) @@ -171,6 +243,26 @@ func loadInputs(clusterPath, jobsPath string) (model.Cluster, model.JobSet, erro return cluster, jobs, nil } +func loadSimulationInputs(clusterPath, jobsPath string) (model.Cluster, model.TimedJobSet, error) { + cluster, err := model.LoadCluster(clusterPath) + if err != nil { + return model.Cluster{}, model.TimedJobSet{}, err + } + jobs, err := model.LoadTimedJobs(jobsPath) + if err != nil { + return model.Cluster{}, model.TimedJobSet{}, err + } + if err := model.ValidateTopologyRequirements(cluster, jobs.StaticJobs()); err != nil { + return model.Cluster{}, model.TimedJobSet{}, fmt.Errorf( + "validate cluster file %q against timed jobs file %q: %w", + clusterPath, + jobsPath, + err, + ) + } + return cluster, jobs, nil +} + func requireInputPaths(clusterPath, jobsPath string) error { if clusterPath == "" { return errors.New("--cluster is required") @@ -190,8 +282,18 @@ func validateOutputFormat(value string) error { } } +func validateSimulationPolicy(value string) error { + if value == policyAll { + return nil + } + if _, err := scheduler.ParsePolicy(value); err != nil { + return fmt.Errorf("unsupported policy %q (supported: %s, %s, %s)", value, scheduler.PolicyStrictFIFO, scheduler.PolicyBackfill, policyAll) + } + return nil +} + func writeRootUsage(w io.Writer) error { - if _, err := fmt.Fprintln(w, "Usage: topoqueue [flags]"); err != nil { + if _, err := fmt.Fprintln(w, "Usage: topoqueue [flags]"); err != nil { return err } if _, err := fmt.Fprintln(w, "Run 'topoqueue -h' for command flags."); err != nil { diff --git a/cmd/topoqueue/main_test.go b/cmd/topoqueue/main_test.go index 06640f4..0f03332 100644 --- a/cmd/topoqueue/main_test.go +++ b/cmd/topoqueue/main_test.go @@ -4,12 +4,14 @@ import ( "bytes" "context" "encoding/json" + "errors" "os" "path/filepath" "strings" "testing" "github.com/Rionlyu/topoqueue/internal/model" + "github.com/Rionlyu/topoqueue/internal/output" "github.com/Rionlyu/topoqueue/internal/scheduler" ) @@ -70,6 +72,32 @@ func TestRunExamples(t *testing.T) { t.Errorf("unexpected ordered job decisions: %#v", got.Jobs) } }) + + t.Run("simulate example JSON", func(t *testing.T) { + t.Parallel() + timedJobsPath := filepath.Join("..", "..", "examples", "timed-jobs.yaml") + stdout, stderr, err := execute(t, []string{ + "simulate", "--cluster", clusterPath, "--jobs", timedJobsPath, + "--policy", "all", "--output", "json", + }) + if err != nil { + t.Fatalf("run() error = %v; stderr = %s", err, stderr) + } + var comparison output.SimulationComparisonJSON + if err := json.Unmarshal([]byte(stdout), &comparison); err != nil { + t.Fatalf("decode simulation comparison JSON: %v\n%s", err, stdout) + } + if len(comparison.Policies) != 2 { + t.Fatalf("policy count = %d, want 2", len(comparison.Policies)) + } + backfill, strict := comparison.Policies[0], comparison.Policies[1] + if backfill.Policy != scheduler.PolicyBackfill || backfill.MakespanTicks != 12 || backfill.TotalQueueDelayTicks != 7 || backfill.MaximumQueueDelayTicks != 7 { + t.Errorf("backfill example summary = %#v", backfill) + } + if strict.Policy != scheduler.PolicyStrictFIFO || strict.MakespanTicks != 14 || strict.TotalQueueDelayTicks != 17 || strict.MaximumQueueDelayTicks != 10 { + t.Errorf("strict example summary = %#v", strict) + } + }) } func TestRunRejectsInvalidArguments(t *testing.T) { @@ -83,6 +111,8 @@ func TestRunRejectsInvalidArguments(t *testing.T) { {name: "command required", args: nil, wantErr: "a command is required"}, {name: "schedule cluster required", args: []string{"schedule"}, wantErr: "--cluster is required"}, {name: "compare jobs required", args: []string{"compare", "--cluster", "cluster.yaml"}, wantErr: "--jobs is required"}, + {name: "simulate cluster required", args: []string{"simulate"}, wantErr: "--cluster is required"}, + {name: "simulate jobs required", args: []string{"simulate", "--cluster", "cluster.yaml"}, wantErr: "--jobs is required"}, { name: "unsupported policy", args: []string{"schedule", "--cluster", "cluster.yaml", "--jobs", "jobs.yaml", "--policy", "largest-first"}, @@ -98,6 +128,21 @@ func TestRunRejectsInvalidArguments(t *testing.T) { args: []string{"compare", "--cluster", "cluster.yaml", "--jobs", "jobs.yaml", "--output", "yaml"}, wantErr: `unsupported output "yaml"`, }, + { + name: "unsupported simulation policy", + args: []string{"simulate", "--cluster", "cluster.yaml", "--jobs", "jobs.yaml", "--policy", "largest-first"}, + wantErr: `unsupported policy "largest-first"`, + }, + { + name: "unsupported simulation output", + args: []string{"simulate", "--cluster", "cluster.yaml", "--jobs", "jobs.yaml", "--output", "yaml"}, + wantErr: `unsupported output "yaml"`, + }, + { + name: "simulation positional arguments", + args: []string{"simulate", "workload", "--cluster", "cluster.yaml", "--jobs", "jobs.yaml"}, + wantErr: "unexpected positional arguments", + }, {name: "unknown command", args: []string{"admit"}, wantErr: `unknown command "admit"`}, } @@ -116,6 +161,154 @@ func TestRunRejectsInvalidArguments(t *testing.T) { } } +func TestRunSimulateModes(t *testing.T) { + t.Parallel() + + clusterPath, _ := examplePaths() + timedJobsPath := writeCLIFile(t, "timed-jobs.yaml", canonicalTimedJobsYAML) + + t.Run("default strict FIFO JSON", func(t *testing.T) { + stdout, stderr, err := execute(t, []string{ + "simulate", "--cluster", clusterPath, "--jobs", timedJobsPath, "--output", "json", + }) + if err != nil { + t.Fatalf("run() error = %v; stderr = %s", err, stderr) + } + var result scheduler.SimulationResult + if err := json.Unmarshal([]byte(stdout), &result); err != nil { + t.Fatalf("decode simulation JSON: %v\n%s", err, stdout) + } + if result.Policy != scheduler.PolicyStrictFIFO || result.MakespanTicks != 14 || result.TotalQueueDelayTicks != 17 { + t.Errorf("strict simulation summary = %#v", result) + } + }) + + t.Run("explicit backfill text", func(t *testing.T) { + stdout, stderr, err := execute(t, []string{ + "simulate", "--cluster", clusterPath, "--jobs", timedJobsPath, + "--policy", "backfill", "--output", "text", + }) + if err != nil { + t.Fatalf("run() error = %v; stderr = %s", err, stderr) + } + for _, wanted := range []string{"backfill", "EVENT TIMELINE", "JOB LIFECYCLE", "2.33"} { + if !strings.Contains(stdout, wanted) { + t.Errorf("simulation text missing %q:\n%s", wanted, stdout) + } + } + }) + + t.Run("all policies JSON", func(t *testing.T) { + stdout, stderr, err := execute(t, []string{ + "simulate", "--cluster", clusterPath, "--jobs", timedJobsPath, + "--policy", "all", "--output", "json", + }) + if err != nil { + t.Fatalf("run() error = %v; stderr = %s", err, stderr) + } + var comparison output.SimulationComparisonJSON + if err := json.Unmarshal([]byte(stdout), &comparison); err != nil { + t.Fatalf("decode simulation comparison JSON: %v\n%s", err, stdout) + } + if len(comparison.Policies) != 2 || comparison.Policies[0].Policy != scheduler.PolicyBackfill || comparison.Policies[1].Policy != scheduler.PolicyStrictFIFO { + t.Errorf("comparison policies = %#v", comparison.Policies) + } + }) + + t.Run("all policies text", func(t *testing.T) { + stdout, stderr, err := execute(t, []string{ + "simulate", "--cluster", clusterPath, "--jobs", timedJobsPath, + "--policy", "all", "--output", "text", + }) + if err != nil { + t.Fatalf("run() error = %v; stderr = %s", err, stderr) + } + if strings.Count(stdout, "EVENT TIMELINE") != 2 || !strings.Contains(stdout, "2.33") || !strings.Contains(stdout, "5.67") { + t.Errorf("unexpected all-policy text:\n%s", stdout) + } + }) +} + +func TestRunSimulateHelpAndCancellation(t *testing.T) { + t.Parallel() + + stdout, stderr, err := execute(t, []string{"help"}) + if err != nil || stderr != "" || !strings.Contains(stdout, "simulate") { + t.Fatalf("root help = stdout %q, stderr %q, error %v", stdout, stderr, err) + } + stdout, stderr, err = execute(t, []string{"simulate", "-h"}) + if err != nil || stdout != "" { + t.Fatalf("simulate help = stdout %q, stderr %q, error %v", stdout, stderr, err) + } + for _, flagName := range []string{"-cluster", "-jobs", "-policy", "-output"} { + if !strings.Contains(stderr, flagName) { + t.Errorf("simulate help missing %q:\n%s", flagName, stderr) + } + } + + clusterPath, _ := examplePaths() + timedJobsPath := writeCLIFile(t, "timed-jobs.yaml", canonicalTimedJobsYAML) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + var canceledStdout, canceledStderr bytes.Buffer + err = run(ctx, []string{"simulate", "--cluster", clusterPath, "--jobs", timedJobsPath}, &canceledStdout, &canceledStderr) + if err == nil || !errors.Is(err, context.Canceled) { + t.Fatalf("canceled simulate error = %v, want context.Canceled", err) + } +} + +func TestRunKeepsStaticAndTimedLoadersSeparate(t *testing.T) { + t.Parallel() + + clusterPath, staticJobsPath := examplePaths() + timedJobsPath := writeCLIFile(t, "timed-jobs.yaml", canonicalTimedJobsYAML) + _, _, err := execute(t, []string{ + "schedule", "--cluster", clusterPath, "--jobs", timedJobsPath, + }) + if err == nil || !strings.Contains(err.Error(), "field arrivalTick not found") { + t.Fatalf("schedule timed workload error = %v", err) + } + _, _, err = execute(t, []string{ + "simulate", "--cluster", clusterPath, "--jobs", staticJobsPath, + }) + if err == nil || !strings.Contains(err.Error(), "arrivalTick is required") { + t.Fatalf("simulate static workload error = %v", err) + } + if _, err := scheduler.ParsePolicy("all"); err == nil { + t.Fatal("scheduler.ParsePolicy(all) error = nil, want all to remain CLI-only") + } +} + +func TestRunSimulateReportsTimedTopologyContext(t *testing.T) { + t.Parallel() + + clusterPath := writeCLIFile(t, "cluster.yaml", `nodes: + - name: node-a + topology: + zone: zone-a + capacity: {cpu: 1, gpu: 1} +`) + jobsPath := writeCLIFile(t, "timed-jobs.yaml", `jobs: + - name: training + arrivalTick: 0 + durationTicks: 1 + replicas: 1 + resourcesPerReplica: {cpu: 1, gpu: 1} + requiredTopology: rack +`) + _, _, err := execute(t, []string{ + "simulate", "--cluster", clusterPath, "--jobs", jobsPath, + }) + if err == nil { + t.Fatal("run() error = nil, want missing timed topology error") + } + for _, wanted := range []string{clusterPath, jobsPath, `missing topology key "rack"`, `job "training"`} { + if !strings.Contains(err.Error(), wanted) { + t.Errorf("run() error = %q, want containing %q", err, wanted) + } + } +} + func TestRunRejectsUnknownYAMLField(t *testing.T) { t.Parallel() @@ -170,3 +363,33 @@ func readREADMEExample(t *testing.T) string { } return string(contents[start:start+end]) + "\n" } + +func writeCLIFile(t *testing.T, name, contents string) string { + t.Helper() + path := filepath.Join(t.TempDir(), name) + if err := os.WriteFile(path, []byte(contents), 0o600); err != nil { + t.Fatalf("write CLI fixture: %v", err) + } + return path +} + +const canonicalTimedJobsYAML = `jobs: + - name: warmup + arrivalTick: 0 + durationTicks: 8 + replicas: 8 + resourcesPerReplica: {cpu: 4, gpu: 1} + requiredTopology: zone + - name: train-xl + arrivalTick: 1 + durationTicks: 4 + replicas: 16 + resourcesPerReplica: {cpu: 4, gpu: 1} + requiredTopology: zone + - name: batch + arrivalTick: 2 + durationTicks: 2 + replicas: 4 + resourcesPerReplica: {cpu: 4, gpu: 1} + requiredTopology: rack +` diff --git a/docs/design.md b/docs/design.md index 643af9b..495465e 100644 --- a/docs/design.md +++ b/docs/design.md @@ -1,8 +1,9 @@ # Design -TopoQueue evaluates a fixed cluster snapshot and an ordered job list. It is a -small teaching tool, so the implementation favors explicit state transitions -and deterministic results over scheduler extensibility. +TopoQueue evaluates a fixed cluster snapshot with either an ordered static job +list or a timed workload. It is a small teaching tool, so the implementation +favors explicit state transitions and deterministic results over scheduler +extensibility. ## Simulation execution @@ -17,6 +18,33 @@ uses the same immutable input ordering. Cancellation is propagated with `context.Context`. Policy results are sorted by policy name after all runs complete, so goroutine completion order cannot affect output. +Timed-policy comparison uses the same ownership rule. Each goroutine receives +deep copies of the cluster and timed jobs, and each simulation remains +single-threaded. The CLI-only policy value `all` selects this comparison; it is +not a scheduler policy. + +## Discrete-event lifecycle + +Timed workloads use signed integer logical ticks. The engine visits only a tick +containing an arrival or completion and advances directly to the next such +tick. It uses no wall-clock time, sleeping, timers, or per-job goroutines. + +At a tick, state changes occur in this fixed order: + +1. all finishing jobs complete in original input order and release resources; +2. all arrivals enter the pending queue in original input order; +3. one admission cycle applies the selected queue policy. + +Future arrivals are ordered by `(arrivalTick, inputIndex)`. Running jobs use a +min-heap ordered by `(finishTick, inputIndex)`. These explicit secondary keys +make simultaneous event order independent of map or heap implementation details. + +Admission records the selected topology domain and exact per-node allocation. +Completion releases that allocation with the job's per-replica request; it does +not rerun placement to infer which nodes supplied resources. Release validates +the full allocation before mutation and prevents remaining capacity from +exceeding original node capacity. + ## Deterministic placement Jobs are considered in YAML order. For a topology-constrained job, placement @@ -39,6 +67,30 @@ Backfill records an unplaceable job as pending and continues through the queue. This can admit a smaller later job, but it does not reorder jobs or reconsider earlier decisions. +For timed strict FIFO, each event tick repeatedly admits the pending head until +one placement fails, then retains the entire remaining queue. Timed backfill +scans the pending queue once, admits every job that currently fits, and retains +failures in relative order. A single pass is sufficient because admission only +consumes resources. Pending jobs are retried after later arrivals or completions. + +When there are no future arrivals or running jobs, pending work becomes +terminally unscheduled. Backfill records every job's actual placement rejection +against the fully released cluster. Strict FIFO records the head's actual reason +and marks later jobs as head-of-line blocked by that head. This terminal rule +prevents permanently impossible workloads from looping. + +## Lifecycle metrics + +Event timelines contain only `arrived`, `admitted`, and `completed` transitions. +Job lifecycles remain in input order and end as `completed` or `unscheduled`. +For completed jobs, queue delay is admission tick minus arrival tick. + +The summary start is the earliest processed event tick, the end is the latest, +and makespan is `end - start`. Total, average, and maximum queue delay include +completed jobs only. Empty workloads report zero metrics. Integer totals remain +exact in scheduler results; only text output rounds average delay to two decimal +places. + ## Rejection categories Insufficient capacity means the cluster-wide number of currently feasible @@ -50,11 +102,17 @@ value of the required topology key has enough slots for all replicas. The distinction exposes whether the job is blocked by total remaining resources or by their distribution across topology domains. +Timed terminal reasons are recomputed after all running allocations have been +released. A job that was temporarily short of total capacity can therefore end +with a topology-fragmentation reason if the empty cluster has enough global +slots but no sufficiently large domain. + ## Deliberate non-goals TopoQueue is not a Kubernetes scheduler, a Kueue implementation, a cluster controller, or a production scheduling system. It does not watch live cluster -state, preempt or resize jobs, model failures, provide fairness guarantees, or -implement Kubernetes resource accounting and exact Kueue semantics. The model -contains only integer CPU and GPU capacity and a single optional topology key -per job. +state, preempt or resize jobs, model failures or retries, provide fairness or +quota guarantees, reserve resources, persist state, or implement Kubernetes +resource accounting and exact Kueue semantics. The model contains only integer +CPU and GPU capacity, identical fixed-duration replicas, and a single optional +topology key per job. diff --git a/docs/event-driven-simulation.md b/docs/event-driven-simulation.md new file mode 100644 index 0000000..2d91269 --- /dev/null +++ b/docs/event-driven-simulation.md @@ -0,0 +1,186 @@ +# Event-driven simulation + +TopoQueue's `simulate` command models job lifecycles on a fixed cluster using +logical integer ticks. It reuses the same deterministic, topology-aware +placement rules as `schedule`, but adds arrivals, queueing, fixed-duration +execution, completion, and resource release. + +```sh +topoqueue simulate \ + --cluster examples/cluster.yaml \ + --jobs examples/timed-jobs.yaml \ + --policy all \ + --output text +``` + +The supported policies are `strict-fifo`, `backfill`, and the CLI-only value +`all`. The default is `strict-fifo`. With `all`, TopoQueue evaluates both real +policies on independent state and orders their results by policy name. + +## Timed workload input + +Timed workloads use a separate strict loader and do not change the meaning of +the static jobs accepted by `schedule` and `compare`: + +```yaml +jobs: + - name: warmup + arrivalTick: 0 + durationTicks: 8 + replicas: 8 + resourcesPerReplica: + cpu: 4 + gpu: 1 + requiredTopology: zone +``` + +`arrivalTick` is required and must be non-negative. `durationTicks` is required +and must be greater than zero. The existing job rules still apply: names are +non-empty and unique, replicas are positive, CPU and GPU requests are +non-negative with at least one positive resource, and checked resource totals +must fit in a signed 64-bit integer. Every node must have a non-empty value for +each topology key required by the workload. + +Decoding rejects unknown fields, missing top-level `jobs`, null collections, +and multiple YAML documents. Static `LoadJobs` continues to reject timed-only +fields such as `arrivalTick` and `durationTicks`. + +## Logical ticks and event ordering + +Ticks are discrete values, not wall-clock time. The simulator uses no timers, +sleeps, randomness, or goroutines per job. It advances directly to the next +arrival or completion and does not poll intermediate ticks. + +At each event tick, processing order is exactly: + +1. Complete all jobs finishing at that tick and release their resources. +2. Enqueue all jobs arriving at that tick. +3. Run one admission cycle for the selected policy. + +Simultaneous completions are ordered by original workload index. Simultaneous +arrivals also preserve original input order. Running jobs are ordered by +`(finishTick, originalInputIndex)`, so heap behavior cannot change the emitted +timeline. Completion before arrival means a newly arriving job can immediately +use resources released at the same tick. + +The timeline emits `arrived`, `admitted`, and `completed` events. Admission +events also contain the selected topology domain and the exact node allocation. + +## Queue policies over time + +Strict FIFO examines the pending head. If it fits, it is admitted and the new +head is examined. If it does not fit, the admission cycle stops immediately; +the head and every later job retain their queue positions. They are reconsidered +at the next arrival or completion event. + +Backfill scans the pending queue once in order. Every job that currently fits +is admitted, while failures remain pending in their existing relative order. +The scan continues after a failure. One pass is sufficient because admission +only consumes resources and cannot make another job newly schedulable. + +Neither policy reserves capacity for a pending job. Backfill can therefore +reduce idle capacity and queue delay, but it provides no fairness guarantee. + +## Completion, release, and terminal jobs + +Admission records how many replicas were placed on each node. Completion +releases exactly `replicas * resourcesPerReplica` for each recorded allocation, +not merely an aggregate estimate. A release is validated in full before any +node is changed: nodes must be known and unique, replica accounting must be +exact, arithmetic must be safe, and remaining resources may not exceed the +node's original capacity. + +The simulation terminates when there are no future arrivals and no running +jobs. Any jobs still pending become `unscheduled`; an impossible job cannot +cause an infinite loop. + +- Backfill gives every remaining job its actual final placement rejection + reason against the fully released cluster. +- Strict FIFO gives the pending head its actual final reason. Every later job + receives `head_of_line_blocked`, naming that head job. + +Final reasons distinguish cluster-wide insufficient capacity from topology +fragmentation, just as static placement does. + +## Results and metrics + +A policy result contains three views of the same run: + +- an ordered event timeline; +- one terminal lifecycle per job in original input order; +- aggregate policy metrics. + +A completed lifecycle contains arrival, duration, start, completion, queue +delay, topology domain, allocation, and requested resources. An unscheduled +lifecycle has no start, completion, or queue-delay tick and includes its final +rejection reason. Pointer-valued tick fields in JSON preserve the distinction +between an absent value and the meaningful tick zero. + +Metrics are defined as follows: + +- `startTick` is the earliest processed event tick. +- `endTick` is the last processed event tick. +- `makespanTicks = endTick - startTick`. +- A completed job's `queueDelayTicks = startTick - arrivalTick`. +- `totalQueueDelayTicks` is the sum of queue delays for completed jobs only. +- `averageQueueDelayTicks` is that exact total divided by completed job count; + it is zero when no job completes. +- `maximumQueueDelayTicks` is the largest completed-job delay and is zero when + no job completes. + +Unscheduled jobs do not contribute to wait aggregates. Empty workloads have +zero counts and metrics and empty event and lifecycle lists. Integer source +metrics remain exact; only human-readable average display is rounded. + +Text output presents a summary, timeline, and lifecycle table. JSON uses the +same declared result structure, is indented, and ends with a newline. A policy +comparison first shows comparative completion and wait metrics, then each +policy's full details in deterministic policy-name order. + +## Determinism + +Identical inputs produce identical decisions and byte-stable output. TopoQueue +uses original input indexes for event ties, lexical names for deterministic +topology and node ties, sorted allocations, ordered lifecycle slices, and +independent mutable state for concurrent policy runs. Map iteration and +goroutine completion order do not determine output order. + +## Complexity + +Let `J` be the number of jobs, `N` the number of nodes, `D` the number of +topology domains, and `P = O(N log N + D log D)` the worst-case cost of one +placement attempt. There are at most `2J` distinct arrival/completion event +ticks. + +Strict FIFO performs each successful admission once and at most one failed +head attempt per event tick, for `O(J * P + J log J)` worst-case time including +arrival sorting and the running-job heap. Backfill may rescan up to `J` pending +jobs at each event tick, giving `O(J^2 * P + J log J)` worst-case time. State, +results, heap entries, and placement scratch space require `O(N + J + D)` +space. Comparing both policies concurrently changes constant factors but not +these asymptotic bounds. + +## Cancellation and checked arithmetic + +Simulation and comparison accept `context.Context` and return contextual +cancellation errors at checked work boundaries. Independent comparison runs +are canceled together and do not share mutable state. + +Input validation checks resource multiplication and aggregate cluster +capacity. At admission, `startTick + durationTicks` is checked before a running +job is recorded; a late job that cannot fit does not fail merely because it +would overflow if admitted. Aggregate queue-delay addition and resource release +are also checked. Arithmetic errors stop the run rather than wrapping values. + +## Scope and limitations + +TopoQueue is an educational simulator over a supplied snapshot. It is not a +Kubernetes scheduler, does not implement Kueue semantics, and is not intended +for production scheduling. Resources are whole integer CPU and GPU units; +replicas are identical; each job has one fixed duration and at most one required +topology key. + +The simulator deliberately does not implement priorities, preemption, elastic +jobs, execution failures or restart policies, multiple queues, quotas, fair +sharing, reservations, live cluster watches, controllers, HTTP services, +persistence, databases, a UI, or real-time execution. diff --git a/docs/exec-plans/v0.2-event-driven-simulation.md b/docs/exec-plans/v0.2-event-driven-simulation.md new file mode 100644 index 0000000..8a3513e --- /dev/null +++ b/docs/exec-plans/v0.2-event-driven-simulation.md @@ -0,0 +1,304 @@ +# TopoQueue v0.2 event-driven simulation ExecPlan + +This document is the living execution plan for the v0.2 event-driven workload +simulation milestone. Update the progress, decisions, validation evidence, and +known limitations as implementation proceeds. + +## Current architecture + +TopoQueue currently evaluates a static `model.Cluster` and ordered +`model.JobSet`. The model package strictly decodes YAML and validates names, +resources, overflow, and required topology labels. The scheduler owns a sorted +slice of mutable node state; `placeJob` computes a complete deterministic +candidate before consuming capacity. `Schedule` applies either strict FIFO or +backfill once, while `Compare` runs isolated policy copies concurrently and +sorts results. The output package has separate stable text and JSON renderers, +and the standard-library CLI wires `schedule` and `compare` without a command +framework. + +The event simulator will reuse placement without changing static command +semantics. Timed inputs, lifecycle results, simulation rendering, and CLI +wiring remain separate so existing YAML, text, and JSON contracts stay intact. + +## Goals + +- Load and validate a strict timed workload with required arrival and duration + fields. +- Simulate arrival, queueing, topology-aware admission, fixed-duration + execution, completion, and exact resource release using logical ticks. +- Retry pending work at arrival and completion event ticks with strict FIFO or + backfill semantics. +- Terminate impossible workloads with deterministic final rejection reasons. +- Expose ordered events, input-ordered job lifecycles, and exact queue-delay + metrics through stable text and JSON. +- Compare timed policies concurrently with isolated mutable state. +- Preserve all existing `schedule` and `compare` behavior byte for byte. + +## Explicit non-goals + +This milestone does not add wall-clock execution, priorities, preemption, +failure retries, failures, elastic jobs, multiple queues, quotas, fair sharing, +reservations, live watches, servers, persistence, databases, a UI, or release +automation. It does not implement Kubernetes or Kueue semantics and is not a +production scheduler. No new third-party dependency is planned. + +## Proposed types and data flow + +1. `internal/model` adds value-based `TimedJob` and `TimedJobSet` types. A + private strict YAML wire type uses pointer timing fields so a missing + `arrivalTick` is distinguishable from a valid zero. Conversion to `JobSet` + reuses static validation and topology checks. Clone helpers isolate policy + runs. +2. Scheduler node state records both original and remaining capacity. A checked + release helper consumes the recorded allocation and originating job request, + rejects unknown/duplicate nodes and replica mismatches, and prevents capacity + from exceeding the original snapshot. +3. Exported simulation result types describe `arrived`, `admitted`, and + `completed` events; completed or unscheduled job lifecycles; and aggregate + metrics. Optional lifecycle ticks use pointers because tick zero is valid. +4. `Simulate` validates and clones inputs, sorts arrivals by + `(arrivalTick,inputIndex)`, maintains pending input indexes, and stores + running jobs in a min-heap ordered by `(finishTick,inputIndex)`. +5. At admission, the existing `placeJob` selects and records the exact domain + and per-node allocation. The finish tick is checked for signed 64-bit + overflow. At completion, release uses that recorded allocation rather than + recomputing placement. +6. `CompareSimulations` mirrors static comparison: two goroutines receive deep + copies, cancellation propagates, all outcomes are collected, and results are + sorted by policy name. +7. New output functions render one result or a sorted comparison. `simulate` + loads the separate timed input and accepts scheduler policies or the CLI-only + value `all`. + +## Event-ordering semantics + +Only ticks containing an arrival or completion are processed. At one tick: + +1. Pop and complete every running job finishing now, ordered by original input + index. Release its exact allocation before emitting `completed`. +2. Enqueue all arrivals now in original input order and emit `arrived`. +3. Run one policy admission cycle and emit each `admitted` event in examination + order. +4. Jump directly to the earlier of the next arrival and next completion. + +Strict FIFO repeatedly admits the pending head until the first failed +placement, then stops that cycle without changing pending order. Backfill scans +the pending queue once, admits every currently fitting job, and retains failures +in relative order. Admission only consumes capacity, so a second scan at the +same tick cannot create a new fit. + +When no future arrival and no running job remains, pending work is terminally +unscheduled. Backfill records each job's actual empty-cluster placement reason. +Strict FIFO records the head's actual reason and marks all later jobs +`head_of_line_blocked` by that head. The event stream contains no synthetic +unscheduled event. + +## Metrics + +- `startTick` is the earliest processed event tick; `endTick` is the latest. +- Empty input has zero for every metric. +- `makespanTicks = endTick - startTick`. +- Queue delay is `startTick - arrivalTick` for completed jobs. +- Total and maximum queue delay are exact integers over completed jobs. +- Average queue delay is derived from exact total/count values and is rounded + only by the human-readable renderer. + +## Milestones and acceptance criteria + +1. **Plan and baseline**: guidance and this plan committed; baseline commands + green; no feature code present. +2. **Timed model**: strict loader, missing-field detection, cloning, validation, + conversion, and model tests pass without changing `LoadJobs`. +3. **Reversible placement and result contract**: checked release invariants and + stable lifecycle types have focused tests; static scheduler remains green. +4. **Event engine and policies**: exact ordering, FIFO/backfill retries, + overflow/cancellation, terminalization, immutability, and canonical metrics + pass scheduler tests and benchmark coverage exists. +5. **Concurrent comparison**: isolated policy runs match independent results, + sort by name, propagate cancellation, and pass race tests. +6. **Output and CLI**: deterministic text/JSON round-trip, `simulate` flags, + help, validation, and both single/all modes pass focused tests. +7. **Example and documentation**: checked-in timed workload and + `demo-simulate` produce required metrics; README/design/user guide are + accurate and retain educational non-goals. +8. **Hardening and final review**: all 24 required categories, repeated package + tests, full validation, triple-render byte comparison, diff review, and clean + signed history succeed. + +## Progress checklist + +- [x] Confirmed clean `feat/event-driven-simulation` branch before feature work. +- [x] Read repository guidance and frozen v0.2 specification. +- [x] Reviewed repository architecture, examples, tests, benchmark, and CI. +- [x] Ran the pre-feature baseline successfully. +- [x] Created the living ExecPlan. +- [x] Add strict timed input model and tests. +- [x] Add checked reversible allocations and lifecycle result types. +- [x] Implement deterministic event processing and timed policies. +- [x] Add concurrent timed-policy comparison and benchmark. +- [x] Add stable simulation text and JSON output. +- [x] Wire the `simulate` CLI and validate flags/output modes. +- [x] Add the timed example and `demo-simulate` target. +- [x] Document semantics, complexity, metrics, and limitations. +- [x] Complete edge-case hardening and all final validation. + +## Decision log + +- **2026-07-13 — Separate timed contract.** Timed YAML and simulation output + receive new types and functions. Static model and output types remain + unchanged to protect strict loading and serialized compatibility. +- **2026-07-13 — Required timing fields via wire pointers.** Public timed values + use `int64`, while the loader decodes pointer timing fields privately to reject + missing values without rejecting a valid arrival at tick zero. +- **2026-07-13 — Input indexes as scheduler identity.** Queue, heap, event tie + breaking, and lifecycle assembly use the original index. Public job names + remain unique through validation. +- **2026-07-13 — Recorded release.** Completion releases the stored per-node + allocation with the original per-replica request; placement is never rerun to + infer released resources. +- **2026-07-13 — Direct result aggregates.** Simulation metrics live alongside + ordered events and jobs, following the existing `PolicyResult` style. Integer + totals/counts remain the source of truth for the derived average. +- **2026-07-13 — No terminal event.** Terminal rejection appears in lifecycle + results only because the specified timeline event set contains exactly + arrived, admitted, and completed. +- **2026-07-13 — Valid maximum tick.** Next-event selection carries an explicit + presence boolean instead of using `math.MaxInt64` as a sentinel, because the + maximum signed tick is a valid arrival or completion. +- **2026-07-13 — Admission-time overflow rollback.** Finish overflow is checked + only after placement succeeds. The just-recorded allocation is released + transactionally before returning the error, so an impossible late job is not + rejected spuriously for a finish it would never reach. +- **2026-07-13 — User-authorized branch pushes.** The frozen prompt initially + prohibited remote operations. During implementation, the user explicitly + overrode that restriction and requested pushes. Only the feature branch is + pushed; no tag, release, merge, or pull request is created. +- **2026-07-13 — Deterministic comparison error precedence.** Concurrent + outcomes are always drained and errors use the same policy-name order as + successful results. A non-cancellation backfill error may cancel strict FIFO + because backfill already has precedence; a strict-FIFO error cannot cancel a + still-running backfill simulation and mask its higher-precedence error. + +## Validation log + +- **2026-07-13 baseline on `main` before branch creation** + - `git pull --ff-only`: up to date. + - `go mod download`: succeeded; its extra unused checksum was removed so the + feature branch started clean. + - `make fmt-check`: passed. + - `make vet`: passed. + - `make test`: passed with the race detector. + - `make build`: passed. + - `make demo`: passed with the checked-in README output. + - `make benchmark`: passed (`BenchmarkScheduleThousandJobs`). +- **2026-07-13 plan milestone** + - `go test ./...`: passed before the initial documentation commit. +- **2026-07-13 timed input milestone** + - `go test ./internal/model`: passed. + - `go test ./...`: passed. + - Strict loader coverage includes required timing fields, explicit empty input, + invalid timing, duplicate names, unknown fields, multiple documents, static + loader separation, ordering, cloning, and conversion independence. +- **2026-07-13 reversible allocation and result-contract milestone** + - `go test ./internal/scheduler`: passed. + - `go test ./...`: passed. + - `make test`: passed with the race detector. + - Checked release tests cover exact multi-node restoration, single-resource + jobs, invalid allocations, overflow, over-release, and transactional failure. +- **2026-07-13 deterministic event-engine milestone** + - `go test ./internal/scheduler`: passed. + - `go test ./...`: passed. + - `go vet ./...`: passed. + - `make test`: passed with the race detector. + - Core lifecycle tests cover canonical FIFO/backfill metrics, exact event + order, simultaneous ties, terminal policy behavior, empty input, maximum + tick handling, admission-time overflow, input immutability, and 50 repeated + deterministic runs. +- **2026-07-13 concurrent comparison milestone** + - `go test ./internal/scheduler`: passed. + - `go test ./...`: passed. + - `go test -race ./...`: passed. + - `BenchmarkSimulateThousandTimedJobs`: completed successfully with a direct, + deterministic workload generator. + - Comparison tests match independent runs, retain sorted policy order, leave + inputs unchanged, repeat deterministically, and propagate cancellation. +- **2026-07-13 simulation output milestone** + - `go test ./internal/output`: passed. + - `go test ./...`: passed. + - `go test -race ./internal/output`: passed. + - `make fmt-check` and `go vet ./...`: passed. + - Text and JSON tests render three times byte-for-byte, round-trip declared + structures, sort reversed policy input, preserve caller-owned slices, and + display canonical averages as 5.67 and 2.33. +- **2026-07-13 simulate CLI milestone** + - `go test ./cmd/topoqueue`: passed. + - `go test ./...`: passed. + - `go test -race ./cmd/topoqueue`: passed. + - `go vet ./...` and `git diff --check`: passed. + - CLI tests cover default and explicit policies, CLI-only `all`, both output + formats, help, path/flag validation, cancellation, policy ordering, and + strict separation between static and timed workload loaders. +- **2026-07-13 lifecycle demo milestone** + - `go test ./cmd/topoqueue` and `go test ./...`: passed. + - `make demo-simulate`: passed. + - Checked-in example assertions confirm backfill makespan/wait/max of 12/7/7 + and strict-FIFO values of 14/17/10, with averages displayed as 2.33/5.67. +- **2026-07-13 documentation milestone** + - `go test ./cmd/topoqueue` and `go test ./...`: passed, including the + byte-for-byte static README demo assertion. + - README documents `simulate`, the canonical policy comparison, complexity, + development commands, and current limitations without changing the static + demo block. + - Design documentation records event ordering, ownership, exact release, + terminalization, final reasons, and lifecycle metric definitions. + - `docs/event-driven-simulation.md` documents the strict input contract, + output model, determinism, complexity, checked arithmetic, and non-goals. +- **2026-07-13 lifecycle hardening milestone** + - Independent scheduler review found no high-severity ordering or release + defects and identified two P2 defensive gaps: final-work cancellation and + full-capacity verification on all-completed runs. Both are fixed. + - Added discriminating coverage for combined same-tick releases, admission + event allocations, final-reason recomputation after release, aggregate + queue-delay overflow, active-work cancellation, direct timed validation, + successful terminal capacity, and CLI topology error context. + - `make fmt-check`, `go vet ./...`, `go test ./...`, and race-tested + `make test`: passed. +- **2026-07-13 independent review and deterministic comparison fix** + - Parallel review and the first `codex review --base main` found one P2 + defect: canceling both timed policy runs on the first intrinsic error made + the reported policy depend on goroutine completion order. + - Comparison now drains both outcomes, selects errors by sorted policy name, + and retains only priority-safe sibling cancellation. The committed + regression repeats a dual finish-overflow comparison 1,000 times. + - Focused final-patch reviews found no remaining correctness, determinism, + cancellation, or goroutine-leak defect. A second full + `codex review --base main` found no actionable defect. + - `go test ./internal/scheduler -run '^TestCompareSimulations' -count=20` and + `go test -race ./internal/scheduler -run '^TestCompareSimulations' -count=10`: + passed. +- **2026-07-13 final validation** + - `make fmt-check`, `make vet`, `make test`, and `make build`: passed; the + full suite ran under the race detector. + - `make demo` and `make demo-simulate`: passed. The timed demo reports + backfill makespan/total-wait/max-wait `12/7/7` and strict FIFO + `14/17/10`, with averages `2.33` and `5.67`. + - `make benchmark`: passed for both the static and 1,000-job timed + benchmarks. + - The required four-package test set passed 20 consecutive repetitions. + - `git diff --check main...HEAD` and the uncommitted-diff whitespace check: + passed. + - Three independently written text files compared byte-for-byte and shared + SHA-256 `482a3d79eebea19edde12fd714c9855dc58eff77829c3f0f1a3dc5f17fd2728d`. + Three JSON files also compared byte-for-byte, passed `jq`, and shared + SHA-256 `22be7435aa74213a95a5612a669439434e59131b6430b71b4434c9ce327a7132`. + +## Known limitations + +- Logical ticks are signed 64-bit integers and durations must be positive. +- Resources remain whole CPU/GPU units with identical replicas and one optional + required topology key. +- Backfill has no reservation guarantee for the queue head and may delay it by + admitting later jobs, matching the deliberately simple policy definition. +- Cancellation returns an error rather than a partial simulation result. +- Results model successful fixed-duration execution only; failures and retries + are outside this milestone. diff --git a/examples/timed-jobs.yaml b/examples/timed-jobs.yaml new file mode 100644 index 0000000..2bad159 --- /dev/null +++ b/examples/timed-jobs.yaml @@ -0,0 +1,27 @@ +jobs: + - name: warmup + arrivalTick: 0 + durationTicks: 8 + replicas: 8 + resourcesPerReplica: + cpu: 4 + gpu: 1 + requiredTopology: zone + + - name: train-xl + arrivalTick: 1 + durationTicks: 4 + replicas: 16 + resourcesPerReplica: + cpu: 4 + gpu: 1 + requiredTopology: zone + + - name: batch + arrivalTick: 2 + durationTicks: 2 + replicas: 4 + resourcesPerReplica: + cpu: 4 + gpu: 1 + requiredTopology: rack diff --git a/internal/model/load.go b/internal/model/load.go index 4f46e49..e899972 100644 --- a/internal/model/load.go +++ b/internal/model/load.go @@ -40,6 +40,54 @@ func LoadJobs(path string) (JobSet, error) { return jobs, nil } +// LoadTimedJobs reads, strictly decodes, and validates a timed jobs YAML file. +func LoadTimedJobs(path string) (TimedJobSet, error) { + var document timedJobDocument + if err := loadYAML(path, "timed jobs", &document); err != nil { + return TimedJobSet{}, err + } + if document.Jobs == nil { + return TimedJobSet{}, fmt.Errorf("validate timed jobs file %q: top-level field %q is required and must be a sequence", path, "jobs") + } + + jobs := TimedJobSet{Jobs: make([]TimedJob, len(document.Jobs))} + for index, input := range document.Jobs { + if input.ArrivalTick == nil { + return TimedJobSet{}, fmt.Errorf("validate timed jobs file %q: %s: arrivalTick is required", path, jobDescription(input.Job, index)) + } + if input.DurationTicks == nil { + return TimedJobSet{}, fmt.Errorf("validate timed jobs file %q: %s: durationTicks is required", path, jobDescription(input.Job, index)) + } + jobs.Jobs[index] = TimedJob{ + Job: input.Job, + ArrivalTick: *input.ArrivalTick, + DurationTicks: *input.DurationTicks, + } + } + + if err := ValidateTimedJobs(jobs); err != nil { + return TimedJobSet{}, fmt.Errorf("validate timed jobs file %q: %w", path, err) + } + return jobs, nil +} + +type timedJobDocument struct { + Jobs []timedJobInput `yaml:"jobs"` +} + +type timedJobInput struct { + Job `yaml:",inline"` + ArrivalTick *int64 `yaml:"arrivalTick"` + DurationTicks *int64 `yaml:"durationTicks"` +} + +func jobDescription(job Job, index int) string { + if job.Name == "" { + return fmt.Sprintf("job at index %d", index) + } + return fmt.Sprintf("job %q", job.Name) +} + func loadYAML(path, kind string, destination any) error { contents, err := os.ReadFile(path) if err != nil { diff --git a/internal/model/model.go b/internal/model/model.go index b6fbd4c..e7b0202 100644 --- a/internal/model/model.go +++ b/internal/model/model.go @@ -24,6 +24,20 @@ type JobSet struct { Jobs []Job `yaml:"jobs" json:"jobs"` } +// TimedJob describes a job together with its logical arrival and execution +// duration. Job is embedded so timed workload files retain the static job +// shape while using a separate loader and model. +type TimedJob struct { + Job `yaml:",inline"` + ArrivalTick int64 `yaml:"arrivalTick" json:"arrivalTick"` + DurationTicks int64 `yaml:"durationTicks" json:"durationTicks"` +} + +// TimedJobSet preserves the input order from a timed jobs file. +type TimedJobSet struct { + Jobs []TimedJob `yaml:"jobs" json:"jobs"` +} + // Job describes a set of identical replicas that must share a topology // domain when RequiredTopology is set. type Job struct { @@ -54,3 +68,19 @@ func (j JobSet) Clone() JobSet { copy(clone.Jobs, j.Jobs) return clone } + +// Clone returns a deep copy of the timed job set. +func (j TimedJobSet) Clone() TimedJobSet { + clone := TimedJobSet{Jobs: make([]TimedJob, len(j.Jobs))} + copy(clone.Jobs, j.Jobs) + return clone +} + +// StaticJobs returns a copy of the timed jobs without lifecycle timing. +func (j TimedJobSet) StaticJobs() JobSet { + jobs := JobSet{Jobs: make([]Job, len(j.Jobs))} + for index, timedJob := range j.Jobs { + jobs.Jobs[index] = timedJob.Job + } + return jobs +} diff --git a/internal/model/timed_jobs_test.go b/internal/model/timed_jobs_test.go new file mode 100644 index 0000000..1d536a6 --- /dev/null +++ b/internal/model/timed_jobs_test.go @@ -0,0 +1,290 @@ +package model_test + +import ( + "reflect" + "strings" + "testing" + + "github.com/Rionlyu/topoqueue/internal/model" +) + +func TestLoadTimedJobsPreservesOrder(t *testing.T) { + t.Parallel() + + path := writeTempFile(t, "timed-jobs.yaml", `jobs: + - name: first + arrivalTick: 0 + durationTicks: 8 + replicas: 2 + resourcesPerReplica: + cpu: 1 + gpu: 0 + - name: second + arrivalTick: 3 + durationTicks: 2 + replicas: 1 + resourcesPerReplica: + cpu: 0 + gpu: 1 + requiredTopology: rack +`) + + want := model.TimedJobSet{Jobs: []model.TimedJob{ + { + Job: model.Job{ + Name: "first", Replicas: 2, + ResourcesPerReplica: model.Resources{CPU: 1}, + }, + ArrivalTick: 0, DurationTicks: 8, + }, + { + Job: model.Job{ + Name: "second", Replicas: 1, + ResourcesPerReplica: model.Resources{GPU: 1}, + RequiredTopology: "rack", + }, + ArrivalTick: 3, DurationTicks: 2, + }, + }} + + got, err := model.LoadTimedJobs(path) + if err != nil { + t.Fatalf("LoadTimedJobs() error = %v", err) + } + if !reflect.DeepEqual(got, want) { + t.Errorf("LoadTimedJobs() = %#v, want %#v", got, want) + } +} + +func TestLoadTimedJobsAcceptsExplicitEmptySequence(t *testing.T) { + t.Parallel() + + path := writeTempFile(t, "timed-jobs.yaml", "jobs: []\n") + got, err := model.LoadTimedJobs(path) + if err != nil { + t.Fatalf("LoadTimedJobs() error = %v", err) + } + if !reflect.DeepEqual(got, model.TimedJobSet{Jobs: []model.TimedJob{}}) { + t.Errorf("LoadTimedJobs() = %#v, want explicit empty job sequence", got) + } +} + +func TestLoadTimedJobsRejectsInvalidInput(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + contents string + wantErrText string + }{ + { + name: "missing jobs field", + contents: "{}\n", + wantErrText: `top-level field "jobs" is required and must be a sequence`, + }, + { + name: "null jobs field", + contents: "jobs: null\n", + wantErrText: `top-level field "jobs" is required and must be a sequence`, + }, + { + name: "missing arrival tick", + contents: `jobs: + - name: job-a + durationTicks: 1 + replicas: 1 + resourcesPerReplica: {cpu: 1, gpu: 0} +`, + wantErrText: "arrivalTick is required", + }, + { + name: "missing duration", + contents: `jobs: + - name: job-a + arrivalTick: 0 + replicas: 1 + resourcesPerReplica: {cpu: 1, gpu: 0} +`, + wantErrText: "durationTicks is required", + }, + { + name: "negative arrival", + contents: `jobs: + - name: job-a + arrivalTick: -1 + durationTicks: 1 + replicas: 1 + resourcesPerReplica: {cpu: 1, gpu: 0} +`, + wantErrText: "arrivalTick must be non-negative", + }, + { + name: "zero duration", + contents: `jobs: + - name: job-a + arrivalTick: 0 + durationTicks: 0 + replicas: 1 + resourcesPerReplica: {cpu: 1, gpu: 0} +`, + wantErrText: "durationTicks must be greater than zero", + }, + { + name: "negative duration", + contents: `jobs: + - name: job-a + arrivalTick: 0 + durationTicks: -1 + replicas: 1 + resourcesPerReplica: {cpu: 1, gpu: 0} +`, + wantErrText: "durationTicks must be greater than zero", + }, + { + name: "duplicate names", + contents: `jobs: + - name: duplicate + arrivalTick: 0 + durationTicks: 1 + replicas: 1 + resourcesPerReplica: {cpu: 1, gpu: 0} + - name: duplicate + arrivalTick: 1 + durationTicks: 1 + replicas: 1 + resourcesPerReplica: {cpu: 1, gpu: 0} +`, + wantErrText: `job "duplicate" at index 1: duplicate name`, + }, + { + name: "unknown root field", + contents: `jobs: [] +version: 1 +`, + wantErrText: "field version not found", + }, + { + name: "unknown job field", + contents: `jobs: + - name: job-a + arrivalTick: 0 + durationTicks: 1 + replicas: 1 + priority: 10 + resourcesPerReplica: {cpu: 1, gpu: 0} +`, + wantErrText: "field priority not found", + }, + { + name: "unknown resource field", + contents: `jobs: + - name: job-a + arrivalTick: 0 + durationTicks: 1 + replicas: 1 + resourcesPerReplica: + cpu: 1 + gpu: 0 + memory: 8 +`, + wantErrText: "field memory not found", + }, + { + name: "multiple documents", + contents: `jobs: [] +--- +jobs: [] +`, + wantErrText: "multiple YAML documents", + }, + } + + for _, test := range tests { + test := test + t.Run(test.name, func(t *testing.T) { + t.Parallel() + path := writeTempFile(t, "timed-jobs.yaml", test.contents) + _, err := model.LoadTimedJobs(path) + if err == nil { + t.Fatalf("LoadTimedJobs() error = nil, want containing %q", test.wantErrText) + } + if !strings.Contains(err.Error(), path) { + t.Errorf("LoadTimedJobs() error = %q, want file path %q", err, path) + } + if !strings.Contains(err.Error(), test.wantErrText) { + t.Errorf("LoadTimedJobs() error = %q, want containing %q", err, test.wantErrText) + } + }) + } +} + +func TestValidateTimedJobsReusesStaticValidation(t *testing.T) { + t.Parallel() + + jobs := model.TimedJobSet{Jobs: []model.TimedJob{ + { + Job: model.Job{ + Name: "job-a", Replicas: 1, + ResourcesPerReplica: model.Resources{CPU: 1}, + }, + DurationTicks: 1, + }, + { + Job: model.Job{ + Name: "job-a", Replicas: 1, + ResourcesPerReplica: model.Resources{CPU: 1}, + }, + DurationTicks: 1, + }, + }} + + err := model.ValidateTimedJobs(jobs) + if err == nil || !strings.Contains(err.Error(), "duplicate name") { + t.Fatalf("ValidateTimedJobs() error = %v, want duplicate-name error", err) + } +} + +func TestStaticLoadJobsRejectsTimedFields(t *testing.T) { + t.Parallel() + + path := writeTempFile(t, "jobs.yaml", `jobs: + - name: job-a + arrivalTick: 0 + durationTicks: 1 + replicas: 1 + resourcesPerReplica: {cpu: 1, gpu: 0} +`) + _, err := model.LoadJobs(path) + if err == nil || !strings.Contains(err.Error(), "field arrivalTick not found") { + t.Fatalf("LoadJobs() error = %v, want rejection of timed fields", err) + } +} + +func TestTimedJobSetCloneAndConversionAreIndependent(t *testing.T) { + t.Parallel() + + original := model.TimedJobSet{Jobs: []model.TimedJob{{ + Job: model.Job{ + Name: "job-a", Replicas: 1, + ResourcesPerReplica: model.Resources{CPU: 2}, + }, + ArrivalTick: 1, DurationTicks: 3, + }}} + + clone := original.Clone() + static := original.StaticJobs() + clone.Jobs[0].Name = "clone" + clone.Jobs[0].ResourcesPerReplica.CPU = 4 + static.Jobs[0].Name = "static" + static.Jobs[0].ResourcesPerReplica.CPU = 8 + + if original.Jobs[0].Name != "job-a" || original.Jobs[0].ResourcesPerReplica.CPU != 2 { + t.Fatalf("copy mutation changed original: %#v", original) + } + if clone.Jobs[0].Name != "clone" || clone.Jobs[0].ResourcesPerReplica.CPU != 4 { + t.Errorf("clone mutation was not retained: %#v", clone) + } + if static.Jobs[0].Name != "static" || static.Jobs[0].ResourcesPerReplica.CPU != 8 { + t.Errorf("converted job mutation was not retained: %#v", static) + } +} diff --git a/internal/model/validation.go b/internal/model/validation.go index 01eeece..cc774c4 100644 --- a/internal/model/validation.go +++ b/internal/model/validation.go @@ -72,6 +72,25 @@ func ValidateJobs(jobSet JobSet) error { return nil } +// ValidateTimedJobs checks the static job fields and lifecycle timing in input +// order. Required-field presence is enforced by LoadTimedJobs because zero is +// a valid arrival tick in the public value model. +func ValidateTimedJobs(jobSet TimedJobSet) error { + if err := ValidateJobs(jobSet.StaticJobs()); err != nil { + return err + } + + for _, job := range jobSet.Jobs { + if job.ArrivalTick < 0 { + return fmt.Errorf("job %q: arrivalTick must be non-negative, got %d", job.Name, job.ArrivalTick) + } + if job.DurationTicks <= 0 { + return fmt.Errorf("job %q: durationTicks must be greater than zero, got %d", job.Name, job.DurationTicks) + } + } + return nil +} + // ValidateTopologyRequirements verifies that every node has every topology key // required by the jobs. func ValidateTopologyRequirements(cluster Cluster, jobSet JobSet) error { diff --git a/internal/output/simulation_json.go b/internal/output/simulation_json.go new file mode 100644 index 0000000..2569a02 --- /dev/null +++ b/internal/output/simulation_json.go @@ -0,0 +1,33 @@ +package output + +import ( + "io" + "sort" + + "github.com/Rionlyu/topoqueue/internal/scheduler" +) + +// SimulationComparisonJSON is the stable top-level JSON representation of a +// timed policy comparison. +type SimulationComparisonJSON struct { + Policies []scheduler.SimulationResult `json:"policies"` +} + +// WriteSimulationJSON writes one timed policy simulation as indented JSON. +func WriteSimulationJSON(w io.Writer, result scheduler.SimulationResult) error { + return writeJSON(w, result) +} + +// WriteSimulationComparisonJSON writes policy-sorted timed simulations as +// indented JSON. +func WriteSimulationComparisonJSON(w io.Writer, results []scheduler.SimulationResult) error { + return writeJSON(w, SimulationComparisonJSON{Policies: sortedSimulationResults(results)}) +} + +func sortedSimulationResults(results []scheduler.SimulationResult) []scheduler.SimulationResult { + ordered := append([]scheduler.SimulationResult(nil), results...) + sort.Slice(ordered, func(i, j int) bool { + return ordered[i].Policy < ordered[j].Policy + }) + return ordered +} diff --git a/internal/output/simulation_output_test.go b/internal/output/simulation_output_test.go new file mode 100644 index 0000000..6bb051b --- /dev/null +++ b/internal/output/simulation_output_test.go @@ -0,0 +1,313 @@ +package output_test + +import ( + "context" + "encoding/json" + "fmt" + "io" + "reflect" + "strings" + "testing" + + "github.com/Rionlyu/topoqueue/internal/model" + "github.com/Rionlyu/topoqueue/internal/output" + "github.com/Rionlyu/topoqueue/internal/scheduler" +) + +func TestSimulationTextOutputIsStableOrderedAndComplete(t *testing.T) { + t.Parallel() + + cluster, jobs := simulationInputs() + strict := mustSimulation(t, cluster, jobs, scheduler.PolicyStrictFIFO) + backfill := mustSimulation(t, cluster, jobs, scheduler.PolicyBackfill) + + // Exercise the allocation formatter's defensive sorting and prove that + // rendering does not reorder the result owned by the caller. + reverseAllocations(strict.Events[1].Allocation) + reverseAllocations(strict.Jobs[0].Allocation) + strictBefore := cloneSimulationResult(t, strict) + + strictText := renderRepeatedly(t, func(w io.Writer) error { + return output.WriteSimulationText(w, strict) + }) + for _, wanted := range []string{ + "POLICY", "COMPLETED", "UNSCHEDULED", "START", "END", "MAKESPAN", + "TOTAL WAIT", "AVERAGE WAIT", "MAX WAIT", "strict-fifo", "5.67", + "EVENT TIMELINE", "TICK", "TYPE", "arrived", "admitted", "completed", + "JOB LIFECYCLE", "ARRIVAL", "DURATION", "COMPLETION", "REQUESTED CPU", + "REQUESTED GPU", "warmup", "train-xl", "batch", + "node-a1=4,node-a2=4", + } { + if !strings.Contains(strictText, wanted) { + t.Errorf("strict simulation text missing %q:\n%s", wanted, strictText) + } + } + wantStrictEvents := []string{ + "0 arrived warmup", "0 admitted warmup", "1 arrived train-xl", + "2 arrived batch", "8 completed warmup", "8 admitted train-xl", + "12 completed train-xl", "12 admitted batch", "14 completed batch", + } + if got := renderedEventOrder(t, strictText); !reflect.DeepEqual(got, wantStrictEvents) { + t.Errorf("strict rendered event order = %#v, want %#v", got, wantStrictEvents) + } + if got := renderedLifecycleOrder(t, strictText); !reflect.DeepEqual(got, []string{"warmup", "train-xl", "batch"}) { + t.Errorf("strict rendered lifecycle order = %#v", got) + } + if !reflect.DeepEqual(strict, strictBefore) { + t.Errorf("WriteSimulationText() mutated its result:\ngot: %#v\nwant: %#v", strict, strictBefore) + } + + comparisonInput := []scheduler.SimulationResult{strict, backfill} + comparisonBefore := cloneSimulationResults(t, comparisonInput) + comparisonText := renderRepeatedly(t, func(w io.Writer) error { + return output.WriteSimulationComparisonText(w, comparisonInput) + }) + lines := strings.Split(comparisonText, "\n") + if len(lines) < 3 { + t.Fatalf("simulation comparison text is too short:\n%s", comparisonText) + } + if got := strings.Fields(lines[1]); len(got) == 0 || got[0] != string(scheduler.PolicyBackfill) { + t.Errorf("first comparison policy row = %q, want backfill", lines[1]) + } + if got := strings.Fields(lines[2]); len(got) == 0 || got[0] != string(scheduler.PolicyStrictFIFO) { + t.Errorf("second comparison policy row = %q, want strict-fifo", lines[2]) + } + for _, wanted := range []string{"COMPLETED", "MAKESPAN", "AVERAGE WAIT", "MAX WAIT", "2.33", "5.67"} { + if !strings.Contains(comparisonText, wanted) { + t.Errorf("simulation comparison text missing %q:\n%s", wanted, comparisonText) + } + } + if strings.Count(comparisonText, "EVENT TIMELINE") != 2 || strings.Count(comparisonText, "JOB LIFECYCLE") != 2 { + t.Errorf("comparison does not contain both full policy details:\n%s", comparisonText) + } + if !reflect.DeepEqual(comparisonInput, comparisonBefore) { + t.Errorf("WriteSimulationComparisonText() mutated its input") + } +} + +func TestSimulationTextUsesDashesForAbsentValues(t *testing.T) { + t.Parallel() + + cluster := model.Cluster{Nodes: []model.Node{{ + Name: "node-a", Capacity: model.Resources{GPU: 1}, + }}} + jobs := model.TimedJobSet{Jobs: []model.TimedJob{{ + Job: model.Job{ + Name: "impossible", Replicas: 2, + ResourcesPerReplica: model.Resources{GPU: 1}, + }, + DurationTicks: 1, + }}} + result := mustSimulation(t, cluster, jobs, scheduler.PolicyBackfill) + text := renderRepeatedly(t, func(w io.Writer) error { + return output.WriteSimulationText(w, result) + }) + normalized := strings.Join(strings.Fields(text), " ") + wanted := "impossible unscheduled 0 1 - - - - - 0 2 insufficient_capacity:" + if !strings.Contains(normalized, wanted) { + t.Errorf("unscheduled lifecycle does not use dashes for absent values; want %q in:\n%s", wanted, text) + } +} + +func TestSimulationJSONOutputRoundTripsStablyWithoutMutation(t *testing.T) { + t.Parallel() + + cluster, jobs := simulationInputs() + backfill := mustSimulation(t, cluster, jobs, scheduler.PolicyBackfill) + strict := mustSimulation(t, cluster, jobs, scheduler.PolicyStrictFIFO) + + backfillBefore := cloneSimulationResult(t, backfill) + policyJSON := renderRepeatedly(t, func(w io.Writer) error { + return output.WriteSimulationJSON(w, backfill) + }) + if !strings.HasSuffix(policyJSON, "\n") || !strings.Contains(policyJSON, "\n ") { + t.Errorf("simulation JSON is not indented with a trailing newline:\n%s", policyJSON) + } + for _, wanted := range []string{ + `"policy": "backfill"`, `"events": [`, `"type": "admitted"`, + `"jobs": [`, `"startTick": 2`, `"queueDelayTicks": 7`, + `"averageQueueDelayTicks": 2.3333333333333335`, + } { + if !strings.Contains(policyJSON, wanted) { + t.Errorf("simulation JSON missing %q:\n%s", wanted, policyJSON) + } + } + var decodedPolicy scheduler.SimulationResult + if err := json.Unmarshal([]byte(policyJSON), &decodedPolicy); err != nil { + t.Fatalf("decode simulation JSON: %v", err) + } + if !reflect.DeepEqual(decodedPolicy, backfill) { + t.Errorf("decoded simulation JSON mismatch:\ngot: %#v\nwant: %#v", decodedPolicy, backfill) + } + if !reflect.DeepEqual(backfill, backfillBefore) { + t.Errorf("WriteSimulationJSON() mutated its result") + } + + comparisonInput := []scheduler.SimulationResult{strict, backfill} + comparisonBefore := cloneSimulationResults(t, comparisonInput) + comparisonJSON := renderRepeatedly(t, func(w io.Writer) error { + return output.WriteSimulationComparisonJSON(w, comparisonInput) + }) + var decodedComparison output.SimulationComparisonJSON + if err := json.Unmarshal([]byte(comparisonJSON), &decodedComparison); err != nil { + t.Fatalf("decode simulation comparison JSON: %v", err) + } + wantComparison := output.SimulationComparisonJSON{Policies: []scheduler.SimulationResult{backfill, strict}} + if !reflect.DeepEqual(decodedComparison, wantComparison) { + t.Errorf("decoded comparison JSON mismatch:\ngot: %#v\nwant: %#v", decodedComparison, wantComparison) + } + backfillIndex := strings.Index(comparisonJSON, `"policy": "backfill"`) + strictIndex := strings.Index(comparisonJSON, `"policy": "strict-fifo"`) + if backfillIndex < 0 || strictIndex < 0 || backfillIndex >= strictIndex { + t.Errorf("comparison JSON policies are not sorted:\n%s", comparisonJSON) + } + if !strings.HasSuffix(comparisonJSON, "\n") || !strings.Contains(comparisonJSON, "\n ") { + t.Errorf("comparison JSON is not indented with a trailing newline:\n%s", comparisonJSON) + } + if !reflect.DeepEqual(comparisonInput, comparisonBefore) { + t.Errorf("WriteSimulationComparisonJSON() mutated its input") + } +} + +func renderRepeatedly(t *testing.T, write func(io.Writer) error) string { + t.Helper() + first := render(t, write) + for iteration := 1; iteration < 3; iteration++ { + if got := render(t, write); got != first { + t.Fatalf("render %d differs from the first render:\nfirst:\n%s\ngot:\n%s", iteration+1, first, got) + } + } + return first +} + +func renderedEventOrder(t *testing.T, text string) []string { + t.Helper() + lines := strings.Split(text, "\n") + heading := lineIndex(lines, "EVENT TIMELINE") + if heading < 0 || heading+1 >= len(lines) { + t.Fatalf("EVENT TIMELINE section missing:\n%s", text) + } + events := make([]string, 0) + for _, line := range lines[heading+2:] { + if line == "" { + break + } + fields := strings.Fields(line) + if len(fields) < 3 { + t.Fatalf("malformed event line %q", line) + } + events = append(events, fmt.Sprintf("%s %s %s", fields[0], fields[1], fields[2])) + } + return events +} + +func renderedLifecycleOrder(t *testing.T, text string) []string { + t.Helper() + lines := strings.Split(text, "\n") + heading := lineIndex(lines, "JOB LIFECYCLE") + if heading < 0 || heading+1 >= len(lines) { + t.Fatalf("JOB LIFECYCLE section missing:\n%s", text) + } + jobs := make([]string, 0) + for _, line := range lines[heading+2:] { + if line == "" { + break + } + fields := strings.Fields(line) + if len(fields) == 0 { + t.Fatalf("malformed lifecycle line %q", line) + } + jobs = append(jobs, fields[0]) + } + return jobs +} + +func lineIndex(lines []string, wanted string) int { + for index, line := range lines { + if line == wanted { + return index + } + } + return -1 +} + +func cloneSimulationResult(t *testing.T, result scheduler.SimulationResult) scheduler.SimulationResult { + t.Helper() + data, err := json.Marshal(result) + if err != nil { + t.Fatalf("clone simulation result: encode: %v", err) + } + var clone scheduler.SimulationResult + if err := json.Unmarshal(data, &clone); err != nil { + t.Fatalf("clone simulation result: decode: %v", err) + } + return clone +} + +func cloneSimulationResults(t *testing.T, results []scheduler.SimulationResult) []scheduler.SimulationResult { + t.Helper() + clones := make([]scheduler.SimulationResult, len(results)) + for index, result := range results { + clones[index] = cloneSimulationResult(t, result) + } + return clones +} + +func reverseAllocations(allocations []scheduler.Allocation) { + for left, right := 0, len(allocations)-1; left < right; left, right = left+1, right-1 { + allocations[left], allocations[right] = allocations[right], allocations[left] + } +} + +func mustSimulation( + t *testing.T, + cluster model.Cluster, + jobs model.TimedJobSet, + policy scheduler.Policy, +) scheduler.SimulationResult { + t.Helper() + result, err := scheduler.Simulate(context.Background(), cluster, jobs, policy) + if err != nil { + t.Fatalf("Simulate() error = %v", err) + } + return result +} + +func simulationInputs() (model.Cluster, model.TimedJobSet) { + cluster := model.Cluster{Nodes: []model.Node{ + simulationNode("node-a1", "rack-a"), + simulationNode("node-a2", "rack-a"), + simulationNode("node-b1", "rack-b"), + simulationNode("node-b2", "rack-b"), + }} + jobs := model.TimedJobSet{Jobs: []model.TimedJob{ + simulationJob("warmup", 0, 8, 8, "zone"), + simulationJob("train-xl", 1, 4, 16, "zone"), + simulationJob("batch", 2, 2, 4, "rack"), + }} + return cluster, jobs +} + +func simulationNode(name, rack string) model.Node { + return model.Node{ + Name: name, + Topology: map[string]string{ + "zone": "zone-a", + "rack": rack, + }, + Capacity: model.Resources{CPU: 32, GPU: 4}, + } +} + +func simulationJob(name string, arrival, duration int64, replicas int, topology string) model.TimedJob { + return model.TimedJob{ + Job: model.Job{ + Name: name, + Replicas: replicas, + ResourcesPerReplica: model.Resources{CPU: 4, GPU: 1}, + RequiredTopology: topology, + }, + ArrivalTick: arrival, + DurationTicks: duration, + } +} diff --git a/internal/output/simulation_text.go b/internal/output/simulation_text.go new file mode 100644 index 0000000..ef5ef69 --- /dev/null +++ b/internal/output/simulation_text.go @@ -0,0 +1,167 @@ +package output + +import ( + "fmt" + "io" + "strconv" + "text/tabwriter" + + "github.com/Rionlyu/topoqueue/internal/scheduler" +) + +// WriteSimulationText writes one timed policy simulation as a summary, an +// ordered event timeline, and original-order job lifecycles. +func WriteSimulationText(w io.Writer, result scheduler.SimulationResult) error { + tw := tabwriter.NewWriter(w, 0, 4, 2, ' ', 0) + if err := writeSimulationSummary(tw, result); err != nil { + return err + } + if err := tw.Flush(); err != nil { + return fmt.Errorf("flush simulation summary: %w", err) + } + + if _, err := fmt.Fprintln(w); err != nil { + return fmt.Errorf("write simulation timeline separator: %w", err) + } + if err := writeSimulationEvents(w, result); err != nil { + return err + } + + if _, err := fmt.Fprintln(w); err != nil { + return fmt.Errorf("write simulation lifecycle separator: %w", err) + } + if err := writeSimulationLifecycles(w, result); err != nil { + return err + } + return nil +} + +// WriteSimulationComparisonText writes a sorted policy comparison followed by +// the complete deterministic output for each simulation. +func WriteSimulationComparisonText(w io.Writer, results []scheduler.SimulationResult) error { + results = sortedSimulationResults(results) + tw := tabwriter.NewWriter(w, 0, 4, 2, ' ', 0) + if _, err := fmt.Fprintln(tw, "POLICY\tCOMPLETED\tUNSCHEDULED\tMAKESPAN\tAVERAGE WAIT\tMAX WAIT"); err != nil { + return fmt.Errorf("write simulation comparison header: %w", err) + } + for _, result := range results { + if _, err := fmt.Fprintf( + tw, + "%s\t%d\t%d\t%d\t%.2f\t%d\n", + result.Policy, + result.CompletedJobCount, + result.UnscheduledJobCount, + result.MakespanTicks, + result.AverageQueueDelayTicks, + result.MaximumQueueDelayTicks, + ); err != nil { + return fmt.Errorf("write simulation comparison policy %q: %w", result.Policy, err) + } + } + if err := tw.Flush(); err != nil { + return fmt.Errorf("flush simulation comparison summary: %w", err) + } + + for _, result := range results { + if _, err := fmt.Fprintln(w); err != nil { + return fmt.Errorf("write simulation comparison separator: %w", err) + } + if err := WriteSimulationText(w, result); err != nil { + return fmt.Errorf("write simulation comparison policy %q details: %w", result.Policy, err) + } + } + return nil +} + +func writeSimulationSummary(w io.Writer, result scheduler.SimulationResult) error { + if _, err := fmt.Fprintln(w, "POLICY\tCOMPLETED\tUNSCHEDULED\tSTART\tEND\tMAKESPAN\tTOTAL WAIT\tAVERAGE WAIT\tMAX WAIT"); err != nil { + return fmt.Errorf("write simulation summary header: %w", err) + } + if _, err := fmt.Fprintf( + w, + "%s\t%d\t%d\t%d\t%d\t%d\t%d\t%.2f\t%d\n", + result.Policy, + result.CompletedJobCount, + result.UnscheduledJobCount, + result.StartTick, + result.EndTick, + result.MakespanTicks, + result.TotalQueueDelayTicks, + result.AverageQueueDelayTicks, + result.MaximumQueueDelayTicks, + ); err != nil { + return fmt.Errorf("write simulation summary for policy %q: %w", result.Policy, err) + } + return nil +} + +func writeSimulationEvents(w io.Writer, result scheduler.SimulationResult) error { + if _, err := fmt.Fprintln(w, "EVENT TIMELINE"); err != nil { + return fmt.Errorf("write simulation event heading for policy %q: %w", result.Policy, err) + } + tw := tabwriter.NewWriter(w, 0, 4, 2, ' ', 0) + if _, err := fmt.Fprintln(tw, "TICK\tTYPE\tJOB\tDOMAIN\tALLOCATION"); err != nil { + return fmt.Errorf("write simulation event header for policy %q: %w", result.Policy, err) + } + for _, event := range result.Events { + if _, err := fmt.Fprintf( + tw, + "%d\t%s\t%s\t%s\t%s\n", + event.Tick, + event.Type, + event.JobName, + displayValue(event.SelectedTopologyDomain), + formatAllocation(event.Allocation), + ); err != nil { + return fmt.Errorf("write simulation event for job %q at tick %d: %w", event.JobName, event.Tick, err) + } + } + if err := tw.Flush(); err != nil { + return fmt.Errorf("flush simulation events for policy %q: %w", result.Policy, err) + } + return nil +} + +func writeSimulationLifecycles(w io.Writer, result scheduler.SimulationResult) error { + if _, err := fmt.Fprintln(w, "JOB LIFECYCLE"); err != nil { + return fmt.Errorf("write simulation lifecycle heading for policy %q: %w", result.Policy, err) + } + tw := tabwriter.NewWriter(w, 0, 4, 2, ' ', 0) + if _, err := fmt.Fprintln( + tw, + "JOB\tSTATUS\tARRIVAL\tDURATION\tSTART\tCOMPLETION\tWAIT\tDOMAIN\tALLOCATION\tREQUESTED CPU\tREQUESTED GPU\tREASON", + ); err != nil { + return fmt.Errorf("write simulation lifecycle header for policy %q: %w", result.Policy, err) + } + for _, job := range result.Jobs { + if _, err := fmt.Fprintf( + tw, + "%s\t%s\t%d\t%d\t%s\t%s\t%s\t%s\t%s\t%d\t%d\t%s\n", + job.JobName, + job.Status, + job.ArrivalTick, + job.DurationTicks, + formatOptionalTick(job.StartTick), + formatOptionalTick(job.CompletionTick), + formatOptionalTick(job.QueueDelayTicks), + displayValue(job.SelectedTopologyDomain), + formatAllocation(job.Allocation), + job.RequestedResources.CPU, + job.RequestedResources.GPU, + formatReason(job.Reason), + ); err != nil { + return fmt.Errorf("write simulation lifecycle for job %q under policy %q: %w", job.JobName, result.Policy, err) + } + } + if err := tw.Flush(); err != nil { + return fmt.Errorf("flush simulation lifecycles for policy %q: %w", result.Policy, err) + } + return nil +} + +func formatOptionalTick(tick *int64) string { + if tick == nil { + return "-" + } + return strconv.FormatInt(*tick, 10) +} diff --git a/internal/scheduler/benchmark_test.go b/internal/scheduler/benchmark_test.go index c6cada1..9c3adbe 100644 --- a/internal/scheduler/benchmark_test.go +++ b/internal/scheduler/benchmark_test.go @@ -24,6 +24,20 @@ func BenchmarkScheduleThousandJobs(b *testing.B) { } } +func BenchmarkSimulateThousandTimedJobs(b *testing.B) { + cluster := benchmarkCluster(100) + jobs := benchmarkTimedJobs(1_000) + b.ResetTimer() + + for iteration := 0; iteration < b.N; iteration++ { + result, err := scheduler.Simulate(context.Background(), cluster, jobs, scheduler.PolicyBackfill) + if err != nil { + b.Fatal(err) + } + runtime.KeepAlive(result) + } +} + func benchmarkCluster(nodeCount int) model.Cluster { cluster := model.Cluster{Nodes: make([]model.Node, 0, nodeCount)} for index := 0; index < nodeCount; index++ { @@ -55,3 +69,24 @@ func benchmarkJobs(jobCount int) model.JobSet { } return jobs } + +func benchmarkTimedJobs(jobCount int) model.TimedJobSet { + jobs := model.TimedJobSet{Jobs: make([]model.TimedJob, 0, jobCount)} + for index := 0; index < jobCount; index++ { + topology := "rack" + if index%3 == 0 { + topology = "" + } + jobs.Jobs = append(jobs.Jobs, model.TimedJob{ + Job: model.Job{ + Name: fmt.Sprintf("timed-job-%04d", index), + Replicas: 1 + index%12, + ResourcesPerReplica: model.Resources{CPU: int64(1 + index%4), GPU: 1}, + RequiredTopology: topology, + }, + ArrivalTick: int64(index / 10), + DurationTicks: int64(1 + index%10), + }) + } + return jobs +} diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go index 6e659ed..42a170e 100644 --- a/internal/scheduler/scheduler.go +++ b/internal/scheduler/scheduler.go @@ -175,6 +175,7 @@ func Schedule(ctx context.Context, cluster model.Cluster, jobs model.JobSet, pol type nodeState struct { name string topology map[string]string + capacity model.Resources remaining model.Resources } @@ -184,6 +185,7 @@ func makeNodeStates(cluster model.Cluster) []nodeState { nodes[index] = nodeState{ name: node.Name, topology: node.Topology, + capacity: node.Capacity, remaining: node.Capacity, } } diff --git a/internal/scheduler/simulation.go b/internal/scheduler/simulation.go new file mode 100644 index 0000000..b285afd --- /dev/null +++ b/internal/scheduler/simulation.go @@ -0,0 +1,409 @@ +package scheduler + +import ( + "container/heap" + "context" + "fmt" + "math" + "sort" + + "github.com/Rionlyu/topoqueue/internal/model" +) + +// Simulate evaluates a timed workload with one deterministic queue policy. +func Simulate(ctx context.Context, cluster model.Cluster, jobs model.TimedJobSet, policy Policy) (SimulationResult, error) { + if ctx == nil { + return SimulationResult{}, fmt.Errorf("simulate policy %q: nil context", policy) + } + if _, err := ParsePolicy(string(policy)); err != nil { + return SimulationResult{}, fmt.Errorf("simulate: %w", err) + } + if err := ctx.Err(); err != nil { + return SimulationResult{}, fmt.Errorf("simulate policy %q: %w", policy, err) + } + if err := model.ValidateCluster(cluster); err != nil { + return SimulationResult{}, fmt.Errorf("simulate policy %q: validate cluster: %w", policy, err) + } + if err := model.ValidateTimedJobs(jobs); err != nil { + return SimulationResult{}, fmt.Errorf("simulate policy %q: validate timed jobs: %w", policy, err) + } + if err := model.ValidateTopologyRequirements(cluster, jobs.StaticJobs()); err != nil { + return SimulationResult{}, fmt.Errorf("simulate policy %q: validate topology requirements: %w", policy, err) + } + + state := newSimulationState(cluster.Clone(), jobs.Clone(), policy) + result, err := state.run(ctx) + if err != nil { + return SimulationResult{}, fmt.Errorf("simulate policy %q: %w", policy, err) + } + return result, nil +} + +type simulationState struct { + policy Policy + jobs model.TimedJobSet + nodes []nodeState + arrivalOrder []int + arrivalCursor int + pending []int + running runningJobHeap + result SimulationResult +} + +func newSimulationState(cluster model.Cluster, jobs model.TimedJobSet, policy Policy) *simulationState { + arrivalOrder := make([]int, len(jobs.Jobs)) + lifecycles := make([]JobLifecycle, len(jobs.Jobs)) + for index, job := range jobs.Jobs { + arrivalOrder[index] = index + lifecycles[index] = JobLifecycle{ + JobName: job.Name, + ArrivalTick: job.ArrivalTick, + DurationTicks: job.DurationTicks, + Allocation: make([]Allocation, 0), + RequestedResources: resourcesForReplicas(job.ResourcesPerReplica, job.Replicas), + } + } + sort.Slice(arrivalOrder, func(i, j int) bool { + left := jobs.Jobs[arrivalOrder[i]] + right := jobs.Jobs[arrivalOrder[j]] + if left.ArrivalTick != right.ArrivalTick { + return left.ArrivalTick < right.ArrivalTick + } + return arrivalOrder[i] < arrivalOrder[j] + }) + + state := &simulationState{ + policy: policy, + jobs: jobs, + nodes: makeNodeStates(cluster), + arrivalOrder: arrivalOrder, + pending: make([]int, 0), + running: make(runningJobHeap, 0), + result: SimulationResult{ + Policy: policy, + Events: make([]SimulationEvent, 0), + Jobs: lifecycles, + }, + } + heap.Init(&state.running) + return state +} + +func (s *simulationState) run(ctx context.Context) (SimulationResult, error) { + for s.arrivalCursor < len(s.arrivalOrder) || s.running.Len() > 0 { + if err := ctx.Err(); err != nil { + return SimulationResult{}, err + } + tick, ok := s.nextEventTick() + if !ok { + return SimulationResult{}, fmt.Errorf("event state has future work but no next event") + } + if err := s.completeJobs(ctx, tick); err != nil { + return SimulationResult{}, err + } + if err := s.arriveJobs(ctx, tick); err != nil { + return SimulationResult{}, err + } + if err := s.admitJobs(ctx, tick); err != nil { + return SimulationResult{}, err + } + } + + if err := ctx.Err(); err != nil { + return SimulationResult{}, err + } + if err := s.terminalizePending(ctx); err != nil { + return SimulationResult{}, err + } + if err := ctx.Err(); err != nil { + return SimulationResult{}, err + } + if len(s.result.Events) > 0 { + s.result.MakespanTicks = s.result.EndTick - s.result.StartTick + } + if s.result.CompletedJobCount > 0 { + s.result.AverageQueueDelayTicks = float64(s.result.TotalQueueDelayTicks) / float64(s.result.CompletedJobCount) + } + if s.result.CompletedJobCount+s.result.UnscheduledJobCount != len(s.jobs.Jobs) { + return SimulationResult{}, fmt.Errorf( + "terminal lifecycle count is %d, want %d", + s.result.CompletedJobCount+s.result.UnscheduledJobCount, + len(s.jobs.Jobs), + ) + } + return s.result, nil +} + +func (s *simulationState) nextEventTick() (int64, bool) { + var tick int64 + found := false + if s.arrivalCursor < len(s.arrivalOrder) { + tick = s.jobs.Jobs[s.arrivalOrder[s.arrivalCursor]].ArrivalTick + found = true + } + if s.running.Len() > 0 && (!found || s.running[0].finishTick < tick) { + tick = s.running[0].finishTick + found = true + } + return tick, found +} + +func (s *simulationState) completeJobs(ctx context.Context, tick int64) error { + for s.running.Len() > 0 && s.running[0].finishTick == tick { + if err := ctx.Err(); err != nil { + return err + } + running := heap.Pop(&s.running).(runningJob) + job := s.jobs.Jobs[running.inputIndex] + if err := releaseAllocation(s.nodes, job.Job, running.allocation); err != nil { + return fmt.Errorf("tick %d complete job %q: %w", tick, job.Name, err) + } + + lifecycle := &s.result.Jobs[running.inputIndex] + if lifecycle.StartTick == nil || lifecycle.QueueDelayTicks == nil { + return fmt.Errorf("tick %d complete job %q: missing admission lifecycle", tick, job.Name) + } + completionTick := tick + lifecycle.CompletionTick = &completionTick + lifecycle.Status = LifecycleCompleted + + delay := *lifecycle.QueueDelayTicks + if delay < 0 || s.result.TotalQueueDelayTicks > math.MaxInt64-delay { + return fmt.Errorf("tick %d complete job %q: total queue delay exceeds %d", tick, job.Name, int64(math.MaxInt64)) + } + s.result.TotalQueueDelayTicks += delay + if delay > s.result.MaximumQueueDelayTicks { + s.result.MaximumQueueDelayTicks = delay + } + s.result.CompletedJobCount++ + s.appendEvent(SimulationEvent{Tick: tick, Type: EventCompleted, JobName: job.Name}) + if err := ctx.Err(); err != nil { + return err + } + } + return nil +} + +func (s *simulationState) arriveJobs(ctx context.Context, tick int64) error { + for s.arrivalCursor < len(s.arrivalOrder) { + inputIndex := s.arrivalOrder[s.arrivalCursor] + job := s.jobs.Jobs[inputIndex] + if job.ArrivalTick != tick { + break + } + if err := ctx.Err(); err != nil { + return err + } + s.pending = append(s.pending, inputIndex) + s.arrivalCursor++ + s.appendEvent(SimulationEvent{Tick: tick, Type: EventArrived, JobName: job.Name}) + } + return nil +} + +func (s *simulationState) admitJobs(ctx context.Context, tick int64) error { + switch s.policy { + case PolicyStrictFIFO: + for len(s.pending) > 0 { + if err := ctx.Err(); err != nil { + return err + } + admitted, err := s.tryAdmit(ctx, tick, s.pending[0]) + if err != nil { + return err + } + if !admitted { + break + } + s.pending = s.pending[1:] + } + case PolicyBackfill: + retained := make([]int, 0, len(s.pending)) + for _, inputIndex := range s.pending { + if err := ctx.Err(); err != nil { + return err + } + admitted, err := s.tryAdmit(ctx, tick, inputIndex) + if err != nil { + return err + } + if !admitted { + retained = append(retained, inputIndex) + } + } + s.pending = retained + default: + return fmt.Errorf("unsupported policy %q", s.policy) + } + return nil +} + +func (s *simulationState) tryAdmit(ctx context.Context, tick int64, inputIndex int) (bool, error) { + job := s.jobs.Jobs[inputIndex] + if tick < job.ArrivalTick { + return false, fmt.Errorf("tick %d admit job %q before arrival tick %d", tick, job.Name, job.ArrivalTick) + } + placed, err := placeJob(ctx, s.nodes, job.Job) + if err != nil { + return false, fmt.Errorf("tick %d place job %q: %w", tick, job.Name, err) + } + if placed.reason != nil { + return false, nil + } + if tick > math.MaxInt64-job.DurationTicks { + if releaseErr := releaseAllocation(s.nodes, job.Job, placed.allocation); releaseErr != nil { + return false, fmt.Errorf( + "tick %d admit job %q: finish tick overflow for duration %d; rollback allocation: %w", + tick, + job.Name, + job.DurationTicks, + releaseErr, + ) + } + return false, fmt.Errorf("tick %d admit job %q: finish tick exceeds %d for duration %d", tick, job.Name, int64(math.MaxInt64), job.DurationTicks) + } + finishTick := tick + job.DurationTicks + queueDelay := tick - job.ArrivalTick + allocation := cloneAllocations(placed.allocation) + lifecycle := &s.result.Jobs[inputIndex] + startTick := tick + lifecycle.StartTick = &startTick + lifecycle.QueueDelayTicks = &queueDelay + lifecycle.SelectedTopologyDomain = placed.domain + lifecycle.Allocation = cloneAllocations(allocation) + heap.Push(&s.running, runningJob{ + inputIndex: inputIndex, + finishTick: finishTick, + allocation: cloneAllocations(allocation), + }) + s.appendEvent(SimulationEvent{ + Tick: tick, + Type: EventAdmitted, + JobName: job.Name, + SelectedTopologyDomain: placed.domain, + Allocation: cloneAllocations(allocation), + }) + return true, nil +} + +func (s *simulationState) terminalizePending(ctx context.Context) error { + if err := ctx.Err(); err != nil { + return err + } + if s.arrivalCursor != len(s.arrivalOrder) || s.running.Len() != 0 { + return fmt.Errorf("terminalize pending jobs while future or running work remains") + } + for _, node := range s.nodes { + if err := validateNodeResourceState(node); err != nil { + return fmt.Errorf("terminal node %q: %w", node.name, err) + } + if node.remaining != node.capacity { + return fmt.Errorf("terminal node %q has remaining %+v, want original capacity %+v", node.name, node.remaining, node.capacity) + } + } + if len(s.pending) == 0 { + return nil + } + + switch s.policy { + case PolicyStrictFIFO: + headIndex := s.pending[0] + headReason, err := s.finalPlacementReason(ctx, headIndex) + if err != nil { + return err + } + s.markUnscheduled(headIndex, headReason) + blockingJob := s.jobs.Jobs[headIndex].Name + for _, inputIndex := range s.pending[1:] { + if err := ctx.Err(); err != nil { + return err + } + s.markUnscheduled(inputIndex, headOfLineReason(blockingJob)) + } + case PolicyBackfill: + for _, inputIndex := range s.pending { + if err := ctx.Err(); err != nil { + return err + } + reason, err := s.finalPlacementReason(ctx, inputIndex) + if err != nil { + return err + } + s.markUnscheduled(inputIndex, reason) + } + default: + return fmt.Errorf("unsupported policy %q", s.policy) + } + return nil +} + +func (s *simulationState) finalPlacementReason(ctx context.Context, inputIndex int) (*Reason, error) { + job := s.jobs.Jobs[inputIndex] + placed, err := placeJob(ctx, s.nodes, job.Job) + if err != nil { + return nil, fmt.Errorf("final placement for job %q: %w", job.Name, err) + } + if placed.reason != nil { + return placed.reason, nil + } + if err := releaseAllocation(s.nodes, job.Job, placed.allocation); err != nil { + return nil, fmt.Errorf("final placement invariant for job %q: unexpectedly fit and rollback failed: %w", job.Name, err) + } + return nil, fmt.Errorf("final placement invariant for job %q: unexpectedly fit on an empty cluster", job.Name) +} + +func (s *simulationState) markUnscheduled(inputIndex int, reason *Reason) { + lifecycle := &s.result.Jobs[inputIndex] + lifecycle.Status = LifecycleUnscheduled + lifecycle.Reason = reason + s.result.UnscheduledJobCount++ +} + +func (s *simulationState) appendEvent(event SimulationEvent) { + if len(s.result.Events) == 0 { + s.result.StartTick = event.Tick + } + s.result.EndTick = event.Tick + s.result.Events = append(s.result.Events, event) +} + +func cloneAllocations(allocation []Allocation) []Allocation { + if len(allocation) == 0 { + return make([]Allocation, 0) + } + clone := make([]Allocation, len(allocation)) + copy(clone, allocation) + return clone +} + +type runningJob struct { + inputIndex int + finishTick int64 + allocation []Allocation +} + +type runningJobHeap []runningJob + +func (h runningJobHeap) Len() int { return len(h) } + +func (h runningJobHeap) Less(i, j int) bool { + if h[i].finishTick != h[j].finishTick { + return h[i].finishTick < h[j].finishTick + } + return h[i].inputIndex < h[j].inputIndex +} + +func (h runningJobHeap) Swap(i, j int) { h[i], h[j] = h[j], h[i] } + +func (h *runningJobHeap) Push(value any) { + *h = append(*h, value.(runningJob)) +} + +func (h *runningJobHeap) Pop() any { + old := *h + last := len(old) - 1 + value := old[last] + old[last] = runningJob{} + *h = old[:last] + return value +} diff --git a/internal/scheduler/simulation_compare.go b/internal/scheduler/simulation_compare.go new file mode 100644 index 0000000..692f5bf --- /dev/null +++ b/internal/scheduler/simulation_compare.go @@ -0,0 +1,82 @@ +package scheduler + +import ( + "context" + "errors" + "fmt" + "sort" + + "github.com/Rionlyu/topoqueue/internal/model" +) + +type simulationComparisonOutcome struct { + policy Policy + result SimulationResult + err error +} + +// CompareSimulations runs strict FIFO and backfill simulations concurrently on +// independent inputs. Results are ordered by policy name, not completion order. +func CompareSimulations(ctx context.Context, cluster model.Cluster, jobs model.TimedJobSet) ([]SimulationResult, error) { + if ctx == nil { + return nil, fmt.Errorf("compare simulations: nil context") + } + if err := ctx.Err(); err != nil { + return nil, fmt.Errorf("compare simulations: %w", err) + } + if err := model.ValidateCluster(cluster); err != nil { + return nil, fmt.Errorf("compare simulations: validate cluster: %w", err) + } + if err := model.ValidateTimedJobs(jobs); err != nil { + return nil, fmt.Errorf("compare simulations: validate timed jobs: %w", err) + } + if err := model.ValidateTopologyRequirements(cluster, jobs.StaticJobs()); err != nil { + return nil, fmt.Errorf("compare simulations: validate topology requirements: %w", err) + } + + ctx, cancel := context.WithCancel(ctx) + defer cancel() + + policies := []Policy{PolicyStrictFIFO, PolicyBackfill} + outcomes := make(chan simulationComparisonOutcome, len(policies)) + for _, policy := range policies { + policy := policy + clusterCopy := cluster.Clone() + jobsCopy := jobs.Clone() + go func() { + result, err := Simulate(ctx, clusterCopy, jobsCopy, policy) + outcomes <- simulationComparisonOutcome{policy: policy, result: result, err: err} + }() + } + + completed := make([]simulationComparisonOutcome, 0, len(policies)) + for range policies { + outcome := <-outcomes + completed = append(completed, outcome) + // Backfill sorts first and therefore has deterministic error precedence. + // Once it fails intrinsically, the sibling cannot change the selected + // error and can be canceled safely. A strict-FIFO failure must not cancel + // a still-running backfill simulation whose error would take precedence. + if outcome.policy == PolicyBackfill && outcome.err != nil && !errors.Is(outcome.err, context.Canceled) { + cancel() + } + } + sort.Slice(completed, func(i, j int) bool { + return completed[i].policy < completed[j].policy + }) + + for _, outcome := range completed { + if outcome.err != nil && !errors.Is(outcome.err, context.Canceled) { + return nil, fmt.Errorf("compare simulation policy %q: %w", outcome.policy, outcome.err) + } + } + + results := make([]SimulationResult, 0, len(completed)) + for _, outcome := range completed { + if outcome.err != nil { + return nil, fmt.Errorf("compare simulation policy %q: %w", outcome.policy, outcome.err) + } + results = append(results, outcome.result) + } + return results, nil +} diff --git a/internal/scheduler/simulation_compare_test.go b/internal/scheduler/simulation_compare_test.go new file mode 100644 index 0000000..d17d76d --- /dev/null +++ b/internal/scheduler/simulation_compare_test.go @@ -0,0 +1,101 @@ +package scheduler_test + +import ( + "context" + "errors" + "math" + "reflect" + "strings" + "testing" + + "github.com/Rionlyu/topoqueue/internal/model" + "github.com/Rionlyu/topoqueue/internal/scheduler" +) + +func TestCompareSimulationsUsesIndependentStateAndStableOrder(t *testing.T) { + t.Parallel() + + cluster, jobs := timedExampleInputs() + originalCluster := cluster.Clone() + originalJobs := jobs.Clone() + + results, err := scheduler.CompareSimulations(context.Background(), cluster, jobs) + if err != nil { + t.Fatalf("CompareSimulations() error = %v", err) + } + if len(results) != 2 { + t.Fatalf("len(results) = %d, want 2", len(results)) + } + if results[0].Policy != scheduler.PolicyBackfill || results[1].Policy != scheduler.PolicyStrictFIFO { + t.Fatalf("policy order = %q, %q; want backfill, strict-fifo", results[0].Policy, results[1].Policy) + } + + want := []scheduler.SimulationResult{ + mustSimulate(t, cluster, jobs, scheduler.PolicyBackfill), + mustSimulate(t, cluster, jobs, scheduler.PolicyStrictFIFO), + } + if !reflect.DeepEqual(results, want) { + t.Errorf("CompareSimulations() results differ from independent simulations") + } + if !reflect.DeepEqual(cluster, originalCluster) { + t.Errorf("CompareSimulations() mutated cluster: got %#v, want %#v", cluster, originalCluster) + } + if !reflect.DeepEqual(jobs, originalJobs) { + t.Errorf("CompareSimulations() mutated timed jobs: got %#v, want %#v", jobs, originalJobs) + } +} + +func TestCompareSimulationsIsRepeatable(t *testing.T) { + t.Parallel() + + cluster, jobs := timedExampleInputs() + want, err := scheduler.CompareSimulations(context.Background(), cluster, jobs) + if err != nil { + t.Fatalf("CompareSimulations() error = %v", err) + } + for iteration := 0; iteration < 50; iteration++ { + got, err := scheduler.CompareSimulations(context.Background(), cluster, jobs) + if err != nil { + t.Fatalf("iteration %d: CompareSimulations() error = %v", iteration, err) + } + if !reflect.DeepEqual(got, want) { + t.Fatalf("iteration %d differs:\ngot: %#v\nwant: %#v", iteration, got, want) + } + } +} + +func TestCompareSimulationsSelectsRuntimeErrorsDeterministically(t *testing.T) { + t.Parallel() + + cluster := model.Cluster{Nodes: []model.Node{{ + Name: "node-a", Capacity: model.Resources{GPU: 1}, + }}} + jobs := model.TimedJobSet{Jobs: []model.TimedJob{ + timedJob("overflow", math.MaxInt64, 1, 1, 0, 1, ""), + }} + const want = `compare simulation policy "backfill": simulate policy "backfill": tick 9223372036854775807 admit job "overflow": finish tick exceeds 9223372036854775807 for duration 1` + + for iteration := 0; iteration < 1_000; iteration++ { + _, err := scheduler.CompareSimulations(context.Background(), cluster, jobs) + if err == nil || err.Error() != want { + t.Fatalf("iteration %d: CompareSimulations() error = %q, want %q", iteration, err, want) + } + } +} + +func TestCompareSimulationsRejectsNilAndCanceledContexts(t *testing.T) { + t.Parallel() + + cluster, jobs := timedExampleInputs() + _, err := scheduler.CompareSimulations(nil, cluster, jobs) + if err == nil || !strings.Contains(err.Error(), "nil context") { + t.Fatalf("CompareSimulations(nil) error = %v, want nil-context error", err) + } + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + _, err = scheduler.CompareSimulations(ctx, cluster, jobs) + if err == nil || !errors.Is(err, context.Canceled) { + t.Fatalf("CompareSimulations(canceled) error = %v, want context.Canceled", err) + } +} diff --git a/internal/scheduler/simulation_result.go b/internal/scheduler/simulation_result.go new file mode 100644 index 0000000..93407d2 --- /dev/null +++ b/internal/scheduler/simulation_result.go @@ -0,0 +1,68 @@ +package scheduler + +import "github.com/Rionlyu/topoqueue/internal/model" + +// SimulationEventType identifies a lifecycle transition in the deterministic +// event timeline. +type SimulationEventType string + +const ( + // EventArrived records that a job entered the pending queue. + EventArrived SimulationEventType = "arrived" + // EventAdmitted records that a job was allocated resources and began running. + EventAdmitted SimulationEventType = "admitted" + // EventCompleted records that a job finished and released its allocation. + EventCompleted SimulationEventType = "completed" +) + +// LifecycleStatus identifies the terminal state of a simulated job. +type LifecycleStatus string + +const ( + // LifecycleCompleted means the job was admitted and ran to completion. + LifecycleCompleted LifecycleStatus = "completed" + // LifecycleUnscheduled means the simulation ended while the job was pending. + LifecycleUnscheduled LifecycleStatus = "unscheduled" +) + +// SimulationEvent is one ordered entry in a policy simulation timeline. +// Admission events include the selected topology domain and allocation. +type SimulationEvent struct { + Tick int64 `json:"tick"` + Type SimulationEventType `json:"type"` + JobName string `json:"jobName"` + SelectedTopologyDomain string `json:"selectedTopologyDomain,omitempty"` + Allocation []Allocation `json:"allocation,omitempty"` +} + +// JobLifecycle contains the terminal result and timing data for one timed job. +// Pointer tick fields distinguish an absent value from the meaningful tick zero. +type JobLifecycle struct { + JobName string `json:"jobName"` + Status LifecycleStatus `json:"status"` + ArrivalTick int64 `json:"arrivalTick"` + DurationTicks int64 `json:"durationTicks"` + StartTick *int64 `json:"startTick,omitempty"` + CompletionTick *int64 `json:"completionTick,omitempty"` + QueueDelayTicks *int64 `json:"queueDelayTicks,omitempty"` + SelectedTopologyDomain string `json:"selectedTopologyDomain,omitempty"` + Allocation []Allocation `json:"allocation"` + RequestedResources model.Resources `json:"requestedResources"` + Reason *Reason `json:"reason,omitempty"` +} + +// SimulationResult contains a deterministic event timeline, original-order job +// lifecycles, and exact aggregate metrics for one queue policy. +type SimulationResult struct { + Policy Policy `json:"policy"` + Events []SimulationEvent `json:"events"` + Jobs []JobLifecycle `json:"jobs"` + CompletedJobCount int `json:"completedJobCount"` + UnscheduledJobCount int `json:"unscheduledJobCount"` + StartTick int64 `json:"startTick"` + EndTick int64 `json:"endTick"` + MakespanTicks int64 `json:"makespanTicks"` + TotalQueueDelayTicks int64 `json:"totalQueueDelayTicks"` + AverageQueueDelayTicks float64 `json:"averageQueueDelayTicks"` + MaximumQueueDelayTicks int64 `json:"maximumQueueDelayTicks"` +} diff --git a/internal/scheduler/simulation_test.go b/internal/scheduler/simulation_test.go new file mode 100644 index 0000000..7f24383 --- /dev/null +++ b/internal/scheduler/simulation_test.go @@ -0,0 +1,493 @@ +package scheduler_test + +import ( + "context" + "errors" + "math" + "reflect" + "strconv" + "strings" + "testing" + + "github.com/Rionlyu/topoqueue/internal/model" + "github.com/Rionlyu/topoqueue/internal/scheduler" +) + +func TestSimulateTimedExamplePolicies(t *testing.T) { + t.Parallel() + + cluster, jobs := timedExampleInputs() + tests := []struct { + name string + policy scheduler.Policy + wantEvents []string + wantTicks [][3]int64 + wantCompleted int + wantUnscheduled int + wantEnd int64 + wantMakespan int64 + wantTotalWait int64 + wantAverageWait float64 + wantMaxWait int64 + wantBatchDomain string + }{ + { + name: "strict FIFO", + policy: scheduler.PolicyStrictFIFO, + wantEvents: []string{ + "0 arrived warmup", "0 admitted warmup", "1 arrived train-xl", + "2 arrived batch", "8 completed warmup", "8 admitted train-xl", + "12 completed train-xl", "12 admitted batch", "14 completed batch", + }, + wantTicks: [][3]int64{{0, 8, 0}, {8, 12, 7}, {12, 14, 10}}, + wantCompleted: 3, + wantEnd: 14, + wantMakespan: 14, + wantTotalWait: 17, + wantAverageWait: float64(17) / 3, + wantMaxWait: 10, + wantBatchDomain: "rack-a", + }, + { + name: "backfill", + policy: scheduler.PolicyBackfill, + wantEvents: []string{ + "0 arrived warmup", "0 admitted warmup", "1 arrived train-xl", + "2 arrived batch", "2 admitted batch", "4 completed batch", + "8 completed warmup", "8 admitted train-xl", "12 completed train-xl", + }, + wantTicks: [][3]int64{{0, 8, 0}, {8, 12, 7}, {2, 4, 0}}, + wantCompleted: 3, + wantEnd: 12, + wantMakespan: 12, + wantTotalWait: 7, + wantAverageWait: float64(7) / 3, + wantMaxWait: 7, + wantBatchDomain: "rack-b", + }, + } + + for _, test := range tests { + test := test + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + result := mustSimulate(t, cluster, jobs, test.policy) + if got := eventSignatures(result.Events); !reflect.DeepEqual(got, test.wantEvents) { + t.Fatalf("events = %#v, want %#v", got, test.wantEvents) + } + if result.CompletedJobCount != test.wantCompleted || result.UnscheduledJobCount != test.wantUnscheduled { + t.Errorf("counts = completed %d, unscheduled %d; want %d, %d", result.CompletedJobCount, result.UnscheduledJobCount, test.wantCompleted, test.wantUnscheduled) + } + if result.StartTick != 0 || result.EndTick != test.wantEnd || result.MakespanTicks != test.wantMakespan { + t.Errorf("ticks = start %d, end %d, makespan %d; want 0, %d, %d", result.StartTick, result.EndTick, result.MakespanTicks, test.wantEnd, test.wantMakespan) + } + if result.TotalQueueDelayTicks != test.wantTotalWait || result.AverageQueueDelayTicks != test.wantAverageWait || result.MaximumQueueDelayTicks != test.wantMaxWait { + t.Errorf("wait metrics = total %d, average %v, max %d; want %d, %v, %d", result.TotalQueueDelayTicks, result.AverageQueueDelayTicks, result.MaximumQueueDelayTicks, test.wantTotalWait, test.wantAverageWait, test.wantMaxWait) + } + for index, lifecycle := range result.Jobs { + if lifecycle.Status != scheduler.LifecycleCompleted { + t.Errorf("job %q status = %q, want completed", lifecycle.JobName, lifecycle.Status) + } + assertLifecycleTicks(t, lifecycle, test.wantTicks[index]) + } + if got := result.Jobs[0].Allocation; !reflect.DeepEqual(got, []scheduler.Allocation{{NodeName: "node-a1", Replicas: 4}, {NodeName: "node-a2", Replicas: 4}}) { + t.Errorf("warmup allocation = %#v", got) + } + warmupAdmission := result.Events[1] + if warmupAdmission.Type != scheduler.EventAdmitted || warmupAdmission.SelectedTopologyDomain != "zone-a" || !reflect.DeepEqual( + warmupAdmission.Allocation, + []scheduler.Allocation{{NodeName: "node-a1", Replicas: 4}, {NodeName: "node-a2", Replicas: 4}}, + ) { + t.Errorf("warmup admission event payload = %#v", warmupAdmission) + } + if result.Jobs[2].SelectedTopologyDomain != test.wantBatchDomain { + t.Errorf("batch domain = %q, want %q", result.Jobs[2].SelectedTopologyDomain, test.wantBatchDomain) + } + }) + } +} + +func TestSimulateProcessesCompletionBeforeSameTickArrivalAndAdmission(t *testing.T) { + t.Parallel() + + cluster := model.Cluster{Nodes: []model.Node{{ + Name: "node-a", Capacity: model.Resources{CPU: 1, GPU: 1}, + }}} + jobs := model.TimedJobSet{Jobs: []model.TimedJob{ + timedJob("first", 0, 2, 1, 1, 1, ""), + timedJob("second", 2, 1, 1, 1, 1, ""), + }} + result := mustSimulate(t, cluster, jobs, scheduler.PolicyStrictFIFO) + want := []string{ + "0 arrived first", "0 admitted first", + "2 completed first", "2 arrived second", "2 admitted second", + "3 completed second", + } + if got := eventSignatures(result.Events); !reflect.DeepEqual(got, want) { + t.Fatalf("events = %#v, want %#v", got, want) + } + assertLifecycleTicks(t, result.Jobs[1], [3]int64{2, 3, 0}) +} + +func TestSimulateReleasesAllSameTickCompletionsBeforeAdmission(t *testing.T) { + t.Parallel() + + cluster := model.Cluster{Nodes: []model.Node{ + {Name: "node-a", Capacity: model.Resources{GPU: 1}}, + {Name: "node-b", Capacity: model.Resources{GPU: 1}}, + }} + jobs := model.TimedJobSet{Jobs: []model.TimedJob{ + timedJob("first", 0, 2, 1, 0, 1, ""), + timedJob("second", 0, 2, 1, 0, 1, ""), + timedJob("whole-cluster", 1, 1, 2, 0, 1, ""), + }} + result := mustSimulate(t, cluster, jobs, scheduler.PolicyStrictFIFO) + wantAtTwo := []string{ + "2 completed first", + "2 completed second", + "2 admitted whole-cluster", + } + gotEvents := eventSignatures(result.Events) + if got := gotEvents[5:8]; !reflect.DeepEqual(got, wantAtTwo) { + t.Fatalf("tick-two events = %#v, want %#v; all events: %#v", got, wantAtTwo, gotEvents) + } + if got := result.Jobs[2].Allocation; !reflect.DeepEqual(got, []scheduler.Allocation{ + {NodeName: "node-a", Replicas: 1}, + {NodeName: "node-b", Replicas: 1}, + }) { + t.Errorf("whole-cluster allocation = %#v", got) + } +} + +func TestSimulateOrdersArrivalsAndCompletionsDeterministically(t *testing.T) { + t.Parallel() + + cluster := model.Cluster{Nodes: []model.Node{ + {Name: "node-a", Capacity: model.Resources{GPU: 1}}, + {Name: "node-b", Capacity: model.Resources{GPU: 1}}, + }} + + t.Run("chronological arrivals and completion input index", func(t *testing.T) { + jobs := model.TimedJobSet{Jobs: []model.TimedJob{ + timedJob("first-index", 1, 2, 1, 0, 1, ""), + timedJob("second-index", 0, 3, 1, 0, 1, ""), + }} + result := mustSimulate(t, cluster, jobs, scheduler.PolicyBackfill) + want := []string{ + "0 arrived second-index", "0 admitted second-index", + "1 arrived first-index", "1 admitted first-index", + "3 completed first-index", "3 completed second-index", + } + if got := eventSignatures(result.Events); !reflect.DeepEqual(got, want) { + t.Fatalf("events = %#v, want %#v", got, want) + } + }) + + t.Run("simultaneous arrivals retain input order", func(t *testing.T) { + jobs := model.TimedJobSet{Jobs: []model.TimedJob{ + timedJob("z-job", 0, 1, 1, 0, 1, ""), + timedJob("a-job", 0, 1, 1, 0, 1, ""), + }} + result := mustSimulate(t, cluster, jobs, scheduler.PolicyBackfill) + wantPrefix := []string{ + "0 arrived z-job", "0 arrived a-job", + "0 admitted z-job", "0 admitted a-job", + } + if got := eventSignatures(result.Events)[:4]; !reflect.DeepEqual(got, wantPrefix) { + t.Fatalf("event prefix = %#v, want %#v", got, wantPrefix) + } + }) +} + +func TestSimulateTerminalPolicySemantics(t *testing.T) { + t.Parallel() + + cluster := model.Cluster{Nodes: []model.Node{{Name: "node-a", Capacity: model.Resources{GPU: 1}}}} + jobs := model.TimedJobSet{Jobs: []model.TimedJob{ + timedJob("impossible", 0, 1, 2, 0, 1, ""), + timedJob("feasible", 0, 1, 1, 0, 1, ""), + }} + + strict := mustSimulate(t, cluster, jobs, scheduler.PolicyStrictFIFO) + if strict.CompletedJobCount != 0 || strict.UnscheduledJobCount != 2 { + t.Fatalf("strict counts = completed %d, unscheduled %d", strict.CompletedJobCount, strict.UnscheduledJobCount) + } + if strict.Jobs[0].Reason == nil || strict.Jobs[0].Reason.Code != scheduler.ReasonInsufficientCapacity { + t.Errorf("strict head reason = %#v", strict.Jobs[0].Reason) + } + if strict.Jobs[1].Reason == nil || strict.Jobs[1].Reason.Code != scheduler.ReasonHeadOfLineBlocked || strict.Jobs[1].Reason.BlockingJob != "impossible" { + t.Errorf("strict follower reason = %#v", strict.Jobs[1].Reason) + } + + backfill := mustSimulate(t, cluster, jobs, scheduler.PolicyBackfill) + if backfill.CompletedJobCount != 1 || backfill.UnscheduledJobCount != 1 { + t.Fatalf("backfill counts = completed %d, unscheduled %d", backfill.CompletedJobCount, backfill.UnscheduledJobCount) + } + if backfill.Jobs[0].Status != scheduler.LifecycleUnscheduled || backfill.Jobs[0].Reason == nil || backfill.Jobs[0].Reason.Code != scheduler.ReasonInsufficientCapacity { + t.Errorf("backfill head lifecycle = %#v", backfill.Jobs[0]) + } + if backfill.Jobs[1].Status != scheduler.LifecycleCompleted { + t.Errorf("backfill follower lifecycle = %#v", backfill.Jobs[1]) + } +} + +func TestSimulateRecomputesFinalReasonAfterAllResourcesAreReleased(t *testing.T) { + t.Parallel() + + cluster := model.Cluster{Nodes: []model.Node{ + {Name: "node-a", Topology: map[string]string{"rack": "rack-a"}, Capacity: model.Resources{GPU: 1}}, + {Name: "node-b", Topology: map[string]string{"rack": "rack-b"}, Capacity: model.Resources{GPU: 1}}, + }} + jobs := model.TimedJobSet{Jobs: []model.TimedJob{ + timedJob("running", 0, 2, 1, 0, 1, "rack"), + timedJob("fragmented", 0, 1, 2, 0, 1, "rack"), + }} + + result := mustSimulate(t, cluster, jobs, scheduler.PolicyBackfill) + blocked := result.Jobs[1] + if blocked.Status != scheduler.LifecycleUnscheduled || blocked.Reason == nil { + t.Fatalf("terminal lifecycle = %#v", blocked) + } + if blocked.Reason.Code != scheduler.ReasonTopologyFragmented { + t.Errorf("final reason = %#v, want topology fragmentation after release", blocked.Reason) + } + if blocked.Reason.TotalAvailableSlots == nil || *blocked.Reason.TotalAvailableSlots != 2 || blocked.Reason.LargestDomainSlots == nil || *blocked.Reason.LargestDomainSlots != 1 { + t.Errorf("final topology capacity details = %#v", blocked.Reason) + } +} + +func TestSimulateEmptyInput(t *testing.T) { + t.Parallel() + + result := mustSimulate(t, model.Cluster{}, model.TimedJobSet{}, scheduler.PolicyBackfill) + if result.Events == nil || result.Jobs == nil { + t.Fatalf("empty result slices must be non-nil: %#v", result) + } + want := scheduler.SimulationResult{ + Policy: scheduler.PolicyBackfill, + Events: []scheduler.SimulationEvent{}, + Jobs: []scheduler.JobLifecycle{}, + } + if !reflect.DeepEqual(result, want) { + t.Errorf("empty result = %#v, want %#v", result, want) + } +} + +func TestSimulateChecksFinishTickOverflowOnlyForFittingJobs(t *testing.T) { + t.Parallel() + + cluster := model.Cluster{Nodes: []model.Node{{Name: "node-a", Capacity: model.Resources{GPU: 1}}}} + overflowing := model.TimedJobSet{Jobs: []model.TimedJob{ + timedJob("blocker", 0, math.MaxInt64, 1, 0, 1, ""), + timedJob("target", 1, 1, 1, 0, 1, ""), + }} + _, err := scheduler.Simulate(context.Background(), cluster, overflowing, scheduler.PolicyStrictFIFO) + if err == nil || !strings.Contains(err.Error(), "target") || !strings.Contains(err.Error(), "finish tick exceeds") { + t.Fatalf("Simulate() overflow error = %v", err) + } + + lateImpossible := model.TimedJobSet{Jobs: []model.TimedJob{ + timedJob("impossible", math.MaxInt64, 1, 2, 0, 1, ""), + }} + result := mustSimulate(t, cluster, lateImpossible, scheduler.PolicyBackfill) + if result.Jobs[0].Status != scheduler.LifecycleUnscheduled || result.StartTick != math.MaxInt64 || result.EndTick != math.MaxInt64 || result.MakespanTicks != 0 { + t.Errorf("late impossible result = %#v", result) + } + + finishesAtMaximum := model.TimedJobSet{Jobs: []model.TimedJob{ + timedJob("maximum", math.MaxInt64-1, 1, 1, 0, 1, ""), + }} + result = mustSimulate(t, cluster, finishesAtMaximum, scheduler.PolicyBackfill) + if result.EndTick != math.MaxInt64 || result.MakespanTicks != 1 || result.Jobs[0].Status != scheduler.LifecycleCompleted { + t.Errorf("maximum-tick result = %#v", result) + } +} + +func TestSimulateRejectsAggregateQueueDelayOverflow(t *testing.T) { + t.Parallel() + + cluster := model.Cluster{Nodes: []model.Node{{Name: "node-a", Capacity: model.Resources{GPU: 2}}}} + jobs := model.TimedJobSet{Jobs: []model.TimedJob{ + timedJob("blocker", 0, math.MaxInt64-1, 2, 0, 1, ""), + timedJob("waiter-a", 0, 1, 1, 0, 1, ""), + timedJob("waiter-b", 0, 1, 1, 0, 1, ""), + }} + _, err := scheduler.Simulate(context.Background(), cluster, jobs, scheduler.PolicyBackfill) + if err == nil || !strings.Contains(err.Error(), "total queue delay exceeds") { + t.Fatalf("Simulate() aggregate wait error = %v", err) + } +} + +func TestSimulateDoesNotMutateInputsAndIsRepeatable(t *testing.T) { + t.Parallel() + + cluster, jobs := timedExampleInputs() + wantCluster := cluster.Clone() + wantJobs := jobs.Clone() + want := mustSimulate(t, cluster, jobs, scheduler.PolicyBackfill) + for iteration := 0; iteration < 50; iteration++ { + got := mustSimulate(t, cluster, jobs, scheduler.PolicyBackfill) + if !reflect.DeepEqual(got, want) { + t.Fatalf("iteration %d differs:\ngot: %#v\nwant: %#v", iteration, got, want) + } + } + if !reflect.DeepEqual(cluster, wantCluster) { + t.Errorf("Simulate() mutated cluster: got %#v, want %#v", cluster, wantCluster) + } + if !reflect.DeepEqual(jobs, wantJobs) { + t.Errorf("Simulate() mutated timed jobs: got %#v, want %#v", jobs, wantJobs) + } +} + +func TestSimulateHonorsCanceledContext(t *testing.T) { + t.Parallel() + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + cluster, jobs := timedExampleInputs() + _, err := scheduler.Simulate(ctx, cluster, jobs, scheduler.PolicyBackfill) + if err == nil || !errors.Is(err, context.Canceled) { + t.Fatalf("Simulate() error = %v, want context canceled", err) + } +} + +func TestSimulateHonorsCancellationDuringActiveWork(t *testing.T) { + t.Parallel() + + cluster, jobs := timedExampleInputs() + ctx := &cancelAfterChecksContext{Context: context.Background(), cancelAt: 6} + _, err := scheduler.Simulate(ctx, cluster, jobs, scheduler.PolicyBackfill) + if err == nil || !errors.Is(err, context.Canceled) { + t.Fatalf("Simulate() error = %v after %d checks, want context canceled", err, ctx.checks) + } + if ctx.checks < ctx.cancelAt { + t.Errorf("context was not checked through active placement: got %d checks, want at least %d", ctx.checks, ctx.cancelAt) + } +} + +func TestSimulateRejectsInvalidTimedInputs(t *testing.T) { + t.Parallel() + + validCluster := model.Cluster{Nodes: []model.Node{{Name: "node-a", Capacity: model.Resources{GPU: 1}}}} + validJobs := model.TimedJobSet{Jobs: []model.TimedJob{ + timedJob("work", 0, 1, 1, 0, 1, ""), + }} + tests := []struct { + name string + cluster model.Cluster + jobs model.TimedJobSet + policy scheduler.Policy + wantMessage string + }{ + { + name: "negative arrival", cluster: validCluster, + jobs: model.TimedJobSet{Jobs: []model.TimedJob{timedJob("work", -1, 1, 1, 0, 1, "")}}, + policy: scheduler.PolicyBackfill, wantMessage: "arrivalTick must be non-negative", + }, + { + name: "zero duration", cluster: validCluster, + jobs: model.TimedJobSet{Jobs: []model.TimedJob{timedJob("work", 0, 0, 1, 0, 1, "")}}, + policy: scheduler.PolicyBackfill, wantMessage: "durationTicks must be greater than zero", + }, + { + name: "missing topology label", + cluster: model.Cluster{Nodes: []model.Node{{ + Name: "node-a", Topology: map[string]string{"zone": "zone-a"}, Capacity: model.Resources{GPU: 1}, + }}}, + jobs: model.TimedJobSet{Jobs: []model.TimedJob{timedJob("work", 0, 1, 1, 0, 1, "rack")}}, + policy: scheduler.PolicyBackfill, wantMessage: `missing topology key "rack"`, + }, + { + name: "unsupported policy", cluster: validCluster, jobs: validJobs, + policy: scheduler.Policy("all"), wantMessage: `unsupported policy "all"`, + }, + } + + for _, test := range tests { + test := test + t.Run(test.name, func(t *testing.T) { + t.Parallel() + _, err := scheduler.Simulate(context.Background(), test.cluster, test.jobs, test.policy) + if err == nil || !strings.Contains(err.Error(), test.wantMessage) { + t.Fatalf("Simulate() error = %v, want containing %q", err, test.wantMessage) + } + }) + } + + if _, err := scheduler.Simulate(nil, validCluster, validJobs, scheduler.PolicyBackfill); err == nil || !strings.Contains(err.Error(), "nil context") { + t.Fatalf("Simulate(nil) error = %v, want nil-context error", err) + } +} + +func assertLifecycleTicks(t *testing.T, lifecycle scheduler.JobLifecycle, want [3]int64) { + t.Helper() + if lifecycle.StartTick == nil || lifecycle.CompletionTick == nil || lifecycle.QueueDelayTicks == nil { + t.Fatalf("job %q lifecycle ticks = start %v, completion %v, delay %v", lifecycle.JobName, lifecycle.StartTick, lifecycle.CompletionTick, lifecycle.QueueDelayTicks) + } + got := [3]int64{*lifecycle.StartTick, *lifecycle.CompletionTick, *lifecycle.QueueDelayTicks} + if got != want { + t.Errorf("job %q lifecycle ticks = %v, want %v", lifecycle.JobName, got, want) + } +} + +func eventSignatures(events []scheduler.SimulationEvent) []string { + signatures := make([]string, len(events)) + for index, event := range events { + signatures[index] = strings.Join([]string{ + strconv.FormatInt(event.Tick, 10), string(event.Type), event.JobName, + }, " ") + } + return signatures +} + +func mustSimulate(t *testing.T, cluster model.Cluster, jobs model.TimedJobSet, policy scheduler.Policy) scheduler.SimulationResult { + t.Helper() + result, err := scheduler.Simulate(context.Background(), cluster, jobs, policy) + if err != nil { + t.Fatalf("Simulate() error = %v", err) + } + return result +} + +func timedExampleInputs() (model.Cluster, model.TimedJobSet) { + cluster := model.Cluster{Nodes: []model.Node{ + node("node-a1", "rack-a", 32, 4), + node("node-a2", "rack-a", 32, 4), + node("node-b1", "rack-b", 32, 4), + node("node-b2", "rack-b", 32, 4), + }} + jobs := model.TimedJobSet{Jobs: []model.TimedJob{ + timedJob("warmup", 0, 8, 8, 4, 1, "zone"), + timedJob("train-xl", 1, 4, 16, 4, 1, "zone"), + timedJob("batch", 2, 2, 4, 4, 1, "rack"), + }} + return cluster, jobs +} + +func timedJob(name string, arrival, duration int64, replicas int, cpu, gpu int64, topology string) model.TimedJob { + return model.TimedJob{ + Job: model.Job{ + Name: name, + Replicas: replicas, + ResourcesPerReplica: model.Resources{CPU: cpu, GPU: gpu}, + RequiredTopology: topology, + }, + ArrivalTick: arrival, + DurationTicks: duration, + } +} + +type cancelAfterChecksContext struct { + context.Context + cancelAt int + checks int +} + +func (c *cancelAfterChecksContext) Err() error { + c.checks++ + if c.checks >= c.cancelAt { + return context.Canceled + } + return c.Context.Err() +} diff --git a/internal/scheduler/state.go b/internal/scheduler/state.go new file mode 100644 index 0000000..ef1f507 --- /dev/null +++ b/internal/scheduler/state.go @@ -0,0 +1,117 @@ +package scheduler + +import ( + "fmt" + "math" + + "github.com/Rionlyu/topoqueue/internal/model" +) + +type pendingRelease struct { + nodeIndex int + resources model.Resources +} + +// releaseAllocation returns the resources represented by allocation to the +// nodes that supplied them. It validates the complete release before changing +// any node so an invalid allocation cannot partially mutate scheduler state. +func releaseAllocation(nodes []nodeState, job model.Job, allocation []Allocation) error { + if job.Replicas <= 0 { + return fmt.Errorf("release job %q: replicas must be greater than zero, got %d", job.Name, job.Replicas) + } + if job.ResourcesPerReplica.CPU < 0 { + return fmt.Errorf("release job %q: resources per replica CPU must be non-negative, got %d", job.Name, job.ResourcesPerReplica.CPU) + } + if job.ResourcesPerReplica.GPU < 0 { + return fmt.Errorf("release job %q: resources per replica GPU must be non-negative, got %d", job.Name, job.ResourcesPerReplica.GPU) + } + if job.ResourcesPerReplica.CPU == 0 && job.ResourcesPerReplica.GPU == 0 { + return fmt.Errorf("release job %q: at least one resource per replica must be positive", job.Name) + } + + nodeIndexes := make(map[string]int, len(nodes)) + for index, node := range nodes { + if _, exists := nodeIndexes[node.name]; exists { + return fmt.Errorf("release job %q: scheduler state contains duplicate node %q", job.Name, node.name) + } + if err := validateNodeResourceState(node); err != nil { + return fmt.Errorf("release job %q: node %q: %w", job.Name, node.name, err) + } + nodeIndexes[node.name] = index + } + + releases := make([]pendingRelease, 0, len(allocation)) + seenNodes := make(map[string]struct{}, len(allocation)) + var allocatedReplicas int64 + for allocationIndex, assigned := range allocation { + if assigned.Replicas <= 0 { + return fmt.Errorf("release job %q allocation at index %d for node %q: replicas must be greater than zero, got %d", job.Name, allocationIndex, assigned.NodeName, assigned.Replicas) + } + nodeIndex, exists := nodeIndexes[assigned.NodeName] + if !exists { + return fmt.Errorf("release job %q allocation at index %d: unknown node %q", job.Name, allocationIndex, assigned.NodeName) + } + if _, duplicate := seenNodes[assigned.NodeName]; duplicate { + return fmt.Errorf("release job %q allocation at index %d: duplicate node %q", job.Name, allocationIndex, assigned.NodeName) + } + seenNodes[assigned.NodeName] = struct{}{} + + replicaCount := int64(assigned.Replicas) + if allocatedReplicas > math.MaxInt64-replicaCount { + return fmt.Errorf("release job %q: allocated replica total exceeds %d", job.Name, int64(math.MaxInt64)) + } + allocatedReplicas += replicaCount + + released, err := checkedResourcesForReplicas(job.ResourcesPerReplica, replicaCount) + if err != nil { + return fmt.Errorf("release job %q allocation for node %q: %w", job.Name, assigned.NodeName, err) + } + node := nodes[nodeIndex] + if released.CPU > node.capacity.CPU-node.remaining.CPU { + return fmt.Errorf("release job %q allocation for node %q: released CPU %d exceeds consumed CPU %d", job.Name, assigned.NodeName, released.CPU, node.capacity.CPU-node.remaining.CPU) + } + if released.GPU > node.capacity.GPU-node.remaining.GPU { + return fmt.Errorf("release job %q allocation for node %q: released GPU %d exceeds consumed GPU %d", job.Name, assigned.NodeName, released.GPU, node.capacity.GPU-node.remaining.GPU) + } + releases = append(releases, pendingRelease{nodeIndex: nodeIndex, resources: released}) + } + + if allocatedReplicas != int64(job.Replicas) { + return fmt.Errorf("release job %q: allocation represents %d replicas, want exactly %d", job.Name, allocatedReplicas, job.Replicas) + } + + for _, release := range releases { + nodes[release.nodeIndex].remaining.CPU += release.resources.CPU + nodes[release.nodeIndex].remaining.GPU += release.resources.GPU + } + return nil +} + +func validateNodeResourceState(node nodeState) error { + if node.capacity.CPU < 0 || node.capacity.GPU < 0 { + return fmt.Errorf("capacity must be non-negative, got CPU %d and GPU %d", node.capacity.CPU, node.capacity.GPU) + } + if node.remaining.CPU < 0 || node.remaining.GPU < 0 { + return fmt.Errorf("remaining capacity must be non-negative, got CPU %d and GPU %d", node.remaining.CPU, node.remaining.GPU) + } + if node.remaining.CPU > node.capacity.CPU || node.remaining.GPU > node.capacity.GPU { + return fmt.Errorf("remaining capacity CPU %d and GPU %d exceeds original capacity CPU %d and GPU %d", node.remaining.CPU, node.remaining.GPU, node.capacity.CPU, node.capacity.GPU) + } + return nil +} + +func checkedResourcesForReplicas(perReplica model.Resources, replicas int64) (model.Resources, error) { + if replicas <= 0 { + return model.Resources{}, fmt.Errorf("replicas must be greater than zero, got %d", replicas) + } + if perReplica.CPU > 0 && replicas > math.MaxInt64/perReplica.CPU { + return model.Resources{}, fmt.Errorf("released CPU exceeds %d", int64(math.MaxInt64)) + } + if perReplica.GPU > 0 && replicas > math.MaxInt64/perReplica.GPU { + return model.Resources{}, fmt.Errorf("released GPU exceeds %d", int64(math.MaxInt64)) + } + return model.Resources{ + CPU: perReplica.CPU * replicas, + GPU: perReplica.GPU * replicas, + }, nil +} diff --git a/internal/scheduler/state_test.go b/internal/scheduler/state_test.go new file mode 100644 index 0000000..5306e52 --- /dev/null +++ b/internal/scheduler/state_test.go @@ -0,0 +1,218 @@ +package scheduler + +import ( + "context" + "math" + "reflect" + "strings" + "testing" + + "github.com/Rionlyu/topoqueue/internal/model" +) + +func TestReleaseAllocationExactlyRestoresPlacedResources(t *testing.T) { + t.Parallel() + + cluster := model.Cluster{Nodes: []model.Node{ + {Name: "node-b", Capacity: model.Resources{CPU: 2, GPU: 1}}, + {Name: "node-a", Capacity: model.Resources{CPU: 4, GPU: 2}}, + }} + job := model.Job{ + Name: "training", + Replicas: 3, + ResourcesPerReplica: model.Resources{CPU: 2, GPU: 1}, + } + nodes := makeNodeStates(cluster) + placed, err := placeJob(context.Background(), nodes, job) + if err != nil { + t.Fatalf("placeJob() error = %v", err) + } + if placed.reason != nil { + t.Fatalf("placeJob() reason = %#v, want admitted", placed.reason) + } + if err := releaseAllocation(nodes, job, placed.allocation); err != nil { + t.Fatalf("releaseAllocation() error = %v", err) + } + + for _, node := range nodes { + if node.remaining != node.capacity { + t.Errorf("node %q remaining = %+v, want original capacity %+v", node.name, node.remaining, node.capacity) + } + } +} + +func TestReleaseAllocationSupportsSingleResourceJobs(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + capacity model.Resources + perReplica model.Resources + }{ + { + name: "CPU only", + capacity: model.Resources{CPU: 8}, + perReplica: model.Resources{CPU: 4}, + }, + { + name: "GPU only", + capacity: model.Resources{GPU: 4}, + perReplica: model.Resources{GPU: 2}, + }, + } + + for _, test := range tests { + test := test + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + job := model.Job{Name: "work", Replicas: 2, ResourcesPerReplica: test.perReplica} + nodes := makeNodeStates(model.Cluster{Nodes: []model.Node{{Name: "node-a", Capacity: test.capacity}}}) + placed, err := placeJob(context.Background(), nodes, job) + if err != nil { + t.Fatalf("placeJob() error = %v", err) + } + if placed.reason != nil { + t.Fatalf("placeJob() reason = %#v, want admitted", placed.reason) + } + if err := releaseAllocation(nodes, job, placed.allocation); err != nil { + t.Fatalf("releaseAllocation() error = %v", err) + } + if nodes[0].remaining != test.capacity { + t.Errorf("remaining = %+v, want %+v", nodes[0].remaining, test.capacity) + } + }) + } +} + +func TestReleaseAllocationRejectsInvalidAllocationWithoutMutation(t *testing.T) { + t.Parallel() + + validJob := model.Job{ + Name: "work", + Replicas: 2, + ResourcesPerReplica: model.Resources{CPU: 2, GPU: 1}, + } + validNodes := []nodeState{ + {name: "node-a", capacity: model.Resources{CPU: 8, GPU: 4}, remaining: model.Resources{CPU: 6, GPU: 3}}, + {name: "node-b", capacity: model.Resources{CPU: 8, GPU: 4}, remaining: model.Resources{CPU: 6, GPU: 3}}, + } + tests := []struct { + name string + nodes []nodeState + job model.Job + allocation []Allocation + wantError string + }{ + { + name: "unknown node after valid entry", + nodes: validNodes, + job: validJob, + allocation: []Allocation{{NodeName: "node-a", Replicas: 1}, {NodeName: "missing", Replicas: 1}}, + wantError: "unknown node", + }, + { + name: "duplicate node", + nodes: validNodes, + job: validJob, + allocation: []Allocation{{NodeName: "node-a", Replicas: 1}, {NodeName: "node-a", Replicas: 1}}, + wantError: "duplicate node", + }, + { + name: "zero replicas", + nodes: validNodes, + job: validJob, + allocation: []Allocation{{NodeName: "node-a", Replicas: 0}}, + wantError: "replicas must be greater than zero", + }, + { + name: "negative replicas", + nodes: validNodes, + job: validJob, + allocation: []Allocation{{NodeName: "node-a", Replicas: -1}}, + wantError: "replicas must be greater than zero", + }, + { + name: "partial replica total", + nodes: validNodes, + job: validJob, + allocation: []Allocation{{NodeName: "node-a", Replicas: 1}}, + wantError: "want exactly 2", + }, + { + name: "resource multiplication overflow", + nodes: []nodeState{{ + name: "node-a", capacity: model.Resources{CPU: math.MaxInt64}, remaining: model.Resources{}, + }}, + job: model.Job{ + Name: "overflow", Replicas: 2, ResourcesPerReplica: model.Resources{CPU: math.MaxInt64}, + }, + allocation: []Allocation{{NodeName: "node-a", Replicas: 2}}, + wantError: "released CPU exceeds", + }, + } + + for _, test := range tests { + test := test + t.Run(test.name, func(t *testing.T) { + t.Parallel() + + nodes := append([]nodeState(nil), test.nodes...) + before := append([]nodeState(nil), nodes...) + err := releaseAllocation(nodes, test.job, test.allocation) + if err == nil { + t.Fatalf("releaseAllocation() error = nil, want containing %q", test.wantError) + } + if !strings.Contains(err.Error(), test.wantError) { + t.Errorf("releaseAllocation() error = %q, want containing %q", err, test.wantError) + } + if !reflect.DeepEqual(nodes, before) { + t.Errorf("releaseAllocation() mutated nodes on error:\ngot: %#v\nwant: %#v", nodes, before) + } + }) + } +} + +func TestReleaseAllocationRejectsOverReleaseWithoutPartialMutation(t *testing.T) { + t.Parallel() + + job := model.Job{ + Name: "work", + Replicas: 2, + ResourcesPerReplica: model.Resources{CPU: 2, GPU: 1}, + } + nodes := []nodeState{ + {name: "node-a", capacity: model.Resources{CPU: 8, GPU: 4}, remaining: model.Resources{CPU: 6, GPU: 3}}, + {name: "node-b", capacity: model.Resources{CPU: 8, GPU: 4}, remaining: model.Resources{CPU: 7, GPU: 4}}, + } + before := append([]nodeState(nil), nodes...) + err := releaseAllocation(nodes, job, []Allocation{ + {NodeName: "node-a", Replicas: 1}, + {NodeName: "node-b", Replicas: 1}, + }) + if err == nil { + t.Fatal("releaseAllocation() error = nil, want over-release error") + } + if !strings.Contains(err.Error(), "exceeds consumed") { + t.Errorf("releaseAllocation() error = %q, want over-release detail", err) + } + if !reflect.DeepEqual(nodes, before) { + t.Errorf("releaseAllocation() partially mutated nodes:\ngot: %#v\nwant: %#v", nodes, before) + } +} + +func TestTerminalizePendingChecksReleasedCapacityWithoutPendingJobs(t *testing.T) { + t.Parallel() + + state := simulationState{ + nodes: []nodeState{{ + name: "node-a", + capacity: model.Resources{CPU: 4, GPU: 1}, + remaining: model.Resources{CPU: 2, GPU: 1}, + }}, + } + err := state.terminalizePending(context.Background()) + if err == nil || !strings.Contains(err.Error(), "want original capacity") { + t.Fatalf("terminalizePending() error = %v, want unreleased-capacity invariant", err) + } +}