Skip to content

Commit 519151b

Browse files
committed
feat(speculation): standard composed speculator
## Summary ### Why? The Generator ranks and the Allocator spends; something must compose them into the Speculator extension the orchestrator calls. ### What? Adds `speculator/standard`. It funds the queue's most promising paths first until the build budget is spent: candidates considered in descending likelihood, in-flight paths kept funded rather than restarted, finished paths never re-proposed. Pairing bestfirst with sticky yields the default speculation policy; changing either behavior means swapping a part, not writing a new Speculator. standard adds no cancellation handling of its own and inherits it from the two parts; a test pins that a cancelled run yields the context error and no actions. Behavior is documented in `standard/README.md`; the speculator README keeps only the extension contract. ## Test Plan ✅ `bazel test //submitqueue/extension/speculation/...` ✅ `make fmt`, `make gazelle`, `make mocks`
1 parent 23a30ad commit 519151b

5 files changed

Lines changed: 223 additions & 0 deletions

File tree

submitqueue/extension/speculation/speculator/README.md

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,8 @@ The `speculator` package defines the one speculation extension the speculate con
88

99
Like the other extensions, a `Speculator` is selected **per queue** by the wiring layer through the `Config` (queue name) and `Factory` interface. Budget, clock, and any extra data are injected at construction by the integrator, not carried on the contract.
1010

11+
[`standard`](standard/README.md) is the default implementation: it funds the most promising paths first until the build budget is spent.
12+
1113
## Adding a backend
1214

1315
Create a package under `speculator/<backend>/` whose `New(...)` returns a `speculator.Speculator`, injecting whatever it needs at construction. Resolve any content it requires internally; do not add a `Config` or `Factory` implementation here — per-queue routing and the factory adapter live in the wiring layer.
Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,30 @@
1+
load("@rules_go//go:def.bzl", "go_library", "go_test")
2+
3+
go_library(
4+
name = "go_default_library",
5+
srcs = ["standard.go"],
6+
importpath = "github.com/uber/submitqueue/submitqueue/extension/speculation/speculator/standard",
7+
visibility = ["//visibility:public"],
8+
deps = [
9+
"//submitqueue/entity:go_default_library",
10+
"//submitqueue/extension/speculation/allocator:go_default_library",
11+
"//submitqueue/extension/speculation/generator:go_default_library",
12+
"//submitqueue/extension/speculation/speculator:go_default_library",
13+
],
14+
)
15+
16+
go_test(
17+
name = "go_default_test",
18+
srcs = ["standard_test.go"],
19+
embed = [":go_default_library"],
20+
deps = [
21+
"//submitqueue/entity:go_default_library",
22+
"//submitqueue/extension/speculation/allocator/mock:go_default_library",
23+
"//submitqueue/extension/speculation/allocator/sticky:go_default_library",
24+
"//submitqueue/extension/speculation/generator/bestfirst:go_default_library",
25+
"//submitqueue/extension/speculation/generator/mock:go_default_library",
26+
"@com_github_stretchr_testify//assert:go_default_library",
27+
"@com_github_stretchr_testify//require:go_default_library",
28+
"@org_uber_go_mock//gomock:go_default_library",
29+
],
30+
)
Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,13 @@
1+
# standard
2+
3+
The `standard` `Speculator` funds the queue's most promising speculation paths first, until the build budget is spent.
4+
5+
Each run it considers candidate paths in descending order of their probability of being the future that actually happens, and proposes builds down that ranking. Paths already pending or building keep the slot they hold rather than restarting; paths whose builds already finished are skipped for as long as their records remain in the supplied path sets, so a finished path can be proposed again — for a retry, say — once retention drops it; new builds fill whatever budget remains.
6+
7+
When the budget runs out, everything below the cut waits for a later run. That is safe because a batch's verdict never depends on what was funded — only on how its dependencies resolve and which builds pass.
8+
9+
Both halves are swappable. The ranking is the `Generator`'s: the default `bestfirst` scores each path by the probability that all its assumptions hold. The budget policy is the `Allocator`'s: the default `sticky` fills only free slots and never preempts, where a preempting allocator would cancel a low-value in-flight path to fund a better one.
10+
11+
`standard` itself decides nothing — it connects the `Generator`'s stream to the `Allocator` — so changing prioritization or budget behavior means swapping a part, not writing a new `Speculator`.
12+
13+
The `Generator`/`Allocator` split is internal to `standard`, not part of the `Speculator` contract: an alternate `Speculator` could compute build and cancel decisions directly.
Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,53 @@
1+
// Copyright (c) 2025 Uber Technologies, Inc.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
// Package standard provides the standard speculator.Speculator: it composes a
16+
// Generator and an Allocator so a queue's candidate scoring and its budget
17+
// rationing policy can vary independently without changing the Speculator. The
18+
// Speculator itself decides nothing — it opens the Generator's candidate stream
19+
// and hands it to the Allocator; all of its behavior comes from those two parts.
20+
package standard
21+
22+
import (
23+
"context"
24+
25+
"github.com/uber/submitqueue/submitqueue/entity"
26+
"github.com/uber/submitqueue/submitqueue/extension/speculation/allocator"
27+
"github.com/uber/submitqueue/submitqueue/extension/speculation/generator"
28+
"github.com/uber/submitqueue/submitqueue/extension/speculation/speculator"
29+
)
30+
31+
// spec is a speculator.Speculator that opens the generator's candidate stream
32+
// and hands it to the allocator to spend the build budget.
33+
type spec struct {
34+
gen generator.Generator
35+
alloc allocator.Allocator
36+
}
37+
38+
// New returns the standard speculator.Speculator, composing a Generator and an
39+
// Allocator.
40+
func New(gen generator.Generator, alloc allocator.Allocator) speculator.Speculator {
41+
return spec{gen: gen, alloc: alloc}
42+
}
43+
44+
// Speculate opens the generator over the queue's batches, then lets the
45+
// allocator spend the budget over the resulting candidate iterator, reconciling
46+
// it against the path sets.
47+
func (s spec) Speculate(ctx context.Context, batches []entity.Batch, pathSets []entity.SpeculationPathSet) ([]entity.Speculation, error) {
48+
iter, err := s.gen.Open(ctx, batches)
49+
if err != nil {
50+
return nil, err
51+
}
52+
return s.alloc.Allocate(ctx, pathSets, iter)
53+
}
Lines changed: 125 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,125 @@
1+
// Copyright (c) 2025 Uber Technologies, Inc.
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License");
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
//
7+
// http://www.apache.org/licenses/LICENSE-2.0
8+
//
9+
// Unless required by applicable law or agreed to in writing, software
10+
// distributed under the License is distributed on an "AS IS" BASIS,
11+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
// See the License for the specific language governing permissions and
13+
// limitations under the License.
14+
15+
package standard
16+
17+
import (
18+
"context"
19+
"errors"
20+
"testing"
21+
22+
"github.com/stretchr/testify/assert"
23+
"github.com/stretchr/testify/require"
24+
"go.uber.org/mock/gomock"
25+
26+
"github.com/uber/submitqueue/submitqueue/entity"
27+
allocatormock "github.com/uber/submitqueue/submitqueue/extension/speculation/allocator/mock"
28+
"github.com/uber/submitqueue/submitqueue/extension/speculation/allocator/sticky"
29+
"github.com/uber/submitqueue/submitqueue/extension/speculation/generator/bestfirst"
30+
generatormock "github.com/uber/submitqueue/submitqueue/extension/speculation/generator/mock"
31+
)
32+
33+
// assumptionFor returns what a path assumes about a given dependency batch.
34+
func assumptionFor(p entity.SpeculationPath, dep string) entity.DependencyAssumption {
35+
for _, d := range p.Dependencies {
36+
if d.Batch == dep {
37+
return d.Assumption
38+
}
39+
}
40+
return entity.DependencyAssumptionUnknown
41+
}
42+
43+
// constScorer is a minimal scorer.Scorer that scores every batch identically.
44+
type constScorer struct{ v float64 }
45+
46+
func (c constScorer) Score(context.Context, entity.Batch) (float64, error) { return c.v, nil }
47+
48+
func TestComposed_EndToEnd_NaivePair(t *testing.T) {
49+
batches := []entity.Batch{
50+
{ID: "q/1", State: entity.BatchStateSpeculating},
51+
{ID: "q/2", State: entity.BatchStateSpeculating, Dependencies: []string{"q/1"}},
52+
}
53+
54+
// bestfirst generator + sticky allocator with a 2-build budget.
55+
spec := New(bestfirst.New(constScorer{0.9}), sticky.New(2))
56+
got, err := spec.Speculate(context.Background(), batches, nil)
57+
require.NoError(t, err)
58+
59+
// Two free budget slots -> two builds. Everything proposed is a build.
60+
require.Len(t, got, 2)
61+
var q2 entity.Speculation
62+
var found bool
63+
for _, s := range got {
64+
assert.Equal(t, entity.PathActionBuild, s.Action)
65+
if s.Path.Head == "q/2" {
66+
q2, found = s, true
67+
}
68+
}
69+
// q/2's funded path is the optimistic one: it assumes q/1 succeeds.
70+
require.True(t, found, "q/2 should have been funded")
71+
assert.Equal(t, entity.DependencyAssumptionSucceeds, assumptionFor(q2.Path, "q/1"))
72+
}
73+
74+
func TestComposed_WiresGeneratorIntoAllocator(t *testing.T) {
75+
ctrl := gomock.NewController(t)
76+
77+
batches := []entity.Batch{{ID: "q/1", State: entity.BatchStateSpeculating}}
78+
pathSets := []entity.SpeculationPathSet{{Head: "q/1"}}
79+
want := []entity.Speculation{{
80+
Path: entity.SpeculationPath{Head: "q/1"},
81+
Action: entity.PathActionBuild,
82+
}}
83+
84+
iter := generatormock.NewMockPathIterator(ctrl)
85+
gen := generatormock.NewMockGenerator(ctrl)
86+
alloc := allocatormock.NewMockAllocator(ctrl)
87+
88+
gen.EXPECT().Open(gomock.Any(), batches).Return(iter, nil)
89+
alloc.EXPECT().Allocate(gomock.Any(), pathSets, iter).Return(want, nil)
90+
91+
got, err := New(gen, alloc).Speculate(context.Background(), batches, pathSets)
92+
require.NoError(t, err)
93+
assert.Equal(t, want, got)
94+
}
95+
96+
func TestComposed_PropagatesGeneratorError(t *testing.T) {
97+
ctrl := gomock.NewController(t)
98+
99+
errOpen := errors.New("open boom")
100+
gen := generatormock.NewMockGenerator(ctrl)
101+
alloc := allocatormock.NewMockAllocator(ctrl)
102+
103+
gen.EXPECT().Open(gomock.Any(), gomock.Any()).Return(nil, errOpen)
104+
// Allocate must not be called when Open fails (no alloc.EXPECT()).
105+
106+
_, err := New(gen, alloc).Speculate(context.Background(), nil, nil)
107+
require.ErrorIs(t, err, errOpen)
108+
}
109+
110+
func TestComposed_PropagatesContextCancellation(t *testing.T) {
111+
// standard adds no cancellation handling of its own — it inherits it from
112+
// the two parts. This pins that the seam actually propagates: a cancelled
113+
// run yields the context error and no actions.
114+
batches := []entity.Batch{
115+
{ID: "q/1", State: entity.BatchStateSpeculating},
116+
{ID: "q/2", State: entity.BatchStateSpeculating, Dependencies: []string{"q/1"}},
117+
}
118+
ctx, cancel := context.WithCancel(context.Background())
119+
cancel()
120+
121+
spec := New(bestfirst.New(constScorer{0.9}), sticky.New(2))
122+
got, err := spec.Speculate(ctx, batches, nil)
123+
require.ErrorIs(t, err, context.Canceled)
124+
assert.Nil(t, got)
125+
}

0 commit comments

Comments
 (0)