diff --git a/cmd/errand/placement.go b/cmd/errand/placement.go index f58232c..b73f0ac 100644 --- a/cmd/errand/placement.go +++ b/cmd/errand/placement.go @@ -52,10 +52,20 @@ 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. + candidates := e.Candidates + // 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() - candidates := e.Candidates + var suppliers []config.RunCandidate + 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. probed := probeCandidates(ctx, candidates, e.Where, q, probe) @@ -86,22 +96,24 @@ 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) { + var supplied []candidateProbe + if len(suppliers) > 0 { + supplied = <-supplierProbes + } + 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..7d081fd 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,37 @@ 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 != "local" { + <-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) + } + // 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