diff --git a/.golangci.yml b/.golangci.yml index 4bd0a5d..487fb1f 100644 --- a/.golangci.yml +++ b/.golangci.yml @@ -22,10 +22,18 @@ linters: - revive linters-settings: + errcheck: + # fmt.Fprint* writes to stdout/stderr; the returned error is + # conventionally ignored because there's nowhere meaningful to + # redirect it. + exclude-functions: + - fmt.Fprint + - fmt.Fprintf + - fmt.Fprintln govet: # These govet checks are disabled by default, but they're useful. enable: - - niliness + - nilness - reflectvaluecompare - sortslice - unusedwrite diff --git a/examples/magic_gen.go b/examples/magic_gen.go index fe05e6e..751098c 100644 --- a/examples/magic_gen.go +++ b/examples/magic_gen.go @@ -240,10 +240,10 @@ func (h *FooHandler) HandleFoo(ctx context.Context, req *Request) (*Response, er _ = p0PanicStacktrace // possibly unused. pred1 := new(struct { ran cff.AtomicBool - run func(context.Context) error + run func(context.Context) (bool, error) job *cff.ScheduledJob }) - pred1.run = func(ctx context.Context) (err error) { + pred1.run = func(ctx context.Context) (result bool, err error) { defer func() { if recovered := recover(); recovered != nil { p0PanicRecover = recovered @@ -251,10 +251,10 @@ func (h *FooHandler) HandleFoo(ctx context.Context, req *Request) (*Response, er } }() p0 = _61_18(v2) - return nil + return p0, nil } - pred1.job = sched.Enqueue(ctx, cff.Job{ + pred1.job = sched.EnqueuePredicate(ctx, cff.PredicateJob{ Run: pred1.run, Dependencies: []*cff.ScheduledJob{ task0.job, @@ -327,10 +327,10 @@ func (h *FooHandler) HandleFoo(ctx context.Context, req *Request) (*Response, er _ = p1PanicStacktrace // possibly unused. pred2 := new(struct { ran cff.AtomicBool - run func(context.Context) error + run func(context.Context) (bool, error) job *cff.ScheduledJob }) - pred2.run = func(ctx context.Context) (err error) { + pred2.run = func(ctx context.Context) (result bool, err error) { defer func() { if recovered := recover(); recovered != nil { p1PanicRecover = recovered @@ -338,10 +338,10 @@ func (h *FooHandler) HandleFoo(ctx context.Context, req *Request) (*Response, er } }() p1 = _74_18(v2) - return nil + return p1, nil } - pred2.job = sched.Enqueue(ctx, cff.Job{ + pred2.job = sched.EnqueuePredicate(ctx, cff.PredicateJob{ Run: pred2.run, Dependencies: []*cff.ScheduledJob{ task0.job, diff --git a/internal/templates/flow/predicate.go.tmpl b/internal/templates/flow/predicate.go.tmpl index b98f07f..726395a 100644 --- a/internal/templates/flow/predicate.go.tmpl +++ b/internal/templates/flow/predicate.go.tmpl @@ -8,7 +8,7 @@ var p{{ predHash . }}PanicRecover interface{} var p{{ predHash . }}PanicStacktrace []byte _ = p{{ predHash . }}PanicStacktrace // possibly unused. {{ $p }} := new({{ template "predicate" }}) -{{ $p }}.run = func(ctx {{ $context }}.Context) (err error) { +{{ $p }}.run = func(ctx {{ $context }}.Context) (result bool, err error) { defer func() { if recovered := recover(); recovered != nil { p{{ predHash . }}PanicRecover = recovered @@ -16,10 +16,10 @@ _ = p{{ predHash . }}PanicStacktrace // possibly unused. } }() p{{ predHash . }} = {{ expr .Function.Node }}{{ template "callTaskArgs" . }} - return nil + return p{{ predHash . }}, nil } -{{ $p }}.job = sched.Enqueue(ctx, {{ $cff }}.Job{ +{{ $p }}.job = sched.EnqueuePredicate(ctx, {{ $cff }}.PredicateJob{ Run: {{ $p }}.run, {{ if .Function.DependsOn -}} Dependencies: []*{{ $cff }}.ScheduledJob{ diff --git a/internal/templates/flow/types.go.tmpl b/internal/templates/flow/types.go.tmpl index 9656c23..9834696 100644 --- a/internal/templates/flow/types.go.tmpl +++ b/internal/templates/flow/types.go.tmpl @@ -16,7 +16,7 @@ struct { ran {{ $cff }}.AtomicBool - run func({{ $context }}.Context) error + run func({{ $context }}.Context) (bool, error) job *{{ $cff }}.ScheduledJob } {{- end -}} diff --git a/internal/tests/benchmark/benchmark_predicate.go b/internal/tests/benchmark/benchmark_predicate.go index e86a727..50b27bc 100644 --- a/internal/tests/benchmark/benchmark_predicate.go +++ b/internal/tests/benchmark/benchmark_predicate.go @@ -81,3 +81,40 @@ func PredicateSplit() float64 { ) return res } + +// PredicateFalseWithSlowDep runs a cff.Flow where a predicate returns false +// while a sibling input task sleeps for slowDelay. Returns the elapsed +// wall-clock time from flow start to consumer dispatch, so the benchmark +// can report consumer_ns/op as a custom metric. +// +// Before scheduler-aware predicates, the consumer waited the full +// slowDelay before its skip gate fired. After, the consumer dispatches +// immediately on predicate=false, even though Wait() still drains the +// slow producer. +func PredicateFalseWithSlowDep(slowDelay time.Duration) time.Duration { + type slowOut struct{} + type fastOut struct{} + type skippedOut struct{} + + var elapsed time.Duration + start := time.Now() + cff.Flow( + context.Background(), + cff.Concurrency(_concurrency), + cff.Results(&elapsed), + cff.Task(func() slowOut { + time.Sleep(slowDelay) + return slowOut{} + }), + cff.Task(func() fastOut { return fastOut{} }), + cff.Task( + func(slowOut) (skippedOut, error) { return skippedOut{}, nil }, + cff.Predicate(func(fastOut) bool { return false }), + cff.FallbackWith(skippedOut{}), + ), + cff.Task(func(skippedOut) time.Duration { + return time.Since(start) + }), + ) + return elapsed +} diff --git a/internal/tests/benchmark/benchmark_predicate_gen.go b/internal/tests/benchmark/benchmark_predicate_gen.go index f7544ac..2ba9d58 100644 --- a/internal/tests/benchmark/benchmark_predicate_gen.go +++ b/internal/tests/benchmark/benchmark_predicate_gen.go @@ -330,10 +330,10 @@ func PredicateSplit() float64 { _ = p0PanicStacktrace // possibly unused. pred1 := new(struct { ran cff.AtomicBool - run func(context.Context) error + run func(context.Context) (bool, error) job *cff.ScheduledJob }) - pred1.run = func(ctx context.Context) (err error) { + pred1.run = func(ctx context.Context) (result bool, err error) { defer func() { if recovered := recover(); recovered != nil { p0PanicRecover = recovered @@ -341,10 +341,10 @@ func PredicateSplit() float64 { } }() p0 = _78_5() - return nil + return p0, nil } - pred1.job = sched.Enqueue(ctx, cff.Job{ + pred1.job = sched.EnqueuePredicate(ctx, cff.PredicateJob{ Run: pred1.run, }) @@ -421,3 +421,329 @@ func PredicateSplit() float64 { }() return res } + +// PredicateFalseWithSlowDep runs a cff.Flow where a predicate returns false +// while a sibling input task sleeps for slowDelay. Returns the elapsed +// wall-clock time from flow start to consumer dispatch, so the benchmark +// can report consumer_ns/op as a custom metric. +// +// Before scheduler-aware predicates, the consumer waited the full +// slowDelay before its skip gate fired. After, the consumer dispatches +// immediately on predicate=false, even though Wait() still drains the +// slow producer. +func PredicateFalseWithSlowDep(slowDelay time.Duration) time.Duration { + type slowOut struct{} + type fastOut struct{} + type skippedOut struct{} + + var elapsed time.Duration + start := time.Now() + func() (err error) { + + _102_3 := context.Background() + + _103_19 := _concurrency + + _104_15 := &elapsed + + _105_12 := func() slowOut { + time.Sleep(slowDelay) + return slowOut{} + } + + _109_12 := func() fastOut { return fastOut{} } + + _111_4 := func(slowOut) (skippedOut, error) { return skippedOut{}, nil } + + _112_18 := func(fastOut) bool { return false } + + _113_21 := skippedOut{} + + _115_12 := func(skippedOut) time.Duration { + return time.Since(start) + } + ctx := _102_3 + emitter := cff.NopEmitter() + + var ( + flowInfo = &cff.FlowInfo{ + File: "go.uber.org/cff/internal/tests/benchmark/benchmark_predicate.go", + Line: 101, + Column: 2, + } + flowEmitter = cff.NopFlowEmitter() + + schedInfo = &cff.SchedulerInfo{ + Name: flowInfo.Name, + Directive: cff.FlowDirective, + File: flowInfo.File, + Line: flowInfo.Line, + Column: flowInfo.Column, + } + + // possibly unused + _ = flowInfo + ) + + startTime := time.Now() + defer func() { flowEmitter.FlowDone(ctx, time.Since(startTime)) }() + + schedEmitter := emitter.SchedulerInit(schedInfo) + + sched := cff.NewScheduler( + cff.SchedulerParams{ + Concurrency: _103_19, Emitter: schedEmitter, + }, + ) + + var tasks []*struct { + emitter cff.TaskEmitter + ran cff.AtomicBool + run func(context.Context) error + job *cff.ScheduledJob + } + defer func() { + for _, t := range tasks { + if !t.ran.Load() { + t.emitter.TaskSkipped(ctx, err) + } + } + }() + + // go.uber.org/cff/internal/tests/benchmark/benchmark_predicate.go:105:12 + var ( + v3 slowOut + ) + task4 := new(struct { + emitter cff.TaskEmitter + ran cff.AtomicBool + run func(context.Context) error + job *cff.ScheduledJob + }) + task4.emitter = cff.NopTaskEmitter() + task4.run = func(ctx context.Context) (err error) { + taskEmitter := task4.emitter + startTime := time.Now() + defer func() { + if task4.ran.Load() { + taskEmitter.TaskDone(ctx, time.Since(startTime)) + } + }() + + defer func() { + recovered := recover() + if recovered != nil { + taskEmitter.TaskPanic(ctx, recovered) + err = &cff.PanicError{ + Value: recovered, + Stacktrace: debug.Stack(), + } + } + }() + + defer task4.ran.Store(true) + + v3 = _105_12() + + taskEmitter.TaskSuccess(ctx) + + return + } + + task4.job = sched.Enqueue(ctx, cff.Job{ + Run: task4.run, + }) + tasks = append(tasks, task4) + + // go.uber.org/cff/internal/tests/benchmark/benchmark_predicate.go:109:12 + var ( + v4 fastOut + ) + task5 := new(struct { + emitter cff.TaskEmitter + ran cff.AtomicBool + run func(context.Context) error + job *cff.ScheduledJob + }) + task5.emitter = cff.NopTaskEmitter() + task5.run = func(ctx context.Context) (err error) { + taskEmitter := task5.emitter + startTime := time.Now() + defer func() { + if task5.ran.Load() { + taskEmitter.TaskDone(ctx, time.Since(startTime)) + } + }() + + defer func() { + recovered := recover() + if recovered != nil { + taskEmitter.TaskPanic(ctx, recovered) + err = &cff.PanicError{ + Value: recovered, + Stacktrace: debug.Stack(), + } + } + }() + + defer task5.ran.Store(true) + + v4 = _109_12() + + taskEmitter.TaskSuccess(ctx) + + return + } + + task5.job = sched.Enqueue(ctx, cff.Job{ + Run: task5.run, + }) + tasks = append(tasks, task5) + + // go.uber.org/cff/internal/tests/benchmark/benchmark_predicate.go:112:4 + var p0 bool + var p0PanicRecover interface{} + var p0PanicStacktrace []byte + _ = p0PanicStacktrace // possibly unused. + pred1 := new(struct { + ran cff.AtomicBool + run func(context.Context) (bool, error) + job *cff.ScheduledJob + }) + pred1.run = func(ctx context.Context) (result bool, err error) { + defer func() { + if recovered := recover(); recovered != nil { + p0PanicRecover = recovered + p0PanicStacktrace = debug.Stack() + } + }() + p0 = _112_18(v4) + return p0, nil + } + + pred1.job = sched.EnqueuePredicate(ctx, cff.PredicateJob{ + Run: pred1.run, + Dependencies: []*cff.ScheduledJob{ + task5.job, + }, + }) + + // go.uber.org/cff/internal/tests/benchmark/benchmark_predicate.go:111:4 + var ( + v5 skippedOut + ) + task6 := new(struct { + emitter cff.TaskEmitter + ran cff.AtomicBool + run func(context.Context) error + job *cff.ScheduledJob + }) + task6.emitter = cff.NopTaskEmitter() + task6.run = func(ctx context.Context) (err error) { + taskEmitter := task6.emitter + startTime := time.Now() + defer func() { + if task6.ran.Load() { + taskEmitter.TaskDone(ctx, time.Since(startTime)) + } + }() + + defer func() { + recovered := recover() + + if recovered == nil && p0PanicRecover != nil { + recovered = p0PanicRecover + } + if recovered != nil { + taskEmitter.TaskPanicRecovered(ctx, recovered) + v5, err = _113_21, nil + } + }() + + if !p0 { + return nil + } + + defer task6.ran.Store(true) + + v5, err = _111_4(v3) + + if err != nil { + taskEmitter.TaskErrorRecovered(ctx, err) + v5, err = _113_21, nil + } else { + taskEmitter.TaskSuccess(ctx) + } + + return + } + + task6.job = sched.Enqueue(ctx, cff.Job{ + Run: task6.run, + Dependencies: []*cff.ScheduledJob{ + task4.job, + pred1.job, + }, + }) + tasks = append(tasks, task6) + + // go.uber.org/cff/internal/tests/benchmark/benchmark_predicate.go:115:12 + var ( + v6 time.Duration + ) + task7 := new(struct { + emitter cff.TaskEmitter + ran cff.AtomicBool + run func(context.Context) error + job *cff.ScheduledJob + }) + task7.emitter = cff.NopTaskEmitter() + task7.run = func(ctx context.Context) (err error) { + taskEmitter := task7.emitter + startTime := time.Now() + defer func() { + if task7.ran.Load() { + taskEmitter.TaskDone(ctx, time.Since(startTime)) + } + }() + + defer func() { + recovered := recover() + if recovered != nil { + taskEmitter.TaskPanic(ctx, recovered) + err = &cff.PanicError{ + Value: recovered, + Stacktrace: debug.Stack(), + } + } + }() + + defer task7.ran.Store(true) + + v6 = _115_12(v5) + + taskEmitter.TaskSuccess(ctx) + + return + } + + task7.job = sched.Enqueue(ctx, cff.Job{ + Run: task7.run, + Dependencies: []*cff.ScheduledJob{ + task6.job, + }, + }) + tasks = append(tasks, task7) + + if err := sched.Wait(ctx); err != nil { + flowEmitter.FlowError(ctx, err) + return err + } + + *(_104_15) = v6 // time.Duration + + flowEmitter.FlowSuccess(ctx) + return nil + }() + return elapsed +} diff --git a/internal/tests/benchmark/benchmark_predicate_test.go b/internal/tests/benchmark/benchmark_predicate_test.go new file mode 100644 index 0000000..f240d91 --- /dev/null +++ b/internal/tests/benchmark/benchmark_predicate_test.go @@ -0,0 +1,20 @@ +package benchmark + +import ( + "testing" + "time" +) + +// BenchmarkPredicateFalseWithSlowDep measures consumer-dispatch latency +// when a predicate returns false and a sibling input task sleeps. +// Per-iteration wall time is bounded by Wait() draining the slow +// producer; the interesting metric is consumer_ns/op (the time from +// flow start to consumer firing), reported as a custom metric. +func BenchmarkPredicateFalseWithSlowDep(b *testing.B) { + slowDelay := 5 * time.Millisecond + var totalConsumer time.Duration + for i := 0; i < b.N; i++ { + totalConsumer += PredicateFalseWithSlowDep(slowDelay) + } + b.ReportMetric(float64(totalConsumer.Nanoseconds())/float64(b.N), "consumer_ns/op") +} diff --git a/internal/tests/predicate/predicate.go b/internal/tests/predicate/predicate.go index d92558e..229b584 100644 --- a/internal/tests/predicate/predicate.go +++ b/internal/tests/predicate/predicate.go @@ -5,6 +5,7 @@ package predicate import ( "context" + "time" "go.uber.org/cff" ) @@ -178,3 +179,43 @@ func PanickedWithFallback() (string, error) { ) return s, err } + +// BlockingInputs builds a flow that probes whether cff.Predicate +// short-circuits the scheduler's wait on the predicated task's input +// dependencies. +// +// Shape: a slow source feeds the predicated task; a fast source feeds +// the predicate. The predicate returns false (so the task body is +// skipped via FallbackWith). A downstream consumer measures the +// elapsed wall-clock time from flow start. +// +// If predicates short-circuit input deps: +// elapsed ≈ 0 (slow path is not on the critical path) +// If predicates only short-circuit the task body: +// elapsed ≈ slowDelay (scheduler still waits on slowOut) +func BlockingInputs(slowDelay time.Duration) (time.Duration, error) { + type slowOut struct{} + type fastOut struct{} + type skippedOut struct{} + + var elapsed time.Duration + start := time.Now() + err := cff.Flow( + context.Background(), + cff.Results(&elapsed), + cff.Task(func() slowOut { + time.Sleep(slowDelay) + return slowOut{} + }), + cff.Task(func() fastOut { return fastOut{} }), + cff.Task( + func(slowOut) (skippedOut, error) { return skippedOut{}, nil }, + cff.Predicate(func(fastOut) bool { return false }), + cff.FallbackWith(skippedOut{}), + ), + cff.Task(func(skippedOut) time.Duration { + return time.Since(start) + }), + ) + return elapsed, err +} diff --git a/internal/tests/predicate/predicate_gen.go b/internal/tests/predicate/predicate_gen.go index 78a6d71..bf6e48c 100644 --- a/internal/tests/predicate/predicate_gen.go +++ b/internal/tests/predicate/predicate_gen.go @@ -17,23 +17,23 @@ func Simple(f func(), pred bool) error { var s string return func() (err error) { - _17_3 := context.Background() + _18_3 := context.Background() - _18_15 := &s + _19_15 := &s - _20_4 := func() string { + _21_4 := func() string { f() return "foo" } - _24_18 := func() bool { return pred } - ctx := _17_3 + _25_18 := func() bool { return pred } + ctx := _18_3 emitter := cff.NopEmitter() var ( flowInfo = &cff.FlowInfo{ File: "go.uber.org/cff/internal/tests/predicate/predicate.go", - Line: 16, + Line: 17, Column: 9, } flowEmitter = cff.NopFlowEmitter() @@ -75,32 +75,32 @@ func Simple(f func(), pred bool) error { } }() - // go.uber.org/cff/internal/tests/predicate/predicate.go:24:4 + // go.uber.org/cff/internal/tests/predicate/predicate.go:25:4 var p0 bool var p0PanicRecover interface{} var p0PanicStacktrace []byte _ = p0PanicStacktrace // possibly unused. pred1 := new(struct { ran cff.AtomicBool - run func(context.Context) error + run func(context.Context) (bool, error) job *cff.ScheduledJob }) - pred1.run = func(ctx context.Context) (err error) { + pred1.run = func(ctx context.Context) (result bool, err error) { defer func() { if recovered := recover(); recovered != nil { p0PanicRecover = recovered p0PanicStacktrace = debug.Stack() } }() - p0 = _24_18() - return nil + p0 = _25_18() + return p0, nil } - pred1.job = sched.Enqueue(ctx, cff.Job{ + pred1.job = sched.EnqueuePredicate(ctx, cff.PredicateJob{ Run: pred1.run, }) - // go.uber.org/cff/internal/tests/predicate/predicate.go:20:4 + // go.uber.org/cff/internal/tests/predicate/predicate.go:21:4 var ( v1 string ) @@ -145,7 +145,7 @@ func Simple(f func(), pred bool) error { defer task0.ran.Store(true) - v1 = _20_4() + v1 = _21_4() taskEmitter.TaskSuccess(ctx) @@ -165,7 +165,7 @@ func Simple(f func(), pred bool) error { return err } - *(_18_15) = v1 // string + *(_19_15) = v1 // string flowEmitter.FlowSuccess(ctx) return nil @@ -178,27 +178,27 @@ func SimpleWithContextTask() error { var s string return func() (err error) { - _34_3 := context.Background() + _35_3 := context.Background() - _35_15 := &s + _36_15 := &s - _36_14 := int64(2) + _37_14 := int64(2) - _38_4 := func(ctx context.Context) string { + _39_4 := func(ctx context.Context) string { return "foo" } - _42_5 := func(int64) bool { + _43_5 := func(int64) bool { return false } - ctx := _34_3 - var v2 int64 = _36_14 + ctx := _35_3 + var v2 int64 = _37_14 emitter := cff.NopEmitter() var ( flowInfo = &cff.FlowInfo{ File: "go.uber.org/cff/internal/tests/predicate/predicate.go", - Line: 33, + Line: 34, Column: 9, } flowEmitter = cff.NopFlowEmitter() @@ -240,32 +240,32 @@ func SimpleWithContextTask() error { } }() - // go.uber.org/cff/internal/tests/predicate/predicate.go:41:4 + // go.uber.org/cff/internal/tests/predicate/predicate.go:42:4 var p0 bool var p0PanicRecover interface{} var p0PanicStacktrace []byte _ = p0PanicStacktrace // possibly unused. pred1 := new(struct { ran cff.AtomicBool - run func(context.Context) error + run func(context.Context) (bool, error) job *cff.ScheduledJob }) - pred1.run = func(ctx context.Context) (err error) { + pred1.run = func(ctx context.Context) (result bool, err error) { defer func() { if recovered := recover(); recovered != nil { p0PanicRecover = recovered p0PanicStacktrace = debug.Stack() } }() - p0 = _42_5(v2) - return nil + p0 = _43_5(v2) + return p0, nil } - pred1.job = sched.Enqueue(ctx, cff.Job{ + pred1.job = sched.EnqueuePredicate(ctx, cff.PredicateJob{ Run: pred1.run, }) - // go.uber.org/cff/internal/tests/predicate/predicate.go:38:4 + // go.uber.org/cff/internal/tests/predicate/predicate.go:39:4 var ( v1 string ) @@ -310,7 +310,7 @@ func SimpleWithContextTask() error { defer task1.ran.Store(true) - v1 = _38_4(ctx) + v1 = _39_4(ctx) taskEmitter.TaskSuccess(ctx) @@ -330,7 +330,7 @@ func SimpleWithContextTask() error { return err } - *(_35_15) = v1 // string + *(_36_15) = v1 // string flowEmitter.FlowSuccess(ctx) return nil @@ -343,27 +343,27 @@ func SimpleWithContextPredicate() error { var s string return func() (err error) { - _54_3 := context.Background() + _55_3 := context.Background() - _55_15 := &s + _56_15 := &s - _56_14 := int64(2) + _57_14 := int64(2) - _58_4 := func() string { + _59_4 := func() string { return "foo" } - _62_5 := func(context.Context, int64) bool { + _63_5 := func(context.Context, int64) bool { return false } - ctx := _54_3 - var v2 int64 = _56_14 + ctx := _55_3 + var v2 int64 = _57_14 emitter := cff.NopEmitter() var ( flowInfo = &cff.FlowInfo{ File: "go.uber.org/cff/internal/tests/predicate/predicate.go", - Line: 53, + Line: 54, Column: 9, } flowEmitter = cff.NopFlowEmitter() @@ -405,32 +405,32 @@ func SimpleWithContextPredicate() error { } }() - // go.uber.org/cff/internal/tests/predicate/predicate.go:61:4 + // go.uber.org/cff/internal/tests/predicate/predicate.go:62:4 var p0 bool var p0PanicRecover interface{} var p0PanicStacktrace []byte _ = p0PanicStacktrace // possibly unused. pred1 := new(struct { ran cff.AtomicBool - run func(context.Context) error + run func(context.Context) (bool, error) job *cff.ScheduledJob }) - pred1.run = func(ctx context.Context) (err error) { + pred1.run = func(ctx context.Context) (result bool, err error) { defer func() { if recovered := recover(); recovered != nil { p0PanicRecover = recovered p0PanicStacktrace = debug.Stack() } }() - p0 = _62_5(ctx, v2) - return nil + p0 = _63_5(ctx, v2) + return p0, nil } - pred1.job = sched.Enqueue(ctx, cff.Job{ + pred1.job = sched.EnqueuePredicate(ctx, cff.PredicateJob{ Run: pred1.run, }) - // go.uber.org/cff/internal/tests/predicate/predicate.go:58:4 + // go.uber.org/cff/internal/tests/predicate/predicate.go:59:4 var ( v1 string ) @@ -475,7 +475,7 @@ func SimpleWithContextPredicate() error { defer task2.ran.Store(true) - v1 = _58_4() + v1 = _59_4() taskEmitter.TaskSuccess(ctx) @@ -495,7 +495,7 @@ func SimpleWithContextPredicate() error { return err } - *(_55_15) = v1 // string + *(_56_15) = v1 // string flowEmitter.FlowSuccess(ctx) return nil @@ -508,27 +508,27 @@ func SimpleWithContextTaskAndPredicate() error { var s string return func() (err error) { - _74_3 := context.Background() + _75_3 := context.Background() - _75_15 := &s + _76_15 := &s - _76_14 := int64(2) + _77_14 := int64(2) - _78_4 := func(ctx context.Context) string { + _79_4 := func(ctx context.Context) string { return "foo" } - _82_5 := func(context.Context, int64) bool { + _83_5 := func(context.Context, int64) bool { return false } - ctx := _74_3 - var v2 int64 = _76_14 + ctx := _75_3 + var v2 int64 = _77_14 emitter := cff.NopEmitter() var ( flowInfo = &cff.FlowInfo{ File: "go.uber.org/cff/internal/tests/predicate/predicate.go", - Line: 73, + Line: 74, Column: 9, } flowEmitter = cff.NopFlowEmitter() @@ -570,32 +570,32 @@ func SimpleWithContextTaskAndPredicate() error { } }() - // go.uber.org/cff/internal/tests/predicate/predicate.go:81:4 + // go.uber.org/cff/internal/tests/predicate/predicate.go:82:4 var p0 bool var p0PanicRecover interface{} var p0PanicStacktrace []byte _ = p0PanicStacktrace // possibly unused. pred1 := new(struct { ran cff.AtomicBool - run func(context.Context) error + run func(context.Context) (bool, error) job *cff.ScheduledJob }) - pred1.run = func(ctx context.Context) (err error) { + pred1.run = func(ctx context.Context) (result bool, err error) { defer func() { if recovered := recover(); recovered != nil { p0PanicRecover = recovered p0PanicStacktrace = debug.Stack() } }() - p0 = _82_5(ctx, v2) - return nil + p0 = _83_5(ctx, v2) + return p0, nil } - pred1.job = sched.Enqueue(ctx, cff.Job{ + pred1.job = sched.EnqueuePredicate(ctx, cff.PredicateJob{ Run: pred1.run, }) - // go.uber.org/cff/internal/tests/predicate/predicate.go:78:4 + // go.uber.org/cff/internal/tests/predicate/predicate.go:79:4 var ( v1 string ) @@ -640,7 +640,7 @@ func SimpleWithContextTaskAndPredicate() error { defer task3.ran.Store(true) - v1 = _78_4(ctx) + v1 = _79_4(ctx) taskEmitter.TaskSuccess(ctx) @@ -660,7 +660,7 @@ func SimpleWithContextTaskAndPredicate() error { return err } - *(_75_15) = v1 // string + *(_76_15) = v1 // string flowEmitter.FlowSuccess(ctx) return nil @@ -677,29 +677,29 @@ func ExtraDependencies() error { var out t3 return func() (err error) { - _98_3 := context.Background() + _99_3 := context.Background() - _99_14 := int(42) + _100_14 := int(42) - _100_15 := &out + _101_15 := &out - _102_4 := func(int) t1 { return t1{} } + _103_4 := func(int) t1 { return t1{} } - _104_4 := func() t2 { return t2{} } + _105_4 := func() t2 { return t2{} } - _106_4 := func(t2) t3 { return t3{} } + _107_4 := func(t2) t3 { return t3{} } - _108_5 := func(int, t1) bool { + _109_5 := func(int, t1) bool { return true } - ctx := _98_3 - var v3 int = _99_14 + ctx := _99_3 + var v3 int = _100_14 emitter := cff.NopEmitter() var ( flowInfo = &cff.FlowInfo{ File: "go.uber.org/cff/internal/tests/predicate/predicate.go", - Line: 97, + Line: 98, Column: 9, } flowEmitter = cff.NopFlowEmitter() @@ -741,7 +741,7 @@ func ExtraDependencies() error { } }() - // go.uber.org/cff/internal/tests/predicate/predicate.go:102:4 + // go.uber.org/cff/internal/tests/predicate/predicate.go:103:4 var ( v4 t1 ) @@ -774,7 +774,7 @@ func ExtraDependencies() error { defer task4.ran.Store(true) - v4 = _102_4(v3) + v4 = _103_4(v3) taskEmitter.TaskSuccess(ctx) @@ -786,7 +786,7 @@ func ExtraDependencies() error { }) tasks = append(tasks, task4) - // go.uber.org/cff/internal/tests/predicate/predicate.go:104:4 + // go.uber.org/cff/internal/tests/predicate/predicate.go:105:4 var ( v5 t2 ) @@ -819,7 +819,7 @@ func ExtraDependencies() error { defer task5.ran.Store(true) - v5 = _104_4() + v5 = _105_4() taskEmitter.TaskSuccess(ctx) @@ -831,35 +831,35 @@ func ExtraDependencies() error { }) tasks = append(tasks, task5) - // go.uber.org/cff/internal/tests/predicate/predicate.go:107:4 + // go.uber.org/cff/internal/tests/predicate/predicate.go:108:4 var p0 bool var p0PanicRecover interface{} var p0PanicStacktrace []byte _ = p0PanicStacktrace // possibly unused. pred1 := new(struct { ran cff.AtomicBool - run func(context.Context) error + run func(context.Context) (bool, error) job *cff.ScheduledJob }) - pred1.run = func(ctx context.Context) (err error) { + pred1.run = func(ctx context.Context) (result bool, err error) { defer func() { if recovered := recover(); recovered != nil { p0PanicRecover = recovered p0PanicStacktrace = debug.Stack() } }() - p0 = _108_5(v3, v4) - return nil + p0 = _109_5(v3, v4) + return p0, nil } - pred1.job = sched.Enqueue(ctx, cff.Job{ + pred1.job = sched.EnqueuePredicate(ctx, cff.PredicateJob{ Run: pred1.run, Dependencies: []*cff.ScheduledJob{ task4.job, }, }) - // go.uber.org/cff/internal/tests/predicate/predicate.go:106:4 + // go.uber.org/cff/internal/tests/predicate/predicate.go:107:4 var ( v6 t3 ) @@ -904,7 +904,7 @@ func ExtraDependencies() error { defer task6.ran.Store(true) - v6 = _106_4(v5) + v6 = _107_4(v5) taskEmitter.TaskSuccess(ctx) @@ -925,7 +925,7 @@ func ExtraDependencies() error { return err } - *(_100_15) = v6 // go.uber.org/cff/internal/tests/predicate.t3 + *(_101_15) = v6 // go.uber.org/cff/internal/tests/predicate.t3 flowEmitter.FlowSuccess(ctx) return nil @@ -939,30 +939,30 @@ func MultiplePredicates() error { var b bool return func() (err error) { - _122_3 := context.Background() + _123_3 := context.Background() - _123_15 := &s + _124_15 := &s - _123_19 := &b + _124_19 := &b - _125_4 := func() string { + _126_4 := func() string { return "foo" } - _128_18 := func() bool { return true } + _129_18 := func() bool { return true } - _131_4 := func() bool { + _132_4 := func() bool { return true } - _134_18 := func() bool { return false } - ctx := _122_3 + _135_18 := func() bool { return false } + ctx := _123_3 emitter := cff.NopEmitter() var ( flowInfo = &cff.FlowInfo{ File: "go.uber.org/cff/internal/tests/predicate/predicate.go", - Line: 121, + Line: 122, Column: 9, } flowEmitter = cff.NopFlowEmitter() @@ -1004,32 +1004,32 @@ func MultiplePredicates() error { } }() - // go.uber.org/cff/internal/tests/predicate/predicate.go:128:4 + // go.uber.org/cff/internal/tests/predicate/predicate.go:129:4 var p0 bool var p0PanicRecover interface{} var p0PanicStacktrace []byte _ = p0PanicStacktrace // possibly unused. pred1 := new(struct { ran cff.AtomicBool - run func(context.Context) error + run func(context.Context) (bool, error) job *cff.ScheduledJob }) - pred1.run = func(ctx context.Context) (err error) { + pred1.run = func(ctx context.Context) (result bool, err error) { defer func() { if recovered := recover(); recovered != nil { p0PanicRecover = recovered p0PanicStacktrace = debug.Stack() } }() - p0 = _128_18() - return nil + p0 = _129_18() + return p0, nil } - pred1.job = sched.Enqueue(ctx, cff.Job{ + pred1.job = sched.EnqueuePredicate(ctx, cff.PredicateJob{ Run: pred1.run, }) - // go.uber.org/cff/internal/tests/predicate/predicate.go:125:4 + // go.uber.org/cff/internal/tests/predicate/predicate.go:126:4 var ( v1 string ) @@ -1074,7 +1074,7 @@ func MultiplePredicates() error { defer task7.ran.Store(true) - v1 = _125_4() + v1 = _126_4() taskEmitter.TaskSuccess(ctx) @@ -1089,32 +1089,32 @@ func MultiplePredicates() error { }) tasks = append(tasks, task7) - // go.uber.org/cff/internal/tests/predicate/predicate.go:134:4 + // go.uber.org/cff/internal/tests/predicate/predicate.go:135:4 var p1 bool var p1PanicRecover interface{} var p1PanicStacktrace []byte _ = p1PanicStacktrace // possibly unused. pred2 := new(struct { ran cff.AtomicBool - run func(context.Context) error + run func(context.Context) (bool, error) job *cff.ScheduledJob }) - pred2.run = func(ctx context.Context) (err error) { + pred2.run = func(ctx context.Context) (result bool, err error) { defer func() { if recovered := recover(); recovered != nil { p1PanicRecover = recovered p1PanicStacktrace = debug.Stack() } }() - p1 = _134_18() - return nil + p1 = _135_18() + return p1, nil } - pred2.job = sched.Enqueue(ctx, cff.Job{ + pred2.job = sched.EnqueuePredicate(ctx, cff.PredicateJob{ Run: pred2.run, }) - // go.uber.org/cff/internal/tests/predicate/predicate.go:131:4 + // go.uber.org/cff/internal/tests/predicate/predicate.go:132:4 var ( v7 bool ) @@ -1159,7 +1159,7 @@ func MultiplePredicates() error { defer task8.ran.Store(true) - v7 = _131_4() + v7 = _132_4() taskEmitter.TaskSuccess(ctx) @@ -1179,8 +1179,8 @@ func MultiplePredicates() error { return err } - *(_123_15) = v1 // string - *(_123_19) = v7 // bool + *(_124_15) = v1 // string + *(_124_19) = v7 // bool flowEmitter.FlowSuccess(ctx) return nil @@ -1192,25 +1192,25 @@ func Panicked() error { var s string return func() (err error) { - _143_3 := context.Background() + _144_3 := context.Background() - _144_15 := &s + _145_15 := &s - _146_4 := func(ctx context.Context) string { + _147_4 := func(ctx context.Context) string { return "foo" } - _150_5 := func() bool { + _151_5 := func() bool { panic("sad times") return true } - ctx := _143_3 + ctx := _144_3 emitter := cff.NopEmitter() var ( flowInfo = &cff.FlowInfo{ File: "go.uber.org/cff/internal/tests/predicate/predicate.go", - Line: 142, + Line: 143, Column: 9, } flowEmitter = cff.NopFlowEmitter() @@ -1252,32 +1252,32 @@ func Panicked() error { } }() - // go.uber.org/cff/internal/tests/predicate/predicate.go:149:4 + // go.uber.org/cff/internal/tests/predicate/predicate.go:150:4 var p0 bool var p0PanicRecover interface{} var p0PanicStacktrace []byte _ = p0PanicStacktrace // possibly unused. pred1 := new(struct { ran cff.AtomicBool - run func(context.Context) error + run func(context.Context) (bool, error) job *cff.ScheduledJob }) - pred1.run = func(ctx context.Context) (err error) { + pred1.run = func(ctx context.Context) (result bool, err error) { defer func() { if recovered := recover(); recovered != nil { p0PanicRecover = recovered p0PanicStacktrace = debug.Stack() } }() - p0 = _150_5() - return nil + p0 = _151_5() + return p0, nil } - pred1.job = sched.Enqueue(ctx, cff.Job{ + pred1.job = sched.EnqueuePredicate(ctx, cff.PredicateJob{ Run: pred1.run, }) - // go.uber.org/cff/internal/tests/predicate/predicate.go:146:4 + // go.uber.org/cff/internal/tests/predicate/predicate.go:147:4 var ( v1 string ) @@ -1322,7 +1322,7 @@ func Panicked() error { defer task9.ran.Store(true) - v1 = _146_4(ctx) + v1 = _147_4(ctx) taskEmitter.TaskSuccess(ctx) @@ -1342,7 +1342,7 @@ func Panicked() error { return err } - *(_144_15) = v1 // string + *(_145_15) = v1 // string flowEmitter.FlowSuccess(ctx) return nil @@ -1355,27 +1355,27 @@ func PanickedWithFallback() (string, error) { var s string err := func() (err error) { - _164_3 := context.Background() + _165_3 := context.Background() - _165_15 := &s + _166_15 := &s - _167_4 := func(ctx context.Context) (string, error) { + _168_4 := func(ctx context.Context) (string, error) { return "foo", nil } - _171_5 := func() bool { + _172_5 := func() bool { panic("sad times") return true } - _176_21 := "predicate-fallback" - ctx := _164_3 + _177_21 := "predicate-fallback" + ctx := _165_3 emitter := cff.NopEmitter() var ( flowInfo = &cff.FlowInfo{ File: "go.uber.org/cff/internal/tests/predicate/predicate.go", - Line: 163, + Line: 164, Column: 9, } flowEmitter = cff.NopFlowEmitter() @@ -1417,32 +1417,32 @@ func PanickedWithFallback() (string, error) { } }() - // go.uber.org/cff/internal/tests/predicate/predicate.go:170:4 + // go.uber.org/cff/internal/tests/predicate/predicate.go:171:4 var p0 bool var p0PanicRecover interface{} var p0PanicStacktrace []byte _ = p0PanicStacktrace // possibly unused. pred1 := new(struct { ran cff.AtomicBool - run func(context.Context) error + run func(context.Context) (bool, error) job *cff.ScheduledJob }) - pred1.run = func(ctx context.Context) (err error) { + pred1.run = func(ctx context.Context) (result bool, err error) { defer func() { if recovered := recover(); recovered != nil { p0PanicRecover = recovered p0PanicStacktrace = debug.Stack() } }() - p0 = _171_5() - return nil + p0 = _172_5() + return p0, nil } - pred1.job = sched.Enqueue(ctx, cff.Job{ + pred1.job = sched.EnqueuePredicate(ctx, cff.PredicateJob{ Run: pred1.run, }) - // go.uber.org/cff/internal/tests/predicate/predicate.go:167:4 + // go.uber.org/cff/internal/tests/predicate/predicate.go:168:4 var ( v1 string ) @@ -1470,7 +1470,7 @@ func PanickedWithFallback() (string, error) { } if recovered != nil { taskEmitter.TaskPanicRecovered(ctx, recovered) - v1, err = _176_21, nil + v1, err = _177_21, nil } }() @@ -1480,11 +1480,11 @@ func PanickedWithFallback() (string, error) { defer task10.ran.Store(true) - v1, err = _167_4(ctx) + v1, err = _168_4(ctx) if err != nil { taskEmitter.TaskErrorRecovered(ctx, err) - v1, err = _176_21, nil + v1, err = _177_21, nil } else { taskEmitter.TaskSuccess(ctx) } @@ -1505,10 +1505,341 @@ func PanickedWithFallback() (string, error) { return err } - *(_165_15) = v1 // string + *(_166_15) = v1 // string flowEmitter.FlowSuccess(ctx) return nil }() return s, err } + +// BlockingInputs builds a flow that probes whether cff.Predicate +// short-circuits the scheduler's wait on the predicated task's input +// dependencies. +// +// Shape: a slow source feeds the predicated task; a fast source feeds +// the predicate. The predicate returns false (so the task body is +// skipped via FallbackWith). A downstream consumer measures the +// elapsed wall-clock time from flow start. +// +// If predicates short-circuit input deps: +// +// elapsed ≈ 0 (slow path is not on the critical path) +// +// If predicates only short-circuit the task body: +// +// elapsed ≈ slowDelay (scheduler still waits on slowOut) +func BlockingInputs(slowDelay time.Duration) (time.Duration, error) { + type slowOut struct{} + type fastOut struct{} + type skippedOut struct{} + + var elapsed time.Duration + start := time.Now() + err := func() (err error) { + + _204_3 := context.Background() + + _205_15 := &elapsed + + _206_12 := func() slowOut { + time.Sleep(slowDelay) + return slowOut{} + } + + _210_12 := func() fastOut { return fastOut{} } + + _212_4 := func(slowOut) (skippedOut, error) { return skippedOut{}, nil } + + _213_18 := func(fastOut) bool { return false } + + _214_21 := skippedOut{} + + _216_12 := func(skippedOut) time.Duration { + return time.Since(start) + } + ctx := _204_3 + emitter := cff.NopEmitter() + + var ( + flowInfo = &cff.FlowInfo{ + File: "go.uber.org/cff/internal/tests/predicate/predicate.go", + Line: 203, + Column: 9, + } + flowEmitter = cff.NopFlowEmitter() + + schedInfo = &cff.SchedulerInfo{ + Name: flowInfo.Name, + Directive: cff.FlowDirective, + File: flowInfo.File, + Line: flowInfo.Line, + Column: flowInfo.Column, + } + + // possibly unused + _ = flowInfo + ) + + startTime := time.Now() + defer func() { flowEmitter.FlowDone(ctx, time.Since(startTime)) }() + + schedEmitter := emitter.SchedulerInit(schedInfo) + + sched := cff.NewScheduler( + cff.SchedulerParams{ + Emitter: schedEmitter, + }, + ) + + var tasks []*struct { + emitter cff.TaskEmitter + ran cff.AtomicBool + run func(context.Context) error + job *cff.ScheduledJob + } + defer func() { + for _, t := range tasks { + if !t.ran.Load() { + t.emitter.TaskSkipped(ctx, err) + } + } + }() + + // go.uber.org/cff/internal/tests/predicate/predicate.go:206:12 + var ( + v8 slowOut + ) + task11 := new(struct { + emitter cff.TaskEmitter + ran cff.AtomicBool + run func(context.Context) error + job *cff.ScheduledJob + }) + task11.emitter = cff.NopTaskEmitter() + task11.run = func(ctx context.Context) (err error) { + taskEmitter := task11.emitter + startTime := time.Now() + defer func() { + if task11.ran.Load() { + taskEmitter.TaskDone(ctx, time.Since(startTime)) + } + }() + + defer func() { + recovered := recover() + if recovered != nil { + taskEmitter.TaskPanic(ctx, recovered) + err = &cff.PanicError{ + Value: recovered, + Stacktrace: debug.Stack(), + } + } + }() + + defer task11.ran.Store(true) + + v8 = _206_12() + + taskEmitter.TaskSuccess(ctx) + + return + } + + task11.job = sched.Enqueue(ctx, cff.Job{ + Run: task11.run, + }) + tasks = append(tasks, task11) + + // go.uber.org/cff/internal/tests/predicate/predicate.go:210:12 + var ( + v9 fastOut + ) + task12 := new(struct { + emitter cff.TaskEmitter + ran cff.AtomicBool + run func(context.Context) error + job *cff.ScheduledJob + }) + task12.emitter = cff.NopTaskEmitter() + task12.run = func(ctx context.Context) (err error) { + taskEmitter := task12.emitter + startTime := time.Now() + defer func() { + if task12.ran.Load() { + taskEmitter.TaskDone(ctx, time.Since(startTime)) + } + }() + + defer func() { + recovered := recover() + if recovered != nil { + taskEmitter.TaskPanic(ctx, recovered) + err = &cff.PanicError{ + Value: recovered, + Stacktrace: debug.Stack(), + } + } + }() + + defer task12.ran.Store(true) + + v9 = _210_12() + + taskEmitter.TaskSuccess(ctx) + + return + } + + task12.job = sched.Enqueue(ctx, cff.Job{ + Run: task12.run, + }) + tasks = append(tasks, task12) + + // go.uber.org/cff/internal/tests/predicate/predicate.go:213:4 + var p0 bool + var p0PanicRecover interface{} + var p0PanicStacktrace []byte + _ = p0PanicStacktrace // possibly unused. + pred1 := new(struct { + ran cff.AtomicBool + run func(context.Context) (bool, error) + job *cff.ScheduledJob + }) + pred1.run = func(ctx context.Context) (result bool, err error) { + defer func() { + if recovered := recover(); recovered != nil { + p0PanicRecover = recovered + p0PanicStacktrace = debug.Stack() + } + }() + p0 = _213_18(v9) + return p0, nil + } + + pred1.job = sched.EnqueuePredicate(ctx, cff.PredicateJob{ + Run: pred1.run, + Dependencies: []*cff.ScheduledJob{ + task12.job, + }, + }) + + // go.uber.org/cff/internal/tests/predicate/predicate.go:212:4 + var ( + v10 skippedOut + ) + task13 := new(struct { + emitter cff.TaskEmitter + ran cff.AtomicBool + run func(context.Context) error + job *cff.ScheduledJob + }) + task13.emitter = cff.NopTaskEmitter() + task13.run = func(ctx context.Context) (err error) { + taskEmitter := task13.emitter + startTime := time.Now() + defer func() { + if task13.ran.Load() { + taskEmitter.TaskDone(ctx, time.Since(startTime)) + } + }() + + defer func() { + recovered := recover() + + if recovered == nil && p0PanicRecover != nil { + recovered = p0PanicRecover + } + if recovered != nil { + taskEmitter.TaskPanicRecovered(ctx, recovered) + v10, err = _214_21, nil + } + }() + + if !p0 { + return nil + } + + defer task13.ran.Store(true) + + v10, err = _212_4(v8) + + if err != nil { + taskEmitter.TaskErrorRecovered(ctx, err) + v10, err = _214_21, nil + } else { + taskEmitter.TaskSuccess(ctx) + } + + return + } + + task13.job = sched.Enqueue(ctx, cff.Job{ + Run: task13.run, + Dependencies: []*cff.ScheduledJob{ + task11.job, + pred1.job, + }, + }) + tasks = append(tasks, task13) + + // go.uber.org/cff/internal/tests/predicate/predicate.go:216:12 + var ( + v11 time.Duration + ) + task14 := new(struct { + emitter cff.TaskEmitter + ran cff.AtomicBool + run func(context.Context) error + job *cff.ScheduledJob + }) + task14.emitter = cff.NopTaskEmitter() + task14.run = func(ctx context.Context) (err error) { + taskEmitter := task14.emitter + startTime := time.Now() + defer func() { + if task14.ran.Load() { + taskEmitter.TaskDone(ctx, time.Since(startTime)) + } + }() + + defer func() { + recovered := recover() + if recovered != nil { + taskEmitter.TaskPanic(ctx, recovered) + err = &cff.PanicError{ + Value: recovered, + Stacktrace: debug.Stack(), + } + } + }() + + defer task14.ran.Store(true) + + v11 = _216_12(v10) + + taskEmitter.TaskSuccess(ctx) + + return + } + + task14.job = sched.Enqueue(ctx, cff.Job{ + Run: task14.run, + Dependencies: []*cff.ScheduledJob{ + task13.job, + }, + }) + tasks = append(tasks, task14) + + if err := sched.Wait(ctx); err != nil { + flowEmitter.FlowError(ctx, err) + return err + } + + *(_205_15) = v11 // time.Duration + + flowEmitter.FlowSuccess(ctx) + return nil + }() + return elapsed, err +} diff --git a/internal/tests/predicate/predicate_test.go b/internal/tests/predicate/predicate_test.go index 7510b37..7317e83 100644 --- a/internal/tests/predicate/predicate_test.go +++ b/internal/tests/predicate/predicate_test.go @@ -2,6 +2,7 @@ package predicate import ( "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -75,3 +76,11 @@ func TestPanicFallback(t *testing.T) { assert.NoError(t, err) assert.Equal(t, s, "predicate-fallback") } + +func TestBlockingInputs_PredicateShortCircuits(t *testing.T) { + slowDelay := 100 * time.Millisecond + elapsed, err := BlockingInputs(slowDelay) + require.NoError(t, err) + assert.Less(t, elapsed, slowDelay/10, + "predicate-false flow should not wait on slow input deps; got elapsed=%s, slowDelay=%s", elapsed, slowDelay) +} diff --git a/internal/tests/shadowedvar/shadowedvar_gen.go b/internal/tests/shadowedvar/shadowedvar_gen.go index 236bbfc..67da97a 100644 --- a/internal/tests/shadowedvar/shadowedvar_gen.go +++ b/internal/tests/shadowedvar/shadowedvar_gen.go @@ -587,10 +587,10 @@ func PredicateCtxConflict(f func(), ctx bool) error { _ = p0PanicStacktrace // possibly unused. pred1 := new(struct { ran cff2.AtomicBool - run func(context.Context) error + run func(context.Context) (bool, error) job *cff2.ScheduledJob }) - pred1.run = func(ctx context.Context) (err error) { + pred1.run = func(ctx context.Context) (result bool, err error) { defer func() { if recovered := recover(); recovered != nil { p0PanicRecover = recovered @@ -598,10 +598,10 @@ func PredicateCtxConflict(f func(), ctx bool) error { } }() p0 = _92_19() - return nil + return p0, nil } - pred1.job = sched.Enqueue(ctx, cff2.Job{ + pred1.job = sched.EnqueuePredicate(ctx, cff2.PredicateJob{ Run: pred1.run, }) diff --git a/scheduler.go b/scheduler.go index 09f3e7b..0b2319e 100644 --- a/scheduler.go +++ b/scheduler.go @@ -16,6 +16,16 @@ import ( // This can change without warning. type Job = scheduler.Job +// PredicateJob is a predicate job prepared to be enqueued to the cff +// scheduler. Predicates returning false trigger early-dispatch of their +// consumers, so the consumer's wrapper can fire its skip gate without +// waiting for unrelated slow data deps. +// +// This is intended to be used by cff's generated code. +// Do not use directly. +// This can change without warning. +type PredicateJob = scheduler.PredicateJob + // AtomicBool is a type-safe means of reading and writing boolean values. // // This is intended to be used by cff's generated code. diff --git a/scheduler/scheduler.go b/scheduler/scheduler.go index a47d23d..f4eb1fb 100644 --- a/scheduler/scheduler.go +++ b/scheduler/scheduler.go @@ -112,6 +112,10 @@ const ( type jobResult struct { Job *ScheduledJob // job that was executed Err error // failure, if any + + // PredicateResult carries the bool returned by a PredicateJob's Run. + // Only meaningful when Job.predicateRun != nil and Err == nil. + PredicateResult bool } // errJobInvalid is a sentinel error to capture jobs that are invalidated @@ -148,6 +152,11 @@ func worker(readyc <-chan *ScheduledJob, donec chan<- jobResult) { } else if j.invalid { // Don't run if marked as invalid. res.Err = errJobInvalid + } else if j.predicateRun != nil { + // Predicate jobs return a bool alongside the error. + // The scheduler reads PredicateResult in the done branch + // to decide whether to early-dispatch consumers. + res.PredicateResult, res.Err = j.predicateRun(j.ctx) } else { res.Err = j.run(j.ctx) } @@ -282,6 +291,31 @@ type Job struct { Dependencies []*ScheduledJob } +// PredicateJob is a job that evaluates a boolean condition and returns +// it to the scheduler so consumers can be early-dispatched when the +// predicate resolves false. +// +// Use PredicateJob (via Scheduler.EnqueuePredicate) for the predicate of +// a gated task. When the predicate's Run returns (false, nil), the +// scheduler immediately dispatches all jobs that depend on the predicate, +// without waiting for any other unrelated data dependencies they have. +// This lets a predicated consumer skip the wait on a slow producer when +// the predicate already says the consumer's body will be skipped. +// +// When the predicate returns (true, nil), behavior is identical to a +// regular Job completing successfully. When the predicate returns a +// non-nil error, it is treated as a regular job failure. +type PredicateJob struct { + // Run evaluates the predicate. The bool is the predicate's outcome; + // the error reports an evaluation failure (e.g., a recovered panic + // converted by the caller). + Run func(context.Context) (bool, error) + + // Dependencies are previously enqueued jobs that must run before this + // predicate. + Dependencies []*ScheduledJob +} + // ScheduledJob is a job that has been scheduled for execution by the // scheduler. type ScheduledJob struct { @@ -292,6 +326,12 @@ type ScheduledJob struct { run func(context.Context) error deps []*ScheduledJob + // predicateRun is set instead of run when this ScheduledJob was + // enqueued via Scheduler.EnqueuePredicate. Non-nil tags this entry + // as a predicate. The worker dispatches predicateRun instead of run + // and captures the bool result in jobResult.PredicateResult. + predicateRun func(context.Context) (bool, error) + // The following fields track the internal state of the job. These are // read-write, but only within Scheduler.run. DO NOT read or write // them outside scheduler.run, as that will introduce a data race. @@ -302,6 +342,12 @@ type ScheduledJob struct { err error // the job error, if encountered when the job ran invalid bool // whether the job is marked invalid and should not run + // predicateResult caches the bool returned by a predicate's Run. + // Only meaningful when predicateRun != nil and done && err == nil. + // Used by the enqueue handler so consumers enqueued AFTER the + // predicate completed can still trigger fast-dispatch. + predicateResult bool + // NOTE: DO NOT add methods to ScheduledJob. There's danger of using // methods that read or write internal state outside the Scheduler.run // function which, as discussed above, introduces a data race. @@ -326,6 +372,24 @@ func (s *Scheduler) Enqueue(ctx context.Context, j Job) *ScheduledJob { return pj } +// EnqueuePredicate queues up a predicate job for execution with the +// scheduler. The returned object may be used as a dependency for other +// jobs. When the predicate's Run returns (false, nil), the scheduler +// will dispatch every job that depends on this predicate immediately, +// even if those jobs still have unfinished data dependencies. +// +// EnqueuePredicate will panic if called after calling Wait. +func (s *Scheduler) EnqueuePredicate(ctx context.Context, p PredicateJob) *ScheduledJob { + // Same constraints as Enqueue: do not access internal state here. + pj := &ScheduledJob{ + ctx: ctx, + predicateRun: p.Run, + deps: p.Dependencies, + } + s.enqueuec <- pj // panics if closed + return pj +} + // run implements the Scheduler Loop. The Scheduler Loop works by maintaining // the ready list, which contains jobs ready to be run, with no outstanding dependencies. // @@ -426,12 +490,18 @@ func (s *Scheduler) run(emitter Emitter, freq time.Duration) { // unless they've already been run. // // If a dependency has already run and errored, mark the job as - // invalid. + // invalid. If a dependency was a predicate that resolved + // false, mark this job for fast-dispatch so we don't wait on + // any other unrelated data deps either. + fastDispatch := false for _, dep := range job.deps { if dep.done { if dep.err != nil { job.invalid = true } + if dep.predicateRun != nil && dep.err == nil && !dep.predicateResult { + fastDispatch = true + } continue } dep.consumers = append(dep.consumers, job) @@ -440,8 +510,14 @@ func (s *Scheduler) run(emitter Emitter, freq time.Duration) { pending++ - // No outstanding dependencies. Ready to run. - if job.remaining == 0 { + // No outstanding dependencies (or fast-dispatch triggered). + // Ready to run. + if fastDispatch { + // Bypass any remaining-dep waits. Consumers' wrappers + // handle the skip via their own gate. + job.remaining = 0 + ready.PushBack(job) + } else if job.remaining == 0 { ready.PushBack(job) } else { waiting++ @@ -451,6 +527,12 @@ func (s *Scheduler) run(emitter Emitter, freq time.Duration) { job := res.Job job.done = true + // Cache the predicate's bool so late-enqueued consumers + // can read it in the enqueue handler below. + if job.predicateRun != nil { + job.predicateResult = res.PredicateResult + } + pending-- ongoing-- @@ -473,6 +555,36 @@ func (s *Scheduler) run(emitter Emitter, freq time.Duration) { } } + // Predicate that returned (false, nil): early-dispatch its + // consumers. They were declared to wait on this predicate + // plus (possibly) unrelated data deps; bypass the data-dep + // wait so the consumer's own wrapper can fire its skip gate + // without sitting through a slow producer. + // + // The data-dep edges still exist: when the producer eventually + // completes, the regular consumer-notification loop below will + // decrement consumer.remaining past zero (no re-push, since + // the equality check `if remaining == 0` no longer matches). + // + // Defensive: skip consumers already done or already in ready. + // Today's cff codegen emits at most one predicate per task, + // so this can't be triggered from generated code — but the + // scheduler API permits a consumer to depend on multiple + // predicates, and a second predicate completing false would + // otherwise double-push a consumer the first already + // fast-dispatched. + if job.predicateRun != nil && res.Err == nil && !res.PredicateResult { + for _, consumer := range job.consumers { + if consumer.done || consumer.remaining == 0 { + continue + } + consumer.remaining = 0 + waiting-- + ready.PushBack(consumer) + } + continue + } + // Notify jobs waiting on this job, adding them to // ready if this was their last outstanding // dependency. diff --git a/scheduler/scheduler_test.go b/scheduler/scheduler_test.go index b6b361a..a2873e7 100644 --- a/scheduler/scheduler_test.go +++ b/scheduler/scheduler_test.go @@ -764,6 +764,128 @@ func TestPredicateScheduling(t *testing.T) { } } +// TestPredicateEarlyDispatch verifies that when a PredicateJob's Run returns +// (false, nil), the scheduler dispatches the consumer immediately, without +// waiting for the consumer's slow data dependency to complete. +func TestPredicateEarlyDispatch(t *testing.T) { + t.Parallel() + + ctx := context.Background() + sched := Config{Concurrency: 2}.New() + + slowDone := make(chan struct{}) + consumerRan := make(chan struct{}, 1) + + slow := sched.Enqueue(ctx, Job{ + Run: func(context.Context) error { + <-slowDone + return nil + }, + }) + pred := sched.EnqueuePredicate(ctx, PredicateJob{ + Run: func(context.Context) (bool, error) { + return false, nil + }, + }) + sched.Enqueue(ctx, Job{ + Run: func(context.Context) error { + consumerRan <- struct{}{} + return nil + }, + Dependencies: []*ScheduledJob{slow, pred}, + }) + + waitErr := make(chan error, 1) + go func() { waitErr <- sched.Wait(ctx) }() + + select { + case <-consumerRan: + // success: consumer dispatched without slow finishing + case <-time.After(time.Second): + t.Fatal("consumer did not run early; still waiting on slow dep") + } + + close(slowDone) + assert.NoError(t, <-waitErr) +} + +// TestPredicateTrueWaitsForDeps verifies that when a PredicateJob returns +// (true, nil), the consumer follows the normal completion path: it waits +// for all of its data deps before running. +func TestPredicateTrueWaitsForDeps(t *testing.T) { + t.Parallel() + + ctx := context.Background() + sched := Config{Concurrency: 2}.New() + + var slowDone atomic.Bool + slowBlocker := make(chan struct{}) + + slow := sched.Enqueue(ctx, Job{ + Run: func(context.Context) error { + <-slowBlocker + slowDone.Store(true) + return nil + }, + }) + pred := sched.EnqueuePredicate(ctx, PredicateJob{ + Run: func(context.Context) (bool, error) { + // Predicate returns true: consumer must NOT be dispatched + // until the slow dep also completes. + return true, nil + }, + }) + sched.Enqueue(ctx, Job{ + Run: func(context.Context) error { + assert.True(t, slowDone.Load(), + "consumer ran before slow dep when predicate returned true") + return nil + }, + Dependencies: []*ScheduledJob{slow, pred}, + }) + + waitErr := make(chan error, 1) + go func() { waitErr <- sched.Wait(ctx) }() + + // Give the predicate time to complete and (incorrectly, if regressed) + // dispatch the consumer. Then unblock the slow dep. + time.Sleep(50 * time.Millisecond) + close(slowBlocker) + + assert.NoError(t, <-waitErr) +} + +// TestPredicateError verifies that when a PredicateJob returns a non-nil +// error, the scheduler treats it as a normal job failure: in default mode, +// Wait returns the error; the consumer is not early-dispatched. +func TestPredicateError(t *testing.T) { + t.Parallel() + + ctx := context.Background() + sched := Config{Concurrency: 2}.New() + + predErr := errors.New("predicate boom") + + pred := sched.EnqueuePredicate(ctx, PredicateJob{ + Run: func(context.Context) (bool, error) { + return false, predErr + }, + }) + + consumerRan := false + sched.Enqueue(ctx, Job{ + Run: func(context.Context) error { + consumerRan = true + return nil + }, + Dependencies: []*ScheduledJob{pred}, + }) + + err := sched.Wait(ctx) + assert.ErrorIs(t, err, predErr) + assert.False(t, consumerRan, "consumer should not run when predicate errored") +} + func TestWorkerTermination(t *testing.T) { t.Parallel()