Skip to content

Submit and supervise a job

By the end of this page you'll have submitted a training job, run a supervisor over it, had the scheduler preempt it, and watched dispatch resubmit it and finish it. Everything runs in one Go program against the in-memory adapter and the in-memory record store, so you don't need a cluster. Allow about fifteen minutes.

The finished program is the runnable example Example_superviseAJob in example_test.go at the root of the repository, and go test -run Example_superviseAJob . runs it.

Before you start

You'll need:

  • Go 1.27.2 or later (the version go.mod names);
  • a new module that requires gitlab.com/phpboyscout/go/dispatch.
mkdir dispatch-tutorial && cd dispatch-tutorial
go mod init example.com/dispatch-tutorial
go get gitlab.com/phpboyscout/go/dispatch

dispatch is in development and has no release yet, so go get resolves a pseudo-version.

Everything below goes in one main.go. Start it with the imports and an empty main:

package main

import (
    "context"
    "fmt"
    "log"
    "time"

    "gitlab.com/phpboyscout/go/dispatch"
    "gitlab.com/phpboyscout/go/dispatch/memory"
    "gitlab.com/phpboyscout/go/dispatch/records/recordstest"
    "gitlab.com/phpboyscout/go/dispatch/supervise"
)

func main() {
    ctx := context.Background()
}

Open a cluster over the memory adapter

A Cluster is an adapter composed with dispatch's core. Here the adapter is the memory adapter, shaped like Slurm, and the record store is the in-memory one from recordstest. Add this to main:

    backend, err := memory.New(memory.Config{Shape: memory.ShapeSlurm, AutoRun: true})
    if err != nil {
        log.Fatal(err)
    }

    store := recordstest.New()
    cluster := dispatch.Compose(backend, dispatch.Config{Store: store})

AutoRun: true makes the memory adapter start a job as soon as it's allowed to, which stands in for a scheduler that has free nodes. Without it you'd call backend.Env().Run yourself.

Compose takes a backend you built. In a real program you'd import an adapter and call dispatch.Open(ctx, "slurm", cfg) instead, which finds the adapter by its registered name: see open the Slurm adapter through dispatch or open the Kubeflow adapter from Go.

Submit a job with a recovery rule

A job is a JobSpec. Its recovery policy is a list of rules, each naming the causes it covers and what to do about them. This one says: if the job is preempted, start it again as a new submission, at most twice.

    spec := dispatch.JobSpec{
        Name:    "finetune",
        Command: []string{"torchrun", "train.py"},
        Nodes:   2,
        Recovery: dispatch.RecoveryPolicy{Rules: []dispatch.Rule{
            {On: []dispatch.CauseKind{dispatch.CausePreempted}, Action: dispatch.ActionResubmit, Budget: 2},
        }},
    }
    me := dispatch.Actor{Kind: dispatch.ActorHuman, ID: "ada"}

    plan, err := cluster.Operator().Submit(ctx, spec, me, dispatch.WithSupervisor())
    if err != nil {
        log.Fatal(err)
    }

    fmt.Println("rule 0 is enforced by the", plan.Rules[0].Enforcer)
    fmt.Println("uncovered:", plan.Uncovered)

Every action names who did it, which is what me is for. WithSupervisor() tells dispatch that you'll run a supervisor for this job, so the supervisor can be the one to enforce the rule. Leave it out and Submit refuses the job, because nobody would enforce the rule.

Run it with go run . and you'll see:

rule 0 is enforced by the supervisor
uncovered: [node-lost out-of-memory time-limit non-zero-exit launch-failed no-capacity requeued unknown]

The plan is dispatch's answer before anything is created. It names who enforces each rule, and it lists every cause that no rule covers. If this job loses a node, it fails, and the plan said so up front.

Submit has recorded the job, but it hasn't started anything. Launching is the supervisor's job.

Run the supervisor

The supervisor reads the record, asks the core what should happen next, and does it. Run supervises one job until its record is settled. Run it in the background, and keep its result:

    sup, err := supervise.New(backend, store)
    if err != nil {
        log.Fatal(err)
    }

    done := make(chan error, 1)
    go func() { done <- sup.Run(ctx, plan.Job) }()

Wait for the job to run

You read a job through the cluster's Observer. Add this helper above main. It polls until the job's status satisfies a condition, which takes milliseconds against the memory adapter. A job on a real scheduler takes minutes, and a host would follow its transitions with dispatch.Watch, as watch a job's state changes shows:

func waitFor(ctx context.Context, o dispatch.Observer, id dispatch.JobID, ok func(dispatch.Status) bool) dispatch.Status {
    ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
    defer cancel()

    for {
        st, err := o.Get(ctx, id)
        if err != nil {
            log.Fatal(err)
        }

        if ok(st) {
            return st
        }

        select {
        case <-ctx.Done():
            log.Fatalf("job stayed %s", st.State)
        case <-time.After(5 * time.Millisecond):
        }
    }
}

Then wait for the job to be running:

    obs := cluster.Observer()

    st := waitFor(ctx, obs, plan.Job, func(s dispatch.Status) bool { return s.State == dispatch.StateRunning })
    fmt.Println(st.State, "as submission", st.Submissions[0].N)

running as submission 1

In those few milliseconds the supervisor numbered the first submission, created the job on the scheduler with its start held back, recorded which object is the real one, and only then let it start. The safety explanation says why it goes the long way round.

Preempt it

The memory adapter's Env plays the scheduler's part. Preempt the running job, then wait until a second submission is running:

    if err := backend.Env().Preempt(*st.Submissions[0].Canonical); err != nil {
        log.Fatal(err)
    }

    st = waitFor(ctx, obs, plan.Job, func(s dispatch.Status) bool {
        return s.State == dispatch.StateRunning && len(s.Submissions) == 2
    })
    fmt.Println(st.State, "as submission", st.Submissions[1].N, "after", st.Plan.Rules[0].Firings, "firing of rule 0")

running as submission 2 after 1 firing of rule 0

The supervisor saw the first object end with cause preempted, found rule 0 covering it with budget left, and started submission 2. In between, Get showed the job as recovering, with cause preempted until submission 2 was authorized. It never showed failed, because nothing failed.

Let it finish

Complete the second execution, wait for the supervisor to return, and read the final status:

    if err := backend.Env().Complete(*st.Submissions[1].Canonical); err != nil {
        log.Fatal(err)
    }

    if err := <-done; err != nil {
        log.Fatal(err)
    }

    st, err = obs.Get(ctx, plan.Job)
    if err != nil {
        log.Fatal(err)
    }

    fmt.Println(st.State, "with cause", st.Cause.Kind)

    for _, e := range st.History {
        fmt.Printf("  %s -> %s (submission %d)\n", e.From, e.To, e.Submission)
    }

succeeded with cause none
  pending -> recovering (submission 2)
  recovering -> succeeded (submission 2)

Run returned nil because the record settled: the job reached an outcome and every object it ever created has ended.

History holds the transitions committed to the record. Running isn't one of them: whether a job is running comes from what the scheduler shows when you ask, so it appears in State and not in History.

What you built

You submitted a job whose plan said, before it started, who would recover it and from what. A supervisor launched it, recovered it from preemption within the rule's budget, and settled it, and every step is in the record.

Next