diff --git a/Makefile b/Makefile index 8f3a57c83..f06d0aea6 100644 --- a/Makefile +++ b/Makefile @@ -78,6 +78,9 @@ FOLDERS ?= 0 FILES ?= 3 CONCURRENCY ?= 5 STACKED ?= false +# BURST=true creates every change first, then enqueues them all at once instead +# of as each is created. Independent changes only; a stack always lands as one. +BURST ?= false SINCE ?= 1h LIMIT ?= 50 LAND ?= true @@ -240,7 +243,7 @@ clean-proto: ## Clean generated proto files @rm -f $(foreach p,$(PROTO_PACKAGES),$(p)/protopb/*.pb.go $(p)/protopb/*.pb.yarpc.go) @echo "Proto clean complete!" -demo-requests: ## Create N changes, enqueue each as it is created, and watch (PROVIDER=fake|git|github COUNT=3 FOLDERS=0 FILES=3 CONCURRENCY=5) +demo-requests: ## Create N changes, enqueue each as it is created, and watch (PROVIDER=fake|git|github COUNT=3 FOLDERS=0 FILES=3 CONCURRENCY=5 BURST=false) @set -e; $(resolve_gateway_addr); $(resolve_provider); \ $(BAZEL) run //service/submitqueue/demo/requests -- \ -provider $$provider \ @@ -251,6 +254,7 @@ demo-requests: ## Create N changes, enqueue each as it is created, and watch (PR -files $(FILES) \ -concurrency $(CONCURRENCY) \ -stacked=$(STACKED) \ + -burst=$(BURST) \ -addr $$addr \ -queue $(QUEUE) \ -strategy $(STRATEGY) \ diff --git a/service/submitqueue/demo/requests/BUILD.bazel b/service/submitqueue/demo/requests/BUILD.bazel index 630398d59..e5a0dc656 100644 --- a/service/submitqueue/demo/requests/BUILD.bazel +++ b/service/submitqueue/demo/requests/BUILD.bazel @@ -46,10 +46,12 @@ go_test( "SUBMITQUEUE_TEST_GIT": "$(location @git//:git)", }, deps = [ + "//api/base/mergestrategy/protopb:go_default_library", "//platform/base/change/git:go_default_library", "//platform/fakemarker:go_default_library", "//platform/git/exec:go_default_library", "//platform/git/exectest:go_default_library", + "//submitqueue/client:go_default_library", "@com_github_stretchr_testify//assert:go_default_library", "@com_github_stretchr_testify//require:go_default_library", ], diff --git a/service/submitqueue/demo/requests/main.go b/service/submitqueue/demo/requests/main.go index afb56898e..c1e3becfa 100644 --- a/service/submitqueue/demo/requests/main.go +++ b/service/submitqueue/demo/requests/main.go @@ -92,6 +92,7 @@ type config struct { files int concurrency int stacked bool + burst bool prefix string land bool watch bool @@ -120,6 +121,8 @@ func parseFlags() config { flag.IntVar(&c.concurrency, "concurrency", 5, "how many changes to create at once; a stack ignores it, being sequential by nature, and -provider git serializes its git commands") flag.BoolVar(&c.stacked, "stacked", false, "chain the changes and enqueue them as one stack") + flag.BoolVar(&c.burst, "burst", false, + "create every change first, then enqueue them all at once instead of as each is created; independent changes only") flag.StringVar(&c.prefix, "prefix", "demo", "branch name prefix") flag.BoolVar(&c.land, "land", true, "enqueue each change as it is created") flag.BoolVar(&c.watch, "watch", true, "watch the requests until they all settle") @@ -242,6 +245,9 @@ func shape(cfg config) string { if cfg.stacked { return "stacked, enqueued as one request once the chain exists" } + if cfg.burst && cfg.land { + return fmt.Sprintf("independent, created %d at a time, then all enqueued at once", cfg.concurrency) + } if cfg.concurrency > 1 { return fmt.Sprintf("independent, %d at a time, each enqueued as soon as it is created", cfg.concurrency) } @@ -432,6 +438,12 @@ func changeFileCount(tag string, change, min int) int { return min + int(sum[0]%4) } +// lander enqueues a request. *client.Client is the real one; a test supplies +// its own to observe when each change is enqueued relative to when it is created. +type lander interface { + Land(ctx context.Context, queue string, uris []string, strategy mergestrategypb.Strategy) (string, error) +} + // createAndEnqueue creates the changes and puts them on the queue, filling // in the tracker's rows as it goes and reporting each step beneath the table. // @@ -451,7 +463,7 @@ func changeFileCount(tag string, change, min int) int { func createAndEnqueue( ctx context.Context, src changeSource, - sq *client.Client, + sq lander, cfg config, strategy mergestrategypb.Strategy, tag, baseSHA string, @@ -480,12 +492,16 @@ func createAndEnqueue( func createIndependent( ctx context.Context, src changeSource, - sq *client.Client, + sq lander, cfg config, strategy mergestrategypb.Strategy, tag, baseSHA string, t *client.Tracker, ) ([]change, error) { + if cfg.burst && cfg.land { + return createBurst(ctx, src, sq, cfg, strategy, tag, baseSHA, t) + } + rows := t.Rows() // Indexed rather than appended: the workers finish in whatever order the // provider answers them, and the caller still wants the run's own order. @@ -521,6 +537,62 @@ func createIndependent( return created, nil } +// createBurst creates every change first and only then enqueues them, firing +// all the Land calls together so the requests arrive at the queue in one burst. +// +// The default path lands each change the moment it exists, so the queue starts +// working while later changes are still being created. Burst trades that early +// overlap for a simultaneous arrival: useful for watching the queue admit a +// large batch at once. It does not make creation faster — with -provider git the +// creation phase is still serialized on a single work tree — it only separates +// creation from enqueuing so the enqueues are not spread across it. +func createBurst( + ctx context.Context, + src changeSource, + sq lander, + cfg config, + strategy mergestrategypb.Strategy, + tag, baseSHA string, + t *client.Tracker, +) ([]change, error) { + rows := t.Rows() + created := make([]change, cfg.count) + + create, createCtx := errgroup.WithContext(ctx) + create.SetLimit(cfg.concurrency) + for i := 1; i <= cfg.count; i++ { + create.Go(func() error { + c, err := createOne(createCtx, src, cfg, tag, baseSHA, cfg.base, i, t, rows[i-1]) + if err != nil { + return err + } + created[i-1] = c + return nil + }) + } + if err := create.Wait(); err != nil { + return nil, err + } + + t.Note("enqueuing %d changes at once", cfg.count) + land, landCtx := errgroup.WithContext(ctx) + land.SetLimit(cfg.concurrency) + for i := 1; i <= cfg.count; i++ { + land.Go(func() error { + sqid, err := sq.Land(landCtx, cfg.queue, urisOf([]change{created[i-1]}), strategy) + if err != nil { + return err + } + t.Update(func() { rows[i-1].SQID, rows[i-1].Submitted = sqid, time.Now() }) + return nil + }) + } + if err := land.Wait(); err != nil { + return nil, err + } + return created, nil +} + // createStack creates the changes one after another, each based on the one // before it, and submits the whole chain as a single request. // @@ -530,7 +602,7 @@ func createIndependent( func createStack( ctx context.Context, src changeSource, - sq *client.Client, + sq lander, cfg config, strategy mergestrategypb.Strategy, tag, baseSHA string, diff --git a/service/submitqueue/demo/requests/main_test.go b/service/submitqueue/demo/requests/main_test.go index 48a856157..7d0a36c89 100644 --- a/service/submitqueue/demo/requests/main_test.go +++ b/service/submitqueue/demo/requests/main_test.go @@ -15,12 +15,18 @@ package main import ( + "context" "fmt" "strings" + "sync" + "sync/atomic" "testing" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + + mergestrategypb "github.com/uber/submitqueue/api/base/mergestrategy/protopb" + "github.com/uber/submitqueue/submitqueue/client" ) func TestChangeFilePath_IsUniquePerFileAcrossChangesAndRuns(t *testing.T) { @@ -206,4 +212,62 @@ func TestShapeReportsConcurrency(t *testing.T) { "one at a time is just sequential; saying so adds nothing") assert.Contains(t, shape(config{count: 10, concurrency: 5, stacked: true}), "stacked", "a stack is sequential whatever the limit says") + assert.Contains(t, shape(config{count: 10, concurrency: 5, land: true, burst: true}), "all enqueued at once") + assert.NotContains(t, shape(config{count: 10, concurrency: 5, burst: true, land: false}), "at once", + "burst is a landing decision; with nothing to land there is no burst") +} + +// recordingSource counts how many changes have been created, so a test can +// observe creation relative to enqueuing. It creates through fakeSource, so it +// stands in for any provider — the two-phase orchestration under test is the +// same whichever one is wired. +type recordingSource struct { + created *int64 +} + +func (recordingSource) baseSHA(context.Context, string) (string, error) { return "base", nil } + +func (r recordingSource) open(ctx context.Context, spec changeSpec) (openedChange, error) { + atomic.AddInt64(r.created, 1) + return fakeSource{}.open(ctx, spec) +} + +// landerFunc adapts a function to the lander interface. +type landerFunc func(context.Context, string, []string, mergestrategypb.Strategy) (string, error) + +func (f landerFunc) Land(ctx context.Context, queue string, uris []string, s mergestrategypb.Strategy) (string, error) { + return f(ctx, queue, uris, s) +} + +// TestCreateBurst_EnqueuesOnlyAfterEveryChangeIsCreated is the property -burst +// exists for: no request reaches the queue until the last change has been +// created, so they all arrive together. It is provider-independent — burst runs +// every source through the same two phases. +func TestCreateBurst_EnqueuesOnlyAfterEveryChangeIsCreated(t *testing.T) { + const count = 8 + cfg := config{count: count, concurrency: 4, files: 3, folders: 3, land: true, burst: true, queue: "q", prefix: "demo"} + + var created int64 + src := recordingSource{created: &created} + + var mu sync.Mutex + createdWhenLanded := make([]int64, 0, count) + lander := landerFunc(func(context.Context, string, []string, mergestrategypb.Strategy) (string, error) { + mu.Lock() + createdWhenLanded = append(createdWhenLanded, atomic.LoadInt64(&created)) + mu.Unlock() + return "sqid", nil + }) + + tracker := client.NewTracker(client.NewRows(count)) + got, err := createBurst(context.Background(), src, lander, cfg, + mergestrategypb.Strategy_SQUASH_REBASE, "run", "base", tracker) + require.NoError(t, err) + require.Len(t, got, count) + + require.Len(t, createdWhenLanded, count, "every change is enqueued exactly once") + for _, seen := range createdWhenLanded { + assert.Equal(t, int64(count), seen, + "burst enqueues nothing until every change has been created") + } }