perf(coro): amortize scheduler hot-path ownership

doc/coro-performance-baseline.md

@@ -1153,3 +1153,62 @@ Wasmtime. These gates freeze two architectural requirements for later work:
 logical G storage may not regain permanent M/P fields, and task/frame pooling
 must demonstrate a benefit beyond this allocation fusion without retaining an
 unbounded embedded-target cache.
+
+### Constant-time source-owner gates checkpoint
+
+The next scheduler checkpoint uses merge `d2eff9bcc` as its exact parent. A
+single bounded source reduction previously re-entered the complete driver
+validator from several nested layers. That validator re-read every configured
+source capacity and walked the complete in-progress poll-resolution cursor.
+The work was valuable at bind, shutdown, compatibility, and diagnostic
+boundaries, but redundant while the same owner P was advancing one already
+selected source slot.
+
+The retained split has three explicit levels. The hot driver/source header
+checks the immutable driver-to-P binding, the channel source back-pointer, and
+the local run cursor in O(1). Every selected source then validates its own scan
+limit, owner, route, slot, generation, and operation state immediately before
+mutation. Complete audits remain at lifecycle and diagnostic boundaries. The
+bounded runner also calls a private already-bound poll reducer after validating
+the owner once; exported and compatibility poll entries retain the checked
+wrapper. Its managed-resume permit probe validates only the active P
+back-pointer and issued-action gate because `NextExecutorRunStep` performs the
+complete hot-header proof before opening the no-return action interval.
+
+Regression tests corrupt a distant manual-source scan tail and an in-progress
+poll cursor: an unrelated observational owner probe remains O(1), while the
+complete audit rejects both. Independent corruption of source-set identity, P
+back-pointer, and run cursor fails at the hot header. Finally, a selected
+manual operation with a damaged exact owner is rejected without changing its
+park or source state, then succeeds after the owner is restored. These are
+architecture gates, not only output tests.
+
+The host changed frequency substantially between two AB/BA-interleaved
+campaigns, so only within-campaign ratios are combined. In the first nine-run
+campaign, replacing complete catalog/cursor audits moved 5,000 unbuffered
+request/ack handoffs from 138.628 ms to 93.204 ms (-32.77%) and spawn 100 by
+100 from 102.044 ms to 99.249 ms (-2.74%). In the second eleven-run campaign,
+removing the nested owner recheck moved handoff from 47.830 ms to 43.712 ms
+(-8.61%) and spawn from 51.389 ms to 50.841 ms (-1.07%). The compounded
+handoff reduction is about 38.6%. The final handoff range in that stable
+campaign was 43.509--44.121 ms.
+
+The same pure-compute function has the same address and byte-for-byte identical
+machine code in parent and candidate binaries. Its noisy timing movement is
+therefore not attributed to this runtime-only change. Against the earlier
+stable same-source Go median of 0.751 ms, the final handoff sample is a
+directional cross-session ratio of about 58.2x, down from about 96.7x at the
+previous checkpoint. It is not promoted to a paired regression budget: the
+next attempted all-workload run coincided with unrelated media rendering and a
+host load average near 35, producing more than 10x ranges, and was rejected.
+
+The stripped workload executable grows from 4,876,592 to 4,877,056 bytes
+(+464, +0.010%). Mach-O `__TEXT`, `__DATA_CONST`, and `__DATA` segment
+reservations are unchanged; `__text` grows by 3,068 bytes. No runtime object or
+coroutine-frame layout changes. The host race/shuffle coroutine suite, complete
+runtime suite, all 20 native-fleet E2Es, and the linked native
+defer/panic/channel-spawn E2Es pass. The remaining handoff gap is no longer a
+catalog-scan problem: an immediately matched channel operation still suspends
+both endpoints and runs the durable A/ack/B plus typed-cleanup transaction.
+Avoiding that suspension on the exact local ready path is the next performance
+gate.

internal/build/coro_native_fleet_e2e_test.go

@@ -1782,6 +1782,38 @@ func TestCoroNativeFleetLockedForeignReleasesQuotaBeforeReplacementStarts(t *tes
 	}
 }
 
+func TestCoroNativeFleetRetirementReleasesHeldLeaseBeforeSuccessorStarts(t *testing.T) {
+	path := filepath.Join(
+		"..", "..", "runtime", "internal", "runtime",
+		"coro_native_m_owner_llgo.go",
+	)
+	raw, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatal("read native physical owner:", err)
+	}
+	source := string(raw)
+	entry := strings.Index(source, "func coroTargetRetirePhysicalOwnerV1(")
+	if entry < 0 {
+		t.Fatal("native physical owner retirement entry is absent")
+	}
+	retirement := source[entry:]
+	release := strings.Index(
+		retirement,
+		"coroTargetReleaseManagedExecutionIfHeldV1(driver)",
+	)
+	start := strings.Index(
+		retirement,
+		"coroNativeMStartPhysicalOwnerV1(successor, successorSlot)",
+	)
+	if release < 0 || start < 0 || release >= start {
+		t.Fatalf(
+			"physical owner retirement must release an inherited P lease before starting its successor: release=%d start=%d",
+			release,
+			start,
+		)
+	}
+}
+
 func TestCoroNativeFleetLockedForeignCompensationE2E(t *testing.T) {
 	runCoroNativeFleetE2E(t, coroNativeFleetLockedForeignCompensationE2ESource, "locked-foreign-compensation", true, 1)
 }

runtime/coro_runnable_distribution_source_test.go

@@ -182,7 +182,8 @@ func TestCoroNativeFleetUsesFixedTopologyLogicalQuotaAndScalarPeerABI(t *testing
 		"coroNativeFleetV1State.execution.TryAcquire(route)",
 		"coroNativeFleetV1State.execution.Release(route)",
 		"func CoroGOMAXPROCS(n int) int",
-		"coroNativeFleetRingExecutionWaitersV1()",
+		"coroNativeFleetRingExecutionWaitersV1(waiters uint32)",
+		"coroNativeFleetV1State.execution.WaiterMask()",
 	} {
 		if !strings.Contains(quota, required) {
 			t.Errorf("native fleet execution quota lacks logical-limit marker %q", required)

runtime/internal/coro/channel_claim_core_test.go

@@ -477,6 +477,82 @@ func externallyCommitChannelCandidateAtRoute(
 	}
 }
 
+func TestOwnerLocalChannelCompletionSkipsExternalSourceEpoch(t *testing.T) {
+	fixture := newChannelClaimCoreFixture(t, "channel-owner-local-peer", []uint32{151}, true, 0)
+
+	// Consume the mandatory initial parked-set visit. The peer is still waiting,
+	// but its record is now owner-idle and therefore eligible for the exact local
+	// completion queue.
+	requestChannelClaimCoreFixture(t, fixture)
+	initial := pollChannelClaimCoreComplete(t, fixture)
+	if initial.Completed != 0 || initial.Promoted != 0 || fixture.wait.work != waitSetWorkIdle ||
+		fixture.p.affectedWaitHead != nil || fixture.p.affectedWaitTail != nil {
+		t.Fatalf("initial owner-local visit = %+v wait=%+v affected=(%p,%p)",
+			initial, fixture.wait, fixture.p.affectedWaitHead, fixture.p.affectedWaitTail)
+	}
+
+	producer := newYieldingTestG(t, "channel-owner-local-producer")
+	if !Enqueue(fixture.p, producer.g) {
+		t.Fatal("enqueue owner-local producer")
+	}
+	if g, ok := NextRunnable(fixture.p); !ok || g != producer.g {
+		t.Fatalf("dequeue owner-local producer = (%p,%t)", g, ok)
+	}
+	producerAction := beginWaitTestResume(t, fixture.p, producer)
+	externallyCommitChannelCandidateAtRoute(t, fixture, 0, RouteID(1))
+
+	local, ok := TryPublishOwnerLocalChannelCompletion(producer.g, fixture.source, fixture.ids[0])
+	ready, readyOK := channelOperationReadyAt(fixture.source, fixture.ids[0].LocalSlot()-1)
+	if !ok || !local || !readyOK || ready || fixture.source.Pending() ||
+		fixture.driver.local.head != &fixture.wait || fixture.driver.local.tail != &fixture.wait ||
+		fixture.wait.work != waitSetWorkQueued || fixture.p.affectedWaitHead != nil ||
+		fixture.p.affectedWaitTail != nil || fixture.registry.ObserveRequested(fixture.handle) ||
+		preemptWordState(loadGPreempt(producer.g)) != preemptRequested {
+		t.Fatalf("owner-local publication = (%t,%t) ready=(%t,%t) pending=%t local=(%p,%p) wait=%+v request=%t preempt=%#x",
+			local, ok, ready, readyOK, fixture.source.Pending(), fixture.driver.local.head,
+			fixture.driver.local.tail, fixture.wait, fixture.registry.ObserveRequested(fixture.handle),
+			loadGPreempt(producer.g))
+	}
+	yieldRunningDriverTask(t, fixture.p, producer, producerAction)
+
+	var complete ExecutorPollProgress
+	for reduction := 0; reduction < 64; reduction++ {
+		step, advanced := NextExecutorRunStep(fixture.driver)
+		if !advanced || step.Kind != ExecutorRunStepSource || step.Poll.Used != 1 ||
+			!step.Poll.AtomicResolve || fixture.driver.poll != (executorPollTransaction{}) {
+			t.Fatalf("owner-local reduction %d = (%+v,%t), poll=%+v", reduction, step, advanced, fixture.driver.poll)
+		}
+		complete = step.Poll
+		if complete.Complete {
+			if !CommitExecutorRunSourceDistribution(fixture.driver, false) {
+				t.Fatal("commit owner-local source distribution")
+			}
+			break
+		}
+	}
+	if !complete.Complete || complete.Promoted != 1 || !emptyOwnerLocalCompletion(&fixture.driver.local) ||
+		fixture.task.g.park.phase != parkReady || fixture.source.Pending() ||
+		fixture.registry.ObserveRequested(fixture.handle) {
+		t.Fatalf("owner-local completion = %+v local=%+v park=%+v pending=%t request=%t",
+			complete, fixture.driver.local, fixture.task.g.park, fixture.source.Pending(),
+			fixture.registry.ObserveRequested(fixture.handle))
+	}
+
+	if !EnterExecutorRunCompatibility(fixture.driver) {
+		t.Fatal("leave owner-local bounded runner")
+	}
+	if g, runnable := NextRunnable(fixture.p); !runnable || g != producer.g {
+		t.Fatalf("dequeue yielded owner-local producer = (%p,%t)", g, runnable)
+	}
+	finishWaitTestTask(t, fixture.p, producer, beginWaitTestResume(t, fixture.p, producer))
+	decision := takeChannelClaimCoreDecision(t, fixture)
+	if decision.outcome != ParkOutcomeCompleted || decision.caseID != 151 || !decision.lease.Valid() {
+		t.Fatalf("owner-local peer decision = %+v", decision)
+	}
+	releaseChannelClaimCoreFixture(t, fixture, decision)
+	runtime.KeepAlive(producer.frame.memory)
+}
+
 func TestSelectClaimLayoutPairAcquisitionAndFrozenSourceID(t *testing.T) {
 	if unsafe.Sizeof(SelectClaim{}) != 4 || unsafe.Alignof(SelectClaim{}) != 4 {
 		t.Fatalf("SelectClaim layout = size:%d align:%d", unsafe.Sizeof(SelectClaim{}), unsafe.Alignof(SelectClaim{}))

runtime/internal/coro/channel_operation_source.go

@@ -1503,6 +1503,101 @@ func (transaction *ChannelExternalCommit) CommitAtRoute(route RouteID) bool {
 	return true
 }
 
+// TryPublishOwnerLocalChannelCompletion consumes the exact Forced mailbox
+// produced by current on its own bound P. published=false, ok=true is the
+// ordinary fallback result: the caller must request the operation's executor
+// and let the durable external protocol service it. ok=false is an invariant
+// failure. A successful call queues only scheduler work and requests a bounded
+// safepoint; it never resolves, materializes, or resumes another G inline.
+//
+// This entry is intentionally G-authenticated rather than route-authenticated.
+// A route is public producer metadata and cannot prove that a callback or
+// foreign thread currently owns scheduler-only P fields.
+func TryPublishOwnerLocalChannelCompletion(
+	current *G,
+	source *ChannelOperationSource,
+	id OperationID,
+) (published, ok bool) {
+	driver, _, route, currentOK := CurrentExecutorDriver(current)
+	if !currentOK || source == nil || source != driver.sources.channel ||
+		source.owner != driver.p || source.route != route || id.Route() != route {
+		return false, true
+	}
+	slot, slotOK := channelOperationSlotFor(source, id)
+	if !slotOK || preemptLoad(&slot.generation) != id.Generation ||
+		producerSourceLifecycle(preemptLoad(&slot.state)) != producerSourceActive &&
+			producerSourceLifecycle(preemptLoad(&slot.state)) != producerSourceClosing {
+		return false, false
+	}
+	record, claim := &slot.record, slot.claim
+	wait := record.link.wait
+	completionRoute, committed := channelPhysicalCompletionRoute(preemptLoad(&slot.physical))
+	// Current operations which have not yet become active, an already queued
+	// wait, and a route-zero/cross-owner completion are valid external-path
+	// cases. Do not disturb their sticky mailbox or pending bit.
+	if !committed || completionRoute != route || wait == nil ||
+		!canAppendOwnerLocalCompletion(driver, wait) {
+		return false, true
+	}
+	ready, readyOK := channelOperationReadyAt(source, id.LocalSlot()-1)
+	state, candidatePublished := operationCandidateState(record), operationCandidateIsPublished(record)
+	if !readyOK || !ready || preemptLoad(&slot.mailbox) != uint32(channelMailboxForced) ||
+		claim == nil || selectClaimLoad(claim) != selectClaimClaimed ||
+		preemptLoad(&slot.external) != uint32(channelExternalExposed) ||
+		preemptLoad(&slot.externalLease)&1 != 0 ||
+		preemptLoad(&slot.inflight)&producerAdmissionCountMask != 0 ||
+		record.id != id || record.phase != operationActive || record.disposition != OperationDispositionPending ||
+		record.resolutionApplied || record.link.operation != record || record.link.park != &wait.g.park ||
+		record.link.ticket != wait.ticket || record.resultState != operationResultEmpty ||
+		!validOperationCandidate(record) || operationCandidateMode(record) != OperationCommitReadyThenTryCommit ||
+		(state != OperationCommitIdle || candidatePublished) &&
+			(state != OperationCommitReady || !candidatePublished) {
+		return false, false
+	}
+	if !candidatePublished {
+		if _, ticketOK := nextParkTicket(record.resultTicket); !ticketOK {
+			return false, false
+		}
+	}
+	// Request before the irreversible owner publication. It coalesces at any
+	// current critical depth and guarantees that the compiler returns to the
+	// scheduler at a bounded legal safepoint if this resume does not park first.
+	if !RequestPreempt(current) {
+		return false, false
+	}
+
+	// Clear pending before removing the exact leaf. A concurrent producer which
+	// was already between leaf and pending publication remains visible through
+	// the refreshed page summary below; one arriving later stores pending itself.
+	preemptStore(&source.pending, 0)
+	index := id.LocalSlot() - 1
+	page := index / ChannelOperationPageCapacity
+	if !source.readyPages.take(page) || !clearChannelOperationReadyAt(source, index) {
+		_ = markChannelOperationReadyAt(source, index)
+		preemptStore(&source.pending, 1)
+		return false, false
+	}
+	mailbox, drainOK := beginChannelMailboxDrain(slot, channelMailboxForced)
+	if !drainOK || mailbox != channelMailboxForced {
+		_ = markChannelOperationReadyAt(source, index)
+		preemptStore(&source.pending, 1)
+		return false, false
+	}
+	if PublishExternallyCommittedReadyThenCandidate(record, id) != OperationCompletionPublished {
+		_ = restoreChannelMailboxDrain(source, slot, channelMailboxForced)
+		return false, false
+	}
+	appendOwnerLocalCompletionUnchecked(driver, wait)
+	if !finishChannelMailboxDrain(source, slot, channelMailboxForced) ||
+		!refreshChannelOperationReadyPage(source, page) {
+		return false, false
+	}
+	if !source.readyPages.empty() {
+		preemptStore(&source.pending, 1)
+	}
+	return true, true
+}
+
 func (source *ChannelOperationSource) publishExternallyCommittedHeld(
 	slot *channelOperationSlot,
 	id OperationID,

runtime/internal/coro/execution_quota.go

@@ -17,15 +17,17 @@
 package coro
 
 // ExecutionQuota is the process-level managed-execution gate shared by a
-// bounded executor fleet. It limits only entry into a physical coroutine
-// resume. An executor which has no permit remains alive and may continue to
-// service its route-local timer, poll, channel, cancellation, and transfer
-// sources.
+// bounded executor fleet. A held route represents one physical owner leasing
+// a logical P across a bounded scheduler run slice, rather than a lock acquired
+// around every individual llvm.coro.resume. An executor which has no lease
+// remains alive and may continue to service route-local timer, poll, channel,
+// cancellation, and transfer sources until it first needs managed execution.
 //
 // This separation keeps every P/route identity stable while GOMAXPROCS changes:
 // shrinking the logical execution limit never destroys a source owner or moves
 // an outstanding operation. The same acquire/release boundary is also the
-// future blocking-compensation handoff point.
+// blocking-call compensation temporarily hands this same lease to a replacement
+// physical owner and reacquires it before the suspended owner resumes Go code.
 //
 // All concurrently observed fields are uint32 atomics. Two packed holder bits
 // per physical route make a double acquire/release fail closed without a
@@ -186,6 +188,52 @@ func (quota *ExecutionQuota) Usage() (limit, active uint32, ok bool) {
 	return limit, active, true
 }
 
+// WaiterMask returns the exact bounded physical routes which published quota
+// contention. It is an advisory wake snapshot, not a second admission gate:
+// each route still rechecks TryAcquire after its retained doorbell fires.
+// A contender which races after this snapshot observes newly available quota
+// in TryAcquire's final recheck and therefore cannot lose a wake.
+func (quota *ExecutionQuota) WaiterMask() (uint32, bool) {
+	if quota == nil || preemptLoad(&quota.lifecycle) != uint32(executionQuotaActive) {
+		return 0, false
+	}
+	mask := preemptLoad(&quota.waiters)
+	if preemptLoad(&quota.lifecycle) != uint32(executionQuotaActive) {
+		return 0, false
+	}
+	return mask & ((uint32(1) << uint32(ExecutorFleetCapacity)) - 1), true
+}
+
+// Held reports whether the exact physical route currently owns its P lease.
+// Idle and Held are stable query results; Claiming/Releasing is an in-flight
+// same-route ownership transition and therefore fails closed for an owner-side
+// handoff decision.
+func (quota *ExecutionQuota) Held(route RouteID) (held, ok bool) {
+	if quota == nil {
+		return false, false
+	}
+	lifecycle := executionQuotaLifecycle(preemptLoad(&quota.lifecycle))
+	if lifecycle != executionQuotaActive && lifecycle != executionQuotaSealed {
+		return false, false
+	}
+	holderMask, _, shift, valid := executionQuotaRouteMasks(route)
+	if !valid {
+		return false, false
+	}
+	state := executionQuotaHolderState((preemptLoad(&quota.holders) & holderMask) >> shift)
+	if current := executionQuotaLifecycle(preemptLoad(&quota.lifecycle)); current != lifecycle {
+		return false, false
+	}
+	switch state {
+	case executionQuotaHolderIdle:
+		return false, true
+	case executionQuotaHolderHeld:
+		return true, true
+	default:
+		return false, false
+	}
+}
+
 // SetLimit atomically changes the logical execution limit and returns the
 // previous value. A shrink never revokes an in-flight resume; it prevents new
 // acquisitions until active falls below the new limit. wake reports whether a
@@ -210,8 +258,8 @@ func (quota *ExecutionQuota) SetLimit(limit uint32) (previous uint32, wake, ok b
 	}
 }
 
-// TryAcquire attempts to grant one exact physical route permission to enter a
-// managed coroutine resume. acquired=false, ok=true is ordinary quota
+// TryAcquire attempts to grant one exact physical route a bounded P lease.
+// acquired=false, ok=true is ordinary quota
 // contention. The waiter bit is published before the final availability
 // recheck, closing the release-before-sleep lost-wake window.
 func (quota *ExecutionQuota) TryAcquire(route RouteID) (acquired, ok bool) {
@@ -307,9 +355,9 @@ func (quota *ExecutionQuota) TryAcquire(route RouteID) (acquired, ok bool) {
 	}
 }
 
-// Release closes the exact route's physical resume interval. wake is a sticky
-// hint: ringing all bounded route doorbells is safe, and each contender clears
-// only its own waiter bit after it successfully rechecks or acquires.
+// Release closes the exact route's bounded P lease. wake is a sticky hint that
+// the caller must snapshot and ring the exact waiter routes. Each contender
+// clears only its own waiter bit after it successfully rechecks or acquires.
 func (quota *ExecutionQuota) Release(route RouteID) (wake, ok bool) {
 	lifecycle := executionQuotaLifecycle(preemptLoad(&quota.lifecycle))
 	if lifecycle != executionQuotaActive && lifecycle != executionQuotaSealed {