gpu: tier selection — one setting, three values, and off means off

cmd/gpu-cuda-profile/main.go

@@ -17,6 +17,7 @@ import (
 	"os"
 	"os/exec"
 	"path/filepath"
+	"strings"
 	"time"
 
 	"github.com/dpsoft/perf-agent/gpu"
@@ -35,9 +36,48 @@ func main() {
 		period   = flag.Int("period", 8, "one-in-N launch sampling period (PERFAGENT_GPU_SAMPLE_PERIOD)")
 		linger   = flag.Int("linger-ms", 30000, "how long the workload may wait to be released after it finishes")
 		out      = flag.String("out", "gpu-cuda.pb.gz", "output pprof profile")
+
+		// One setting, three values, and no way to ask for two. The default
+		// is the empty string rather than "off" so that an unspecified flag
+		// DEFERS to an inherited PERFAGENT_GPU_PC_SAMPLING instead of
+		// contradicting it — an explicit --gpu-pc-sampling=off against an
+		// exported "serialized" is a disagreement and is refused, but not
+		// setting the flag at all is not.
+		pcSampling = flag.String("gpu-pc-sampling", "",
+			"GPU PC-sampling tier: "+strings.Join(gpu.PCSamplingTierNames, " | ")+
+				" (default off; also read from "+gpu.PCSamplingEnvVar+"). "+
+				"\"continuous\" does not serialize kernels; \"serialized\" does, and requires "+
+				"-gpu-pc-sampling-acknowledge-perturbation")
+		pcAck = flag.Bool("gpu-pc-sampling-acknowledge-perturbation", false,
+			"acknowledge that the \"serialized\" tier perturbs the workload: it inflates GPU "+
+				"kernel durations inside a burst, it distorts any CPU and off-CPU profile taken "+
+				"alongside it with no marking in those profiles at all, and it is unavailable "+
+				"where CUDA graphs are in use")
 	)
 	flag.Parse()
 
+	// Tier selection, and it happens BEFORE anything is attached or launched.
+	// Every refusal here is a startup error: an unknown value, a value naming
+	// two tiers, the flag and the environment naming two tiers, or Tier A
+	// without its acknowledgement. None of them is resolved to a tier — a
+	// profile produced under a tier nobody chose is worse than no profile,
+	// because nothing in it says which one ran.
+	tier, err := gpu.PCSamplingRequest{
+		Flag:                    *pcSampling,
+		Env:                     os.Getenv(gpu.PCSamplingEnvVar),
+		AcknowledgePerturbation: *pcAck,
+	}.Select()
+	if err != nil {
+		log.Fatalf("gpu pc sampling: %v", err)
+	}
+	// Printed at startup as well as standing in every JoinHealth render
+	// below. The startup copy is for the operator who is watching the run
+	// begin; the standing copy is for the one who reads the profile an hour
+	// later, which is the reader the warning is actually for.
+	for _, line := range gpu.PCSamplingStandingWarning(tier) {
+		log.Print(line)
+	}
+
 	shimPath, err := filepath.Abs(*shim)
 	if err != nil {
 		log.Fatalf("shim path: %v", err)
@@ -50,7 +90,11 @@ func main() {
 		log.Fatalf("adapter %s: %v (build it with: make -C shim nvidia)", shimPath, err)
 	}
 
-	timeline := gpu.NewTimeline(gpu.TimelineConfig{})
+	// The selected tier reaches the agent's own join here and the producer's
+	// environment below, from ONE variable. Two copies that could disagree
+	// about which tier ran is how a profile ends up disclosing one thing and
+	// doing another.
+	timeline := gpu.NewTimeline(gpu.TimelineConfig{PCSampling: tier})
 	// Without a symbolizer the sampled launch stacks still arrive and are
 	// still accounted for, but every one of them degrades to no stack — the
 	// profile would then be honest and useless, all GPU time unattributed.
@@ -113,6 +157,12 @@ func main() {
 		"CUDA_INJECTION64_PATH="+shimPath,
 		fmt.Sprintf("PERFAGENT_GPU_SAMPLE_PERIOD=%d", *period),
 		"PERFAGENT_GPU_LOG=stderr",
+		// Set EXPLICITLY on every run including an off one, never left to be
+		// inherited. os.Environ() may already carry this variable from the
+		// operator's shell; appending the resolved value last is what keeps a
+		// stale export from turning a run this agent believes is off into a
+		// producer that serializes the workload's kernels.
+		gpu.PCSamplingEnvVar+"="+tier.EnvValue(),
 	)
 	cmd.Stdout, cmd.Stderr = os.Stdout, os.Stderr
 	// Same release protocol as the stub: the workload's CPU stacks are

gpu/conformance_test.go

@@ -307,7 +307,7 @@ func assertPCAttribAccompaniesSamples(t *testing.T, snap Snapshot) {
 // three outcomes, so they sum to len(snap.Executions).
 //
 // It also asserts the negative that matters more than the sum: with the
-// default harness configuration (SerializedSampling unset — nothing is ever
+// default harness configuration (PCSampling off — nothing is ever
 // serialized) NO execution may read "true" or "unknown". A conformance run
 // that started reporting perturbation nobody caused would be as wrong as one
 // that stopped reporting perturbation that happened.

gpu/joinhealth.go

@@ -22,9 +22,18 @@ func plural(n uint64, one, many string) string {
 }
 
 // JoinHealth renders a Snapshot's join and loss counters as operator-facing
-// lines: element 0 is always a one-line summary, and every element after it
-// is one anomaly that the summary's trailing count agrees with. Callers log
-// them one per line (see cmd/gpu-stub-profile, cmd/gpu-cuda-profile).
+// lines: element 0 is always a one-line summary, then any STANDING WARNINGS
+// (see PCSamplingStandingWarning — Tier A's whole-run perturbation notice),
+// then one line per anomaly. The summary's trailing counts agree with both
+// groups, so len(lines)-1 is always warnings+anomalies. Callers log them one
+// per line (see cmd/gpu-stub-profile, cmd/gpu-cuda-profile).
+//
+// A standing warning is not an anomaly and is kept apart from them on
+// purpose: an anomaly is something that went wrong, while a warning is a
+// consequence of what the operator deliberately asked for. Both are printed
+// on every render — the warning especially, because a perturbation notice
+// shown once at startup has scrolled away long before the profile it applies
+// to is read.
 //
 // The shape is deliberate. Printing the whole counter set on every run - the
 // obvious `%+v` - is what makes a rising UnmatchedExecutionCount invisible:
@@ -75,8 +84,18 @@ func JoinHealth(snap Snapshot) []string {
 // ProjectExecutionsWith reports no suppression - correctly, since without that
 // call nothing suppressed anything.
 func JoinHealthWith(snap Snapshot, proj ProjectionStats) []string {
+	// Warnings BEFORE anomalies, for the same reason the serialization
+	// anomaly is raised before the join ones: they qualify the numbers
+	// themselves rather than what those numbers were attributed to. A
+	// perturbed measurement joined perfectly is still a perturbed
+	// measurement, and a reader who stops after the first two lines must have
+	// been told that.
+	warnings := PCSamplingStandingWarning(snap.PCSampling)
 	anomalies := joinAnomalies(snap, proj)
-	return append([]string{joinSummary(snap, len(anomalies))}, anomalies...)
+	out := make([]string, 0, 1+len(warnings)+len(anomalies))
+	out = append(out, joinSummary(snap, len(warnings), len(anomalies)))
+	out = append(out, warnings...)
+	return append(out, anomalies...)
 }
 
 // joinSummary is the always-printed line. len(snap.Executions), not a
@@ -85,7 +104,7 @@ func JoinHealthWith(snap Snapshot, proj ProjectionStats) []string {
 // parenthesised breakdown checkable against a figure that does not come
 // from the same counters it is auditing. joinAnomalies performs that check
 // rather than leaving it to the reader's arithmetic.
-func joinSummary(snap Snapshot, anomalies int) string {
+func joinSummary(snap Snapshot, warnings, anomalies int) string {
 	js := snap.JoinStats
 	execs := uint64(len(snap.Executions))
 
@@ -147,6 +166,15 @@ func joinSummary(snap Snapshot, anomalies int) string {
 			plural(uint64(snap.PendingModuleGroups), "kernel group", "kernel groups"))
 	}
 
+	// Which tier this run selected, and only when one was. "off" is the
+	// default and printing it on every run is the zero-valued noise this
+	// format exists to avoid; "continuous" and "serialized" are both facts a
+	// reader needs before they read anything else, because they decide what a
+	// PC sample can be attributed to and whether the durations are perturbed.
+	if snap.PCSampling != PCSamplingOff {
+		fmt.Fprintf(&b, "; pc sampling %s", snap.PCSampling)
+	}
+
 	// The serialization disclosure, and only when there is one to make. With
 	// PC sampling off or in continuous collection every execution is "false"
 	// and nothing was ever serialized, so a permanent "0 serialized" clause
@@ -159,6 +187,20 @@ func joinSummary(snap Snapshot, anomalies int) string {
 			plural(uint64(snap.SamplingWindowsHeld), "burst", "bursts"))
 	}
 
+	// Counted, and counted in LINES, so that the trailing figures still add
+	// up to exactly what follows the summary: len(lines)-1 == warnings +
+	// anomalies. A standing warning is not an anomaly — nothing went wrong,
+	// the operator asked for it — but it is not free either, and a summary
+	// that said "no anomalies" while four lines of perturbation warning
+	// followed it would be the reassuring half of a contradiction.
+	switch warnings {
+	case 0:
+	case 1:
+		b.WriteString("; 1 standing warning line")
+	default:
+		fmt.Fprintf(&b, "; %d standing warning lines", warnings)
+	}
+
 	switch anomalies {
 	case 0:
 		b.WriteString("; no anomalies")
@@ -231,12 +273,23 @@ func joinAnomalies(snap Snapshot, proj ProjectionStats) []string {
 			"distorted too and carry no marking at all",
 			snap.ExecutionsSerialized, execs)
 	}
-	if snap.ExecutionsSerializationUnknown > 0 && snap.SamplingWindowsReceived > 0 {
+	// No `&& SamplingWindowsReceived > 0` guard. Unknown executions are
+	// reachable only under Tier A (the other two tiers answer "false"
+	// unconditionally and never consult the store), and the case the guard
+	// would suppress — Tier A selected, not one window record arrived, so
+	// EVERY execution is "unknown" — is the worst one available, not the
+	// uninteresting one. The clause names whether any window arrived instead
+	// of hiding the line when none did.
+	if snap.ExecutionsSerializationUnknown > 0 {
+		cause := "a dropped batch, a late attach, a sequence gap, or a burst that never closed"
+		if snap.SamplingWindowsReceived == 0 {
+			cause = "NOT ONE window record reached the agent, though Tier A was selected — the " +
+				"producer never bursted, the probe never attached, or every batch was lost"
+		}
 		add("%d of %d executions cannot be said to have run unperturbed — no sampling window "+
-			"covers them (a dropped batch, a late attach, a sequence gap, or a burst that "+
-			"never closed). They are marked gpu_serialized=\"unknown\" and MUST NOT be read "+
+			"covers them (%s). They are marked gpu_serialized=\"unknown\" and MUST NOT be read "+
 			"as \"false\"",
-			snap.ExecutionsSerializationUnknown, execs)
+			snap.ExecutionsSerializationUnknown, execs, cause)
 	}
 	if snap.SamplingWindowsOpen > 0 {
 		add("%s still open — the producer stopped reporting mid-burst (a hard exit), so the "+

gpu/joinhealth_test.go

@@ -67,6 +67,14 @@ func anomalousSnapshot() Snapshot {
 		// Tier A gone wrong in all three ways at once: bursts perturbed some
 		// executions, an unbroken history proved others were untouched, and a
 		// window that never closed leaves the rest unplaceable.
+		//
+		// The tier is set, and it has to be: the three counters below are
+		// reachable ONLY under PCSamplingSerialized (the other two tiers
+		// answer "false" unconditionally and never consult the window store),
+		// so a fixture that carried them with the tier off would be a run that
+		// cannot happen — and would quietly stop exercising the standing
+		// warning that a real one carries.
+		PCSampling:                     PCSamplingSerialized,
 		SamplingWindowsReceived:        41,
 		SamplingWindowsHeld:            21,
 		SamplingWindowsOpen:            1,
@@ -89,10 +97,17 @@ func TestJoinHealthAnomaliesEachGetTheirOwnLine(t *testing.T) {
 	lines := JoinHealth(anomalousSnapshot())
 
 	require.Greater(t, len(lines), 1)
+	warnings := 0
 	for _, l := range lines[1:] {
+		if strings.HasPrefix(l, PCSamplingWarningPrefix+": ") {
+			warnings++
+			continue
+		}
 		assert.True(t, strings.HasPrefix(l, joinAnomalyPrefix+": "), "line %q", l)
 		assert.Contains(t, l, " — ", "every anomaly says what the number means when it is bad")
 	}
+	assert.Equal(t, len(PCSamplingStandingWarning(PCSamplingSerialized)), warnings,
+		"a Tier A snapshot carries its whole standing warning, every render")
 
 	joined := strings.Join(lines, "\n")
 	for _, want := range []string{
@@ -118,28 +133,107 @@ func TestJoinHealthAnomaliesEachGetTheirOwnLine(t *testing.T) {
 	}
 }
 
-// The summary's anomaly count is the one derived figure here; it must never
-// read green when things are worst.
+// The summary's trailing counts are the only derived figures here; they must
+// never read green when things are worst, and together they must account for
+// EVERY line below the summary. A standing warning line that no count covered
+// would be a line the summary implicitly denies exists.
 func TestJoinHealthSummaryCountMatchesTheLinesBelowIt(t *testing.T) {
+	tierA := healthySnapshot()
+	tierA.PCSampling = PCSamplingSerialized
 	for name, snap := range map[string]Snapshot{
-		"healthy":   healthySnapshot(),
-		"anomalous": anomalousSnapshot(),
-		"empty":     {},
+		"healthy":          healthySnapshot(),
+		"anomalous":        anomalousSnapshot(),
+		"empty":            {},
+		"tier A, no fault": tierA,
 	} {
 		t.Run(name, func(t *testing.T) {
 			lines := JoinHealth(snap)
-			switch n := len(lines) - 1; n {
+
+			warnings := 0
+			for _, l := range lines[1:] {
+				if strings.HasPrefix(l, PCSamplingWarningPrefix+": ") {
+					warnings++
+				}
+			}
+			anomalies := len(lines) - 1 - warnings
+
+			switch warnings {
+			case 0:
+				assert.NotContains(t, lines[0], "standing warning")
+			case 1:
+				assert.Contains(t, lines[0], "; 1 standing warning line")
+			default:
+				assert.Contains(t, lines[0], "; "+strconv.Itoa(warnings)+" standing warning lines")
+			}
+
+			switch anomalies {
 			case 0:
 				assert.Contains(t, lines[0], "; no anomalies")
 			case 1:
 				assert.Contains(t, lines[0], "; 1 anomaly")
 			default:
-				assert.Contains(t, lines[0], "; "+strconv.Itoa(n)+" anomalies")
+				assert.Contains(t, lines[0], "; "+strconv.Itoa(anomalies)+" anomalies")
 			}
 		})
 	}
 }
 
+// The warning STANDS. It is rendered on a Tier A run in which nothing at all
+// went wrong — no perturbed execution yet, no unknown, no window even — and
+// that is the case it exists for: a burst-free interval, a graph refusal that
+// stopped bursts, or simply the first snapshot of a run. A disclosure that
+// appeared only once some counter moved would be absent from exactly the
+// profiles whose readers had no other way to learn the tier was on.
+func TestTheTierAWarningStandsOnAnOtherwisePerfectRun(t *testing.T) {
+	snap := healthySnapshot()
+	snap.PCSampling = PCSamplingSerialized
+	// Nothing is amiss: every join exact, no window, no serialized execution.
+	snap.ExecutionsNotSerialized = uint64(len(snap.Executions))
+
+	lines := JoinHealth(snap)
+	joined := strings.Join(lines, "\n")
+	assert.Contains(t, lines[0], "; pc sampling serialized")
+	assert.Contains(t, lines[0], "; no anomalies")
+	assert.Contains(t, joined, "CARRY NO MARKING AT ALL")
+	assert.Contains(t, joined, "CUDA GRAPHS")
+	assert.Equal(t, PCSamplingStandingWarning(PCSamplingSerialized), lines[1:],
+		"the warning is the whole warning, in order, immediately under the summary")
+}
+
+// And it is absent for the two tiers that do not perturb anything. A warning
+// on every run is a warning readers learn to skip.
+func TestNoStandingWarningWhenNothingIsPerturbed(t *testing.T) {
+	for _, tier := range []PCSamplingTier{PCSamplingOff, PCSamplingContinuous} {
+		t.Run(tier.String(), func(t *testing.T) {
+			snap := healthySnapshot()
+			snap.PCSampling = tier
+			lines := JoinHealth(snap)
+			require.Len(t, lines, 1)
+			assert.NotContains(t, lines[0], "standing warning")
+			if tier == PCSamplingOff {
+				assert.NotContains(t, lines[0], "pc sampling")
+			} else {
+				assert.Contains(t, lines[0], "; pc sampling continuous")
+			}
+		})
+	}
+}
+
+// Tier A selected and NOT ONE window record received: every execution is
+// "unknown", and that is the worst available state of the disclosure rather
+// than a quiet one. It must be raised, and the line must say that no window
+// arrived at all rather than offering the ordinary lossy-transport causes.
+func TestJoinHealthRaisesTierAWithNoWindowAtAll(t *testing.T) {
+	snap := healthySnapshot()
+	snap.PCSampling = PCSamplingSerialized
+	snap.ExecutionsSerializationUnknown = uint64(len(snap.Executions))
+	snap.ExecutionsNotSerialized = 0
+
+	joined := strings.Join(JoinHealth(snap), "\n")
+	assert.Contains(t, joined, "NOT ONE window record reached the agent")
+	assert.Contains(t, joined, "MUST NOT be read")
+}
+
 // A snapshot with nothing in it is the degenerate worst case: every ratio a
 // health figure could compute is 0/0. It must read as an anomaly, not as a
 // clean run.
@@ -258,6 +352,12 @@ func TestJoinHealthRenderedOutput(t *testing.T) {
 		{"healthy", healthySnapshot()},
 		{"anomalous", anomalousSnapshot()},
 		{"empty", Snapshot{}},
+		{"tier A, nothing wrong", func() Snapshot {
+			s := healthySnapshot()
+			s.PCSampling = PCSamplingSerialized
+			s.ExecutionsNotSerialized = uint64(len(s.Executions))
+			return s
+		}()},
 	} {
 		t.Log(c.name + ":\n" + strings.Join(JoinHealth(c.snap), "\n"))
 	}

gpu/serialization_test.go

@@ -21,7 +21,7 @@ import (
 const tierAPID = uint32(4242)
 
 func tierATimeline() *Timeline {
-	return NewTimeline(TimelineConfig{SerializedSampling: true})
+	return NewTimeline(TimelineConfig{PCSampling: PCSamplingSerialized})
 }
 
 // serializedExec is an execution from tierAPID over [startNs, endNs]. The PID
@@ -357,7 +357,7 @@ func TestSerializationWindowsNeverCrossProcesses(t *testing.T) {
 // That can only turn "false" into "unknown" — never the reverse — and this
 // asserts the direction rather than trusting the comment.
 func TestSerializationEvictionDegradesTowardsUnknownNeverTowardsFalse(t *testing.T) {
-	tl := NewTimeline(TimelineConfig{SerializedSampling: true, MaxSamplingWindowsPerPID: 4})
+	tl := NewTimeline(TimelineConfig{PCSampling: PCSamplingSerialized, MaxSamplingWindowsPerPID: 4})
 	// The gap between bursts 1 and 2 is provable while both are held.
 	emitBurst(t, tl, uint64(tierAPID), 1000, 2000)
 	emitBurst(t, tl, uint64(tierAPID), 3000, 4000)
@@ -382,7 +382,7 @@ func TestSerializationEvictionDegradesTowardsUnknownNeverTowardsFalse(t *testing
 // A process past the PID bound gets no window history, so its executions read
 // "unknown". It must not inherit somebody else's.
 func TestSerializationRefusedPIDReadsUnknown(t *testing.T) {
-	tl := NewTimeline(TimelineConfig{SerializedSampling: true, MaxSamplingWindowPIDs: 1})
+	tl := NewTimeline(TimelineConfig{PCSampling: PCSamplingSerialized, MaxSamplingWindowPIDs: 1})
 	emitBurst(t, tl, 1, 1000, 2000)
 	emitBurst(t, tl, 2, 1000, 2000) // refused: the store already holds one PID