perf(coro): bound hot scheduler queue audits

doc/coro-performance-baseline.md

@@ -1018,3 +1018,70 @@ panic and recover IR tests, and the linked native ExplicitStatus panic E2E
 completed. The legacy PCLN signal integration command remains blocked before
 its fault subtest by the same unrelated `pclntab_external.go` uintptr-retention
 audit on both this candidate and the exact parent.
+
+### Bounded hot-queue audit checkpoint
+
+The next scheduler checkpoint uses `a3e1a09e2` as its exact parent. Three
+ordinary paths still revisited an entire owner queue: every completed unlocked
+Yield/Park passed the physical-owner handoff gate through full ready and wait
+audits, surplus distribution audited the complete source ready queue before
+selecting at most eight entries, and destination import audited all existing
+local runnables before draining at most eight frozen mailbox entries. A burst
+of N parked goroutines therefore retained quadratic validation work after the
+spawn-owner fix.
+
+The unlocked handoff now validates the O(1) scheduler endpoints plus the exact
+completed continuation. A Yield must be the newly appended ready tail; a Park
+must have an exact locally valid active wait record. Full queue audits remain
+mandatory for the exceptional locked-M handoff. Fleet distribution validates
+only the selected frame chain or a mailbox-bounded prefix, and fleet drain
+validates only the frozen incoming entries while rechecking the destination
+endpoints under its Try gate. Public exact import, physical-owner transitions,
+lifecycle, shutdown, tests, and diagnostics retain complete audits.
+
+Regression fixtures independently corrupt a distant source runnable, a
+distant destination runnable, unrelated active waits, ready tail links, wait
+endpoints, and the ready-count overflow sentinel. They prove that hot paths do
+not walk unrelated payloads while every local ownership header and selected
+continuation still fails closed.
+
+The same standard-Go workload was rebuilt on Darwin arm64 with Go 1.26.5 and
+LLVM 19.1.7. Five process-start `GOMAXPROCS=1` runs of each 10,000-parked-G
+artifact were interleaved with reversed order. Medians were:
+
+| Metric | `a3e1a09e2` parent | bounded-audit candidate | Change |
+| --- | ---: | ---: | ---: |
+| workload time | 1,904.885 ms | 157.035 ms | -91.76% (12.13x faster) |
+| retired instructions | 26.449 billion | 6.297 billion | -76.19% (4.20x fewer) |
+| peak RSS | 32,735,232 | 30,326,784 | -2,408,448 (-7.36%) |
+| stripped file size | 4,875,776 | 4,875,600 | -176 bytes |
+
+Segment reservations for `__TEXT`, `__DATA_CONST`, and `__DATA` are unchanged.
+Seven interleaved short-workload runs also showed no displaced regression:
+spawn 100 x 100 moved from a 56.690 ms median to 56.087 ms, and 5,000
+unbuffered request/ack handoffs moved from 72.557 ms to 71.484 ms.
+
+Five-run peak-RSS medians against the same-source Go build were:
+
+| parked Gs | Go gc RSS | candidate RSS | Go incremental / G | candidate incremental / G |
+| ---: | ---: | ---: | ---: | ---: |
+| 0 | 3,522,560 | 7,716,864 | - | - |
+| 1,000 | 6,471,680 | 9,535,488 | 2,949 B | 1,819 B |
+| 5,000 | 17,596,416 | 18,989,056 | 2,815 B | 2,254 B |
+| 10,000 | 31,539,200 | 30,326,784 | 2,802 B | 2,261 B |
+
+At 10,000 parked goroutines the stackless candidate is 1,212,416 bytes (3.84%)
+below Go in total RSS, not only in slope. Its incremental resident cost is
+about 19.3% lower than Go's; the larger fixed LLGo runtime still makes the
+candidate larger at 5,000, so the observed total-footprint crossover lies
+between those scales. This checkpoint changes no logical frame or G layout;
+the extra RSS reduction comes from avoiding repeated touches of cold queue and
+frame metadata, and should be treated as a working-set result rather than a
+second object-size reduction.
+
+The remaining 5,000-to-10,000 instruction growth is about 2.46x rather than
+the former near-4x queue-scan signature. A larger diagnostic profile contains
+no `validReadyQueue` under normal handoff, distribution, or drain. Its active
+cost is now dominated by channel registration/cleanup, fixed source and fleet
+dispatch, frame allocation/free, and BDWGC marking/locking. Those are the next
+performance gates before broadening capability coverage.

runtime/internal/coro/executor_fleet.go

@@ -576,7 +576,8 @@ func (fleet *ExecutorFleet) DistributePNeutralRunnable(
 ) (distribution RunnableDistribution, ok bool) {
 	sourceSlot, _, sourceOK := executorFleetSlotFor(fleet, sourceHandle)
 	if !sourceOK || preemptLoad(&sourceSlot.state) != uint32(executorFleetSlotActive) ||
-		sourceSlot.p != source || source == nil || !validReadyQueueHeader(source) {
+		sourceSlot.p != source || source == nil || !validReadyQueueHeader(source) ||
+		source.readyCount == ^uint32(0) {
 		return RunnableDistribution{}, false
 	}
 	if source.readyHead == nil {
@@ -585,9 +586,6 @@ func (fleet *ExecutorFleet) DistributePNeutralRunnable(
 	if !stableRunnableTransferP(source) {
 		return RunnableDistribution{}, true
 	}
-	if !validReadyQueue(source) {
-		return RunnableDistribution{}, false
-	}
 	batchLimit := source.readyCount / 2
 	if source.readyCount == 1 {
 		batchLimit = 1
@@ -689,16 +687,21 @@ func (fleet *ExecutorFleet) DistributeMaterializedRunnableToPreferredRoute(
 ) (distribution RunnableDistribution, ok bool) {
 	sourceSlot, _, sourceOK := executorFleetSlotFor(fleet, sourceHandle)
 	if !sourceOK || preemptLoad(&sourceSlot.state) != uint32(executorFleetSlotActive) ||
-		sourceSlot.p != source || source == nil || !validReadyQueueHeader(source) {
+		sourceSlot.p != source || source == nil || !validReadyQueueHeader(source) ||
+		source.readyCount == ^uint32(0) {
 		return RunnableDistribution{}, false
 	}
 	if source.readyHead == nil || !stableRunnableTransferP(source) {
 		return RunnableDistribution{}, true
 	}
-	if !validReadyQueue(source) {
-		return RunnableDistribution{}, false
-	}
 	candidate := source.readyHead
+	// This exact causal route needs only the selected continuation. Validate its
+	// complete frame chain, but do not walk unrelated runnable work. The
+	// prepared publication rechecks its O(1) queue/G header under the target
+	// mailbox gate.
+	if !pNeutralRunnable(candidate, true) {
+		return RunnableDistribution{}, true
+	}
 	preferred, preferredOK := MaterializedRunnablePreferredRoute(candidate)
 	if !preferredOK {
 		return RunnableDistribution{}, true
@@ -717,17 +720,23 @@ func (fleet *ExecutorFleet) DistributeMaterializedRunnableToPreferredRoute(
 		) {
 		return RunnableDistribution{}, true
 	}
-	id, request, published := fleet.PublishPNeutralRunnableAndRequest(
+	var candidates [RunnableTransferMailboxCapacity]*G
+	candidates[0] = candidate
+	id, count, request, published := fleet.publishPreparedPNeutralRunnableBatchAndRequest(
 		target.handle,
 		source,
-		candidate,
+		&candidates,
+		1,
 	)
 	if !published {
 		if !restoreRunnableDemandAfterFailedClaim(target) {
 			return RunnableDistribution{}, false
 		}
 		return RunnableDistribution{}, true
 	}
+	if count != 1 {
+		return RunnableDistribution{}, false
+	}
 	if !preemptCompareAndSwap(
 		&target.runnableDemand,
 		uint32(runnableDemandClaimed),
@@ -738,7 +747,7 @@ func (fleet *ExecutorFleet) DistributeMaterializedRunnableToPreferredRoute(
 	distribution = RunnableDistribution{
 		Target:   target.handle,
 		Transfer: id,
-		Count:    1,
+		Count:    count,
 		Request:  request,
 	}
 	return distribution, distribution.Valid()

runtime/internal/coro/executor_fleet_test.go

@@ -825,6 +825,130 @@ func TestExecutorFleetDemandDistributesBoundedHalfBatch(t *testing.T) {
 	}
 }
 
+func TestExecutorFleetDistributionAuditsOnlyBoundedSelectedPrefix(t *testing.T) {
+	fleet := new(ExecutorFleet)
+	source := bindExecutorFleetManualFixture(t, fleet)
+	target := bindExecutorFleetManualFixture(t, fleet)
+	tasks := make([]*yieldingTestG, 18)
+	for index := range tasks {
+		tasks[index] = newYieldingTestG(t, "fleet-bounded-audit")
+		scratch := new(P)
+		yieldRunnableForTransfer(t, scratch, tasks[index])
+		runnable, ok := NextRunnable(scratch)
+		if !ok || runnable != tasks[index].g || !Enqueue(source.p, tasks[index].g) {
+			t.Fatalf("prepare bounded-audit task %d", index)
+		}
+	}
+	if !fleet.RequestPNeutralRunnable(target.handle, target.p) {
+		t.Fatal("request bounded-audit target")
+	}
+
+	// The owner-maintained endpoint is still a mandatory O(1) gate.
+	source.p.readyTail.nextReady = source.p.readyTail
+	if distribution, ok := fleet.DistributePNeutralRunnable(source.handle, source.p); ok ||
+		distribution != (RunnableDistribution{}) {
+		t.Fatalf("distribution accepted corrupt source header = %+v/%t", distribution, ok)
+	}
+	source.p.readyTail.nextReady = nil
+	readyCount := source.p.readyCount
+	source.p.readyCount = ^uint32(0)
+	if distribution, ok := fleet.DistributePNeutralRunnable(source.handle, source.p); ok ||
+		distribution != (RunnableDistribution{}) {
+		t.Fatalf("distribution accepted overflowing source count = %+v/%t", distribution, ok)
+	}
+	source.p.readyCount = readyCount
+
+	// Corrupt a task beyond the maximum eight-entry transfer prefix. The full
+	// diagnostic catches it, while the hot path validates and moves only the
+	// exact bounded prefix selected for this transaction.
+	distant := tasks[12].g
+	distantState := distant.state
+	distant.state = GDead
+	if validReadyQueue(source.p) {
+		t.Fatal("full diagnostic accepted distant runnable corruption")
+	}
+	distribution, ok := fleet.DistributePNeutralRunnable(source.handle, source.p)
+	if !ok || !distribution.Valid() || distribution.Target != target.handle ||
+		distribution.Count != RunnableTransferMailboxCapacity ||
+		source.p.readyHead != tasks[RunnableTransferMailboxCapacity].g ||
+		source.p.readyCount != uint32(len(tasks))-RunnableTransferMailboxCapacity {
+		t.Fatalf("bounded selected-prefix distribution = %+v/%t count:%d head:%p",
+			distribution, ok, source.p.readyCount, source.p.readyHead)
+	}
+	distant.state = distantState
+	if !validReadyQueue(source.p) {
+		t.Fatal("restored source queue failed full diagnostic")
+	}
+	targetTasks := []*yieldingTestG{
+		newYieldingTestG(t, "fleet-target-first"),
+		newYieldingTestG(t, "fleet-target-middle"),
+		newYieldingTestG(t, "fleet-target-last"),
+	}
+	for index, task := range targetTasks {
+		scratch := new(P)
+		yieldRunnableForTransfer(t, scratch, task)
+		runnable, nextOK := NextRunnable(scratch)
+		if !nextOK || runnable != task.g || !Enqueue(target.p, task.g) {
+			t.Fatalf("prepare target bounded-audit task %d", index)
+		}
+	}
+	targetDistantState := targetTasks[1].g.state
+	targetTasks[1].g.state = GDead
+	if validReadyQueue(target.p) {
+		t.Fatal("full diagnostic accepted distant destination corruption")
+	}
+	moved, more, status := fleet.TryDrainPNeutralRunnables(
+		target.handle,
+		target.p,
+		RunnableTransferMailboxCapacity,
+	)
+	targetTasks[1].g.state = targetDistantState
+	if status != RunnableTransferDrainComplete || more || moved != RunnableTransferMailboxCapacity ||
+		target.p.readyHead != targetTasks[0].g ||
+		target.p.readyCount != uint32(len(targetTasks))+RunnableTransferMailboxCapacity ||
+		!validReadyQueue(target.p) {
+		t.Fatalf("drain bounded selected prefix = (%d,%t,%d)", moved, more, status)
+	}
+}
+
+func TestExecutorFleetPreferredDistributionAuditsOnlyExactCandidate(t *testing.T) {
+	fleet := new(ExecutorFleet)
+	source := bindExecutorFleetManualFixture(t, fleet)
+	target := bindExecutorFleetManualFixture(t, fleet)
+	selected := newYieldingTestG(t, "fleet-preferred-exact")
+	distant := newYieldingTestG(t, "fleet-preferred-distant")
+	enqueueMaterializedFleetRunnable(t, source.p, selected, RouteID(target.handle.Route))
+	scratch := new(P)
+	yieldRunnableForTransfer(t, scratch, distant)
+	if runnable, ok := NextRunnable(scratch); !ok || runnable != distant.g ||
+		!Enqueue(source.p, distant.g) {
+		t.Fatal("prepare preferred distant runnable")
+	}
+	if !fleet.RequestPNeutralRunnable(target.handle, target.p) {
+		t.Fatal("request preferred exact target")
+	}
+	distantState := distant.g.state
+	distant.g.state = GDead
+	if validReadyQueue(source.p) {
+		t.Fatal("full diagnostic accepted preferred distant corruption")
+	}
+	distribution, ok := fleet.DistributeMaterializedRunnableToPreferredRoute(
+		source.handle,
+		source.p,
+	)
+	if !ok || !distribution.Valid() || distribution.Target != target.handle ||
+		distribution.Count != 1 || source.p.readyHead != distant.g || source.p.readyCount != 1 {
+		t.Fatalf("preferred exact-candidate distribution = %+v/%t count:%d head:%p",
+			distribution, ok, source.p.readyCount, source.p.readyHead)
+	}
+	distant.g.state = distantState
+	if !validReadyQueue(source.p) ||
+		!fleet.ImportPNeutralRunnable(target.handle, target.p, distribution.Transfer) ||
+		target.p.readyHead != selected.g {
+		t.Fatal("restore/import preferred exact-candidate distribution")
+	}
+}
+
 func TestExecutorFleetBatchPreparationDoesNotAllocate(t *testing.T) {
 	fleet := new(ExecutorFleet)
 	source := bindExecutorFleetManualFixture(t, fleet)

runtime/internal/coro/os_thread_affinity.go

@@ -255,6 +255,24 @@ func osThreadSuspendPeerReady(p *P, owner *G) bool {
 	return false
 }
 
+func validCompletedOSThreadSuspendAction(p *P, task *G, kind ActionKind) bool {
+	if !completedExecutorRunAction(p, task, Action{Kind: kind}) {
+		return false
+	}
+	switch kind {
+	case ActionYield:
+		// Resumed appends the just-yielded continuation at the owner tail.
+		return p.readyTail == task && task.nextReady == nil
+	case ActionPark:
+		// Validate the exact newly parked record and its two active-list
+		// neighbors without traversing unrelated parked tasks or candidates.
+		return task.active != nil &&
+			validActiveWaitSetRecordFast(p, task.active.parkWait)
+	default:
+		return true
+	}
+}
+
 // PrepareOSThreadSuspendHandoff converts one already committed locked
 // ActionYield or ActionPark into a P-local detached phase. It records no
 // physical owner. The target must publish its generation-bound M baton only
@@ -275,9 +293,8 @@ func PrepareOSThreadSuspendHandoff(
 	}
 	p := driver.p
 	if p == nil || task == nil ||
-		!idleExecutorScheduler(p) || !validReadyQueue(p) ||
-		!validSchedulerWaitQueues(p) ||
-		!completedExecutorRunAction(p, task, Action{Kind: kind}) {
+		!idleExecutorScheduler(p) ||
+		!validCompletedOSThreadSuspendAction(p, task, kind) {
 		return false, false
 	}
 	// The target observes every committed Yield/Park. An unlocked task needs no
@@ -287,6 +304,13 @@ func PrepareOSThreadSuspendHandoff(
 	if task.osThreadLockDepth == 0 {
 		return false, true
 	}
+	// The overwhelmingly common unlocked path above needs only the scheduler's
+	// owner-maintained O(1) headers and the exact completed task. Retain the
+	// complete queue audits for the exceptional physical-owner transition,
+	// where they protect a detached LockOSThread island across another M.
+	if !validReadyQueue(p) || !validSchedulerWaitQueues(p) {
+		return false, false
+	}
 	if p.osThreadLockOwner != task ||
 		p.osThreadSuspend != osThreadSuspendAttached {
 		return false, false