From 112ef4909b4fe4bb2e5c71fe28326ca74223c987 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 7 Oct 2026 02:53:32 +0000 Subject: [PATCH 1/2] Ask lease suppliers alongside the candidates, under one deadline #67 put the local supplier's probe under the candidates' deadline, but it still ran after them, so whenever a peer hung until the deadline the local runner got no time and --where could not rent while any peer was asleep. Suppliers are now probed in the same call as the candidates, first so the fan-out limit never holds them back, and their answers are used only when renting. The bare wildcard skips them. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_013BQxiCNDXMnjv5pvmgo5E6 --- cmd/errand/placement.go | 26 +++++++++++++++----------- cmd/errand/placement_test.go | 28 +++++++++++++++++++--------- 2 files changed, 34 insertions(+), 20 deletions(-) diff --git a/cmd/errand/placement.go b/cmd/errand/placement.go index f58232c..eb82cbe 100644 --- a/cmd/errand/placement.go +++ b/cmd/errand/placement.go @@ -52,13 +52,19 @@ func chooseRunners(ctx context.Context, e config.EffectiveRun, probe placementPr if err != nil { return placementSelection{}, err } - // Every probe, a supplier's asked later included, shares one deadline. - ctx, cancel := context.WithTimeout(ctx, 2*time.Second) - defer cancel() candidates := e.Candidates + // Suppliers, such as the runner installed here, are asked alongside the + // candidates, so selection keeps one deadline and a peer that hangs + // cannot use up a supplier's time. They go first so the fan-out limit + // never holds them back. The bare wildcard never rents. + var suppliers []config.RunCandidate + if !q.Any() { + suppliers = e.Suppliers + } // A ready lease of yours is not a candidate here: it is reused by asking // its cloud peer, which hands it to this run like a new lease. - probed := probeCandidates(ctx, candidates, e.Where, q, probe) + all := probeCandidates(ctx, append(slices.Clip(suppliers), candidates...), e.Where, q, probe) + supplied, probed := all[:len(suppliers)], all[len(suppliers):] var eligible []placementChoice var exclusions []string result := placementSelection{Probed: map[string]proto.Info{}} @@ -86,22 +92,20 @@ func chooseRunners(ctx context.Context, e config.EffectiveRun, probe placementPr // equal ones are tried in random order. How busy a cloud peer's own // runner is does not matter; the job runs on the rented machine. if !matched && !q.Any() { - // Suppliers are asked only now, when renting is needed. - suppliers := append(slices.Clone(candidates), e.Suppliers...) - for i, p := range probeCandidates(ctx, e.Suppliers, e.Where, q, probe) { + brokers := append(slices.Clip(candidates), suppliers...) + for i, p := range supplied { if p.info == nil { - name := e.Suppliers[i].Name + name := suppliers[i].Name result.Excluded = append(result.Excluded, placementExclusion{Peer: name, Reason: p.reason}) exclusions = append(exclusions, fmt.Sprintf("%s: %s", terminalSafeField(name), terminalSafeField(p.reason))) } - probed = append(probed, p) } - for i, p := range probed { + for i, p := range append(slices.Clip(probed), supplied...) { if p.info == nil { continue } if offer, ok := matchingOffer(q, p.info.Offers); ok { - c := suppliers[i] + c := brokers[i] target := client.ConfigureSSHPeer(c.URL, c.Name, c.RemoteCommand, c.RemoteSocket) result.Leases = append(result.Leases, leaseOption{Broker: placementChoice{RunCandidate: c, Info: *p.info, Target: target}, Offer: offer}) } diff --git a/cmd/errand/placement_test.go b/cmd/errand/placement_test.go index 3eb4266..8f3465d 100644 --- a/cmd/errand/placement_test.go +++ b/cmd/errand/placement_test.go @@ -185,10 +185,8 @@ func TestReadyLeasesAreReusedThroughTheirCloudPeer(t *testing.T) { // is asked only once renting is needed, and only for its offers. func TestLocalCloudPeerOnlySuppliesLeases(t *testing.T) { gpu := proto.Facts{OS: "linux", GPUs: []proto.GPU{{Name: "H100", MemoryMiB: 81920}}} - var probed []string cabalFacts := proto.Facts{OS: "linux"} probe := func(_ context.Context, target, _ string, _ time.Duration) (proto.Info, error) { - probed = append(probed, target) if target == "local" { return proto.Info{MaxJobs: 1, Facts: gpu, Offers: []proto.Offer{{Name: "h100", Facts: gpu}}}, nil } @@ -200,10 +198,11 @@ func TestLocalCloudPeerOnlySuppliesLeases(t *testing.T) { if err != nil || len(s.Choices) != 0 || len(s.Leases) != 1 || s.Leases[0].Broker.Name != "local" { t.Fatalf("rent through local: %v %+v", err, s) } - // A runner of yours that matches is used, and local is not asked. - cabalFacts, probed = gpu, nil - if s, err := chooseRunners(context.Background(), e, probe); err != nil || len(s.Choices) != 1 || s.Choices[0].Name != "cabal" || strings.Join(probed, " ") != "cabal" { - t.Fatalf("matching runner: %v %+v probed %q", err, s, probed) + // A runner of yours that matches is used; local, whose own facts match + // too, is never a choice. + cabalFacts = gpu + if s, err := chooseRunners(context.Background(), e, probe); err != nil || len(s.Choices) != 1 || s.Choices[0].Name != "cabal" || len(s.Leases) != 0 { + t.Fatalf("matching runner: %v %+v", err, s) } // With no peers at all, local alone can still rent. e.Candidates = nil @@ -224,9 +223,9 @@ func TestLocalCloudPeerOnlySuppliesLeases(t *testing.T) { } } -// A supplier asked once no runner matched shares the selection's one -// deadline: a slow peer and a slow local runner together still take about -// two seconds, not two each. +// Suppliers are asked alongside the candidates, under one deadline: a slow +// peer and a slow local runner together take about two seconds, not two +// each, and a peer that hangs leaves the local runner its answer. func TestSupplierProbeSharesSelectionDeadline(t *testing.T) { slow := func(ctx context.Context, _, _ string, _ time.Duration) (proto.Info, error) { <-ctx.Done() @@ -238,6 +237,17 @@ func TestSupplierProbeSharesSelectionDeadline(t *testing.T) { if took := time.Since(start); err == nil || took > 3*time.Second { t.Fatalf("took %s: %v", took, err) } + gpu := proto.Facts{OS: "linux", GPUs: []proto.GPU{{Name: "H100", MemoryMiB: 81920}}} + asleep := func(ctx context.Context, target, _ string, _ time.Duration) (proto.Info, error) { + if target == "cabal" { + <-ctx.Done() + return proto.Info{}, ctx.Err() + } + return proto.Info{MaxJobs: 1, Offers: []proto.Offer{{Name: "h100", Facts: gpu}}}, nil + } + if s, err := chooseRunners(context.Background(), e, asleep); err != nil || len(s.Leases) != 1 || s.Leases[0].Broker.Name != "local" { + t.Fatalf("peer asleep: %v %+v", err, s) + } } // Every reachable cloud peer offering a match is a supplier. The cheapest From 4847cc9e34901b4e07c18017830ad8c6d8bdc05e Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 7 Oct 2026 03:05:11 +0000 Subject: [PATCH 2/2] Probe lease suppliers on their own and wait for them only to rent Suppliers shared the candidates' fan-out slots, so enough hung peers could still keep the local runner from being asked, and a hung local runner delayed every placement until the deadline, even when one of your runners matched at once. Suppliers now get their own probe call, started with the candidates under the same deadline, and are awaited only when renting; otherwise the deadline's cancel ends them. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_013BQxiCNDXMnjv5pvmgo5E6 --- cmd/errand/placement.go | 22 +++++++++++++++------- cmd/errand/placement_test.go | 22 +++++++++++++++++++++- 2 files changed, 36 insertions(+), 8 deletions(-) diff --git a/cmd/errand/placement.go b/cmd/errand/placement.go index eb82cbe..b73f0ac 100644 --- a/cmd/errand/placement.go +++ b/cmd/errand/placement.go @@ -53,18 +53,22 @@ func chooseRunners(ctx context.Context, e config.EffectiveRun, probe placementPr return placementSelection{}, err } candidates := e.Candidates - // Suppliers, such as the runner installed here, are asked alongside the - // candidates, so selection keeps one deadline and a peer that hangs - // cannot use up a supplier's time. They go first so the fan-out limit - // never holds them back. The bare wildcard never rents. + // Suppliers, such as the runner installed here, are asked at the same + // time as the candidates and under the same deadline, but on their own, + // so peers that hang can neither use up a supplier's time nor take its + // probe slots. Their answers are awaited only when renting; otherwise + // they are cancelled. The bare wildcard never rents. + ctx, cancel := context.WithTimeout(ctx, 2*time.Second) + defer cancel() var suppliers []config.RunCandidate - if !q.Any() { + supplierProbes := make(chan []candidateProbe, 1) + if !q.Any() && len(e.Suppliers) > 0 { suppliers = e.Suppliers + go func() { supplierProbes <- probeCandidates(ctx, suppliers, e.Where, q, probe) }() } // A ready lease of yours is not a candidate here: it is reused by asking // its cloud peer, which hands it to this run like a new lease. - all := probeCandidates(ctx, append(slices.Clip(suppliers), candidates...), e.Where, q, probe) - supplied, probed := all[:len(suppliers)], all[len(suppliers):] + probed := probeCandidates(ctx, candidates, e.Where, q, probe) var eligible []placementChoice var exclusions []string result := placementSelection{Probed: map[string]proto.Info{}} @@ -92,6 +96,10 @@ func chooseRunners(ctx context.Context, e config.EffectiveRun, probe placementPr // equal ones are tried in random order. How busy a cloud peer's own // runner is does not matter; the job runs on the rented machine. if !matched && !q.Any() { + var supplied []candidateProbe + if len(suppliers) > 0 { + supplied = <-supplierProbes + } brokers := append(slices.Clip(candidates), suppliers...) for i, p := range supplied { if p.info == nil { diff --git a/cmd/errand/placement_test.go b/cmd/errand/placement_test.go index 8f3465d..7d081fd 100644 --- a/cmd/errand/placement_test.go +++ b/cmd/errand/placement_test.go @@ -239,7 +239,7 @@ func TestSupplierProbeSharesSelectionDeadline(t *testing.T) { } gpu := proto.Facts{OS: "linux", GPUs: []proto.GPU{{Name: "H100", MemoryMiB: 81920}}} asleep := func(ctx context.Context, target, _ string, _ time.Duration) (proto.Info, error) { - if target == "cabal" { + if target != "local" { <-ctx.Done() return proto.Info{}, ctx.Err() } @@ -248,6 +248,26 @@ func TestSupplierProbeSharesSelectionDeadline(t *testing.T) { if s, err := chooseRunners(context.Background(), e, asleep); err != nil || len(s.Leases) != 1 || s.Leases[0].Broker.Name != "local" { t.Fatalf("peer asleep: %v %+v", err, s) } + // Peers that hang cannot take the supplier's probe slot either. + for i := 0; i < 12; i++ { + e.Candidates = append(e.Candidates, config.RunCandidate{Name: fmt.Sprint("asleep", i), URL: fmt.Sprint("asleep", i)}) + } + if s, err := chooseRunners(context.Background(), e, asleep); err != nil || len(s.Leases) != 1 || s.Leases[0].Broker.Name != "local" { + t.Fatalf("many peers asleep: %v %+v", err, s) + } + // A supplier that hangs never delays a runner of yours that matches. + e.Candidates = []config.RunCandidate{{Name: "cabal", URL: "cabal"}} + hungSupplier := func(ctx context.Context, target, _ string, _ time.Duration) (proto.Info, error) { + if target == "local" { + <-ctx.Done() + return proto.Info{}, ctx.Err() + } + return proto.Info{MaxJobs: 1, Facts: gpu}, nil + } + start = time.Now() + if s, err := chooseRunners(context.Background(), e, hungSupplier); err != nil || len(s.Choices) != 1 || time.Since(start) > time.Second { + t.Fatalf("matching runner took %s: %v %+v", time.Since(start), err, s) + } } // Every reachable cloud peer offering a match is a supplier. The cheapest