Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion submitqueue/orchestrator/controller/speculate/BUILD.bazel
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,8 @@ go_library(
"check.go",
"dispatch.go",
"doc.go",
"finalize.go",
"outcome.go",
"run.go",
"snapshot.go",
"speculate.go",
Expand All @@ -29,6 +31,7 @@ go_test(
name = "go_default_test",
srcs = [
"check_test.go",
"outcome_test.go",
"run_test.go",
"snapshot_test.go",
"speculate_test.go",
Expand All @@ -37,7 +40,6 @@ go_test(
deps = [
"//platform/base/messagequeue:go_default_library",
"//platform/consumer:go_default_library",
"//platform/errs:go_default_library",
"//platform/extension/messagequeue/mock:go_default_library",
"//submitqueue/core/topickey:go_default_library",
"//submitqueue/entity:go_default_library",
Expand Down
4 changes: 2 additions & 2 deletions submitqueue/orchestrator/controller/speculate/check.go
Original file line number Diff line number Diff line change
Expand Up @@ -125,8 +125,8 @@ func rejectionReason(proposal entity.Speculation, snap snapshot) (rejection, boo
// fewer, and every assumption a real value.
//
// A malformed path is not merely suboptimal, it is unmergeable — the merge
// preconditions are read off the path's assumptions, so a path missing a
// dependency would let its head merge without waiting for it.
// preconditions are read off the path's assumptions (see mergeablePath), so a
// path missing a dependency would let its head merge without waiting for it.
func isWellFormed(path entity.SpeculationPath, head entity.Batch) bool {
if path.Head != head.ID {
return false
Expand Down
34 changes: 22 additions & 12 deletions submitqueue/orchestrator/controller/speculate/dispatch.go
Original file line number Diff line number Diff line change
Expand Up @@ -26,10 +26,18 @@ import (
"github.com/uber/submitqueue/submitqueue/extension/storage"
)

// dispatch saves each head's decisions and hands the build stage its work.
// Everything decided this run — build results, broken-path cancellations and
// dispatch saves what finalize left over and hands the build stage its work.
// Everything still outstanding — build results, broken-path cancellations and
// accepted proposals — is folded together per head, so a head costs one
// compare-and-swap however many of its paths changed.
//
// It walks every in-flight batch, not only the speculating ones. Proposals
// apply to speculating heads alone and are simply absent for the rest, but
// observations are not: a merging or cancelling head's paths keep holding CI
// slots until their builds stop, and this is the only writer that can record
// that they have. Batches already finalized arrive here clean —
// commitOutcome persisted their set with their outcome — so only their
// dispatch is left to do.
func (c *Controller) dispatch(ctx context.Context, queue string, snap snapshot, kept []entity.Speculation) error {
nowMs := time.Now().UnixMilli()

Expand All @@ -39,7 +47,7 @@ func (c *Controller) dispatch(ctx context.Context, queue string, snap snapshot,
byHead[proposal.Path.Head] = append(byHead[proposal.Path.Head], proposal)
}

for _, batch := range snap.speculating {
for _, batch := range snap.inFlight {
// A head with no stored set is one nothing has been funded for yet. It
// gets an empty set to fold this run's proposals into, which persist
// then creates; a head that ends the run with no paths writes nothing
Expand All @@ -57,7 +65,7 @@ func (c *Controller) dispatch(ctx context.Context, queue string, snap snapshot,
}

if changed {
if err := c.persist(ctx, set, exists); err != nil {
if _, err := c.persist(ctx, set, exists); err != nil {
if errors.Is(err, storage.ErrVersionMismatch) {
// Skipped rather than failed, and nothing is lost by that.
//
Expand Down Expand Up @@ -101,8 +109,9 @@ func (c *Controller) dispatch(ctx context.Context, queue string, snap snapshot,
}

// persist writes a head's path set, creating it if this run is the first to
// fund the head.
func (c *Controller) persist(ctx context.Context, set entity.SpeculationPathSet, exists bool) error {
// fund the head. It returns the set as stored, with its version advanced, so
// a caller that keeps the set around goes on holding a current copy.
func (c *Controller) persist(ctx context.Context, set entity.SpeculationPathSet, exists bool) (entity.SpeculationPathSet, error) {
store := c.store.GetSpeculationPathSetStore()

if !exists {
Expand All @@ -111,23 +120,24 @@ func (c *Controller) persist(ctx context.Context, set entity.SpeculationPathSet,
if errors.Is(err, storage.ErrAlreadyExists) {
// Another writer created it between this run's read and now.
// Treat it as a lost race: the next run reads the winner.
return storage.ErrVersionMismatch
return set, storage.ErrVersionMismatch
}
metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1)
return fmt.Errorf("failed to create path set for batch %s: %w", set.Head, err)
return set, fmt.Errorf("failed to create path set for batch %s: %w", set.Head, err)
}
return nil
return set, nil
}

newVersion := set.Version + 1
if err := store.Update(ctx, set, set.Version, newVersion); err != nil {
if errors.Is(err, storage.ErrVersionMismatch) {
return err
return set, err
}
metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1)
return fmt.Errorf("failed to update path set for batch %s: %w", set.Head, err)
return set, fmt.Errorf("failed to update path set for batch %s: %w", set.Head, err)
}
return nil
set.Version = newVersion
return set, nil
}

// applyProposal folds one accepted proposal into the set and reports whether
Expand Down
90 changes: 61 additions & 29 deletions submitqueue/orchestrator/controller/speculate/doc.go
Original file line number Diff line number Diff line change
Expand Up @@ -12,23 +12,19 @@
// See the License for the specific language governing permissions and
// limitations under the License.

// Package speculate plans a queue's speculative builds: which guesses about
// the queue's future are worth building, within the queue's cap on concurrent
// builds (the build budget).
//
// Batch outcomes — merge or fail — are still decided by the legacy per-batch
// finalizer in speculate.go, which waits on every dependency. Deriving them
// from the paths planned here replaces it in the next change; this package
// doc grows with it.
// Package speculate plans a queue's speculative builds and finalizes each
// batch's outcome from their results. It is the orchestrator's decision
// stage: builds are started by the build stage and watched — and stopped —
// by the buildsignal stage, but what to build and what a finished build
// means are decided here.
//
// # Why speculation
//
// Batches in a queue depend on the batches ahead of them, so without
// speculation everything is serial: C waits for B, B waits for A. Speculation
// builds a batch against a guess about how its dependencies turn out. When
// the guess holds, the head's build has already run by the time its
// dependencies resolve — it never waits for a build of its own to start
// afterwards.
// the guess holds, the batch merges the moment the guessed-on dependencies
// land — it never waits for a build of its own to start afterwards.
//
// # Paths
//
Expand All @@ -38,6 +34,26 @@
// a path *is* its guess; building the same guess again is a new attempt of
// the same path, and (path ID, attempt) names the resulting build.
//
// # A worked example
//
// A two-batch queue, where B depends on A and A is still building:
//
// queue: A ← B
//
// B's speculation space is two paths:
// P1 = [A succeeds] B built on top of A's result
// P2 = [A fails] B built without A
//
// Fund both and every future is covered:
//
// - A succeeds and P1 passed: B merges the moment A lands. P2's guess
// ("A fails") is broken — it can no longer come true — so its build is
// cancelled to free the slot.
// - A fails and P2 passed: B merges without A, again with no new build.
// P1's guess is broken.
// - A resolved either way, and every unbroken path failed: no future
// remains in which B passes, so B fails.
//
// # The life of a path
//
// A path's status tracks its current attempt:
Expand All @@ -46,10 +62,11 @@
// (no entry) ─────────► pending ─────────► building ──────┬──► passed
// │ │ └──► failed
// "stop this": │ │
// broken or ▼ ▼
// preempted ─────► cancelling ◄────┘
// │
// │ build observed stopped
// broken, ▼ ▼
// superseded, ─────► cancelling ◄────┘
// head cancelled, │
// or preempted │ build observed stopped, or
// │ nothing was ever dispatched
// ▼
// cancelled
//
Expand All @@ -58,14 +75,27 @@
// building, and every pending, building, and cancelling path holds its slot
// until its build stops. A path is broken once a dependency's actual result
// proves one of its assumptions wrong: its guess can no longer come true, so
// its build is cancelled to free the slot.
// its build is cancelled to free the slot. A path is superseded when a
// sibling path of the same head passes — that sibling will carry the head out
// of the queue, so the others are cancelled too.
//
// Cancelling is intent, not fact: the build keeps its slot until CI actually
// stops it, and only an observation of that stop moves the path to cancelled.
// The intent needs no dispatch of its own — the poll loop reads it off the set
// and asks the runner to stop the build. A terminal path can be resurrected by
// a new build proposal — status returns to pending and Attempt increments, the
// one backwards step in the diagram.
// stops it, and only an observation of that stop (or proof nothing was ever
// dispatched) moves the path to cancelled. The intent needs no dispatch of its
// own — the poll loop reads it off the set and asks the runner to stop the
// build. A terminal path can be resurrected by a new build proposal — status
// returns to pending and Attempt increments, the one backwards step in the
// diagram.
//
// # The life of a batch, as seen from here
//
// Created ──admit──► Speculating ──┬── merge ──► Merging (merge stage takes over)
// └── fail ───► Failed
// user cancel (cancel stage):
// ... ──► Cancelling ── every path stopped ──► Cancelled
//
// Failed and Cancelled fan out to the conclude stage, which reconciles the
// batch's requests.
//
// # How a run works
//
Expand All @@ -75,16 +105,18 @@
// reordered signals are harmless, and a later run repairs whatever an
// earlier one left half-done.
//
// signal ──► read ──► cancel ──► ask ──► check ──► dispatch
// one broken the filter save changes,
// read of paths Specu- its hand builds to
// queue + lator proposals the build stage
// paths
// signal ──► read ──► finalize ──► ask ──► check ──► dispatch
// one enact the the filter save changes,
// read of outcomes Specu- its hand builds to
// queue + the facts lator proposals the build stage
// paths already
// decide
//
// The Speculator is the extension that proposes which paths to fund or
// preempt. It only ever proposes: check.go filters its answer, and broken
// paths are cancelled before it is asked, so it reasons over facts as they
// now stand rather than over a picture the run is about to invalidate.
// preempt. It only ever proposes: check.go filters its answer, and outcomes
// are computed here, never by the extension. Finalize runs before ask so the
// Speculator reasons over facts as they now stand, not over a picture the run
// is about to invalidate.
//
// # Ownership
//
Expand Down
Loading
Loading