scheduler: don't wait on input dependencies when cff.Predicate returns false

.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

internal/templates/flow/predicate.go.tmpl

@@ -8,18 +8,18 @@ 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
         p{{ predHash . }}PanicStacktrace = {{ import "runtime/debug" }}.Stack()
 	}
     }()
     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{

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 -}}

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
+}