diff --git a/doc/coro-performance-baseline.md b/doc/coro-performance-baseline.md index e26972398a..57e0b8841d 100644 --- a/doc/coro-performance-baseline.md +++ b/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. diff --git a/internal/build/coro_native_fleet_e2e_test.go b/internal/build/coro_native_fleet_e2e_test.go index 2596d80380..bbcdcda575 100644 --- a/internal/build/coro_native_fleet_e2e_test.go +++ b/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) } diff --git a/runtime/coro_runnable_distribution_source_test.go b/runtime/coro_runnable_distribution_source_test.go index e49902ae97..28d79dedd5 100644 --- a/runtime/coro_runnable_distribution_source_test.go +++ b/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) diff --git a/runtime/internal/coro/channel_claim_core_test.go b/runtime/internal/coro/channel_claim_core_test.go index 40e70267d9..d4e0af6460 100644 --- a/runtime/internal/coro/channel_claim_core_test.go +++ b/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{})) diff --git a/runtime/internal/coro/channel_operation_source.go b/runtime/internal/coro/channel_operation_source.go index 257fb5a389..129fd90b44 100644 --- a/runtime/internal/coro/channel_operation_source.go +++ b/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, diff --git a/runtime/internal/coro/execution_quota.go b/runtime/internal/coro/execution_quota.go index 7017bf80e8..da48ca7765 100644 --- a/runtime/internal/coro/execution_quota.go +++ b/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("a.lifecycle) != uint32(executionQuotaActive) { + return 0, false + } + mask := preemptLoad("a.waiters) + if preemptLoad("a.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("a.lifecycle)) + if lifecycle != executionQuotaActive && lifecycle != executionQuotaSealed { + return false, false + } + holderMask, _, shift, valid := executionQuotaRouteMasks(route) + if !valid { + return false, false + } + state := executionQuotaHolderState((preemptLoad("a.holders) & holderMask) >> shift) + if current := executionQuotaLifecycle(preemptLoad("a.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("a.lifecycle)) if lifecycle != executionQuotaActive && lifecycle != executionQuotaSealed { diff --git a/runtime/internal/coro/execution_quota_test.go b/runtime/internal/coro/execution_quota_test.go index c5220ae469..9134771431 100644 --- a/runtime/internal/coro/execution_quota_test.go +++ b/runtime/internal/coro/execution_quota_test.go @@ -38,15 +38,27 @@ func TestExecutionQuotaLifecycleAndStickyWake(t *testing.T) { if acquired, ok := quota.TryAcquire(1); !acquired || !ok { t.Fatal("route 1 did not acquire the only permit") } + if held, ok := quota.Held(1); !ok || !held { + t.Fatalf("route 1 held lease = (%t, %t)", held, ok) + } + if held, ok := quota.Held(2); !ok || held { + t.Fatalf("idle route 2 held lease = (%t, %t)", held, ok) + } if acquired, ok := quota.TryAcquire(2); acquired || !ok { t.Fatalf("contended route 2 acquire = (%t, %t)", acquired, ok) } + if waiters, ok := quota.WaiterMask(); !ok || waiters != 1<<(2-1) { + t.Fatalf("exact execution waiter mask = (%08b, %t)", waiters, ok) + } if previous, wake, ok := quota.SetLimit(4); !ok || previous != 1 || !wake { t.Fatalf("grow execution quota = (%d, %t, %t)", previous, wake, ok) } if acquired, ok := quota.TryAcquire(2); !acquired || !ok { t.Fatal("route 2 did not acquire after growth") } + if waiters, ok := quota.WaiterMask(); !ok || waiters != 0 { + t.Fatalf("acquired route retained waiter = (%08b, %t)", waiters, ok) + } if previous, wake, ok := quota.SetLimit(1); !ok || previous != 4 || wake { t.Fatalf("shrink execution quota = (%d, %t, %t)", previous, wake, ok) } @@ -62,6 +74,9 @@ func TestExecutionQuotaLifecycleAndStickyWake(t *testing.T) { if wake, ok := quota.Release(1); !ok || !wake { t.Fatalf("route 1 release = (%t, %t), want sticky wake", wake, ok) } + if held, ok := quota.Held(1); !ok || held { + t.Fatalf("released route 1 held lease = (%t, %t)", held, ok) + } if acquired, ok := quota.TryAcquire(3); !acquired || !ok { t.Fatal("route 3 did not acquire released permit") } diff --git a/runtime/internal/coro/executor_driver.go b/runtime/internal/coro/executor_driver.go index 72be5bf92f..3c047fd66e 100644 --- a/runtime/internal/coro/executor_driver.go +++ b/runtime/internal/coro/executor_driver.go @@ -43,6 +43,7 @@ type ExecutorDriver struct { route RouteID sources ExecutorSourceSet poll executorPollTransaction + local ownerLocalCompletionCursor run executorRunCursor prepareNow int64 hasPrepareNow bool @@ -62,7 +63,12 @@ const ( const executorDriverMagic uint32 = 0x45584431 // "EXD1" -func validExecutorDriver(driver *ExecutorDriver) bool { +// validExecutorDriverHeader is the O(1) owner gate for a selected runner +// reduction. It checks the immutable binding, reciprocal P/source identities, +// and the local run cursor, but deliberately does not re-audit every source +// capacity or walk the in-progress logical resolution cursor. The concrete +// source or resolution reduction performs those exact checks before mutation. +func validExecutorDriverHeader(driver *ExecutorDriver) bool { if driver == nil || driver.magic != executorDriverMagic || driver.state == executorDriverUnbound { return false } @@ -81,16 +87,28 @@ func validExecutorDriver(driver *ExecutorDriver) bool { driver.p != nil && driver.registry != nil && driver.handle.Slot != 0 && driver.handle.Generation != 0 && driver.route.Valid() && driver.sources.route == driver.route && driver.p.executor == driver && preemptLoad(&driver.p.executorMode) == executorModeBound && - validExecutorSourceSet(&driver.sources, driver.p) && validExecutorPollTransaction(&driver.poll, &driver.sources) && + validExecutorSourceSetHeader(&driver.sources, driver.p) && + validOwnerLocalCompletionHeader(&driver.local, driver.p) && validExecutorRunCursor(&driver.run, driver.p) } +func validExecutorDriver(driver *ExecutorDriver) bool { + return validExecutorDriverHeader(driver) && + validExecutorSourceSet(&driver.sources, driver.p) && + validExecutorPollTransaction(&driver.poll, &driver.sources) && + validOwnerLocalCompletion(&driver.local, driver.p) +} + func validExecutorDriverForP(driver *ExecutorDriver, p *P) bool { return validExecutorDriver(driver) && driver.state == executorDriverActive && driver.p == p } +func validExecutorDriverHeaderForP(driver *ExecutorDriver, p *P) bool { + return validExecutorDriverHeader(driver) && driver.state == executorDriverActive && driver.p == p +} + func validRunningExecutorOwner(driver *ExecutorDriver) bool { - if !validExecutorDriver(driver) || driver.state != executorDriverActive { + if !validExecutorDriverHeader(driver) || driver.state != executorDriverActive { return false } p := driver.p @@ -149,7 +167,7 @@ func currentExecutorParkDriver(g *G) (*ExecutorDriver, ExecutorHandle, RouteID, driver := p.executor handle := g.active.handle header := g.active.header - if !validExecutorDriverForP(driver, p) || p.current != g || !p.inResume || + if !validExecutorDriverHeaderForP(driver, p) || p.current != g || !p.inResume || !expectedAction(p, g, p.action, ActionResume) || !activeResumeOwnedByAction(g) || g.state != GRunning || g.active.state != FrameActive || g.active.handle != handle || g.active.header != header || @@ -329,6 +347,7 @@ func bindExecutorAtRoute(driver *ExecutorDriver, p *P, registry *ExecutorRegistr if driver == nil || driver.magic != 0 || driver.state != executorDriverUnbound || driver.p != nil || driver.registry != nil || driver.handle != (ExecutorHandle{}) || driver.route != 0 || driver.sources != (ExecutorSourceSet{}) || driver.poll != (executorPollTransaction{}) || + driver.local != (ownerLocalCompletionCursor{}) || driver.run != (executorRunCursor{}) || driver.prepareNow != 0 || driver.hasPrepareNow || driver.terminalKind != ActionInvalid || diff --git a/runtime/internal/coro/executor_driver_test.go b/runtime/internal/coro/executor_driver_test.go index 6ec213df1b..3fd167060c 100644 --- a/runtime/internal/coro/executor_driver_test.go +++ b/runtime/internal/coro/executor_driver_test.go @@ -111,6 +111,71 @@ func TestExecutorDriverBindCloseLifecycle(t *testing.T) { } } +func TestExecutorDriverHotHeaderDefersDeepCatalogAndPollAudits(t *testing.T) { + p := new(P) + driver, _, manual, _ := bindTestExecutorDriverWithManual(t, p) + if !validExecutorDriverHeader(driver) || !validExecutorDriver(driver) { + t.Fatal("fresh driver failed header or complete audit") + } + + // A selected owner reduction does not walk an unrelated source catalog. + // The complete diagnostic boundary must still reject its invalid scan tail. + manual.scanLimit = ManualOperationConfiguredCapacity(manual) + 1 + if !validExecutorDriverHeader(driver) { + t.Fatal("hot header inspected the manual catalog scan tail") + } + if validExecutorDriver(driver) { + t.Fatal("complete driver audit accepted an invalid manual scan tail") + } + if pending, ok := ExecutorRunManagedResumePending(driver); !ok || pending { + t.Fatalf("observational hot gate over distant catalog damage = (%t, %t)", pending, ok) + } + manual.scanLimit = 0 + + // The in-progress poll cursor is likewise validated by its exact reducer and + // by complete lifecycle diagnostics, not by an unrelated owner observation. + driver.poll = executorPollTransaction{phase: executorPollAcknowledge} + if !validExecutorDriverHeader(driver) { + t.Fatal("hot header inspected the logical poll cursor") + } + if validExecutorDriver(driver) { + t.Fatal("complete driver audit accepted an invalid logical poll cursor") + } + driver.poll = executorPollTransaction{} + if !validExecutorDriver(driver) { + t.Fatal("restored driver failed complete audit") + } + closeTestExecutorDriver(t, driver) +} + +func TestExecutorDriverHotHeaderRejectsLocalBindingDamage(t *testing.T) { + p := new(P) + driver, _, _ := bindTestExecutorDriver(t, p) + + driver.sources.magic = 0 + if validExecutorDriverHeader(driver) { + t.Fatal("hot header accepted damaged source-set identity") + } + driver.sources.magic = executorSourceSetMagic + + p.executor = nil + if validExecutorDriverHeader(driver) { + t.Fatal("hot header accepted damaged P back-pointer") + } + p.executor = driver + + driver.run.actionsSinceSource = executorRunSourceQuantum + 1 + if validExecutorDriverHeader(driver) { + t.Fatal("hot header accepted damaged local run cursor") + } + driver.run.actionsSinceSource = 0 + + if !validExecutorDriver(driver) { + t.Fatal("restored local binding failed complete audit") + } + closeTestExecutorDriver(t, driver) +} + func drainTimerAwareExecutorRunSources(t *testing.T, driver *ExecutorDriver, now int64) { t.Helper() for reduction := 0; reduction < 4096; reduction++ { diff --git a/runtime/internal/coro/executor_progress.go b/runtime/internal/coro/executor_progress.go index 3cd279a3bb..6d957859fa 100644 --- a/runtime/internal/coro/executor_progress.go +++ b/runtime/internal/coro/executor_progress.go @@ -18,10 +18,12 @@ package coro // ExecutorPollProgress is the pointer-free host boundary for one bounded // source-service entry. Counts are cumulative for the current A/ack/B -// transaction; Used is charged only for this call. Every production catalog -// slot, affected wait-set decision, candidate scan/settle/apply, promotion, and -// legacy-G visit is resumable and charged as one reduction. ApplyVisits counts -// only source-specific candidate ApplyOne actions. Complete means that +// transaction; Used is charged only for this call. One production catalog +// reduction visits at most executorCatalogBatchQuantum adjacent timer or poll +// slots (one slot for other sources); every affected wait-set decision, +// candidate scan/settle/apply, promotion, and legacy-G visit remains one +// reduction. ApplyVisits counts only source-specific candidate ApplyOne +// actions. Complete means that // the transaction reached the end of epoch B. More requests a later, // non-recursive scheduler entry, while Blocked means that only a new external // fact (or a reported future deadline) can make progress. More and Blocked are @@ -71,6 +73,13 @@ const ( executorCatalogDone ) +// Timer expiry and reactor readiness commonly arrive in bursts. Keeping this +// quantum small amortizes runner/validation dispatch without turning one host +// reduction into an unbounded catalog walk on embedded or single-threaded +// targets. Other source types retain one-entry reductions because they may run +// stronger admission or control protocols per entry. +const executorCatalogBatchQuantum uint32 = 8 + // executorPollTransaction is scheduler-owner-only continuation state. It has // no callback-visible pointer and is embedded at a stable address in the // driver. now is captured once per logical epoch and is intentionally @@ -234,10 +243,11 @@ func beginExecutorPollEpoch(transaction *executorPollTransaction, sources *Execu return true } -// executorMinPollBudget returns a conservative full-transaction budget. Most -// catalogs charge every slot in their binding-local active prefix; indexed -// catalogs may skip empty regions and finish earlier. Configured-but-never- -// allocated tails remain covered by structural audits, not routine service. +// executorMinPollBudget returns a conservative full-transaction budget. It +// deliberately counts every slot in the binding-local active prefix even +// though timer/poll batching and indexed catalogs can finish earlier. +// Configured-but-never-allocated tails remain covered by structural audits, +// not routine service. func executorMinPollBudget(sources *ExecutorSourceSet) (uint32, bool) { if sources == nil { return 0, false @@ -254,10 +264,10 @@ func executorMinPollBudget(sources *ExecutorSourceSet) (uint32, bool) { } // MinExecutorPollBudget is a sufficient base budget for one idle driver's -// fixed A/ack/B catalog and two empty common-resolution actions. An indexed -// catalog can complete below this bound. Non-empty affected waits and legacy -// waiters add explicitly charged reductions; smaller budgets retain exact -// source and resolution cursors for a later entry. +// fixed A/ack/B catalog and two empty common-resolution actions. Batched or +// indexed catalogs can complete below this bound. Non-empty affected waits and +// legacy waiters add explicitly charged reductions; smaller budgets retain +// exact source and resolution cursors for a later entry. func MinExecutorPollBudget(driver *ExecutorDriver) (uint32, bool) { if !validExecutorDriver(driver) || driver.state != executorDriverActive || driver.poll.phase != executorPollIdle { return 0, false @@ -401,6 +411,30 @@ func publishExecutorCatalogEntry(driver *ExecutorDriver) bool { return true } +func publishExecutorCatalogReduction(driver *ExecutorDriver) bool { + if driver == nil || + (driver.poll.phase != executorPollEpochAPublish && driver.poll.phase != executorPollEpochBPublish) || + driver.poll.source >= executorCatalogDone { + return false + } + source := driver.poll.source + limit := uint32(1) + if source == executorCatalogTimers || source == executorCatalogPoll { + limit = executorCatalogBatchQuantum + } + for visited := uint32(0); visited < limit; visited++ { + if !publishExecutorCatalogEntry(driver) { + return false + } + // A source transition is a stable fairness boundary. The next reducer + // starts the next source even when both are batchable. + if driver.poll.source != source { + break + } + } + return true +} + func executorProgressFromScan(scan executorSourceScan, used, budget uint32, complete, more, blocked bool) (ExecutorPollProgress, bool) { if scan.completed < 0 || scan.timers < 0 || scan.poll < 0 || scan.manual < 0 || scan.manualLost < 0 || scan.worker < 0 || scan.workerLost < 0 || @@ -432,14 +466,14 @@ func executorProgressFromScan(scan executorSourceScan, used, budget uint32, comp }, true } -// pollExecutorSliceAt advances the first production-bounded part of one -// A/ack/B transaction without recursively re-entering the scheduler. Every -// source entry, acknowledgement, candidate action, promotion, and legacy-G -// visit costs exactly one reduction. Administrative phase transitions are -// folded into the action they expose and never hide a collection scan. -func pollExecutorSliceAt(driver *ExecutorDriver, now int64, withDeadline bool, budget uint32) (scan executorSourceScan, progress ExecutorPollProgress, ok bool) { - if budget == 0 || !validExecutorDriver(driver) || driver.state != executorDriverActive || !idleExecutorScheduler(driver.p) || - !driver.sources.acceptsScan(driver.p, now, withDeadline) { +// pollBoundExecutorSliceAt advances a driver whose immutable binding was +// validated by the current owner reduction. It still checks the exact idle +// scheduler boundary and every transaction/source cursor before mutation. +// Keeping this private avoids repeating the complete owner header when +// nextExecutorRunStepAt immediately delegates one source reduction here. +func pollBoundExecutorSliceAt(driver *ExecutorDriver, now int64, withDeadline bool, budget uint32) (scan executorSourceScan, progress ExecutorPollProgress, ok bool) { + if budget == 0 || driver == nil || driver.state != executorDriverActive || !idleExecutorScheduler(driver.p) || + withDeadline != driver.sources.usesMonotonicTime() || withDeadline && now < 0 { return executorSourceScan{}, ExecutorPollProgress{}, false } if driver.poll.phase == executorPollIdle { @@ -469,7 +503,7 @@ func pollExecutorSliceAt(driver *ExecutorDriver, now int64, withDeadline bool, b } continue } - if !publishExecutorCatalogEntry(driver) { + if !publishExecutorCatalogReduction(driver) { return transaction.total, ExecutorPollProgress{}, false } used++ @@ -527,6 +561,18 @@ func pollExecutorSliceAt(driver *ExecutorDriver, now int64, withDeadline bool, b return scan, progress, ok } +// pollExecutorSliceAt is the checked host/compatibility boundary for the +// production-bounded A/ack/B transaction. Every source entry, +// acknowledgement, candidate action, promotion, and legacy-G visit costs +// exactly one reduction. Administrative phase transitions are folded into the +// action they expose and never hide a collection scan. +func pollExecutorSliceAt(driver *ExecutorDriver, now int64, withDeadline bool, budget uint32) (scan executorSourceScan, progress ExecutorPollProgress, ok bool) { + if !validExecutorDriverHeader(driver) { + return executorSourceScan{}, ExecutorPollProgress{}, false + } + return pollBoundExecutorSliceAt(driver, now, withDeadline, budget) +} + // PollExecutorSlice services a no-deadline source catalog for at most budget // catalog, resolution, and acknowledgement reductions. More never authorizes // direct recursion; a target schedules a later host entry and returns first. diff --git a/runtime/internal/coro/executor_progress_test.go b/runtime/internal/coro/executor_progress_test.go index 223064cfff..b9f0975b13 100644 --- a/runtime/internal/coro/executor_progress_test.go +++ b/runtime/internal/coro/executor_progress_test.go @@ -57,6 +57,24 @@ func TestExecutorPollProgressPODLayout(t *testing.T) { } } +func beginExecutorTimerBatchTestPark(t *testing.T, p *P, seed uint32) *timerV2TestPark { + t.Helper() + task := newYieldingTestG(t, "timer-batch") + if !Enqueue(p, task.g) { + t.Fatal("enqueue batched timer") + } + if g, ok := NextRunnableAt(p, 0); !ok || g != task.g { + t.Fatal("dequeue batched timer") + } + action := beginWaitTestResume(t, p, task) + ticket, ok := BeginParkSet(&task.g.park, 1, seed) + wait := new(WaitSetRecord) + if !ok || !PrepareWaitSetRecord(wait, task.g, ticket) { + t.Fatal("prepare batched timer park") + } + return &timerV2TestPark{task: task, ticket: ticket, wait: wait, action: action} +} + func TestExecutorPollEpochBPreservesAExternalBlockOnly(t *testing.T) { sources := &ExecutorSourceSet{} transaction := executorPollTransaction{ @@ -136,6 +154,103 @@ func TestExecutorPollProgressSkipsConfiguredUnallocatedTails(t *testing.T) { } } +func TestExecutorPollReductionBatchesDueTimers(t *testing.T) { + p := new(P) + driver := new(ExecutorDriver) + registry := new(ExecutorRegistry) + timers := new(TimerRegistrationTable) + handle := registerTestExecutor(t, registry) + if !BindExecutorSourceCatalogAtRoute( + driver, + p, + registry, + handle, + RouteID(1), + ExecutorSourceCatalog{Timers: timers}, + ) { + t.Fatal("bind batched timer executor") + } + + const count = executorCatalogBatchQuantum + parks := make([]*timerV2TestPark, int(count)) + handles := make([]TimerRegistrationHandle, int(count)) + for index := 0; index < int(count); index++ { + park := beginExecutorTimerBatchTestPark(t, p, uint32(301+index)) + timer, attached := timers.ReserveAndAttachTimerV2( + p, + &park.task.g.park, + park.ticket, + park.wait, + uint32(index+1), + 1, + ) + if !attached { + t.Fatalf("reserve batched timer %d", index) + } + commitTimerV2TestPark(t, p, park) + parks[index], handles[index] = park, timer + } + if result := registry.Request(handle); result != ExecutorRequestPublished { + t.Fatalf("request batched timer executor = %d", result) + } + first, ok := PollExecutorSliceAt(driver, 1, 1) + if !ok || first.Used != 1 || first.Complete || first.Timers != count || first.Completed != count || + driver.poll.phase != executorPollEpochAPublish || driver.poll.source != executorCatalogDone { + t.Fatalf("first batched timer reduction = (%+v,%t), poll=%+v", first, ok, driver.poll) + } + + var complete ExecutorPollProgress + for reduction := 0; reduction < 1000; reduction++ { + progress, advanced := PollExecutorSliceAt(driver, 1, 1) + if !advanced || progress.Used != 1 { + t.Fatalf("continue batched timer reduction %d = (%+v,%t)", reduction, progress, advanced) + } + if progress.Complete { + complete = progress + break + } + } + if !complete.Complete || complete.Timers != count || complete.Promoted != count || + driver.poll != (executorPollTransaction{}) { + t.Fatalf("complete batched timers = %+v poll=%+v", complete, driver.poll) + } + sentinel := newYieldingTestG(t, "timer-batch-sentinel") + if !Enqueue(p, sentinel.g) { + t.Fatal("enqueue batched timer terminal sentinel") + } + parkIndex := make(map[*G]int, len(parks)) + for index, park := range parks { + parkIndex[park.task.g] = index + } + for resumed := 0; resumed < len(parks); resumed++ { + g, ok := NextRunnableAt(p, 1) + index, found := parkIndex[g] + if !ok || !found { + t.Fatalf("dequeue promoted batched timer %d = (%p,%t)", resumed, g, ok) + } + delete(parkIndex, g) + park := parks[index] + action := beginWaitTestResume(t, p, park.task) + outcome, caseID, lease, taskCancel, decisionOK := TakeRunDecision(park.task.g, park.ticket) + if outcome != ParkOutcomeCompleted || caseID != uint32(index+1) || taskCancel != TaskCancelNone || + !decisionOK || + !timers.DiscardTimerV2Result(p, handles[index], lease) || + !timers.RecycleTimerV2(p, handles[index]) { + t.Fatalf("consume batched timer %d = outcome:%d case:%d lease:%+v task:%d", + index, outcome, caseID, lease, taskCancel) + } + finishWaitTestTask(t, p, park.task, action) + } + closeTestExecutorDriver(t, driver) + if g, ok := NextRunnable(p); !ok || g != sentinel.g { + t.Fatalf("dequeue batched timer terminal sentinel = (%p,%t)", g, ok) + } + finishWaitTestTask(t, p, sentinel, beginWaitTestResume(t, p, sentinel)) + if !timers.CanRelease() || !registry.CanRelease() { + t.Fatal("batched timer fixture retained source state") + } +} + func TestExecutorPollSliceBudgetOneCompletesExactAcknowledgeTransaction(t *testing.T) { p := new(P) driver, registry, handle := bindTestExecutorDriver(t, p) diff --git a/runtime/internal/coro/executor_resume_handoff.go b/runtime/internal/coro/executor_resume_handoff.go index 60587c2723..5bd4fd6323 100644 --- a/runtime/internal/coro/executor_resume_handoff.go +++ b/runtime/internal/coro/executor_resume_handoff.go @@ -183,7 +183,7 @@ func DetachExecutorResume( // requests may remain for the returning M. A target must additionally settle // its route mailbox, admission and physical-owner directory before FinishReturn. func ExecutorResumeHandoffReturnable(driver *ExecutorDriver) bool { - if !validExecutorDriver(driver) || driver.state != executorDriverActive || + if !validExecutorDriverHeader(driver) || driver.state != executorDriverActive || driver.run.issued != ActionInvalid || driver.poll.phase != executorPollIdle { return false } @@ -218,7 +218,7 @@ func ExecutorResumeHandoffContext( // RestoreExecutorResume reattaches the exact active LLVM resume after the // replacement owner has finished and the target has strongly joined it. The -// caller must reacquire its managed-execution permit before calling Restore. +// caller must reacquire its managed-execution P lease before calling Restore. // Success consumes and zeroes handoff; a duplicate or mismatched restore is // rejected without changing scheduler state. func RestoreExecutorResume(handoff *ExecutorResumeHandoff) bool { diff --git a/runtime/internal/coro/executor_source_set.go b/runtime/internal/coro/executor_source_set.go index 4140fe0611..3cc6828b49 100644 --- a/runtime/internal/coro/executor_source_set.go +++ b/runtime/internal/coro/executor_source_set.go @@ -98,11 +98,25 @@ func (scan *executorSourceScan) add(other executorSourceScan) { scan.hasDeadline = other.hasDeadline } +// validExecutorSourceSetHeader checks only the immutable binding and reciprocal +// owner identities maintained by BindExecutorSourceCatalog. It is the O(1) +// gate for one already-selected scheduler reduction: the selected source then +// validates its exact scan limit, slot, generation, and operation state before +// it mutates anything. +// +// Do not add catalog-capacity walks here. Dynamic pages are published +// monotonically and every concrete catalog access rechecks its own current +// bound. Complete source-set audits remain in validExecutorSourceSet for +// lifecycle, shutdown, compatibility, test, and diagnostic boundaries. +func validExecutorSourceSetHeader(sources *ExecutorSourceSet, p *P) bool { + return sources != nil && sources.magic == executorSourceSetMagic && p != nil && sources.owner == p && + sources.route.Valid() && + (sources.channel == nil) == (p.channelSource == nil) && + (sources.channel == nil || p.channelSource == sources.channel) +} + func validExecutorSourceSet(sources *ExecutorSourceSet, p *P) bool { - if sources == nil || sources.magic != executorSourceSetMagic || p == nil || sources.owner != p || - !sources.route.Valid() || - (sources.channel == nil) != (p.channelSource == nil) || - sources.channel != nil && p.channelSource != sources.channel { + if !validExecutorSourceSetHeader(sources, p) { return false } _, timerScanOK := timerRegistrationScanLimit(sources.timers) @@ -340,7 +354,7 @@ func (sources *ExecutorSourceSet) publishPass(p *P, now int64, withDeadline bool // bit to become quiet. Keeping this separate prevents static source order from // becoming a select tie breaker. func (sources *ExecutorSourceSet) applyOne(p *P, link *ParkLink) OperationApplyResult { - if !validExecutorSourceSet(sources, p) || link == nil || link.operation == nil || + if !validExecutorSourceSetHeader(sources, p) || link == nil || link.operation == nil || link.operation.link.operation != link.operation || &link.operation.link != link || link.operation.phase != operationActive || link.operation.id.Source() == OperationSourceInvalid { return OperationApplyInvalid @@ -560,7 +574,7 @@ func (sources *ExecutorSourceSet) resolvePublishedEpoch(p *P) (promoted, applyVi // Deadline sources are sampled by drain and represented by the aggregate // deadline; future deadlines are not pending runnable work. func (sources *ExecutorSourceSet) pending(p *P) bool { - return validExecutorSourceSet(sources, p) && + return validExecutorSourceSetHeader(sources, p) && (p.affectedWaitHead != nil || sources.poll != nil && sources.poll.Pending() || sources.manual != nil && sources.manual.Pending() || sources.worker != nil && sources.worker.Pending() || diff --git a/runtime/internal/coro/executor_source_set_test.go b/runtime/internal/coro/executor_source_set_test.go index 0fb77d3f85..5aee613114 100644 --- a/runtime/internal/coro/executor_source_set_test.go +++ b/runtime/internal/coro/executor_source_set_test.go @@ -56,6 +56,48 @@ func TestExecutorSourceSetRejectsStandaloneAffectedOperationBeforeResolution(t * } } +func TestExecutorSourceSetHotDispatchRevalidatesSelectedSource(t *testing.T) { + p := new(P) + driver, _, manual, _ := bindTestExecutorDriverWithManual(t, p) + state, ticket, ids := reserveManualWaitSet(t, manual, p, 87, []uint32{13}) + id := ids[0] + slot, _ := manualOperationSlotFor(manual, id) + if result := manual.Post(id); result != ManualOperationPosted { + t.Fatalf("post exact hot-dispatch operation = %d", result) + } + if published, lost, ok := manual.PublishPass(p); !ok || published != 1 || lost != 0 { + t.Fatalf("publish exact hot-dispatch operation = (%d, %d, %t)", published, lost, ok) + } + if resolution, duplicates, ok := manual.ResolveAffectedPublishedEpoch(p); !ok || duplicates != 0 || + resolution != (CompletionResolution{WaitSets: 1, Completed: 1, Winners: 1}) { + t.Fatalf("resolve exact hot-dispatch operation = (%+v, %d, %t)", resolution, duplicates, ok) + } + + // The aggregate header deliberately trusts its frozen source pointer, but + // the selected direct-call source must reject a damaged exact owner before + // touching its slot or park state. + manual.owner = new(P) + if !validExecutorSourceSetHeader(&driver.sources, p) { + t.Fatal("aggregate hot header inspected selected-source internals") + } + if result := driver.sources.applyOne(p, &slot.record.link); result != OperationApplyInvalid { + t.Fatalf("selected source accepted damaged exact owner: %d", result) + } + if slot.record.phase != operationActive || slot.record.resolutionApplied || ParkReady(state, ticket) { + t.Fatal("rejected selected-source dispatch changed live operation") + } + manual.owner = p + if result := driver.sources.applyOne(p, &slot.record.link); result != OperationApplyDetached { + t.Fatalf("restored selected-source dispatch = %d", result) + } + outcome, _, lease, consumed := ConsumeParkSet(state, ticket) + if !consumed || outcome != ParkOutcomeCompleted || !lease.Valid() { + t.Fatalf("consume selected-source winner = (%d, %+v, %t)", outcome, lease, consumed) + } + finishManualOperations(t, manual, p, ids, lease) + closeTestExecutorDriver(t, driver) +} + func TestExecutorSourceSetRetryBudgetAndExternalFactHaveDistinctScheduling(t *testing.T) { p := new(P) manual := new(ManualOperationSource) diff --git a/runtime/internal/coro/os_thread_affinity.go b/runtime/internal/coro/os_thread_affinity.go index 69c4bee1d3..63cf220cac 100644 --- a/runtime/internal/coro/os_thread_affinity.go +++ b/runtime/internal/coro/os_thread_affinity.go @@ -286,7 +286,7 @@ func PrepareOSThreadSuspendHandoff( task *G, kind ActionKind, ) (required, ok bool) { - if !validExecutorDriver(driver) || driver.state != executorDriverActive || + if !validExecutorDriverHeader(driver) || driver.state != executorDriverActive || driver.run.issued != ActionInvalid || driver.poll.phase != executorPollIdle { return false, false @@ -337,7 +337,7 @@ func PrepareOSThreadSuspendHandoff( func OSThreadSuspendHandoffStatus( driver *ExecutorDriver, ) (detached, returnable, ok bool) { - if !validExecutorDriver(driver) || driver.state != executorDriverActive { + if !validExecutorDriverHeader(driver) || driver.state != executorDriverActive { return false, false, false } p := driver.p diff --git a/runtime/internal/coro/owner_local_completion.go b/runtime/internal/coro/owner_local_completion.go new file mode 100644 index 0000000000..37aaf5f9d1 --- /dev/null +++ b/runtime/internal/coro/owner_local_completion.go @@ -0,0 +1,160 @@ +/* + * Copyright (c) 2026 The XGo Authors (xgo.dev). All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package coro + +// ownerLocalCompletionCursor is the scheduler-owned fast lane for a physical +// completion produced by the G which currently owns this exact P. The producer +// publishes the typed source result while it already owns the source's runtime +// synchronization domain, then appends only the affected logical wait here. +// The scheduler resolves it after llvm.coro.resume returns; it never resumes a +// peer recursively or performs typed cleanup inside the producer's lock. +// +// Cross-P, callback, timer, poller, and foreign-thread producers cannot enter +// this queue. They retain the durable source pending + executor request + +// A/ack/B protocol. workNext is sufficient because a locally queued record was +// required to be owner-idle before publication. +type ownerLocalCompletionCursor struct { + head *WaitSetRecord + tail *WaitSetRecord + resolve publishedEpochResolveCursor +} + +func validOwnerLocalCompletionHeader(local *ownerLocalCompletionCursor, p *P) bool { + if local == nil || p == nil || (local.head == nil) != (local.tail == nil) || + local.tail != nil && local.tail.workNext != nil { + return false + } + if local.resolve.phase == publishedEpochResolveIdle { + if local.resolve != (publishedEpochResolveCursor{}) { + return false + } + } else if local.head != nil || !validPublishedEpochResolveCursor(&local.resolve, p) { + return false + } + return local.head == nil || local.head.work == waitSetWorkQueued && + validActiveWaitSetRecordFast(p, local.head) +} + +func validOwnerLocalCompletion(local *ownerLocalCompletionCursor, p *P) bool { + if !validOwnerLocalCompletionHeader(local, p) { + return false + } + var tail *WaitSetRecord + for slow, fast := local.head, local.head; fast != nil && fast.workNext != nil; { + slow = slow.workNext + fast = fast.workNext.workNext + if slow == fast { + return false + } + } + for record := local.head; record != nil; record = record.workNext { + if record.work != waitSetWorkQueued || !validActiveWaitSetRecordFast(p, record) { + return false + } + tail = record + } + return tail == local.tail +} + +func emptyOwnerLocalCompletion(local *ownerLocalCompletionCursor) bool { + return local != nil && *local == (ownerLocalCompletionCursor{}) +} + +func canAppendOwnerLocalCompletion(driver *ExecutorDriver, record *WaitSetRecord) bool { + return driver != nil && validRunningExecutorOwner(driver) && + validOwnerLocalCompletionHeader(&driver.local, driver.p) && + driver.local.resolve == (publishedEpochResolveCursor{}) && + record != nil && record.work == waitSetWorkIdle && record.workNext == nil && + validActiveWaitSetRecordFast(driver.p, record) +} + +func appendOwnerLocalCompletionUnchecked(driver *ExecutorDriver, record *WaitSetRecord) { + record.work = waitSetWorkQueued + if driver.local.tail == nil { + driver.local.head = record + } else { + driver.local.tail.workNext = record + } + driver.local.tail = record +} + +func ownerLocalCompletionPending(driver *ExecutorDriver) bool { + return driver != nil && (driver.local.head != nil || + driver.local.resolve.phase != publishedEpochResolveIdle) +} + +func initializeOwnerLocalCompletionResolution( + driver *ExecutorDriver, + step *publishedEpochResolveStep, +) bool { + if driver == nil || step == nil || driver.local.resolve != (publishedEpochResolveCursor{}) || + !validOwnerLocalCompletionHeader(&driver.local, driver.p) || driver.local.head == nil || + !validAffectedWaitQueueHeader(driver.p) { + return false + } + head, tail := driver.local.head, driver.local.tail + driver.local.resolve.batchTail = tail + if !startPublishedEpochWait(&driver.sources, &driver.local.resolve, head) { + driver.local.resolve.batchTail = nil + return false + } + driver.local.head, driver.local.tail = nil, nil + return true +} + +// resolveOwnerLocalCompletionStep performs one ordinary published-epoch +// resolution reduction, but starts from the exact owner-local FIFO instead of +// rescanning every source and acknowledging a doorbell which was never needed. +// Source-specific ApplyOne and typed materialization remain identical to the +// external path. +func resolveOwnerLocalCompletionStep( + driver *ExecutorDriver, + step *publishedEpochResolveStep, +) (ok bool) { + if driver == nil || step == nil { + return false + } + *step = publishedEpochResolveStep{} + p, cursor := driver.p, &driver.local.resolve + if p == nil || !validReadyQueueHeader(p) || !validParkWaitQueueHeader(p) || + !validAffectedWaitQueueHeader(p) || !validOwnerLocalCompletionHeader(&driver.local, p) { + return false + } + if cursor.phase == publishedEpochResolveIdle { + if !initializeOwnerLocalCompletionResolution(driver, step) { + return false + } + } else if !validPublishedEpochResolveCursor(cursor, p) { + return false + } + + switch cursor.phase { + case publishedEpochResolveDiscover: + ok = resolvePublishedEpochDiscoverStep(&driver.sources, p, cursor, step) + case publishedEpochResolvePark: + ok = resolvePublishedEpochParkStep(&driver.sources, p, cursor, step) + case publishedEpochResolveApply: + ok = resolvePublishedEpochApplyStep(&driver.sources, p, cursor, step) + case publishedEpochResolveFinish: + ok = resolvePublishedEpochFinishStep(cursor, step) + case publishedEpochResolvePromote: + ok = resolvePublishedEpochPromoteStep(&driver.sources, p, cursor, step) + default: + ok = false + } + return ok +} diff --git a/runtime/internal/coro/published_epoch_resolution.go b/runtime/internal/coro/published_epoch_resolution.go index 5601b2fdc0..394abd36b2 100644 --- a/runtime/internal/coro/published_epoch_resolution.go +++ b/runtime/internal/coro/published_epoch_resolution.go @@ -199,7 +199,7 @@ func initializePublishedEpochResolution(sources *ExecutorSourceSet, p *P, cursor return false } if sources != nil { - if !validExecutorSourceSet(sources, p) { + if !validExecutorSourceSetHeader(sources, p) { return false } if sources.manual != nil { diff --git a/runtime/internal/coro/resume_cleanup.go b/runtime/internal/coro/resume_cleanup.go index 8f20e87cd7..80caa61c6a 100644 --- a/runtime/internal/coro/resume_cleanup.go +++ b/runtime/internal/coro/resume_cleanup.go @@ -403,11 +403,11 @@ type ResumeCleanupStep struct { plan *ResumeCleanupPlan } -func pendingResumeCleanupStep(driver *ExecutorDriver) (ResumeCleanupStep, bool) { - if driver == nil || driver.poll.resolve.phase != publishedEpochResolvePromote { +func pendingResumeCleanupStepForCursor(cursor *publishedEpochResolveCursor) (ResumeCleanupStep, bool) { + if cursor == nil || cursor.phase != publishedEpochResolvePromote { return ResumeCleanupStep{}, false } - record := driver.poll.resolve.wait + record := cursor.wait if record == nil || record.resumeKind != resumeBindingCleanup { return ResumeCleanupStep{}, false } @@ -425,6 +425,16 @@ func pendingResumeCleanupStep(driver *ExecutorDriver) (ResumeCleanupStep, bool) }, true } +func pendingResumeCleanupStep(driver *ExecutorDriver) (ResumeCleanupStep, bool) { + if driver == nil { + return ResumeCleanupStep{}, false + } + if step, pending := pendingResumeCleanupStepForCursor(&driver.poll.resolve); pending { + return step, true + } + return pendingResumeCleanupStepForCursor(&driver.local.resolve) +} + // CommitResumeCleanupStep completes the exact outstanding typed runtime step. // small is invalid for source-neutral hooks. Channel hooks alone publish the // closed runtime wait status for their physical winner. @@ -570,7 +580,7 @@ func advanceResumeCleanupCore( record *WaitSetRecord, plan *ResumeCleanupPlan, ) (finalized bool, ok bool) { - if sources == nil || p == nil || !validExecutorSourceSet(sources, p) || + if sources == nil || p == nil || !validExecutorSourceSetHeader(sources, p) || !validResumeCleanupPlan(record, plan) { return false, false } diff --git a/runtime/internal/coro/resume_packet.go b/runtime/internal/coro/resume_packet.go index d26e5b275e..ecb2f137b8 100644 --- a/runtime/internal/coro/resume_packet.go +++ b/runtime/internal/coro/resume_packet.go @@ -334,7 +334,7 @@ func materializePollResume( // logical detach and before ready-queue publication. No source identity is // written to the packet after cleanup succeeds. func materializeSingleResumePacket(sources *ExecutorSourceSet, p *P, record *WaitSetRecord) bool { - if sources == nil || !validExecutorSourceSet(sources, p) || record == nil || + if sources == nil || !validExecutorSourceSetHeader(sources, p) || record == nil || !validActiveWaitSetRecordFast(p, record) || record.work != waitSetWorkResolving || record.g.park.phase != parkReady || record.resumeKind != resumeBindingSingle || !validBoundResumePacket((*ResumePacket)(record.resume), record.ticket) { diff --git a/runtime/internal/coro/run_slice.go b/runtime/internal/coro/run_slice.go index 300dca0f76..c60fd264ce 100644 --- a/runtime/internal/coro/run_slice.go +++ b/runtime/internal/coro/run_slice.go @@ -49,7 +49,8 @@ func validExecutorRunCursor(cursor *executorRunCursor, p *P) bool { } func emptyExecutorRunCursor(driver *ExecutorDriver) bool { - return driver != nil && driver.run == (executorRunCursor{}) + return driver != nil && driver.run == (executorRunCursor{}) && + emptyOwnerLocalCompletion(&driver.local) } // EnterExecutorRunCompatibility is the only supported stable-idle switch from @@ -61,6 +62,7 @@ func emptyExecutorRunCursor(driver *ExecutorDriver) bool { func EnterExecutorRunCompatibility(driver *ExecutorDriver) bool { if !validExecutorDriver(driver) || driver.state != executorDriverActive || driver.run.issued != ActionInvalid || driver.poll.phase != executorPollIdle || + !emptyOwnerLocalCompletion(&driver.local) || !idleExecutorScheduler(driver.p) { return false } @@ -109,9 +111,9 @@ func serviceExecutorRunSource(driver *ExecutorDriver, now int64, withDeadline bo var progress ExecutorPollProgress var ok bool if withDeadline { - _, progress, ok = pollExecutorSliceAt(driver, now, true, 1) + _, progress, ok = pollBoundExecutorSliceAt(driver, now, true, 1) } else { - _, progress, ok = pollExecutorSliceAt(driver, 0, false, 1) + _, progress, ok = pollBoundExecutorSliceAt(driver, 0, false, 1) } if !ok || progress.Used != 1 { return ExecutorRunStep{}, false @@ -130,6 +132,39 @@ func serviceExecutorRunSource(driver *ExecutorDriver, now int64, withDeadline bo return ExecutorRunStep{Kind: ExecutorRunStepSource, Poll: progress}, true } +// serviceExecutorRunLocal advances one completion which was published by the +// currently owning G on this same P. It deliberately uses the ordinary Source +// host step so target-side ready distribution and readyDebt keep one common +// boundary, but AtomicResolve identifies that no catalog scan or executor +// acknowledgement was performed. +func serviceExecutorRunLocal(driver *ExecutorDriver) (ExecutorRunStep, bool) { + var resolved publishedEpochResolveStep + if !resolveOwnerLocalCompletionStep(driver, &resolved) || + resolved.applyVisits < 0 || resolved.promoted < 0 { + return ExecutorRunStep{}, false + } + complete := resolved.complete + if complete { + if ownerLocalCompletionPending(driver) { + return ExecutorRunStep{}, false + } + driver.run.blocked = false + driver.run.actionsSinceSource = 0 + if runnableForOSThreadOwner(driver.p) { + driver.run.readyDebt = true + } + } + progress := ExecutorPollProgress{ + Used: 1, + ApplyVisits: uint32(resolved.applyVisits), + Promoted: uint32(resolved.promoted), + Complete: complete, + More: !complete || runnableForOSThreadOwner(driver.p) || executorRunExternalSourceRequested(driver), + AtomicResolve: true, + } + return ExecutorRunStep{Kind: ExecutorRunStepSource, Poll: progress}, true +} + // CommitExecutorRunSourceDistribution closes the optional target-side ready // distribution boundary after one complete Source reduction. Source completion // records readyDebt before returning so a hot source cannot starve a newly @@ -147,13 +182,13 @@ func CommitExecutorRunSourceDistribution(driver *ExecutorDriver, distributed boo return false } if !distributed || runnableForOSThreadOwner(driver.p) { - return validExecutorDriver(driver) + return validExecutorDriverHeader(driver) } if !driver.run.readyDebt { return false } driver.run.readyDebt = false - if validExecutorDriver(driver) { + if validExecutorDriverHeader(driver) { return true } // Preserve the fail-closed diagnostic state if an unrelated invariant was @@ -189,7 +224,7 @@ func dispatchExecutorRunReady(driver *ExecutorDriver) (ExecutorRunStep, bool) { } func nextExecutorRunStepAt(driver *ExecutorDriver, now int64, withDeadline bool) (ExecutorRunStep, bool) { - if !validExecutorDriver(driver) || driver.state != executorDriverActive || + if !validExecutorDriverHeader(driver) || driver.state != executorDriverActive || driver.sources.usesMonotonicTime() != withDeadline || withDeadline && now < 0 || driver.run.issued != ActionInvalid { return ExecutorRunStep{}, false @@ -217,11 +252,21 @@ func nextExecutorRunStepAt(driver *ExecutorDriver, now int64, withDeadline bool) // Once epoch A starts, acknowledgement and epoch B finish before any G. if driver.poll.phase != executorPollIdle { - if cleanup, pending := pendingResumeCleanupStep(driver); pending { + if cleanup, pending := pendingResumeCleanupStepForCursor(&driver.poll.resolve); pending { return ExecutorRunStep{Kind: ExecutorRunStepMaterialize, Cleanup: cleanup}, true } return serviceExecutorRunSource(driver, now, withDeadline) } + // An owner-local completion has already published its exact typed source + // fact. Resolve it before starting an unrelated external A/ack/B epoch and + // before dispatching another G; typed cleanup still returns through the + // ordinary direct-runtime Materialize boundary. + if ownerLocalCompletionPending(driver) { + if cleanup, pending := pendingResumeCleanupStepForCursor(&driver.local.resolve); pending { + return ExecutorRunStep{Kind: ExecutorRunStepMaterialize, Cleanup: cleanup}, true + } + return serviceExecutorRunLocal(driver) + } if driver.run.readyDebt { if runnableForOSThreadOwner(p) { return dispatchExecutorRunReady(driver) @@ -263,11 +308,15 @@ func NextExecutorRunStepAt(driver *ExecutorDriver, now int64) (ExecutorRunStep, // reduction will enter a managed llvm.coro.resume. It is deliberately // observational and must be called before NextExecutorRunStep marks the // physical Action interval issued. A target can therefore acquire its -// process-level execution permit without returning across an issued action or +// process-level P lease without returning across an issued action or // teaching the target-neutral driver about threads and GOMAXPROCS. func ExecutorRunManagedResumePending(driver *ExecutorDriver) (pending, ok bool) { - if !validExecutorDriver(driver) || driver.state != executorDriverActive || - driver.run.issued != ActionInvalid { + // This is only a pre-step lease probe. NextExecutorRunStep performs the + // complete owner-header validation before it can issue an action, so the + // probe needs only the exact active P back-pointer and issued-state gate. + if driver == nil || driver.magic != executorDriverMagic || driver.state != executorDriverActive || + driver.p == nil || driver.p.executor != driver || + preemptLoad(&driver.p.executorMode) != executorModeBound || driver.run.issued != ActionInvalid { return false, false } p := driver.p @@ -297,14 +346,14 @@ func ExecutorRunManagedResumePending(driver *ExecutorDriver) (pending, ok bool) // and retains no platform state; asynchronous producers must still publish a // durable source and use the registry request/doorbell protocol. func RequestExecutorSourceService(driver *ExecutorDriver) bool { - if !validExecutorDriver(driver) || driver.state != executorDriverActive || + if !validExecutorDriverHeader(driver) || driver.state != executorDriverActive || driver.run.issued != ActionInvalid || driver.poll.phase != executorPollIdle || !idleExecutorScheduler(driver.p) { return false } driver.run.sourceMore = true driver.run.blocked = false - return validExecutorDriver(driver) + return validExecutorDriverHeader(driver) } // ExecutorOwnerWaitPending closes the running-owner-to-physical-wait window @@ -313,13 +362,13 @@ func RequestExecutorSourceService(driver *ExecutorDriver) bool { // G, source fact, registry request, or scheduler request makes blocking // unnecessary, while a later producer observes that marker and rings. func ExecutorOwnerWaitPending(driver *ExecutorDriver) (pending, ok bool) { - if !validExecutorDriver(driver) || driver.state != executorDriverActive || + if !validExecutorDriverHeader(driver) || driver.state != executorDriverActive || driver.run.issued != ActionInvalid || driver.poll.phase != executorPollIdle || !idleExecutorScheduler(driver.p) { return false, false } return runnableForOSThreadOwner(driver.p) || - executorRunSourceRequested(driver), true + ownerLocalCompletionPending(driver) || executorRunSourceRequested(driver), true } func completedExecutorRunAction(p *P, g *G, action Action) bool { @@ -352,7 +401,7 @@ func completedExecutorRunAction(p *P, g *G, action Action) bool { // the completed G nor its old handle, so a runtime may reclaim a dynamic G // immediately after a successful ActionComplete commit. func commitExecutorRunAction(driver *ExecutorDriver, g *G, next Action, placement executorRunQueuePlacement) bool { - if !validExecutorDriver(driver) || driver.state != executorDriverActive || + if !validExecutorDriverHeader(driver) || driver.state != executorDriverActive || driver.run.issued == ActionInvalid || g == nil || !validOSThreadPeerActionCommit(driver.p, g) { return false diff --git a/runtime/internal/runtime/coro_channel_adapter_test.go b/runtime/internal/runtime/coro_channel_adapter_test.go index 2b6c51bccd..a4e3950981 100644 --- a/runtime/internal/runtime/coro_channel_adapter_test.go +++ b/runtime/internal/runtime/coro_channel_adapter_test.go @@ -255,6 +255,11 @@ func yieldCoroChannelAdapterFrame(t *testing.T, p *coro.P, frame *coroChannelAda } func pollCoroChannelAdapterExecutor(t *testing.T, driver *coro.ExecutorDriver) { + t.Helper() + _ = pollCoroChannelAdapterExecutorProgress(t, driver) +} + +func pollCoroChannelAdapterExecutorProgress(t *testing.T, driver *coro.ExecutorDriver) (atomicResolve bool) { t.Helper() for step := 0; ; step++ { runStep, ok := coro.NextExecutorRunStep(driver) @@ -263,11 +268,12 @@ func pollCoroChannelAdapterExecutor(t *testing.T, driver *coro.ExecutorDriver) { } switch runStep.Kind { case coro.ExecutorRunStepSource: + atomicResolve = atomicResolve || runStep.Poll.AtomicResolve if runStep.Poll.Complete { if !coro.EnterExecutorRunCompatibility(driver) { t.Fatalf("leave bounded channel adapter executor at step %d", step) } - return + return atomicResolve } case coro.ExecutorRunStepMaterialize: if !coroMaterializeResumeCleanupStepV1(runStep.Cleanup) { @@ -480,7 +486,10 @@ func TestCoroChannelAdapterCleanupCursorBoundsPeerWork(t *testing.T) { func TestCoroChannelAdapterPairCommitAndResume(t *testing.T) { coroCurrentTaskRouteTestV1 = 7 - defer func() { coroCurrentTaskRouteTestV1 = 0 }() + defer func() { + coroCurrentTaskTestV1 = nil + coroCurrentTaskRouteTestV1 = 0 + }() p := new(coro.P) driver := new(coro.ExecutorDriver) handle, ok := coroProgramExecutorRegistryState.Register() @@ -642,6 +651,9 @@ func TestCoroChannelAdapterPairCommitAndResume(t *testing.T) { t.Fatalf("channel select waiters not published: first=%p second=%p", selectChannels[0].recvq.first, selectChannels[1].recvq.first) } + // Clear the mandatory initial parked-set visit so a later same-P matcher + // can publish the exact selected endpoint directly to the owner-local FIFO. + pollCoroChannelAdapterExecutor(t, driver) deferredG, deferredOK := coro.NextRunnable(p) if !deferredOK || deferredG == nil || deferredG == selectFrame.g { t.Fatalf("dequeue unrelated ready G before select completion = (%p, %t)", deferredG, deferredOK) @@ -656,12 +668,18 @@ func TestCoroChannelAdapterPairCommitAndResume(t *testing.T) { t.Fatalf("unexpected direct select sender G %p", deferredG) } selectedValue := uint32(0xa5b6c7d8) - var directSendState CoroChanParkV1 directSendAction := activateCoroChannelAdapterFrame(t, p, directSender) - parkCoroChannelAdapterFrame( - t, p, directSender, directSendAction, selectChannels[1], unsafe.Pointer(&selectedValue), &directSendState, true, - ) - pollCoroChannelAdapterExecutor(t, driver) + coroCurrentTaskTestV1 = directSender.g + coroCurrentTaskRouteTestV1 = 1 + if !CoroChanTrySend(selectChannels[1], unsafe.Pointer(&selectedValue), int(unsafe.Sizeof(selectedValue))) { + t.Fatal("same-P direct send did not match channel selector") + } + coroCurrentTaskTestV1 = nil + coroCurrentTaskRouteTestV1 = 7 + yieldCoroChannelAdapterFrame(t, p, directSender, directSendAction) + if !pollCoroChannelAdapterExecutorProgress(t, driver) { + t.Fatal("same-P channel match did not use owner-local completion reduction") + } completed := map[*coro.G]bool{} for len(completed) != 2 { next, nextOK := coro.NextRunnable(p) @@ -694,10 +712,7 @@ func TestCoroChannelAdapterPairCommitAndResume(t *testing.T) { selectFrame.header.Lifecycle = uint16(coro.FrameActive) yieldCoroChannelAdapterFrame(t, p, selectFrame, selectAction) case directSender.g: - directSendAction, directStatus := resumeCoroChannelAdapterFrame(t, p, directSender, &directSendState) - if directStatus != coroChanResumeSendOK { - t.Fatalf("direct select sender resume status = %d, want %d", directStatus, coroChanResumeSendOK) - } + directSendAction = activateCoroChannelAdapterFrame(t, p, directSender) yieldCoroChannelAdapterFrame(t, p, directSender, directSendAction) default: t.Fatalf("unexpected completed select pair G %p", next) diff --git a/runtime/internal/runtime/coro_current_task_route.go b/runtime/internal/runtime/coro_current_task_route.go index f04cd1e53c..dc7c1f11b8 100644 --- a/runtime/internal/runtime/coro_current_task_route.go +++ b/runtime/internal/runtime/coro_current_task_route.go @@ -25,18 +25,23 @@ import "github.com/goplus/llgo/runtime/internal/coro" // foreign-thread callbacks, timers, IO reactors, and teardown paths // deliberately return route zero. A synchronous same-G C-to-Go reentry is // still inside that managed resume and may inherit its route. -func coroCurrentTaskRouteV1() coro.RouteID { +func coroCurrentTaskV1() (*coro.G, coro.RouteID) { gp := getg() if gp == nil || gp.startfn != nil || gp.startarg == nil { - return 0 + return nil, 0 } task := (*coro.G)(gp.startarg) if ctx := (*coroRuntimeContext)(coro.TaskLocal(task)); ctx != gp.context || !validCoroRuntimeTaskContext(task, ctx) { - return 0 + return nil, 0 } _, _, route, current := coro.CurrentExecutorDriver(task) if !current { - return 0 + return nil, 0 } + return task, route +} + +func coroCurrentTaskRouteV1() coro.RouteID { + _, route := coroCurrentTaskV1() return route } diff --git a/runtime/internal/runtime/coro_current_task_route_default.go b/runtime/internal/runtime/coro_current_task_route_default.go index ad8eef3e87..b9eab2eaf4 100644 --- a/runtime/internal/runtime/coro_current_task_route_default.go +++ b/runtime/internal/runtime/coro_current_task_route_default.go @@ -23,6 +23,11 @@ import "github.com/goplus/llgo/runtime/internal/coro" // Targets without a multi-route fleet have no useful producer-locality // destination. Keeping this a compile-time zero also avoids treating the // command/host executor's route-1 identity as a migration contract. +func coroCurrentTaskV1() (*coro.G, coro.RouteID) { + return nil, 0 +} + func coroCurrentTaskRouteV1() coro.RouteID { - return 0 + _, route := coroCurrentTaskV1() + return route } diff --git a/runtime/internal/runtime/coro_current_task_route_test_adapter.go b/runtime/internal/runtime/coro_current_task_route_test_adapter.go index 128fca22f8..c44a0811cf 100644 --- a/runtime/internal/runtime/coro_current_task_route_test_adapter.go +++ b/runtime/internal/runtime/coro_current_task_route_test_adapter.go @@ -20,8 +20,16 @@ package runtime import "github.com/goplus/llgo/runtime/internal/coro" -var coroCurrentTaskRouteTestV1 coro.RouteID +var ( + coroCurrentTaskTestV1 *coro.G + coroCurrentTaskRouteTestV1 coro.RouteID +) + +func coroCurrentTaskV1() (*coro.G, coro.RouteID) { + return coroCurrentTaskTestV1, coroCurrentTaskRouteTestV1 +} func coroCurrentTaskRouteV1() coro.RouteID { - return coroCurrentTaskRouteTestV1 + _, route := coroCurrentTaskV1() + return route } diff --git a/runtime/internal/runtime/coro_execution_quota_native_llgo.go b/runtime/internal/runtime/coro_execution_quota_native_llgo.go index 26cf5eb7a5..8cd307ceb0 100644 --- a/runtime/internal/runtime/coro_execution_quota_native_llgo.go +++ b/runtime/internal/runtime/coro_execution_quota_native_llgo.go @@ -47,13 +47,17 @@ func coroNativeFleetExecutionDomainV1( domain.handle.Route == uint32(route) } -func coroNativeFleetRingExecutionWaitersV1() bool { +func coroNativeFleetRingExecutionWaitersV1(waiters uint32) bool { state := &coroNativeFleetV1State if state.lifecycle != coroNativeFleetActiveV1 || - state.domainCount != coroNativeFleetDomainCapacityV1 { + state.domainCount != coroNativeFleetDomainCapacityV1 || + waiters>>state.domainCount != 0 { return false } for index := uint32(0); index < state.domainCount; index++ { + if waiters&(uint32(1)<