diff --git a/doc/rfc/index.md b/doc/rfc/index.md index aa4f41311..2974cd970 100644 --- a/doc/rfc/index.md +++ b/doc/rfc/index.md @@ -19,6 +19,7 @@ Design documents and technical proposals, grouped by scope. Shared/cross-cutting - [Extension Contract](submitqueue/extension-contract.md) - When extensions take orchestrator identity (request/batch) and resolve granular content themselves vs. take controller-resolved data; revises the BuildRunner base/head contract - [Gateway Status and List APIs](submitqueue/status-list-api.md) - Gateway-owned request context, materialized current status, sqid or change-URI status lookup, and queue admission listing - [Speculation](submitqueue/speculation.md) - Why SubmitQueue speculates, the path/tree model, and the two pluggable seams: speculation-tree enumeration and path selection +- [Outcome Predictor](submitqueue/outcome-predictor.md) - How likely a batch is to succeed: a predictor built with a Scorer that multiplies its price by what the pipeline has observed (a build passed, the batch is merging), factors written by hand first and fitted later - [Best-First Speculation Path Generation](submitqueue/speculation-generator-best-first.md) - The default Generator: per-head lazy streams of flip subsets merged best-first across heads, log-probability ranking, and the strict snapshot contract - [Modular Queue Wiring](submitqueue/modular-queue-wiring.md) - Declare-don't-assemble engine (`pipeline.Construct`) that unifies topic registry, controller registration, DLQ pairing, and lifecycle ordering into one typed call; services self-declare via Deps struct + Stages slice, hosts own per-queue profiles and transport diff --git a/doc/rfc/submitqueue/outcome-predictor.md b/doc/rfc/submitqueue/outcome-predictor.md new file mode 100644 index 000000000..8202157c3 --- /dev/null +++ b/doc/rfc/submitqueue/outcome-predictor.md @@ -0,0 +1,103 @@ +# Outcome Predictor + +How likely a batch is to succeed, from a Scorer's price plus what the pipeline has since observed about the batch. + +## Problem + +Nothing the pipeline learns about a batch changes its price. A batch whose build has passed, whose dependencies have landed, and which is being merged is priced exactly as it was before anything was known about it — on the size of its diff. The queue holds that evidence in memory during the run that needs it, and throws it away. + +The prices are also fixed numbers someone typed. The heuristic scorer (`submitqueue/extension/speculation/scorer/heuristic`) maps lines changed onto a bucket table, an unconfigured queue prices everything at 0.5, and the bestfirst generator (`submitqueue/extension/speculation/generator/bestfirst`) uses 0.95 for anything it could not price. Nothing improves them over time. + +This document covers the first problem only. Pricing a change from its content is a separate question, and it stays in whatever Scorer a queue configures for it. + +## What it does + +The predictor is its own contract, separate from the Scorer. It is handed a batch and what that batch's builds have done, and it returns how likely the batch is to reach Succeeded. It is built with a Scorer: it asks the Scorer for the batch's price, then multiplies the **odds** of that price by one factor per piece of evidence. + +The generator depends on the predictor, not on the Scorer. + +A batch the base prices at 0.6 has odds of 1.5 to 1. Its build passes, worth ten times the odds: 15 to 1, or 0.94. It reaches merging, worth another twelve: 180 to 1, or 0.995. The generator already resolves a batch to certainty once it is terminal, so the predictor never has to handle that case. + +Multiplying odds keeps the answer a probability without clamping, and it makes each factor mean the same thing wherever it applies — the same evidence is worth the same amount whether the base said 0.5 or 0.95. Adding to the probability directly has neither property. + +The factors are what gets configured, and later learned. Written as logs and summed, this is ordinary logistic regression, so fitting it needs nothing special. + +**All factors at one, the predictor returns the base price unchanged.** Turning it on is a no-op until someone sets a factor. + +## Where the line falls + +**The scorer prices the change. The predictor prices the situation.** + +Lines, files, which directories, who wrote it — that describes the change, and it belongs in the scorer. Builds, batch state, dependencies, time waiting — that describes where the batch sits right now, and it belongs in the predictor. + +That is also why they are two contracts rather than one. The evidence is an input the scorer has no use for: put it on `Score` and every implementation that prices content — all three that exist — takes a parameter it discards. A parameter every implementation throws away belongs to a different contract. + +Keeping them apart means either side can be replaced without redoing the other, and it leaves the scorer per-queue configurable, which it already is. + +## What it looks at + +| Evidence | Direction | +| --- | --- | +| A build passed for this batch | Strongly up — the biggest factor, with the caveat below | +| Builds failed for this batch, counted | Down | +| How old the passing build is | Down as it ages — trunk moves | +| A build is running | Slightly up: better than never tried | +| The batch is merging | Strongly up — it cleared speculation and is pushing, but can still lose a race | +| The batch is cancelling | Strongly down, though not to zero — cancelling is best effort | +| How many dependencies it has | Down — more assumptions that can break | +| How long it has been in the queue | Down — usually it has been invalidated before | +| The queue's recent landing rate | Moves everything, so a bad week does not need re-fitting | + +Two of these are decisions rather than just numbers. + +**A passing build only counts under the assumptions it was built with.** A dependency can hold a green build for a path assuming *its* dependency succeeds, while the candidate being priced assumes that one fails. That build says nothing about the second case. So the evidence is narrower than "a build passed": it is "a build passed on the all-succeed path", which is right whenever the candidate assumes the same — the common case — and must not count otherwise. + +**Merging and cancelling belong here and nowhere else.** The generator may only tell terminal from non-terminal; the allocator reads path status only. Pinning a merging dependency to certain inside the generator was tried and reverted, correctly — a merge can fail, so nothing is settled. How much a state is worth is a price, and that price is the predictor's. + +## Getting the evidence + +Batch state comes free: the predictor is handed the batch, which carries it. The build evidence does not. The speculate controller reads each in-flight head's path set once per run and hands it to the Speculator; the Generator never sees it. + +So the Generator takes the path sets alongside the batches, and hands each batch its own set when it asks for a prediction. The Scorer contract does not change at all. + +The alternative — the predictor reads the path-set store itself — needs no contract change but re-reads what the run already holds, once per dependency, and can see a newer version than the rest of the run is working from. The run reads once so that two decisions in it can never disagree about the world, and that is worth more than the plumbing costs. + +## Fitting the factors + +Written by hand first. A factor is "how much does this multiply the odds", which is a number an engineer can propose and a reviewer can argue with, so the first version needs no data at all. + +Learning them later needs three things: + +- **A record written when the price is set** — the batch, the base price, each piece of evidence, the result, and the predictor version. Writing it when the outcome arrives instead would record values the serving path never has. +- **Outcomes joined to it.** Succeeded is a yes, failed is a no, and a batch the author cancelled is neither, so it is left out — counting it as a failure would teach the predictor that abandoned work is bad code. Batch outcomes are not logged today; they can be derived from request logs through the request-batch store. +- **An awareness that the data is biased.** Only paths that got funded produce outcomes, so the records show what the previous ranking already believed. Fitted factors will lean toward agreeing with it. + +The fitted result is a small versioned file of named factors loaded at wiring time, with the evidence list hashed so a file that does not match the code fails at startup rather than mispricing quietly. + +## Judging it + +Whether the probabilities are honest is measurable, but it is not the point. The point is builds per landed change, share of builds spent on paths later thrown away, and how long the queue takes. Measuring those before production means replaying recorded runs against a candidate predictor. No such harness exists, and that is the main thing between fitted factors and trusting them. + +## Rollout + +1. Start recording prices and outcomes. Nothing else works without it. +2. Ship the predictor with all factors at one — identical output — then set factors for builds and batch state by hand. This is the largest win and it needs no data. +3. Fit the factors. Run it alongside the current ranking first, then one queue, then everywhere. + +## Not in scope + +Worth doing, separately: predicting how long and how large a build will be, so the allocator can rank by value per CI-minute rather than by probability alone; per-directory history in the base; and pricing a dependency against the specific assumptions a path makes about its own dependencies, which would remove the caveat on build evidence. + +## Rejected + +**One contract, with the evidence on `Score`.** Tried: every scorer that prices content took a parameter it discarded, and the composite forwarded one it never read. The evidence is not part of pricing a change. + +**One estimate over content and evidence together.** They change at different rates and need different amounts of data, and it would force every queue onto the same content scorer. + +**Adding to the probability instead of multiplying odds.** Leaves the range, needs clamping, and the same increment means different things at different prices. + +**More dimensions on the bucket table.** A second dimension squares it, a third makes it unwritable, and every cell is still a guess. + +**Putting state in the generator.** Tried and reverted: a merge can fail, so nothing is settled, and how much a state is worth is a price. + +**A scoring stage.** Prices only mean anything inside the run that produced them; storing them would make them stale by construction. diff --git a/service/submitqueue/orchestrator/server/BUILD.bazel b/service/submitqueue/orchestrator/server/BUILD.bazel index c1f3332e3..956f4de7e 100644 --- a/service/submitqueue/orchestrator/server/BUILD.bazel +++ b/service/submitqueue/orchestrator/server/BUILD.bazel @@ -49,6 +49,8 @@ go_library( "//submitqueue/extension/conflict/pathoverlap:go_default_library", "//submitqueue/extension/speculation/allocator/sticky:go_default_library", "//submitqueue/extension/speculation/generator/bestfirst:go_default_library", + "//submitqueue/extension/speculation/predictor:go_default_library", + "//submitqueue/extension/speculation/predictor/regression:go_default_library", "//submitqueue/extension/speculation/scorer:go_default_library", "//submitqueue/extension/speculation/scorer/composite:go_default_library", "//submitqueue/extension/speculation/scorer/fake:go_default_library", @@ -118,6 +120,7 @@ go_test( "//submitqueue/extension/buildrunner:go_default_library", "//submitqueue/extension/changeprovider:go_default_library", "//submitqueue/extension/conflict:go_default_library", + "//submitqueue/extension/speculation/predictor:go_default_library", "//submitqueue/extension/speculation/scorer:go_default_library", "//submitqueue/extension/speculation/speculator:go_default_library", "//submitqueue/extension/storage:go_default_library", diff --git a/service/submitqueue/orchestrator/server/config.go b/service/submitqueue/orchestrator/server/config.go index 945374522..4e6d51a81 100644 --- a/service/submitqueue/orchestrator/server/config.go +++ b/service/submitqueue/orchestrator/server/config.go @@ -67,6 +67,22 @@ const ( // Ways a composite scorer combines its components. const combineAvg = "avg" +// Predictor types selectable from configuration. +const predictorTypeRegression = "regression" + +// Evidence a regression predictor prices, as named in configuration. The set is +// closed: a factor under any other name would be applied to nothing and never +// noticed. +const ( + factorPathPassed = "pathPassed" + factorPathFailed = "pathFailed" + factorMerging = "merging" + factorCancelling = "cancelling" +) + +// neutralFactor leaves the scorer's price untouched: odds multiplied by one. +const neutralFactor = 1.0 + // defaultBuildBudget is how many builds a queue may have occupying CI at once // when it states no budget of its own. Four is enough for speculation to be // visible — a queue that can only build one path never speculates — while @@ -110,6 +126,7 @@ type namedQueueProfileConfig struct { Analyzer *analyzerConfig `yaml:"analyzer"` Scorer *scorerConfig `yaml:"scorer"` Speculator *speculatorConfig `yaml:"speculator"` + Predictor *predictorConfig `yaml:"predictor"` } // queueProfileConfig is the full set of extensions a queue resolves to. @@ -119,6 +136,7 @@ type queueProfileConfig struct { Analyzer analyzerConfig `yaml:"analyzer"` Scorer scorerConfig `yaml:"scorer"` Speculator speculatorConfig `yaml:"speculator"` + Predictor predictorConfig `yaml:"predictor"` } // changeProviderConfig selects how change metadata is fetched. The github and @@ -239,6 +257,17 @@ type speculatorConfig struct { BuildBudget int `yaml:"buildBudget"` } +// predictorConfig tunes how a queue turns its scorer's price into the +// probability the generator ranks on. The scorer being revised is the queue's +// own, so it is not named again here. +type predictorConfig struct { + Type string `yaml:"type"` + // Factors multiply the odds of the scorer's price, one per piece of + // evidence, keyed by evidence name. An omitted factor is neutral, so an + // omitted block ranks on the scorer's price alone. + Factors map[string]float64 `yaml:"factors"` +} + // loadProfilesConfig reads and validates the profiles configuration at path. func loadProfilesConfig(path string) (profilesConfig, error) { data, err := os.ReadFile(path) @@ -300,6 +329,11 @@ func (c *profilesConfig) normalizeAndValidate() error { return err } } + if q.Predictor != nil { + if err := q.Predictor.normalizeAndValidate(where); err != nil { + return err + } + } } return c.validateGitRepoPaths() } @@ -358,6 +392,9 @@ func (c profilesConfig) resolve(q namedQueueProfileConfig) queueProfileConfig { if q.Speculator != nil { profile.Speculator = *q.Speculator } + if q.Predictor != nil { + profile.Predictor = *q.Predictor + } return profile } @@ -374,7 +411,10 @@ func (p *queueProfileConfig) normalizeAndValidate(where string) error { if err := p.Scorer.normalizeAndValidate(where); err != nil { return err } - return p.Speculator.normalizeAndValidate(where) + if err := p.Speculator.normalizeAndValidate(where); err != nil { + return err + } + return p.Predictor.normalizeAndValidate(where) } func (c *changeProviderConfig) normalizeAndValidate(where string) error { @@ -556,6 +596,31 @@ func (s *scorerConfig) normalizeAndValidate(where string) error { return nil } +// normalizeAndValidate applies defaults and rejects a predictor that could not +// be built. An empty block is a regression predictor with every factor neutral, +// which prices a batch at exactly its scorer's price. +func (p *predictorConfig) normalizeAndValidate(where string) error { + if p.Type == "" { + p.Type = predictorTypeRegression + } + if p.Type != predictorTypeRegression { + return fmt.Errorf("%s: unknown predictor type %q", where, p.Type) + } + for name, factor := range p.Factors { + switch name { + case factorPathPassed, factorPathFailed, factorMerging, factorCancelling: + default: + return fmt.Errorf("%s: unknown predictor factor %q", where, name) + } + // Zero would pin every batch carrying the evidence to a probability of + // zero, and a negative multiplier on odds means nothing at all. + if factor <= 0 { + return fmt.Errorf("%s: predictor factor %q is %v, must be positive", where, name, factor) + } + } + return nil +} + func (s *speculatorConfig) normalizeAndValidate(where string) error { // A negative budget is rejected rather than clamped: sticky would compute no // free slots from it, so the queue would batch and then never build anything, diff --git a/service/submitqueue/orchestrator/server/config_test.go b/service/submitqueue/orchestrator/server/config_test.go index 2de8278c5..deaacc6c1 100644 --- a/service/submitqueue/orchestrator/server/config_test.go +++ b/service/submitqueue/orchestrator/server/config_test.go @@ -666,3 +666,49 @@ func TestLoadProfilesConfig_RejectsBadScorers(t *testing.T) { }) } } + +func TestLoadProfilesConfig_RejectsBadPredictors(t *testing.T) { + tests := []struct { + name string + contents string + }{ + {name: "unknown predictor type", contents: "defaults:\n predictor: {type: vibes}\n"}, + {name: "unknown factor", contents: "defaults:\n predictor:\n factors: {pathPased: 2}\n"}, + {name: "zero factor", contents: "defaults:\n predictor:\n factors: {merging: 0}\n"}, + {name: "negative factor", contents: "defaults:\n predictor:\n factors: {pathFailed: -1}\n"}, + {name: "bad factor on a queue override", contents: "defaults: {}\nqueues:\n - name: q\n predictor:\n factors: {merging: 0}\n"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + _, err := loadProfilesConfig(writeProfiles(t, tt.contents)) + require.Error(t, err) + }) + } +} + +// An omitted predictor block leaves the queue ranking on its scorer's price +// alone, which is what every queue does until someone states a factor. +func TestLoadProfilesConfig_DefaultsThePredictorToNeutral(t *testing.T) { + cfg, err := loadProfilesConfig(writeProfiles(t, "defaults: {}\nqueues:\n - name: q\n")) + require.NoError(t, err) + + assert.Equal(t, predictorTypeRegression, cfg.Defaults.Predictor.Type) + + factors := factorsFrom(cfg.resolve(cfg.Queues[0]).Predictor) + assert.Equal(t, neutralFactor, factors.PathPassed) + assert.Equal(t, neutralFactor, factors.PathFailed) + assert.Equal(t, neutralFactor, factors.Merging) + assert.Equal(t, neutralFactor, factors.Cancelling) +} + +func TestLoadProfilesConfig_ReadsPredictorFactors(t *testing.T) { + cfg, err := loadProfilesConfig(writeProfiles(t, + "defaults:\n predictor:\n factors: {pathPassed: 10, pathFailed: 0.3, merging: 12, cancelling: 0.1}\n")) + require.NoError(t, err) + + factors := factorsFrom(cfg.Defaults.Predictor) + assert.Equal(t, 10.0, factors.PathPassed) + assert.Equal(t, 0.3, factors.PathFailed) + assert.Equal(t, 12.0, factors.Merging) + assert.Equal(t, 0.1, factors.Cancelling) +} diff --git a/service/submitqueue/orchestrator/server/profiles.go b/service/submitqueue/orchestrator/server/profiles.go index 31b72c58c..0e8807089 100644 --- a/service/submitqueue/orchestrator/server/profiles.go +++ b/service/submitqueue/orchestrator/server/profiles.go @@ -45,12 +45,14 @@ import ( conflictfake "github.com/uber/submitqueue/submitqueue/extension/conflict/fake" "github.com/uber/submitqueue/submitqueue/extension/conflict/none" "github.com/uber/submitqueue/submitqueue/extension/conflict/pathoverlap" + "github.com/uber/submitqueue/submitqueue/extension/speculation/allocator/sticky" + "github.com/uber/submitqueue/submitqueue/extension/speculation/generator/bestfirst" + "github.com/uber/submitqueue/submitqueue/extension/speculation/predictor" + "github.com/uber/submitqueue/submitqueue/extension/speculation/predictor/regression" "github.com/uber/submitqueue/submitqueue/extension/speculation/scorer" "github.com/uber/submitqueue/submitqueue/extension/speculation/scorer/composite" scorerfake "github.com/uber/submitqueue/submitqueue/extension/speculation/scorer/fake" "github.com/uber/submitqueue/submitqueue/extension/speculation/scorer/heuristic" - "github.com/uber/submitqueue/submitqueue/extension/speculation/allocator/sticky" - "github.com/uber/submitqueue/submitqueue/extension/speculation/generator/bestfirst" "github.com/uber/submitqueue/submitqueue/extension/speculation/speculator" specstandard "github.com/uber/submitqueue/submitqueue/extension/speculation/speculator/standard" "github.com/uber/submitqueue/submitqueue/extension/storage" @@ -80,6 +82,10 @@ type Profile struct { // likely their assumptions are to hold. Scorer scorer.Factory + // Predictor turns this queue's scorer price into the probability the + // generator ranks on, revising it with the batch's observed progress. + Predictor predictor.Factory + // Speculator decides which of this queue's speculation paths to build and // which running ones to preempt, within the build budget. Speculator speculator.Factory @@ -142,6 +148,14 @@ func (p Profiles) ScorerFactory() scorer.Factory { }) } +// PredictorFactory returns a predictor.Factory that resolves the +// OutcomePredictor for each queue from the profile registry. +func (p Profiles) PredictorFactory() predictor.Factory { + return predictorFunc(func(c predictor.Config) (predictor.OutcomePredictor, error) { + return p.For(c.QueueName).Predictor.For(c) + }) +} + // StorageFactory returns a storage.Factory that routes each queue to its // profile's storage backend before binding the queue-scoped store aggregate. func (p Profiles) StorageFactory() storage.Factory { @@ -176,6 +190,10 @@ type scorerFunc func(scorer.Config) (scorer.Scorer, error) func (f scorerFunc) For(c scorer.Config) (scorer.Scorer, error) { return f(c) } +type predictorFunc func(predictor.Config) (predictor.OutcomePredictor, error) + +func (f predictorFunc) For(c predictor.Config) (predictor.OutcomePredictor, error) { return f(c) } + type speculatorFunc func(speculator.Config) (speculator.Speculator, error) func (f speculatorFunc) For(c speculator.Config) (speculator.Speculator, error) { return f(c) } @@ -265,33 +283,71 @@ func (b *profileBuilder) build(cfg queueProfileConfig, where string) (Profile, e if err != nil { return Profile{}, err } - // The speculator is composed last, because it is built from whatever scorer - // the profile ended up with. - return withSpeculator(Profile{ + // The predictor and the speculator are composed last, because each is built + // from what the profile ended up with one level below it. + return withSpeculator(withPredictor(Profile{ ChangeProvider: provider, BuildRunner: runner, Analyzer: analyzer, Storage: b.stores, Scorer: sc, - }, cfg.Speculator.BuildBudget), nil + }, cfg.Predictor, b.scope), cfg.Speculator.BuildBudget), nil +} + +// withPredictor returns the profile with its predictor composed over its own +// scorer: the scorer prices the batch's change, and the predictor revises that +// price with what the batch's builds have done. +// +// The scorer is resolved lazily, at the queue the predictor itself was asked +// for, so the queue's identity reaches one level down into the scorer too. +func withPredictor(p Profile, cfg predictorConfig, scope tally.Scope) Profile { + p.Predictor = predictorFunc(func(c predictor.Config) (predictor.OutcomePredictor, error) { + sc, err := p.Scorer.For(scorer.Config{QueueName: c.QueueName}) + if err != nil { + return nil, fmt.Errorf("failed to resolve scorer for queue %q: %w", c.QueueName, err) + } + return regression.New(c, sc, factorsFrom(cfg), scope.SubScope("predictor")), nil + }) + return p +} + +// factorsFrom reads the configured factors onto the named fields the predictor +// takes, leaving an unstated one neutral. Names are validated when the config +// is loaded. +func factorsFrom(cfg predictorConfig) regression.Factors { + factors := regression.AllOnes() + for name, factor := range cfg.Factors { + switch name { + case factorPathPassed: + factors.PathPassed = factor + case factorPathFailed: + factors.PathFailed = factor + case factorMerging: + factors.Merging = factor + case factorCancelling: + factors.Cancelling = factor + } + } + return factors } // withSpeculator returns the profile with its speculator composed from its own -// scorer: bestfirst ranks a queue's candidate paths by how likely all their +// predictor: bestfirst ranks a queue's candidate paths by how likely all their // assumptions are to hold, and sticky spends buildBudget down that ranking // without preempting builds already running. Swapping either part changes the // policy without touching the speculate controller, which depends only on the // Speculator contract. // -// The scorer is resolved lazily, at the queue the speculator itself was asked -// for, so the queue's identity reaches one level down into the scorer too. +// The predictor is resolved lazily, at the queue the speculator itself was +// asked for, so the queue's identity reaches down through the predictor to the +// scorer under it. func withSpeculator(p Profile, buildBudget int) Profile { p.Speculator = speculatorFunc(func(c speculator.Config) (speculator.Speculator, error) { - sc, err := p.Scorer.For(scorer.Config{QueueName: c.QueueName}) + pred, err := p.Predictor.For(predictor.Config{QueueName: c.QueueName}) if err != nil { - return nil, fmt.Errorf("failed to resolve scorer for queue %q: %w", c.QueueName, err) + return nil, fmt.Errorf("failed to resolve predictor for queue %q: %w", c.QueueName, err) } - return specstandard.New(c, bestfirst.New(sc), sticky.New(buildBudget)), nil + return specstandard.New(c, bestfirst.New(pred), sticky.New(buildBudget)), nil }) return p } diff --git a/service/submitqueue/orchestrator/server/profiles_test.go b/service/submitqueue/orchestrator/server/profiles_test.go index 25a7fb7ec..8a44c8629 100644 --- a/service/submitqueue/orchestrator/server/profiles_test.go +++ b/service/submitqueue/orchestrator/server/profiles_test.go @@ -15,15 +15,19 @@ package main import ( + "context" "errors" "testing" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "github.com/uber-go/tally" + "github.com/uber/submitqueue/submitqueue/entity" "github.com/uber/submitqueue/submitqueue/extension/buildrunner" "github.com/uber/submitqueue/submitqueue/extension/changeprovider" "github.com/uber/submitqueue/submitqueue/extension/conflict" + "github.com/uber/submitqueue/submitqueue/extension/speculation/predictor" "github.com/uber/submitqueue/submitqueue/extension/speculation/scorer" "github.com/uber/submitqueue/submitqueue/extension/speculation/speculator" "github.com/uber/submitqueue/submitqueue/extension/storage" @@ -37,8 +41,15 @@ type recorder struct { analyzer string storage string scorer string + predictor string } +// stubScorer stands in wherever a Profile's scorer has to be real rather than +// nil, because something is composed over it. +type stubScorer struct{} + +func (stubScorer) Score(_ context.Context, _ entity.Batch) (float64, error) { return 0.5, nil } + // profileRecording returns a Profile whose every factory records the queue name // it receives into rec and returns a nil implementation. Nil is fine: these // tests are about what reaches the factory, not what it builds. @@ -62,6 +73,10 @@ func profileRecording(rec *recorder) Profile { }), Scorer: scorerFunc(func(c scorer.Config) (scorer.Scorer, error) { rec.scorer = c.QueueName + return stubScorer{}, nil + }), + Predictor: predictorFunc(func(c predictor.Config) (predictor.OutcomePredictor, error) { + rec.predictor = c.QueueName return nil, nil }), } @@ -113,10 +128,10 @@ func TestProfilesForwardQueueNameToFactories(t *testing.T) { } } -// TestWithSpeculatorResolvesScorerAtSameQueue covers the one seam that resolves -// another seam: the speculator is composed from the profile's scorer, and must -// ask for it at the queue it was itself asked for. -func TestWithSpeculatorResolvesScorerAtSameQueue(t *testing.T) { +// The speculator is composed from the profile's predictor, which is itself +// composed over the profile's scorer. Each has to be asked for at the queue the +// one above it was asked for, or an implementation is built for the wrong one. +func TestWithSpeculatorResolvesPredictorAtSameQueue(t *testing.T) { var rec recorder profile := withSpeculator(profileRecording(&rec), defaultBuildBudget) profiles := Profiles{defaultProfile: profile} @@ -124,20 +139,39 @@ func TestWithSpeculatorResolvesScorerAtSameQueue(t *testing.T) { spec, err := profiles.SpeculatorFactory().For(speculator.Config{QueueName: "unlisted-queue"}) require.NoError(t, err) assert.NotNil(t, spec) + assert.Equal(t, "unlisted-queue", rec.predictor) +} + +func TestWithPredictorResolvesScorerAtSameQueue(t *testing.T) { + var rec recorder + profile := withPredictor(profileRecording(&rec), predictorConfig{}, tally.NoopScope) + + pred, err := profile.Predictor.For(predictor.Config{QueueName: "unlisted-queue"}) + require.NoError(t, err) + assert.NotNil(t, pred) assert.Equal(t, "unlisted-queue", rec.scorer) } -// TestWithSpeculatorPropagatesScorerError covers the error path the factory -// conversion introduced: resolving the scorer can now fail where reading a -// struct field could not, and the failure must surface rather than yielding a -// speculator built over a nil scorer. -func TestWithSpeculatorPropagatesScorerError(t *testing.T) { - sentinel := errors.New("scorer unavailable") +// Resolving either level can fail where reading a struct field could not, and +// the failure must surface rather than yielding something built over a nil. +func TestWithSpeculatorPropagatesPredictorError(t *testing.T) { + sentinel := errors.New("predictor unavailable") profile := withSpeculator(Profile{ - Scorer: scorerFunc(func(scorer.Config) (scorer.Scorer, error) { return nil, sentinel }), + Predictor: predictorFunc(func(predictor.Config) (predictor.OutcomePredictor, error) { return nil, sentinel }), }, defaultBuildBudget) spec, err := profile.Speculator.For(speculator.Config{QueueName: "any-queue"}) require.ErrorIs(t, err, sentinel) assert.Nil(t, spec) } + +func TestWithPredictorPropagatesScorerError(t *testing.T) { + sentinel := errors.New("scorer unavailable") + profile := withPredictor(Profile{ + Scorer: scorerFunc(func(scorer.Config) (scorer.Scorer, error) { return nil, sentinel }), + }, predictorConfig{}, tally.NoopScope) + + pred, err := profile.Predictor.For(predictor.Config{QueueName: "any-queue"}) + require.ErrorIs(t, err, sentinel) + assert.Nil(t, pred) +} diff --git a/submitqueue/extension/speculation/generator/bestfirst/BUILD.bazel b/submitqueue/extension/speculation/generator/bestfirst/BUILD.bazel index 3fbed48cb..e6d8d4f8d 100644 --- a/submitqueue/extension/speculation/generator/bestfirst/BUILD.bazel +++ b/submitqueue/extension/speculation/generator/bestfirst/BUILD.bazel @@ -8,7 +8,7 @@ go_library( deps = [ "//submitqueue/entity:go_default_library", "//submitqueue/extension/speculation/generator:go_default_library", - "//submitqueue/extension/speculation/scorer:go_default_library", + "//submitqueue/extension/speculation/predictor:go_default_library", ], ) @@ -19,7 +19,7 @@ go_test( deps = [ "//submitqueue/entity:go_default_library", "//submitqueue/extension/speculation/generator:go_default_library", - "//submitqueue/extension/speculation/scorer:go_default_library", + "//submitqueue/extension/speculation/predictor:go_default_library", "@com_github_stretchr_testify//assert:go_default_library", "@com_github_stretchr_testify//require:go_default_library", ], diff --git a/submitqueue/extension/speculation/generator/bestfirst/bestfirst.go b/submitqueue/extension/speculation/generator/bestfirst/bestfirst.go index e0e4d45bc..dab1d7ee8 100644 --- a/submitqueue/extension/speculation/generator/bestfirst/bestfirst.go +++ b/submitqueue/extension/speculation/generator/bestfirst/bestfirst.go @@ -14,8 +14,8 @@ // Package bestfirst provides a probability-ordered speculation path generator. // -// Throughout, a probability is the [0, 1] value a scorer gives; a score is its -// logarithm. Scores are summed and compared, never exponentiated, so wide +// Throughout, a probability is the [0, 1] value a predictor gives; a score is +// its logarithm. Scores are summed and compared, never exponentiated, so wide // heads cannot underflow into ties. The algorithm — per-head streams // enumerating flip subsets lazily, merged through one global heap — is // documented in doc/rfc/submitqueue/speculation-generator-best-first.md. @@ -30,29 +30,29 @@ import ( "slices" "github.com/uber/submitqueue/submitqueue/entity" - "github.com/uber/submitqueue/submitqueue/extension/speculation/scorer" "github.com/uber/submitqueue/submitqueue/extension/speculation/generator" + "github.com/uber/submitqueue/submitqueue/extension/speculation/predictor" ) // bestFirst generates candidate paths using independent dependency -// probabilities supplied by scorer. +// probabilities supplied by predictor. type bestFirst struct { - scorer scorer.Scorer + predictor predictor.OutcomePredictor } var _ generator.Generator = (*bestFirst)(nil) // New returns a Generator that ranks paths by the probability that every -// unresolved dependency assumption holds. The scorer is called at most once +// unresolved dependency assumption holds. The predictor is called at most once // per unresolved dependency batch in each Generate call. -func New(s scorer.Scorer) generator.Generator { - return &bestFirst{scorer: s} +func New(p predictor.OutcomePredictor) generator.Generator { + return &bestFirst{predictor: p} } -// Generate scores the unresolved dependencies of the snapshot's Speculating +// Generate prices the unresolved dependencies of the snapshot's Speculating // heads and opens a lazy global best-first iterator. The snapshot is taken as // given: it is the caller's to keep well formed, and nothing here re-checks it. -func (g *bestFirst) Generate(ctx context.Context, batches []entity.Batch) (generator.Iterator, error) { +func (g *bestFirst) Generate(ctx context.Context, batches []entity.Batch, pathSets []entity.SpeculationPathSet) (generator.Iterator, error) { if err := ctx.Err(); err != nil { return nil, err } @@ -61,9 +61,13 @@ func (g *bestFirst) Generate(ctx context.Context, batches []entity.Batch) (gener for _, batch := range batches { batchByID[batch.ID] = batch } + pathsByHead := make(map[string]entity.SpeculationPathSet, len(pathSets)) + for _, set := range pathSets { + pathsByHead[set.Head] = set + } heads, unresolvedIDs := speculatingHeads(batches, batchByID) - probabilityByID, err := g.score(ctx, unresolvedIDs, batchByID) + probabilityByID, err := g.predict(ctx, unresolvedIDs, batchByID, pathsByHead) if err != nil { return nil, err } @@ -101,13 +105,14 @@ func speculatingHeads(batches []entity.Batch, batchByID map[string]entity.Batch) return heads, slices.Sorted(maps.Keys(unresolved)) } -// score asks the scorer for each unresolved dependency exactly once, however -// many heads wait on it. +// predict asks the predictor for each unresolved dependency exactly once, +// however many heads wait on it. Each dependency is priced against its own path +// set, zero-valued for one that has never speculated. // // A dependency that cannot be priced takes defaultProbability rather than // ending the run — one unusable number must not cost the queue every candidate // it had. Only cancellation is an error. -func (g *bestFirst) score(ctx context.Context, ids []string, batchByID map[string]entity.Batch) (map[string]float64, error) { +func (g *bestFirst) predict(ctx context.Context, ids []string, batchByID map[string]entity.Batch, pathsByHead map[string]entity.SpeculationPathSet) (map[string]float64, error) { probabilityByID := make(map[string]float64, len(ids)) for _, id := range ids { if err := ctx.Err(); err != nil { @@ -116,16 +121,16 @@ func (g *bestFirst) score(ctx context.Context, ids []string, batchByID map[strin batch, known := batchByID[id] if !known { // A batch the snapshot never carried is zero in every field, not - // just missing — scoring it would price some other batch entirely, + // just missing — pricing it would price some other batch entirely, // or fail on its empty queue. It is unpriceable, not cheap. probabilityByID[id] = defaultProbability continue } - probability, err := g.scorer.Score(ctx, batch) + probability, err := g.predictor.Predict(ctx, batch, pathsByHead[id]) if err != nil { - // A scorer that failed because the caller went away has not found - // an unpriceable dependency — it has found a dead ctx, which ends - // the run. The loop's own check would not catch it on the last + // A predictor that failed because the caller went away has not + // found an unpriceable dependency — it has found a dead ctx, which + // ends the run. The loop's own check would not catch it on the last // dependency, and a cancelled Generate must never hand back an // iterator. if ctxErr := ctx.Err(); ctxErr != nil { @@ -133,17 +138,17 @@ func (g *bestFirst) score(ctx context.Context, ids []string, batchByID map[strin } probability = defaultProbability } - probabilityByID[id] = asProbability(probability) + probabilityByID[id] = asProbability(float64(probability)) } return probabilityByID, nil } // defaultProbability stands in for a score that is not a probability, one the -// scorer could not produce at all, and one for a dependency the snapshot never -// carried. It is optimistic on purpose: a dependency nobody could estimate is -// treated as very likely to succeed, which keeps its head's preferred path near -// the front rather than burying it or dropping the queue's whole snapshot on -// one bad number. +// predictor could not produce at all, and one for a dependency the snapshot +// never carried. It is optimistic on purpose: a dependency nobody could +// estimate is treated as very likely to succeed, which keeps its head's preferred +// path near the front rather than burying it or dropping the queue's whole +// snapshot on one bad number. const defaultProbability = 0.95 // asProbability keeps a usable score and substitutes the default for anything diff --git a/submitqueue/extension/speculation/generator/bestfirst/bestfirst_test.go b/submitqueue/extension/speculation/generator/bestfirst/bestfirst_test.go index 7dd2ccdfd..ef45214c6 100644 --- a/submitqueue/extension/speculation/generator/bestfirst/bestfirst_test.go +++ b/submitqueue/extension/speculation/generator/bestfirst/bestfirst_test.go @@ -28,25 +28,26 @@ import ( "github.com/stretchr/testify/require" "github.com/uber/submitqueue/submitqueue/entity" - "github.com/uber/submitqueue/submitqueue/extension/speculation/scorer" "github.com/uber/submitqueue/submitqueue/extension/speculation/generator" + "github.com/uber/submitqueue/submitqueue/extension/speculation/predictor" ) -// stubScorer scores each batch by ID, defaulting to 0.5 for unknown batches. It -// is a minimal scorer.Scorer for exercising the generator without a resolver. -type stubScorer struct { +// stubPredictor prices each batch by ID, defaulting to 0.5 for unknown batches. +// It is a minimal predictor.OutcomePredictor for exercising the generator +// without a scorer under it. +type stubPredictor struct { scores map[string]float64 } -func (s stubScorer) Score(_ context.Context, b entity.Batch) (float64, error) { +func (s stubPredictor) Predict(_ context.Context, b entity.Batch, _ entity.SpeculationPathSet) (predictor.Probability, error) { if v, ok := s.scores[b.ID]; ok { - return v, nil + return predictor.Probability(v), nil } return 0.5, nil } -func scored(scores map[string]float64) scorer.Scorer { - return stubScorer{scores: scores} +func scored(scores map[string]float64) predictor.OutcomePredictor { + return stubPredictor{scores: scores} } // drainAll pulls every candidate from an iterator. @@ -125,41 +126,43 @@ func iteratorOf(t *testing.T, iter generator.Iterator) *candidateIterator { return it } -// countingScorer records how many times each batch is scored. -type countingScorer struct { +// countingPredictor records how many times each batch is priced. +type countingPredictor struct { scores map[string]float64 calls map[string]int total int } -func newCountingScorer(scores map[string]float64) *countingScorer { - return &countingScorer{scores: scores, calls: map[string]int{}} +func newCountingPredictor(scores map[string]float64) *countingPredictor { + return &countingPredictor{scores: scores, calls: map[string]int{}} } -func (c *countingScorer) Score(_ context.Context, b entity.Batch) (float64, error) { +func (c *countingPredictor) Predict(_ context.Context, b entity.Batch, _ entity.SpeculationPathSet) (predictor.Probability, error) { c.calls[b.ID]++ c.total++ if v, ok := c.scores[b.ID]; ok { - return v, nil + return predictor.Probability(v), nil } return 0.5, nil } -// errScorer always fails, to exercise error propagation from scoring. -type errScorer struct{} +// errPredictor always fails, to exercise error propagation from pricing. +type errPredictor struct{} -func (errScorer) Score(context.Context, entity.Batch) (float64, error) { +func (errPredictor) Predict(context.Context, entity.Batch, entity.SpeculationPathSet) (predictor.Probability, error) { return 0, assert.AnError } -// constScorer scores every batch identically, regardless of ID. -type constScorer struct{ v float64 } +// constPredictor prices every batch identically, regardless of ID. +type constPredictor struct{ v float64 } -func (c constScorer) Score(context.Context, entity.Batch) (float64, error) { return c.v, nil } +func (c constPredictor) Predict(context.Context, entity.Batch, entity.SpeculationPathSet) (predictor.Probability, error) { + return predictor.Probability(c.v), nil +} // wideHead builds one Speculating head over n unresolved dependencies, each at a // distinct score so no two combinations tie. -func wideHead(n int) ([]entity.Batch, scorer.Scorer) { +func wideHead(n int) ([]entity.Batch, predictor.OutcomePredictor) { head := entity.Batch{ID: "q/head", State: entity.BatchStateSpeculating} batches := []entity.Batch{} scores := map[string]float64{} @@ -181,7 +184,7 @@ func TestBestFirst_OrderingAndEnumeration(t *testing.T) { } sc := scored(map[string]float64{"q/A": 0.9, "q/B": 0.8}) - iter, err := New(sc).Generate(context.Background(), batches) + iter, err := New(sc).Generate(context.Background(), batches, nil) require.NoError(t, err) cands := forHead(drainAll(t, iter), "q/C") @@ -222,7 +225,7 @@ func TestBestFirst_PinsResolvedDependencies(t *testing.T) { {ID: "q/A", State: tt.state}, {ID: "q/H", State: entity.BatchStateSpeculating, Dependencies: []string{"q/A"}}, } - iter, err := New(scored(nil)).Generate(context.Background(), batches) + iter, err := New(scored(nil)).Generate(context.Background(), batches, nil) require.NoError(t, err) cands := forHead(drainAll(t, iter), "q/H") @@ -245,7 +248,7 @@ func TestBestFirst_ResolvedDependenciesDropOutOfSearch(t *testing.T) { Dependencies: []string{"q/succeeded", "q/failed", "q/open"}}, } iter, err := New(scored(map[string]float64{"q/open": 0.7})). - Generate(context.Background(), batches) + Generate(context.Background(), batches, nil) require.NoError(t, err) cands := forHead(drainAll(t, iter), "q/H") @@ -269,7 +272,7 @@ func TestBestFirst_EmitsExactSequenceAcrossHeads(t *testing.T) { } sc := scored(map[string]float64{"q/A": 0.9, "q/B": 0.8}) - iter, err := New(sc).Generate(context.Background(), batches) + iter, err := New(sc).Generate(context.Background(), batches, nil) require.NoError(t, err) cands := drainAll(t, iter) @@ -304,7 +307,7 @@ func TestBestFirst_PreferredAssumptionFollowsScore(t *testing.T) { } sc := scored(map[string]float64{"q/high": 0.8, "q/low": 0.3}) - iter, err := New(sc).Generate(context.Background(), batches) + iter, err := New(sc).Generate(context.Background(), batches, nil) require.NoError(t, err) cands := drainAll(t, iter) @@ -352,7 +355,7 @@ func TestBestFirst_OnlySpeculatingHeadsProduceCandidates(t *testing.T) { t.Run(name, func(t *testing.T) { batches := []entity.Batch{{ID: "q/H", State: tt.state}} - iter, err := New(scored(nil)).Generate(context.Background(), batches) + iter, err := New(scored(nil)).Generate(context.Background(), batches, nil) require.NoError(t, err) cands := drainAll(t, iter) @@ -371,7 +374,7 @@ func TestBestFirst_HeadWithNoDependencies(t *testing.T) { {ID: "q/H", State: entity.BatchStateSpeculating}, } - iter, err := New(scored(nil)).Generate(context.Background(), batches) + iter, err := New(scored(nil)).Generate(context.Background(), batches, nil) require.NoError(t, err) cands := drainAll(t, iter) @@ -390,7 +393,7 @@ func TestBestFirst_AbsorbsScorerError(t *testing.T) { {ID: "q/H", State: entity.BatchStateSpeculating, Dependencies: []string{"q/A"}}, } - iter, err := New(errScorer{}).Generate(context.Background(), batches) + iter, err := New(errPredictor{}).Generate(context.Background(), batches, nil) require.NoError(t, err) cands := forHead(drainAll(t, iter), "q/H") @@ -406,9 +409,9 @@ func TestBestFirst_NeverScoresAnAbsentDependency(t *testing.T) { batches := []entity.Batch{ {ID: "q/H", State: entity.BatchStateSpeculating, Dependencies: []string{"q/missing"}}, } - sc := newCountingScorer(map[string]float64{}) + sc := newCountingPredictor(map[string]float64{}) - iter, err := New(sc).Generate(context.Background(), batches) + iter, err := New(sc).Generate(context.Background(), batches, nil) require.NoError(t, err) cands := drainAll(t, iter) @@ -417,6 +420,39 @@ func TestBestFirst_NeverScoresAnAbsentDependency(t *testing.T) { assert.InDelta(t, math.Log(defaultProbability), cands[0].RankingScore, 1e-9) } +// recordingPredictor keeps the path set each batch was priced against. +type recordingPredictor struct { + seen map[string]entity.SpeculationPathSet +} + +func (r *recordingPredictor) Predict(_ context.Context, b entity.Batch, paths entity.SpeculationPathSet) (predictor.Probability, error) { + r.seen[b.ID] = paths + return 0.5, nil +} + +// Each dependency is priced against its own progress, not the queue's. A +// dependency with no set has simply never speculated, which is silence rather +// than an error. +func TestBestFirst_PricesEachDependencyAgainstItsOwnPathSet(t *testing.T) { + batches := []entity.Batch{ + {ID: "q/built", State: entity.BatchStateSpeculating}, + {ID: "q/fresh", State: entity.BatchStateSpeculating}, + {ID: "q/H", State: entity.BatchStateSpeculating, Dependencies: []string{"q/built", "q/fresh"}}, + } + built := entity.SpeculationPathSet{ + Queue: "q", + Head: "q/built", + Paths: []entity.SpeculationPathEntry{{ID: "p1", Status: entity.SpeculationPathStatusPassed}}, + } + pred := &recordingPredictor{seen: map[string]entity.SpeculationPathSet{}} + + _, err := New(pred).Generate(context.Background(), batches, []entity.SpeculationPathSet{built}) + require.NoError(t, err) + + assert.Equal(t, built, pred.seen["q/built"], "a dependency is priced against its own set") + assert.Equal(t, entity.SpeculationPathSet{}, pred.seen["q/fresh"], "a dependency that never speculated has no set") +} + // A merging dependency is still in progress — the merge can fail — so it stays // an open question here like any other. Whether a path betting against it is // worth funding is a matter of price, which is the scorer's to say, not a @@ -426,9 +462,9 @@ func TestBestFirst_MergingDependencyStaysOpen(t *testing.T) { {ID: "q/landing", State: entity.BatchStateMerging}, {ID: "q/H", State: entity.BatchStateSpeculating, Dependencies: []string{"q/landing"}}, } - sc := newCountingScorer(map[string]float64{"q/landing": 0.9}) + sc := newCountingPredictor(map[string]float64{"q/landing": 0.9}) - iter, err := New(sc).Generate(context.Background(), batches) + iter, err := New(sc).Generate(context.Background(), batches, nil) require.NoError(t, err) cands := drainAll(t, iter) @@ -443,7 +479,7 @@ func TestBestFirst_GeneratesOnlyWhatIsPulled(t *testing.T) { const deps, space = 12, 1 << 12 batches, sc := wideHead(deps) - iter, err := New(sc).Generate(context.Background(), batches) + iter, err := New(sc).Generate(context.Background(), batches, nil) require.NoError(t, err) it := iteratorOf(t, iter) @@ -480,7 +516,7 @@ func TestBestFirst_DrainYieldsEveryCombinationOnce(t *testing.T) { } sc := scored(map[string]float64{"q/A": 0.9, "q/B": 0.7, "q/C": 0.6}) - iter, err := New(sc).Generate(context.Background(), batches) + iter, err := New(sc).Generate(context.Background(), batches, nil) require.NoError(t, err) cands := drainAll(t, iter) @@ -530,7 +566,7 @@ func TestBestFirst_ScoresGloballyNonIncreasing(t *testing.T) { } sc := scored(map[string]float64{"q/A": 0.85, "q/B": 0.3, "q/C": 0.65}) - iter, err := New(sc).Generate(context.Background(), batches) + iter, err := New(sc).Generate(context.Background(), batches, nil) require.NoError(t, err) cands := drainAll(t, iter) @@ -554,7 +590,7 @@ func TestBestFirst_EqualScoresOrderDeterministically(t *testing.T) { {ID: "q/c", State: entity.BatchStateSpeculating}, } - iter, err := New(scored(nil)).Generate(context.Background(), batches) + iter, err := New(scored(nil)).Generate(context.Background(), batches, nil) require.NoError(t, err) cands := drainAll(t, iter) @@ -577,7 +613,7 @@ func TestBestFirst_EqualScoresOrderDeterministically(t *testing.T) { } sc := scored(map[string]float64{"q/coinA": 0.5, "q/coinB": 0.5}) - iter, err := New(sc).Generate(context.Background(), batches) + iter, err := New(sc).Generate(context.Background(), batches, nil) require.NoError(t, err) cands := drainAll(t, iter) @@ -612,9 +648,9 @@ func TestBestFirst_EqualScoresOrderDeterministically(t *testing.T) { } sc := scored(map[string]float64{"q/A": 0.5, "q/B": 0.5, "q/C": 0.5}) - first, err := New(sc).Generate(context.Background(), batches) + first, err := New(sc).Generate(context.Background(), batches, nil) require.NoError(t, err) - second, err := New(sc).Generate(context.Background(), batches) + second, err := New(sc).Generate(context.Background(), batches, nil) require.NoError(t, err) a, b := drainAll(t, first), drainAll(t, second) @@ -625,9 +661,9 @@ func TestBestFirst_EqualScoresOrderDeterministically(t *testing.T) { func TestBestFirst_NextMakesNoScorerCalls(t *testing.T) { batches, _ := wideHead(6) - sc := newCountingScorer(map[string]float64{}) + sc := newCountingPredictor(map[string]float64{}) - iter, err := New(sc).Generate(context.Background(), batches) + iter, err := New(sc).Generate(context.Background(), batches, nil) require.NoError(t, err) afterGenerate := sc.total @@ -645,9 +681,9 @@ func TestBestFirst_MemoizesDependencyScoresAcrossHeads(t *testing.T) { {ID: "q/H2", State: entity.BatchStateSpeculating, Dependencies: []string{"q/shared"}}, {ID: "q/H3", State: entity.BatchStateSpeculating, Dependencies: []string{"q/shared"}}, } - sc := newCountingScorer(map[string]float64{"q/shared": 0.7}) + sc := newCountingPredictor(map[string]float64{"q/shared": 0.7}) - _, err := New(sc).Generate(context.Background(), batches) + _, err := New(sc).Generate(context.Background(), batches, nil) require.NoError(t, err) assert.Equal(t, 1, sc.calls["q/shared"], "a shared dependency is scored once") @@ -679,7 +715,7 @@ func TestBestFirst_MatchesBruteForceEnumeration(t *testing.T) { ID: "q/H", State: entity.BatchStateSpeculating, Dependencies: deps, }) - iter, err := New(scored(scores)).Generate(context.Background(), batches) + iter, err := New(scored(scores)).Generate(context.Background(), batches, nil) require.NoError(t, err) got := drainAll(t, iter) @@ -750,7 +786,7 @@ func TestBestFirst_WideHeadsRankWithoutUnderflow(t *testing.T) { wide("q/narrow", narrowWidth) wide("q/wide", wideWidth) - iter, err := New(constScorer{depScore}).Generate(context.Background(), batches) + iter, err := New(constPredictor{depScore}).Generate(context.Background(), batches, nil) require.NoError(t, err) first, ok, err := iter.Next(context.Background()) @@ -788,7 +824,7 @@ func TestBestFirst_HonorsCancelledContext(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) cancel() - iter, err := New(scored(nil)).Generate(ctx, batches) + iter, err := New(scored(nil)).Generate(ctx, batches, nil) require.ErrorIs(t, err, context.Canceled) assert.Nil(t, iter) }) @@ -797,7 +833,7 @@ func TestBestFirst_HonorsCancelledContext(t *testing.T) { ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(-time.Minute)) defer cancel() - _, err := New(scored(nil)).Generate(ctx, batches) + _, err := New(scored(nil)).Generate(ctx, batches, nil) require.ErrorIs(t, err, context.DeadlineExceeded) }) @@ -805,7 +841,7 @@ func TestBestFirst_HonorsCancelledContext(t *testing.T) { // Generate on a live context so the stream has candidates waiting; the // cancel lands between pulls, which is where a caller that has given up // actually stops. - iter, err := New(scored(nil)).Generate(context.Background(), batches) + iter, err := New(scored(nil)).Generate(context.Background(), batches, nil) require.NoError(t, err) _, ok, err := iter.Next(context.Background()) @@ -829,17 +865,17 @@ func TestBestFirst_HonorsCancelledContext(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - iter, err := New(cancellingScorer{cancel: cancel}).Generate(ctx, batches) + iter, err := New(cancellingPredictor{cancel: cancel}).Generate(ctx, batches, nil) require.ErrorIs(t, err, context.Canceled) assert.Nil(t, iter) }) } -// cancellingScorer kills the context and then fails, the way a scorer whose -// own call was cancelled would. -type cancellingScorer struct{ cancel context.CancelFunc } +// cancellingPredictor kills the context and then fails, the way a predictor +// whose own call was cancelled would. +type cancellingPredictor struct{ cancel context.CancelFunc } -func (s cancellingScorer) Score(context.Context, entity.Batch) (float64, error) { +func (s cancellingPredictor) Predict(context.Context, entity.Batch, entity.SpeculationPathSet) (predictor.Probability, error) { s.cancel() return 0, context.Canceled } @@ -870,7 +906,7 @@ func TestBestFirst_DefaultsScoreOutsideUnitInterval(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - iter, err := New(constScorer{tt.score}).Generate(context.Background(), batches) + iter, err := New(constPredictor{tt.score}).Generate(context.Background(), batches, nil) require.NoError(t, err) cands := drainAll(t, iter) @@ -898,7 +934,7 @@ func TestBestFirst_ImpossibleFlipScoresNegativeInfinity(t *testing.T) { } sc := scored(map[string]float64{"q/certain": 1.0, "q/toss": 0.6}) - iter, err := New(sc).Generate(context.Background(), batches) + iter, err := New(sc).Generate(context.Background(), batches, nil) require.NoError(t, err) cands := drainAll(t, iter) @@ -930,9 +966,9 @@ func TestBestFirst_ResolvedDependenciesAreNeverScored(t *testing.T) { {ID: "q/H", State: entity.BatchStateSpeculating, Dependencies: []string{"q/passed", "q/broke", "q/stopped", "q/running"}}, } - sc := newCountingScorer(map[string]float64{"q/running": 0.8}) + sc := newCountingPredictor(map[string]float64{"q/running": 0.8}) - iter, err := New(sc).Generate(context.Background(), batches) + iter, err := New(sc).Generate(context.Background(), batches, nil) require.NoError(t, err) cands := drainAll(t, iter) @@ -953,7 +989,7 @@ func TestBestFirst_ReturnedPathsAreIndependent(t *testing.T) { // scribbling on what it was handed must not reach the paths still to come. batches, _ := wideHead(3) iter, err := New(scored(map[string]float64{"q/dep00": 0.9, "q/dep01": 0.8, "q/dep02": 0.7})). - Generate(context.Background(), batches) + Generate(context.Background(), batches, nil) require.NoError(t, err) first, ok, err := iter.Next(context.Background()) @@ -1010,7 +1046,7 @@ func TestBestFirst_ScoresAreSummedFromTheHeadsBestScore(t *testing.T) { want[s.scoreFor(taken)]++ } - iter, err := New(scored(scores)).Generate(context.Background(), batches) + iter, err := New(scored(scores)).Generate(context.Background(), batches, nil) require.NoError(t, err) got := map[float64]int{} for _, c := range drainAll(t, iter) { @@ -1038,7 +1074,7 @@ func TestBestFirst_UntouchedHeadsNeverWorkOutFlips(t *testing.T) { scores[dep] = 0.6 + 0.005*float64(i) } - iter, err := New(scored(scores)).Generate(context.Background(), batches) + iter, err := New(scored(scores)).Generate(context.Background(), batches, nil) require.NoError(t, err) it := iteratorOf(t, iter) @@ -1078,7 +1114,7 @@ func TestBestFirst_AFailedPullConsumesNothing(t *testing.T) { } sc := scored(map[string]float64{"q/dep": 0.8}) - iter, err := New(sc).Generate(context.Background(), batches) + iter, err := New(sc).Generate(context.Background(), batches, nil) require.NoError(t, err) cancelled, cancel := context.WithCancel(context.Background()) diff --git a/submitqueue/extension/speculation/generator/generator.go b/submitqueue/extension/speculation/generator/generator.go index 3f7a6f8cf..3e8da561c 100644 --- a/submitqueue/extension/speculation/generator/generator.go +++ b/submitqueue/extension/speculation/generator/generator.go @@ -45,7 +45,11 @@ type Generator interface { // duplicate, or self dependency. That is a precondition the caller owns: a // generator may assume it and is not required to detect a breach, so a // malformed snapshot yields undefined candidates rather than an error. - Generate(ctx context.Context, batches []entity.Batch) (Iterator, error) + // + // pathSets is what each batch's builds have done so far, at most one set per + // head and none for a batch nothing has speculated on. It is part of the + // same snapshot as batches and carries the same holding rules. + Generate(ctx context.Context, batches []entity.Batch, pathSets []entity.SpeculationPathSet) (Iterator, error) } // Iterator is a pull-based stream of candidate paths. Beyond what ranking diff --git a/submitqueue/extension/speculation/generator/mock/generator_mock.go b/submitqueue/extension/speculation/generator/mock/generator_mock.go index 22740a6bb..14ceb5559 100644 --- a/submitqueue/extension/speculation/generator/mock/generator_mock.go +++ b/submitqueue/extension/speculation/generator/mock/generator_mock.go @@ -43,18 +43,18 @@ func (m *MockGenerator) EXPECT() *MockGeneratorMockRecorder { } // Generate mocks base method. -func (m *MockGenerator) Generate(ctx context.Context, batches []entity.Batch) (generator.Iterator, error) { +func (m *MockGenerator) Generate(ctx context.Context, batches []entity.Batch, pathSets []entity.SpeculationPathSet) (generator.Iterator, error) { m.ctrl.T.Helper() - ret := m.ctrl.Call(m, "Generate", ctx, batches) + ret := m.ctrl.Call(m, "Generate", ctx, batches, pathSets) ret0, _ := ret[0].(generator.Iterator) ret1, _ := ret[1].(error) return ret0, ret1 } // Generate indicates an expected call of Generate. -func (mr *MockGeneratorMockRecorder) Generate(ctx, batches any) *gomock.Call { +func (mr *MockGeneratorMockRecorder) Generate(ctx, batches, pathSets any) *gomock.Call { mr.mock.ctrl.T.Helper() - return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Generate", reflect.TypeOf((*MockGenerator)(nil).Generate), ctx, batches) + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Generate", reflect.TypeOf((*MockGenerator)(nil).Generate), ctx, batches, pathSets) } // MockIterator is a mock of Iterator interface. diff --git a/submitqueue/extension/speculation/predictor/BUILD.bazel b/submitqueue/extension/speculation/predictor/BUILD.bazel new file mode 100644 index 000000000..fc689f24c --- /dev/null +++ b/submitqueue/extension/speculation/predictor/BUILD.bazel @@ -0,0 +1,9 @@ +load("@rules_go//go:def.bzl", "go_library") + +go_library( + name = "go_default_library", + srcs = ["predictor.go"], + importpath = "github.com/uber/submitqueue/submitqueue/extension/speculation/predictor", + visibility = ["//visibility:public"], + deps = ["//submitqueue/entity:go_default_library"], +) diff --git a/submitqueue/extension/speculation/predictor/mock/BUILD.bazel b/submitqueue/extension/speculation/predictor/mock/BUILD.bazel new file mode 100644 index 000000000..fd84517d1 --- /dev/null +++ b/submitqueue/extension/speculation/predictor/mock/BUILD.bazel @@ -0,0 +1,13 @@ +load("@rules_go//go:def.bzl", "go_library") + +go_library( + name = "go_default_library", + srcs = ["predictor_mock.go"], + importpath = "github.com/uber/submitqueue/submitqueue/extension/speculation/predictor/mock", + visibility = ["//visibility:public"], + deps = [ + "//submitqueue/entity:go_default_library", + "//submitqueue/extension/speculation/predictor:go_default_library", + "@org_uber_go_mock//gomock:go_default_library", + ], +) diff --git a/submitqueue/extension/speculation/predictor/mock/predictor_mock.go b/submitqueue/extension/speculation/predictor/mock/predictor_mock.go new file mode 100644 index 000000000..05dccc8b9 --- /dev/null +++ b/submitqueue/extension/speculation/predictor/mock/predictor_mock.go @@ -0,0 +1,97 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: predictor.go +// +// Generated by this command: +// +// mockgen -source=predictor.go -destination=mock/predictor_mock.go -package=mock +// + +// Package mock is a generated GoMock package. +package mock + +import ( + context "context" + reflect "reflect" + + entity "github.com/uber/submitqueue/submitqueue/entity" + predictor "github.com/uber/submitqueue/submitqueue/extension/speculation/predictor" + gomock "go.uber.org/mock/gomock" +) + +// MockOutcomePredictor is a mock of OutcomePredictor interface. +type MockOutcomePredictor struct { + ctrl *gomock.Controller + recorder *MockOutcomePredictorMockRecorder + isgomock struct{} +} + +// MockOutcomePredictorMockRecorder is the mock recorder for MockOutcomePredictor. +type MockOutcomePredictorMockRecorder struct { + mock *MockOutcomePredictor +} + +// NewMockOutcomePredictor creates a new mock instance. +func NewMockOutcomePredictor(ctrl *gomock.Controller) *MockOutcomePredictor { + mock := &MockOutcomePredictor{ctrl: ctrl} + mock.recorder = &MockOutcomePredictorMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockOutcomePredictor) EXPECT() *MockOutcomePredictorMockRecorder { + return m.recorder +} + +// Predict mocks base method. +func (m *MockOutcomePredictor) Predict(ctx context.Context, batch entity.Batch, paths entity.SpeculationPathSet) (predictor.Probability, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Predict", ctx, batch, paths) + ret0, _ := ret[0].(predictor.Probability) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// Predict indicates an expected call of Predict. +func (mr *MockOutcomePredictorMockRecorder) Predict(ctx, batch, paths any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Predict", reflect.TypeOf((*MockOutcomePredictor)(nil).Predict), ctx, batch, paths) +} + +// MockFactory is a mock of Factory interface. +type MockFactory struct { + ctrl *gomock.Controller + recorder *MockFactoryMockRecorder + isgomock struct{} +} + +// MockFactoryMockRecorder is the mock recorder for MockFactory. +type MockFactoryMockRecorder struct { + mock *MockFactory +} + +// NewMockFactory creates a new mock instance. +func NewMockFactory(ctrl *gomock.Controller) *MockFactory { + mock := &MockFactory{ctrl: ctrl} + mock.recorder = &MockFactoryMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *MockFactory) EXPECT() *MockFactoryMockRecorder { + return m.recorder +} + +// For mocks base method. +func (m *MockFactory) For(cfg predictor.Config) (predictor.OutcomePredictor, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "For", cfg) + ret0, _ := ret[0].(predictor.OutcomePredictor) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// For indicates an expected call of For. +func (mr *MockFactoryMockRecorder) For(cfg any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "For", reflect.TypeOf((*MockFactory)(nil).For), cfg) +} diff --git a/submitqueue/extension/speculation/predictor/predictor.go b/submitqueue/extension/speculation/predictor/predictor.go new file mode 100644 index 000000000..7ab1e9763 --- /dev/null +++ b/submitqueue/extension/speculation/predictor/predictor.go @@ -0,0 +1,56 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package predictor defines how likely a batch is to succeed, given both what +// it changes and what has happened to it so far. A Scorer prices the change; an +// OutcomePredictor is built over one and revises its price with the batch's +// observed progress. +package predictor + +//go:generate mockgen -source=predictor.go -destination=mock/predictor_mock.go -package=mock + +import ( + "context" + + "github.com/uber/submitqueue/submitqueue/entity" +) + +// Probability is how likely an outcome is, from 0.0 to 1.0. +type Probability float64 + +// OutcomePredictor estimates a batch's final outcome. +type OutcomePredictor interface { + // Predict returns how likely the batch is to reach Succeeded with its + // changes landed. A passing build is necessary but not sufficient. + // + // paths is the batch's own build progress, zero-valued for a batch nothing + // has speculated on. Callers may predict every batch a queue waits on, so + // anything expensive belongs behind the implementation's own cache. + Predict(ctx context.Context, batch entity.Batch, paths entity.SpeculationPathSet) (Probability, error) +} + +// Config carries the per-queue identity handed to a Factory. The system knows +// only the queue name; everything an implementation needs is injected at +// construction by the integrator. +type Config struct { + // QueueName identifies the queue this OutcomePredictor serves. + QueueName string +} + +// Factory builds the OutcomePredictor for a queue. Implementations inject what +// they need at construction, including the Scorer whose price they revise. +type Factory interface { + // For returns the OutcomePredictor for the given queue. + For(cfg Config) (OutcomePredictor, error) +} diff --git a/submitqueue/extension/speculation/predictor/regression/BUILD.bazel b/submitqueue/extension/speculation/predictor/regression/BUILD.bazel new file mode 100644 index 000000000..254e211fb --- /dev/null +++ b/submitqueue/extension/speculation/predictor/regression/BUILD.bazel @@ -0,0 +1,29 @@ +load("@rules_go//go:def.bzl", "go_library", "go_test") + +go_library( + name = "go_default_library", + srcs = ["regression.go"], + importpath = "github.com/uber/submitqueue/submitqueue/extension/speculation/predictor/regression", + visibility = ["//visibility:public"], + deps = [ + "//platform/metrics:go_default_library", + "//submitqueue/entity:go_default_library", + "//submitqueue/extension/speculation/predictor:go_default_library", + "//submitqueue/extension/speculation/scorer:go_default_library", + "@com_github_uber_go_tally//:go_default_library", + ], +) + +go_test( + name = "go_default_test", + srcs = ["regression_test.go"], + embed = [":go_default_library"], + deps = [ + "//submitqueue/entity:go_default_library", + "//submitqueue/extension/speculation/predictor:go_default_library", + "//submitqueue/extension/speculation/scorer:go_default_library", + "@com_github_stretchr_testify//assert:go_default_library", + "@com_github_stretchr_testify//require:go_default_library", + "@com_github_uber_go_tally//:go_default_library", + ], +) diff --git a/submitqueue/extension/speculation/predictor/regression/regression.go b/submitqueue/extension/speculation/predictor/regression/regression.go new file mode 100644 index 000000000..3ba287654 --- /dev/null +++ b/submitqueue/extension/speculation/predictor/regression/regression.go @@ -0,0 +1,178 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package regression revises a Scorer's price by multiplying its odds by one +// factor per piece of evidence about the batch's progress. +// +// Odds rather than the probability itself, because a factor then means the same +// thing wherever it applies and the result cannot leave [0, 1]. Written as logs +// and summed, the same arithmetic is a logistic regression, which is what lets +// hand-written factors later be replaced by fitted ones without changing the +// form. See doc/rfc/submitqueue/outcome-predictor.md. +package regression + +import ( + "fmt" + "math" + + "context" + + "github.com/uber-go/tally" + "github.com/uber/submitqueue/platform/metrics" + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/speculation/predictor" + "github.com/uber/submitqueue/submitqueue/extension/speculation/scorer" +) + +// Factors are the odds multipliers, one per piece of evidence. A factor of 1 +// leaves the price alone. Named fields rather than a keyed map, so an evidence +// name that does not exist fails to compile instead of being ignored. +type Factors struct { + // PathPassed applies once when a build has passed on the batch's + // all-succeed path. + PathPassed float64 + // PathFailed applies once per failed path, compounding. + PathFailed float64 + // Merging applies while the batch is merging. + Merging float64 + // Cancelling applies while the batch is cancelling. + Cancelling float64 +} + +// AllOnes is the neutral set: the prediction is the scorer's price. +func AllOnes() Factors { + return Factors{PathPassed: 1, PathFailed: 1, Merging: 1, Cancelling: 1} +} + +// epsilon bounds the price away from 0 and 1, which have no finite odds. +// Without it a certain scorer could never be revised by any evidence — and +// certainty about an unfinished batch is the scorer overstating what it sees. +const epsilon = 1e-6 + +// regression is a predictor.OutcomePredictor that revises a scorer's price. +type regression struct { + // cfg is the per-queue identity this predictor was built for. + cfg predictor.Config + // base prices the batch's change; its price is what the factors revise. + base scorer.Scorer + // factors are the odds multipliers applied to that price. + factors Factors + // scope is the tally scope for emitting metrics. + scope tally.Scope +} + +// New creates a regression predictor bound to the queue named in cfg, revising +// base's price by factors. +// Panics if base is nil or any factor is not positive. +func New(cfg predictor.Config, base scorer.Scorer, factors Factors, scope tally.Scope) predictor.OutcomePredictor { + if base == nil { + panic("regression.New: base must not be nil") + } + for name, factor := range map[string]float64{ + "PathPassed": factors.PathPassed, + "PathFailed": factors.PathFailed, + "Merging": factors.Merging, + "Cancelling": factors.Cancelling, + } { + // Zero would pin the prediction to 0 and negative has no meaning as a + // multiplier on odds. Configuration rejects both, so reaching here is a + // wiring bug rather than an operator's mistake. + if !(factor > 0) { + panic(fmt.Sprintf("regression.New: factor %s must be positive, got %v", name, factor)) + } + } + return ®ression{cfg: cfg, base: base, factors: factors, scope: scope} +} + +// Predict prices the batch's change through the base scorer, then multiplies +// the odds of that price by one factor per piece of evidence. +func (r *regression) Predict(ctx context.Context, batch entity.Batch, paths entity.SpeculationPathSet) (ret predictor.Probability, retErr error) { + op := metrics.Begin(r.scope, "predict", metrics.FastLatencyBuckets) + defer func() { op.Complete(retErr) }() + + price, err := r.base.Score(ctx, batch) + if err != nil { + return 0, err + } + // A price that is not a probability is a broken scorer, not a low opinion of + // the batch. Saying so leaves the caller to fall back on its own default, + // where clamping would hand back a number that looks deliberate. + if !(price >= 0 && price <= 1) { + return 0, fmt.Errorf("base scorer returned %v, which is not a probability", price) + } + + odds := oddsOf(math.Min(math.Max(price, epsilon), 1-epsilon)) + if hasPassedAllSucceedPath(paths) { + odds *= r.factors.PathPassed + } + odds *= math.Pow(r.factors.PathFailed, float64(countFailed(paths))) + switch batch.State { + case entity.BatchStateMerging: + odds *= r.factors.Merging + case entity.BatchStateCancelling: + odds *= r.factors.Cancelling + } + return probabilityOf(odds), nil +} + +// oddsOf converts a probability to odds. p is bounded away from 1, so this is +// finite. +func oddsOf(p float64) float64 { + return p / (1 - p) +} + +// probabilityOf converts odds back to a probability. Overflowed odds read as +// certainty rather than the NaN the division would produce. +func probabilityOf(odds float64) predictor.Probability { + if math.IsInf(odds, 1) { + return 1 + } + return predictor.Probability(odds / (1 + odds)) +} + +// hasPassedAllSucceedPath reports a passed build on the batch's all-succeed +// path. Only that path counts: one built without a dependency's changes says +// nothing about a candidate that assumes the dependency lands. +func hasPassedAllSucceedPath(paths entity.SpeculationPathSet) bool { + for _, entry := range paths.Paths { + if entry.Status != entity.SpeculationPathStatusPassed { + continue + } + if assumesAllSucceed(entry.Path) { + return true + } + } + return false +} + +// assumesAllSucceed reports whether every dependency is assumed to succeed. +func assumesAllSucceed(path entity.SpeculationPath) bool { + for _, dep := range path.Dependencies { + if dep.Assumption != entity.DependencyAssumptionSucceeds { + return false + } + } + return true +} + +// countFailed counts the batch's failed paths; each one compounds. +func countFailed(paths entity.SpeculationPathSet) int { + failed := 0 + for _, entry := range paths.Paths { + if entry.Status == entity.SpeculationPathStatusFailed { + failed++ + } + } + return failed +} diff --git a/submitqueue/extension/speculation/predictor/regression/regression_test.go b/submitqueue/extension/speculation/predictor/regression/regression_test.go new file mode 100644 index 000000000..c21837688 --- /dev/null +++ b/submitqueue/extension/speculation/predictor/regression/regression_test.go @@ -0,0 +1,234 @@ +// Copyright (c) 2025 Uber Technologies, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package regression + +import ( + "context" + "fmt" + "math" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/uber-go/tally" + "github.com/uber/submitqueue/submitqueue/entity" + "github.com/uber/submitqueue/submitqueue/extension/speculation/predictor" + "github.com/uber/submitqueue/submitqueue/extension/speculation/scorer" +) + +// testCfg is the per-queue identity used by every case in this file. +var testCfg = predictor.Config{QueueName: "test-queue"} + +// fixedScorer always returns the same price. +type fixedScorer struct{ price float64 } + +func (f fixedScorer) Score(_ context.Context, _ entity.Batch) (float64, error) { + return f.price, nil +} + +// errorScorer always fails. +type errorScorer struct{} + +func (errorScorer) Score(_ context.Context, _ entity.Batch) (float64, error) { + return 0, fmt.Errorf("scorer failed") +} + +// pathSet builds a set whose entries carry the given statuses, every path +// assuming all of its dependencies succeed. +func pathSet(statuses ...entity.SpeculationPathStatus) entity.SpeculationPathSet { + set := entity.SpeculationPathSet{Queue: "q", Head: "q/batch/1"} + for i, status := range statuses { + set.Paths = append(set.Paths, entity.SpeculationPathEntry{ + ID: fmt.Sprintf("path-%d", i), + Status: status, + Path: entity.SpeculationPath{ + Head: "q/batch/1", + Dependencies: []entity.PathDependency{{Batch: "q/batch/0", Assumption: entity.DependencyAssumptionSucceeds}}, + }, + }) + } + return set +} + +// predict runs one prediction with all-neutral factors except those overridden. +func predict(t *testing.T, price float64, factors Factors, batch entity.Batch, paths entity.SpeculationPathSet) float64 { + t.Helper() + p := New(testCfg, fixedScorer{price: price}, factors, tally.NoopScope) + got, err := p.Predict(context.Background(), batch, paths) + require.NoError(t, err) + return float64(got) +} + +func TestPredict_NeutralFactorsReturnTheScorersPrice(t *testing.T) { + for _, price := range []float64{0.01, 0.25, 0.5, 0.6, 0.9, 0.99} { + t.Run(fmt.Sprintf("price %v", price), func(t *testing.T) { + got := predict(t, price, AllOnes(), entity.Batch{}, pathSet(entity.SpeculationPathStatusPassed, entity.SpeculationPathStatusFailed)) + assert.InDelta(t, price, got, 1e-9) + }) + } +} + +func TestPredict_AppliesOneFactorPerEvidence(t *testing.T) { + // 0.5 has odds of exactly 1, so the resulting odds are the factor itself and + // the expected probability is factor/(1+factor). + tests := []struct { + name string + factors Factors + batch entity.Batch + paths entity.SpeculationPathSet + want float64 + }{ + { + name: "a passed path", + factors: Factors{PathPassed: 9, PathFailed: 1, Merging: 1, Cancelling: 1}, + paths: pathSet(entity.SpeculationPathStatusPassed), + want: 0.9, + }, + { + name: "no passed path leaves the price alone", + factors: Factors{PathPassed: 9, PathFailed: 1, Merging: 1, Cancelling: 1}, + paths: pathSet(entity.SpeculationPathStatusBuilding), + want: 0.5, + }, + { + name: "one failed path", + factors: Factors{PathPassed: 1, PathFailed: 0.25, Merging: 1, Cancelling: 1}, + paths: pathSet(entity.SpeculationPathStatusFailed), + want: 0.2, + }, + { + name: "failed paths compound", + factors: Factors{PathPassed: 1, PathFailed: 0.5, Merging: 1, Cancelling: 1}, + paths: pathSet(entity.SpeculationPathStatusFailed, entity.SpeculationPathStatusFailed), + want: 0.2, + }, + { + name: "merging", + factors: Factors{PathPassed: 1, PathFailed: 1, Merging: 19, Cancelling: 1}, + batch: entity.Batch{State: entity.BatchStateMerging}, + want: 0.95, + }, + { + name: "cancelling", + factors: Factors{PathPassed: 1, PathFailed: 1, Merging: 1, Cancelling: 0.25}, + batch: entity.Batch{State: entity.BatchStateCancelling}, + want: 0.2, + }, + { + name: "a state with no factor leaves the price alone", + factors: Factors{PathPassed: 1, PathFailed: 1, Merging: 19, Cancelling: 0.25}, + batch: entity.Batch{State: entity.BatchStateSpeculating}, + want: 0.5, + }, + { + name: "evidence compounds across kinds", + factors: Factors{PathPassed: 4, PathFailed: 1, Merging: 3, Cancelling: 1}, + batch: entity.Batch{State: entity.BatchStateMerging}, + paths: pathSet(entity.SpeculationPathStatusPassed), + want: 0.923076923, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.InDelta(t, tt.want, predict(t, 0.5, tt.factors, tt.batch, tt.paths), 1e-9) + }) + } +} + +// A path built without one of its dependencies proves nothing about a candidate +// that assumes the dependency lands, which is what stacking on this batch means. +func TestPredict_IgnoresAPassedPathThatAssumesAFailure(t *testing.T) { + paths := pathSet(entity.SpeculationPathStatusPassed) + paths.Paths[0].Path.Dependencies[0].Assumption = entity.DependencyAssumptionFails + + factors := Factors{PathPassed: 9, PathFailed: 1, Merging: 1, Cancelling: 1} + assert.InDelta(t, 0.5, predict(t, 0.5, factors, entity.Batch{}, paths), 1e-9) +} + +func TestPredict_APathWithNoDependenciesCounts(t *testing.T) { + paths := pathSet(entity.SpeculationPathStatusPassed) + paths.Paths[0].Path.Dependencies = nil + + factors := Factors{PathPassed: 9, PathFailed: 1, Merging: 1, Cancelling: 1} + assert.InDelta(t, 0.9, predict(t, 0.5, factors, entity.Batch{}, paths), 1e-9) +} + +func TestPredict_AnEmptyPathSetIsNoEvidence(t *testing.T) { + factors := Factors{PathPassed: 9, PathFailed: 0.1, Merging: 1, Cancelling: 1} + assert.InDelta(t, 0.5, predict(t, 0.5, factors, entity.Batch{}, entity.SpeculationPathSet{}), 1e-9) +} + +// A scorer certain either way still has to be movable, or no evidence could ever +// revise a price the scorer had no business being certain about. +func TestPredict_CertainPricesStayInRangeAndStillMove(t *testing.T) { + tests := []struct { + name string + price float64 + factor float64 + wantAbove float64 + wantBelow float64 + }{ + {name: "certain success, evidence against", price: 1, factor: 0.5, wantAbove: 0.99, wantBelow: 1}, + {name: "certain failure, evidence for", price: 0, factor: 2, wantAbove: 0, wantBelow: 0.01}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + factors := AllOnes() + factors.PathPassed = tt.factor + got := predict(t, tt.price, factors, entity.Batch{}, pathSet(entity.SpeculationPathStatusPassed)) + assert.Greater(t, got, tt.wantAbove) + assert.Less(t, got, tt.wantBelow) + }) + } +} + +func TestPredict_RejectsAPriceThatIsNotAProbability(t *testing.T) { + for _, price := range []float64{-0.1, 1.5, math.NaN()} { + t.Run(fmt.Sprintf("price %v", price), func(t *testing.T) { + p := New(testCfg, fixedScorer{price: price}, AllOnes(), tally.NoopScope) + _, err := p.Predict(context.Background(), entity.Batch{}, entity.SpeculationPathSet{}) + require.Error(t, err) + }) + } +} + +func TestPredict_PropagatesAScorerError(t *testing.T) { + p := New(testCfg, errorScorer{}, AllOnes(), tally.NoopScope) + _, err := p.Predict(context.Background(), entity.Batch{}, entity.SpeculationPathSet{}) + require.Error(t, err) +} + +func TestNew_RejectsUnusableConstruction(t *testing.T) { + zeroed := AllOnes() + zeroed.Merging = 0 + negative := AllOnes() + negative.PathFailed = -1 + + tests := []struct { + name string + base scorer.Scorer + factors Factors + }{ + {name: "nil base", base: nil, factors: AllOnes()}, + {name: "zero factor", base: fixedScorer{price: 0.5}, factors: zeroed}, + {name: "negative factor", base: fixedScorer{price: 0.5}, factors: negative}, + {name: "unset factors", base: fixedScorer{price: 0.5}, factors: Factors{}}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Panics(t, func() { New(testCfg, tt.base, tt.factors, tally.NoopScope) }) + }) + } +} diff --git a/submitqueue/extension/speculation/speculator/standard/BUILD.bazel b/submitqueue/extension/speculation/speculator/standard/BUILD.bazel index 1f79c617b..286eca2a5 100644 --- a/submitqueue/extension/speculation/speculator/standard/BUILD.bazel +++ b/submitqueue/extension/speculation/speculator/standard/BUILD.bazel @@ -23,6 +23,7 @@ go_test( "//submitqueue/extension/speculation/allocator/sticky:go_default_library", "//submitqueue/extension/speculation/generator/bestfirst:go_default_library", "//submitqueue/extension/speculation/generator/mock:go_default_library", + "//submitqueue/extension/speculation/predictor:go_default_library", "//submitqueue/extension/speculation/speculator:go_default_library", "@com_github_stretchr_testify//assert:go_default_library", "@com_github_stretchr_testify//require:go_default_library", diff --git a/submitqueue/extension/speculation/speculator/standard/standard.go b/submitqueue/extension/speculation/speculator/standard/standard.go index c9b4ecd34..e61083c4a 100644 --- a/submitqueue/extension/speculation/speculator/standard/standard.go +++ b/submitqueue/extension/speculation/speculator/standard/standard.go @@ -47,7 +47,7 @@ func New(cfg speculator.Config, gen generator.Generator, alloc allocator.Allocat // allocator spend the budget over the resulting candidate iterator, reconciling // it against the path sets. func (s spec) Speculate(ctx context.Context, batches []entity.Batch, pathSets []entity.SpeculationPathSet) ([]entity.Speculation, error) { - iter, err := s.gen.Generate(ctx, batches) + iter, err := s.gen.Generate(ctx, batches, pathSets) if err != nil { return nil, err } diff --git a/submitqueue/extension/speculation/speculator/standard/standard_test.go b/submitqueue/extension/speculation/speculator/standard/standard_test.go index 78aea9b23..a4a55be23 100644 --- a/submitqueue/extension/speculation/speculator/standard/standard_test.go +++ b/submitqueue/extension/speculation/speculator/standard/standard_test.go @@ -28,6 +28,7 @@ import ( "github.com/uber/submitqueue/submitqueue/extension/speculation/allocator/sticky" "github.com/uber/submitqueue/submitqueue/extension/speculation/generator/bestfirst" generatormock "github.com/uber/submitqueue/submitqueue/extension/speculation/generator/mock" + "github.com/uber/submitqueue/submitqueue/extension/speculation/predictor" "github.com/uber/submitqueue/submitqueue/extension/speculation/speculator" ) @@ -44,10 +45,13 @@ func assumptionFor(p entity.SpeculationPath, dep string) entity.DependencyAssump return entity.DependencyAssumptionUnknown } -// constScorer is a minimal scorer.Scorer that scores every batch identically. -type constScorer struct{ v float64 } +// constPredictor is a minimal predictor.OutcomePredictor that prices every +// batch identically. +type constPredictor struct{ v float64 } -func (c constScorer) Score(context.Context, entity.Batch) (float64, error) { return c.v, nil } +func (c constPredictor) Predict(context.Context, entity.Batch, entity.SpeculationPathSet) (predictor.Probability, error) { + return predictor.Probability(c.v), nil +} func TestComposed_EndToEnd_NaivePair(t *testing.T) { batches := []entity.Batch{ @@ -56,7 +60,7 @@ func TestComposed_EndToEnd_NaivePair(t *testing.T) { } // bestfirst generator + sticky allocator with a 2-build budget. - spec := New(testCfg, bestfirst.New(constScorer{0.9}), sticky.New(2)) + spec := New(testCfg, bestfirst.New(constPredictor{0.9}), sticky.New(2)) got, err := spec.Speculate(context.Background(), batches, nil) require.NoError(t, err) @@ -89,7 +93,7 @@ func TestComposed_WiresGeneratorIntoAllocator(t *testing.T) { gen := generatormock.NewMockGenerator(ctrl) alloc := allocatormock.NewMockAllocator(ctrl) - gen.EXPECT().Generate(gomock.Any(), batches).Return(iter, nil) + gen.EXPECT().Generate(gomock.Any(), batches, gomock.Any()).Return(iter, nil) alloc.EXPECT().Allocate(gomock.Any(), pathSets, iter).Return(want, nil) got, err := New(testCfg, gen, alloc).Speculate(context.Background(), batches, pathSets) @@ -104,7 +108,7 @@ func TestComposed_PropagatesGeneratorError(t *testing.T) { gen := generatormock.NewMockGenerator(ctrl) alloc := allocatormock.NewMockAllocator(ctrl) - gen.EXPECT().Generate(gomock.Any(), gomock.Any()).Return(nil, errGenerate) + gen.EXPECT().Generate(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, errGenerate) // Allocate must not be called when Generate fails (no alloc.EXPECT()). _, err := New(testCfg, gen, alloc).Speculate(context.Background(), nil, nil) @@ -122,7 +126,7 @@ func TestComposed_PropagatesContextCancellation(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) cancel() - spec := New(testCfg, bestfirst.New(constScorer{0.9}), sticky.New(2)) + spec := New(testCfg, bestfirst.New(constPredictor{0.9}), sticky.New(2)) got, err := spec.Speculate(ctx, batches, nil) require.ErrorIs(t, err, context.Canceled) assert.Nil(t, got)