diff --git a/.github/workflows/coroutine.yml b/.github/workflows/coroutine.yml index 0aa8bac8fd..f1b2d4f25b 100644 --- a/.github/workflows/coroutine.yml +++ b/.github/workflows/coroutine.yml @@ -108,6 +108,7 @@ jobs: ./internal/runtime/coro_execution_quota_default.go \ ./internal/runtime/coro_ready_distribution_default.go \ ./internal/runtime/coro_native_fleet.go \ + ./internal/runtime/coro_native_deferred_replacement_default.go \ ./internal/runtime/coro_native_fleet_reactor.go \ ./internal/runtime/coro_native_atomic_host.go \ ./internal/runtime/coro_native_fleet_test.go \ diff --git a/doc/coro-performance-baseline.md b/doc/coro-performance-baseline.md index be8455a887..a6e8aebaf5 100644 --- a/doc/coro-performance-baseline.md +++ b/doc/coro-performance-baseline.md @@ -1805,3 +1805,46 @@ runtime package, 20-repeat native standby, and focused compiler syscall/worker tests also pass. The next file-performance target is the remaining coroutine frame chain through `io.ReadFull`, `os.File` and `internal/poll`, not another file-specific runtime path. + +### Request-driven deferred replacement checkpoint + +The follow-up on 2026-08-14 closes the direct-only cross-route liveness gap +without restoring eager compensation. A blocking native syscall now reserves +the exact logical handoff and a replacement directory slot, releases its +managed-execution permit, and arms one pointer-free 32-bit dispatch gate. It +does not consume a physical M unless a durable executor request wins the gate. +Every native source uses one common request tail: preserve the existing +doorbell transport first, then activate an armed replacement for an accepted +request. A quick syscall return withdraws the arm with one CAS; a request winner +uses the existing generation-bound cancel, return, and strong-recycle path. + +The new linked gate constructs the dependency directly. Route A's sole current +M blocks in a real compiler-certified `llgo.syscall`; route B completes a direct +channel rendezvous for a waiter on A; the accepted request starts A's deferred +replacement; the resumed waiter releases the C call. There is no timer, poll, +worker, or additional runtime waiter in the cycle. The scenario and the nearby +same-route/timer/poll/nested/retirement replacement set pass five repeated runs. + +The exact comparison parent is merge `a9bb968259c791c3eaaefd8a0db9a327647449f0`. +The parent and candidate use the same LLVM 22 compiler/workload artifacts and +31-process medians. A background Docker build was active on the machine, so +these measurements are a regression guard rather than a new Go comparison: + +| 500-operation workload | parent | deferred candidate | delta | +| --- | ---: | ---: | ---: | +| direct `syscall.Seek/Write/Seek/Read` | 2.853167 ms | 3.043458 ms | +6.7% | +| standard `os.File`/`io.ReadFull` chain | 7.441375 ms | 7.461333 ms | effectively neutral | + +The direct boundary pays roughly 95 ns per syscall in this fixture for logical +handoff/slot prepare plus the uncontended Arm/Withdraw race; no physical thread +is started. The standard-library path is unchanged within process noise. The +stripped artifact grows from 7,050,672 to 7,129,760 bytes (+79,088, about 1.1%). +This is accepted as the temporary correctness gate; follow-up profiling should +reduce the prepare/rollback cost without weakening the direct-only liveness +proof. + +One pre-existing `LockedOrdinarySuspend` stress case remains flaky: the merge +parent failed 1 of 20 isolated runs with the same bare `abort trap`, while the +candidate reproduced a similar rate. It does not exercise the deferred syscall +path, so this checkpoint does not claim to fix or regress that independent +native-fleet race. diff --git a/doc/llvm-coro-runtime-design.md b/doc/llvm-coro-runtime-design.md index 6017a17aff..5cb8bcd01c 100644 --- a/doc/llvm-coro-runtime-design.md +++ b/doc/llvm-coro-runtime-design.md @@ -3178,13 +3178,38 @@ event source 只要复用 common park/detach 协议，就自动进入这项判 未匹配 direct-channel waiter 不能独立完成，因此不能等同 timer/poll，否则一个 常驻 runtime waiter 会使每个短 syscall 都启动 M。producer 完成匹配后必须先发布 -typed completion，再发布 exact executor request。完整的跨 route 闭环还要求：若 -目标 route 正处于“已释放 lease、原 M 阻塞、尚无 replacement”的窗口，这个 request -必须能按需激活预留的 replacement。该 follow-up 应复用 -`ExecutionDomainHandoff` 和 active-M directory；不得重新把所有 channel waiter -加入 eager gate，也不得引入第二套 channel 调度器。在此闭环验收前，timer/poll -驱动的 blocking compensation 已验证，但“remote producer 唤醒目标 G，而该 G -反过来解除原 syscall”的 direct-only 环不能标为完成。 +typed completion，再发布 exact executor request。该 request 现在通过公共物理 tail +按需激活预留的 replacement，不把未匹配 waiter 加入 eager gate，也不 +引入第二套 channel 调度器。 + +这个闭环复用 `ExecutionDomainHandoff` 和 active-M directory，只在稳定的 +parent M owner 中增加一个 pointer-free 32-bit `DeferredExecutorHandoff`： + +1. syscall owner 预先建立 exact logical handoff 并保留 replacement directory slot， + 但不请求物理 M； +2. 释放 managed-execution lease 后发布 `Armed`，再重查一次已存在 + demand，关闭 request-before-arm 窗口； +3. 每个 source 在事实与 executor request 都持久化后，先完成原有 + doorbell transport，再调用公共 activation tail； +4. accepted request 以单次 CAS 把 `Armed -> Starting`，随后发布 + `Queued` 或 `Started`；syscall 快速返回则以同一个 CAS 尝试 + `Armed -> Idle`； +5. request 胜出时继续复用 generation-bound return/cancel/recycle，完成 strong + recycle 后才清回 `Idle`。 + +`Starting` 只是一个有界的发布窗口，不是新的 scheduler wait state。slot 只是 +路由 hint；延迟到达的 accepted request 可与同一稳定 parent 下的下一次 +armed call 合并，而真正防止 stale physical owner 的 authority 仍是 +`ExecutionDomainHandoff` generation 和 replacement baton。该 gate 不保存 stack +boundary、G/P、LLVM coroutine handle 或函数地址，因此 request 从其他 route +到达时不需要反向解析编译器对象。 + +已接入该公共 tail 的生产者包括 Manual/Worker/Poll/TaskControl ingress、 +keyed post、direct channel completion、controlled timer、ready distribution 和普通 +executor request。linked E2E 已验证：目标 route 的 sole M 在真实 +`llgo.syscall` C 调用中阻塞，另一 route 的 direct-channel producer 发布 +completion/request，request-driven replacement 恢复目标 waiter，waiter 又解除 +原 syscall。该测试不借助 timer、poll、worker 或额外 runtime waiter。 物理 M 使用 generation 化的异步 standby request/cancel：request 只把 clean M 从 standby 转入 `DISPATCHING`，不等待 C->Go；原 syscall 快速返回时只能取消仍处于 @@ -3199,5 +3224,5 @@ owner 的硬门槛。 - sole-M blocking pipe + timer 必须继续前进； - TCP nonblocking poll 路径结构和性能不变； - queued-cancel、dispatch-wins、handoff return、shutdown 各竞态重复通过； -- direct-only 跨 route 依赖在 request-driven replacement 完成前保持显式未完成， - 不得用常驻 waiter eager compensation 掩盖。 +- direct-only 跨 route 依赖必须由 request-driven replacement 闭环，不得用 + 常驻 waiter eager compensation 掩盖。 diff --git a/internal/build/coro_native_fleet_e2e_test.go b/internal/build/coro_native_fleet_e2e_test.go index 7375b749af..0cf45eabb1 100644 --- a/internal/build/coro_native_fleet_e2e_test.go +++ b/internal/build/coro_native_fleet_e2e_test.go @@ -953,6 +953,125 @@ func Check() int32 { } ` +// coroNativeFleetDeferredDirectChannelReplacementE2ESource isolates the one +// liveness edge which cannot be inferred before a syscall starts: a source-free +// one-case channel waiter on the blocked route is completed later by a peer +// route. The raw llgo.syscall call must stay on the current M, while that +// durable direct-channel request starts the pre-armed replacement which runs +// the waiter and releases the C block. +const coroNativeFleetDeferredDirectChannelReplacementE2ESource = `package main + +import _ "unsafe" + +var Failed uint32 +var Ready chan uint32 +var Signal chan uint32 +var MainThread uintptr +var WaiterBefore uintptr +var WaiterAfter uintptr +var SenderThread uintptr +var Got uint32 + +//llgo:link funcPCABI0 llgo.funcPCABI0 +func funcPCABI0(fn any) uintptr + +//llgo:link raw llgo.syscall +func raw(fn uintptr) (uintptr, uintptr, uintptr) + +//llgo:coro contract foreign.v1 scope=declaration progress=may-block affinity=any-thread reentry=none memory=by-value abi=word-call.v1/0 +func libc___llgo_coro_native_fleet_e2e_block_v1_trampoline() + +//go:linkname schedulerYield llgo.coroYield +func schedulerYield() + +//llgo:coro noblock +//go:linkname threadID C.__llgo_coro_native_fleet_e2e_thread_id_v1 +func threadID() uintptr + +//llgo:coro noblock +//go:linkname resetState C.__llgo_coro_native_fleet_e2e_block_reset_v1 +func resetState() + +//llgo:coro noblock +//go:linkname isWaiting C.__llgo_coro_native_fleet_e2e_blocked_v1 +func isWaiting() uintptr + +//llgo:coro noblock +//go:linkname unblock C.__llgo_coro_native_fleet_e2e_release_v1 +func unblock() + +func directBlock() { + raw(funcPCABI0(libc___llgo_coro_native_fleet_e2e_block_v1_trampoline)) +} + +func waiter() { + thread := threadID() + if thread != MainThread { + go waiter() + return + } + WaiterBefore = thread + Ready <- 1 + Got = <-Signal + WaiterAfter = threadID() + unblock() +} + +func sender() { + thread := threadID() + if thread == MainThread { + go sender() + return + } + SenderThread = thread + for isWaiting() == 0 { + } + Signal <- 0xd1ec7 +} + +func Setup() { + Failed = 0 + Ready = make(chan uint32, 1) + Signal = make(chan uint32) + MainThread = threadID() + WaiterBefore = 0 + WaiterAfter = 0 + SenderThread = 0 + Got = 0 +} + +func main() { + resetState() + go waiter() + <-Ready + // The buffered Ready send lets waiter continue directly into its one-case + // receive. Yield is an explicit stable boundary which proves that receive + // has parked before this route enters C. + schedulerYield() + go sender() + before := threadID() + directBlock() + after := threadID() + if isWaiting() == 0 { + Failed = 151 + return + } + if before != MainThread || after != before { + Failed = 152 + return + } + if WaiterBefore != MainThread || WaiterAfter == 0 || + WaiterAfter == MainThread || SenderThread == 0 || + SenderThread == MainThread || Got != 0xd1ec7 { + Failed = 153 + } +} + +func Check() int32 { + return int32(Failed) +} +` + const coroNativeFleetLockedOrdinarySuspendE2ESource = `package main import _ "unsafe" @@ -1836,6 +1955,16 @@ func TestCoroNativeFleetSameRouteReplacementE2E(t *testing.T) { runCoroNativeFleetE2E(t, coroNativeFleetSameRouteReplacementE2ESource, "same-route-replacement", true, 1) } +func TestCoroNativeFleetDeferredDirectChannelReplacementE2E(t *testing.T) { + runCoroNativeFleetE2E( + t, + coroNativeFleetDeferredDirectChannelReplacementE2ESource, + "deferred-direct-channel-replacement", + true, + 2, + ) +} + func TestCoroNativeFleetLockedOrdinarySuspendE2E(t *testing.T) { runCoroNativeFleetE2E(t, coroNativeFleetLockedOrdinarySuspendE2ESource, "locked-ordinary-suspend", true, 2) } @@ -2006,6 +2135,7 @@ func buildCoroNativeFleetE2ERuntimeIsland(t *testing.T, temp string) []string { filepath.Join("..", "..", "runtime", "internal", "runtime", "coro_panic_trace_release.go"), filepath.Join("..", "..", "runtime", "internal", "runtime", "coro_spawn.go"), filepath.Join("..", "..", "runtime", "internal", "runtime", "coro_native_atomic_llgo.go"), + filepath.Join("..", "..", "runtime", "internal", "runtime", "coro_native_deferred_replacement_llgo.go"), filepath.Join("..", "..", "runtime", "internal", "runtime", "coro_native_fleet.go"), filepath.Join("..", "..", "runtime", "internal", "runtime", "coro_native_fleet_owner_llgo.go"), filepath.Join("..", "..", "runtime", "internal", "runtime", "coro_native_fleet_program_llgo.go"), diff --git a/internal/build/coro_spawn_native_e2e_test.go b/internal/build/coro_spawn_native_e2e_test.go index 2ee6c8368f..b7fe62e00f 100644 --- a/internal/build/coro_spawn_native_e2e_test.go +++ b/internal/build/coro_spawn_native_e2e_test.go @@ -490,6 +490,16 @@ func buildCoroSpawnNativeE2EUserSource( semantics, intrinsic, err := coroIntrinsicCallSiteSemanticsForTest(universe, call) return intrinsic && semantics.ElidesManagedCall(), err }, + ClassifyElidedCallCertificate: func(_ *ssa.Function, call ssa.CallInstruction) (string, error) { + certificate, certified, err := universe.CoroWorkerSyscallCertificate(call) + if err != nil || !certified { + return "", err + } + return certificate.ID, nil + }, + ClassifyStaticCodeAddressCallArgument: func(_ *ssa.Function, call ssa.CallInstruction, argument int) (bool, error) { + return universe.CoroStaticCodeAddressCallArgument(call, argument) + }, }) if err != nil { t.Fatal(err) diff --git a/internal/build/coro_stdlib_sync_acceptance_test.go b/internal/build/coro_stdlib_sync_acceptance_test.go index d186a22c2c..cb91d2be54 100644 --- a/internal/build/coro_stdlib_sync_acceptance_test.go +++ b/internal/build/coro_stdlib_sync_acceptance_test.go @@ -256,6 +256,7 @@ func assertCoroStdlibSyncRuntimeSelection(t *testing.T, fixture coroStdlibSyncFi t.Helper() const runtimePackage = "github.com/goplus/llgo/runtime/internal/runtime" required := map[string]bool{ + "coro_native_deferred_replacement_llgo.go": false, "coro_execution_quota_native_llgo.go": false, "coro_executor_driver_timer_llgo.go": false, "coro_keyed_registry_atomic_llgo.go": false, diff --git a/runtime/coro_runnable_distribution_source_test.go b/runtime/coro_runnable_distribution_source_test.go index 3a93df6fc0..dd917717bd 100644 --- a/runtime/coro_runnable_distribution_source_test.go +++ b/runtime/coro_runnable_distribution_source_test.go @@ -189,8 +189,9 @@ func TestCoroNativeFleetUsesFixedTopologyLogicalQuotaAndScalarPeerABI(t *testing for _, required := range []string{ "coroNativeMDirectoryCapacityV1 uint32 = 10_000", "coroNativeMPageCapacityV1 uint32 = 64", - "handoff coro.ExecutionDomainHandoff", - "resume coro.ExecutorResumeHandoff", + "handoff coro.ExecutionDomainHandoff", + "deferred coro.DeferredExecutorHandoff", + "resume coro.ExecutorResumeHandoff", "token uint32", "owners [coroNativeFleetDomainCapacityV1]coroNativeMOwnerV1", "pages [coroNativeMPageCountV1]unsafe.Pointer", @@ -292,7 +293,7 @@ func TestCoroNativeFleetUsesFixedTopologyLogicalQuotaAndScalarPeerABI(t *testing for _, required := range []string{ "coroNativeFleetActiveDomainForRouteV1(id.Route())", "coroNativeFleetV1State.fleet.PostManualAndRequest(id)", - "coroNativeFleetRequestNeedsRingV1(domain, result.Executor)", + "coroNativeFleetFinishExecutorRequestV1(domain, result.Executor)", } { if !strings.Contains(keyed, required) { t.Errorf("managed keyed completion lacks owner-joined post marker %q", required) diff --git a/runtime/coro_target_selection_test.go b/runtime/coro_target_selection_test.go index 28928d1e55..24f5e8ebcd 100644 --- a/runtime/coro_target_selection_test.go +++ b/runtime/coro_target_selection_test.go @@ -129,6 +129,7 @@ func TestCoroNativeFleetTargetBuildSelection(t *testing.T) { } for _, required := range []string{ "coro_target_native_fleet_llgo.go", + "coro_native_deferred_replacement_llgo.go", "coro_physical_thread_capacity_native_llgo.go", "coro_native_fleet_owner_llgo.go", "coro_native_fleet_program_llgo.go", @@ -142,6 +143,7 @@ func TestCoroNativeFleetTargetBuildSelection(t *testing.T) { } } for _, forbidden := range []string{ + "coro_native_deferred_replacement_default.go", "coro_target_native_llgo.go", "coro_ready_distribution_default.go", "coro_target_executor_retired_default.go", diff --git a/runtime/internal/coro/deferred_executor_handoff.go b/runtime/internal/coro/deferred_executor_handoff.go new file mode 100644 index 0000000000..6932156e3c --- /dev/null +++ b/runtime/internal/coro/deferred_executor_handoff.go @@ -0,0 +1,194 @@ +/* + * 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 + +// DeferredExecutorHandoffPhase is the physical-start state of one prepared +// execution-domain replacement. The logical ownership generation remains in +// ExecutionDomainHandoff; this gate only decides whether a durable executor +// request has made physical dispatch necessary. +type DeferredExecutorHandoffPhase uint8 + +const ( + DeferredExecutorHandoffIdle DeferredExecutorHandoffPhase = iota + DeferredExecutorHandoffArmed + DeferredExecutorHandoffStarting + DeferredExecutorHandoffQueued + DeferredExecutorHandoffStarted +) + +const ( + deferredExecutorHandoffPhaseBits = 3 + deferredExecutorHandoffPhaseMask = uint32(1<> deferredExecutorHandoffPhaseBits, + DeferredExecutorHandoffPhase(state & deferredExecutorHandoffPhaseMask) +} + +func deferredExecutorHandoffValid(slot uint32, phase DeferredExecutorHandoffPhase) bool { + if phase == DeferredExecutorHandoffIdle { + return slot == 0 + } + return slot != 0 && slot <= deferredExecutorHandoffSlotMask && + phase <= DeferredExecutorHandoffStarted +} + +// Arm publishes one prepared replacement after its managed-execution permit +// has been released. The zero value is reusable Idle. +func (handoff *DeferredExecutorHandoff) Arm(slot uint32) bool { + return handoff != nil && deferredExecutorHandoffValid(slot, DeferredExecutorHandoffArmed) && + preemptCompareAndSwap( + &handoff.state, + 0, + deferredExecutorHandoffPack(slot, DeferredExecutorHandoffArmed), + ) +} + +// BeginStart lets one accepted durable executor request become the unique +// physical-start publisher. A false result means there is no armed replacement +// to start; Idle and an already-starting/started request are both benign. +func (handoff *DeferredExecutorHandoff) BeginStart() (slot uint32, started bool) { + if handoff == nil { + return 0, false + } + state := preemptLoad(&handoff.state) + slot, phase := deferredExecutorHandoffUnpack(state) + if !deferredExecutorHandoffValid(slot, phase) || phase != DeferredExecutorHandoffArmed { + return 0, false + } + return slot, preemptCompareAndSwap( + &handoff.state, + state, + deferredExecutorHandoffPack(slot, DeferredExecutorHandoffStarting), + ) +} + +// PublishStart completes the unique Starting interval. queued records whether +// the cached-thread dispatch can still be withdrawn through its C token. +func (handoff *DeferredExecutorHandoff) PublishStart(slot uint32, queued bool) bool { + if handoff == nil || !deferredExecutorHandoffValid(slot, DeferredExecutorHandoffStarting) { + return false + } + phase := DeferredExecutorHandoffStarted + if queued { + phase = DeferredExecutorHandoffQueued + } + return preemptCompareAndSwap( + &handoff.state, + deferredExecutorHandoffPack(slot, DeferredExecutorHandoffStarting), + deferredExecutorHandoffPack(slot, phase), + ) +} + +// RetryStart returns a failed physical-start publication to Armed. The durable +// request caller must report failure; a later request may retry, while a +// concurrently returning owner may withdraw the restored arm. +func (handoff *DeferredExecutorHandoff) RetryStart(slot uint32) bool { + return handoff != nil && deferredExecutorHandoffValid(slot, DeferredExecutorHandoffStarting) && + preemptCompareAndSwap( + &handoff.state, + deferredExecutorHandoffPack(slot, DeferredExecutorHandoffStarting), + deferredExecutorHandoffPack(slot, DeferredExecutorHandoffArmed), + ) +} + +// Withdraw wins only before a durable request has begun physical dispatch. +func (handoff *DeferredExecutorHandoff) Withdraw(slot uint32) bool { + return handoff != nil && deferredExecutorHandoffValid(slot, DeferredExecutorHandoffArmed) && + preemptCompareAndSwap( + &handoff.state, + deferredExecutorHandoffPack(slot, DeferredExecutorHandoffArmed), + 0, + ) +} + +// Observe returns one atomic state snapshot. ok rejects an impossible packed +// value rather than allowing target code to treat corruption as Idle. +func (handoff *DeferredExecutorHandoff) Observe() ( + slot uint32, + phase DeferredExecutorHandoffPhase, + ok bool, +) { + if handoff == nil { + return 0, DeferredExecutorHandoffIdle, false + } + state := preemptLoad(&handoff.state) + slot, phase = deferredExecutorHandoffUnpack(state) + return slot, phase, deferredExecutorHandoffValid(slot, phase) +} + +// Complete clears a dispatched replacement only after its exact logical +// return and physical recycle have completed. +func (handoff *DeferredExecutorHandoff) Complete(slot uint32) bool { + if handoff == nil || slot == 0 || slot > deferredExecutorHandoffSlotMask { + return false + } + for _, phase := range [...]DeferredExecutorHandoffPhase{ + DeferredExecutorHandoffQueued, + DeferredExecutorHandoffStarted, + } { + if preemptCompareAndSwap( + &handoff.state, + deferredExecutorHandoffPack(slot, phase), + 0, + ) { + return true + } + } + return false +} + +// Idle reports whether no prepared physical dispatch remains. +func (handoff *DeferredExecutorHandoff) Idle() bool { + return handoff != nil && preemptLoad(&handoff.state) == 0 +} diff --git a/runtime/internal/coro/deferred_executor_handoff_test.go b/runtime/internal/coro/deferred_executor_handoff_test.go new file mode 100644 index 0000000000..48fee0840f --- /dev/null +++ b/runtime/internal/coro/deferred_executor_handoff_test.go @@ -0,0 +1,122 @@ +/* + * 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 + +import ( + "sync" + "testing" +) + +func TestDeferredExecutorHandoffWithdraw(t *testing.T) { + var handoff DeferredExecutorHandoff + if !handoff.Idle() || !handoff.Arm(17) || handoff.Arm(18) { + t.Fatal("arm deferred executor handoff") + } + if slot, phase, ok := handoff.Observe(); !ok || slot != 17 || phase != DeferredExecutorHandoffArmed { + t.Fatalf("armed snapshot = (%d, %d, %t)", slot, phase, ok) + } + if handoff.Withdraw(18) || !handoff.Withdraw(17) || !handoff.Idle() { + t.Fatal("withdraw deferred executor handoff") + } +} + +func TestDeferredExecutorHandoffStartOutcomes(t *testing.T) { + for _, queued := range []bool{false, true} { + var handoff DeferredExecutorHandoff + if !handoff.Arm(23) { + t.Fatal("arm deferred executor handoff") + } + slot, begun := handoff.BeginStart() + if !begun || slot != 23 || handoff.Withdraw(23) || + !handoff.PublishStart(slot, queued) { + t.Fatalf("publish deferred start queued=%t", queued) + } + want := DeferredExecutorHandoffStarted + if queued { + want = DeferredExecutorHandoffQueued + } + if gotSlot, phase, ok := handoff.Observe(); !ok || gotSlot != slot || phase != want { + t.Fatalf("started snapshot queued=%t = (%d, %d, %t)", queued, gotSlot, phase, ok) + } + if !handoff.Complete(slot) || !handoff.Idle() || handoff.Complete(slot) { + t.Fatalf("complete deferred start queued=%t", queued) + } + } +} + +func TestDeferredExecutorHandoffRetry(t *testing.T) { + var handoff DeferredExecutorHandoff + if !handoff.Arm(29) { + t.Fatal("arm deferred executor handoff") + } + slot, begun := handoff.BeginStart() + if !begun || slot != 29 || !handoff.RetryStart(slot) { + t.Fatal("retry deferred executor start") + } + if !handoff.Withdraw(slot) || !handoff.Idle() { + t.Fatal("withdraw retried deferred executor handoff") + } +} + +func TestDeferredExecutorHandoffStartWithdrawRace(t *testing.T) { + const iterations = 2_000 + for iteration := 0; iteration < iterations; iteration++ { + var handoff DeferredExecutorHandoff + if !handoff.Arm(31) { + t.Fatal("arm deferred executor handoff") + } + var wait sync.WaitGroup + wait.Add(2) + started := make(chan bool, 1) + withdrawn := make(chan bool, 1) + go func() { + defer wait.Done() + slot, ok := handoff.BeginStart() + if ok && !handoff.PublishStart(slot, true) { + t.Errorf("publish winning start at iteration %d", iteration) + } + started <- ok + }() + go func() { + defer wait.Done() + withdrawn <- handoff.Withdraw(31) + }() + wait.Wait() + startWon, withdrawWon := <-started, <-withdrawn + if startWon == withdrawWon { + t.Fatalf("race winners at iteration %d = start:%t withdraw:%t", iteration, startWon, withdrawWon) + } + if startWon { + if !handoff.Complete(31) { + t.Fatalf("complete race winner at iteration %d", iteration) + } + } + if !handoff.Idle() { + t.Fatalf("race did not return idle at iteration %d", iteration) + } + } +} + +func TestDeferredExecutorHandoffRejectsInvalidSlots(t *testing.T) { + var handoff DeferredExecutorHandoff + if handoff.Arm(0) || handoff.Arm(deferredExecutorHandoffSlotMask+1) { + t.Fatal("accepted invalid deferred executor slot") + } + if _, begun := handoff.BeginStart(); begun { + t.Fatal("started idle deferred executor handoff") + } +} diff --git a/runtime/internal/runtime/coro_keyed_post_native_llgo.go b/runtime/internal/runtime/coro_keyed_post_native_llgo.go index db0d190b82..cee3295a3e 100644 --- a/runtime/internal/runtime/coro_keyed_post_native_llgo.go +++ b/runtime/internal/runtime/coro_keyed_post_native_llgo.go @@ -35,8 +35,7 @@ func coroTargetPostKeyedOperationV2(id coro.OperationID) bool { // backend retirement wait forever. Raw C/host producers continue to use the // exported coroNativeFleetPostV1 path and its strong ingress join. result := coroNativeFleetV1State.fleet.PostManualAndRequest(id) - ringOK := !coroNativeFleetRequestNeedsRingV1(domain, result.Executor) || - domain.doorbell.Ring() return result.Route == coro.OperationRoutePosted && - coro.ExecutorRequestAccepted(result.Executor) && ringOK + coro.ExecutorRequestAccepted(result.Executor) && + coroNativeFleetFinishExecutorRequestV1(domain, result.Executor) } diff --git a/runtime/internal/runtime/coro_native_deferred_replacement_default.go b/runtime/internal/runtime/coro_native_deferred_replacement_default.go new file mode 100644 index 0000000000..8c0a37df8c --- /dev/null +++ b/runtime/internal/runtime/coro_native_deferred_replacement_default.go @@ -0,0 +1,25 @@ +//go:build (darwin || linux) && !baremetal && (!llgo || (llgo_coro && llgo_coro_native_pipe && (!llgo_coro_native_timer || coro_runtime_adapter_test))) + +/* + * 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 runtime + +// Profiles without the native timer fleet do not detach one physical owner +// per execution route, so they have no request-driven replacement to start. +func coroNativeMActivateDeferredReplacementV1(*coroNativeFleetDomainV1) bool { + return true +} diff --git a/runtime/internal/runtime/coro_native_deferred_replacement_llgo.go b/runtime/internal/runtime/coro_native_deferred_replacement_llgo.go new file mode 100644 index 0000000000..19e1a53ed6 --- /dev/null +++ b/runtime/internal/runtime/coro_native_deferred_replacement_llgo.go @@ -0,0 +1,111 @@ +//go:build llgo && llgo_coro && llgo_coro_native_pipe && llgo_coro_native_timer && (darwin || linux) && !baremetal && !coro_runtime_adapter_test + +/* + * 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 runtime + +import "github.com/goplus/llgo/runtime/internal/coro" + +// coroNativeMActivateDeferredReplacementV1 is the request-side half of a +// demand-free native syscall handoff. It resolves only stable directory state; +// neither the caller's stack boundary nor an LLVM coroutine handle is +// published. The exact execution-domain generation remains authoritative in +// parent.handoff and replacement.baton. +func coroNativeMActivateDeferredReplacementV1( + domain *coroNativeFleetDomainV1, +) bool { + if domain == nil || domain.driverOwnerV1() == nil || + domain.handle.Route == 0 || + domain.handle.Route > coroNativeFleetDomainCapacityV1 { + return false + } + parentSlot := coroNativeAtomicLoadV1( + &coroNativeMDirectoryV1State.active[domain.handle.Route-1], + ) + parent, ownerOK := coroNativeMOwnerForSlotV1(parentSlot) + if !ownerOK || parent == nil || parentSlot == 0 || + parent.handle != domain.handle { + return false + } + for { + slot, phase, valid := parent.deferred.Observe() + if !valid { + return false + } + switch phase { + case coro.DeferredExecutorHandoffIdle, + coro.DeferredExecutorHandoffStarting, + coro.DeferredExecutorHandoffQueued, + coro.DeferredExecutorHandoffStarted: + return true + case coro.DeferredExecutorHandoffArmed: + active, resolved, activeSlot, _, activeOK := + coroNativeMActiveOwnerV1(domain.driverOwnerV1()) + if !activeOK || active != parent || resolved != domain || + activeSlot != parentSlot { + return false + } + startedSlot, won := parent.deferred.BeginStart() + if !won { + continue + } + // BeginStart reloads the one-word gate and is authoritative. A delayed + // accepted request may safely coalesce into a later armed call on the + // same stable parent even when that call obtained a different slot. + slot = startedSlot + if slot > coroNativeMDirectoryCapacityV1 { + _ = parent.deferred.RetryStart(startedSlot) + return false + } + replacement, replacementOK := coroNativeMOwnerForSlotV1(slot) + released, releasedOK := parent.handoff.Released() + if !replacementOK || replacement == nil || + coroNativeMOwnerLifecycleLoadV1(replacement) != + coroNativeMOwnerReplacementPublishedV1 || + replacement.parentSlot != parentSlot || + replacement.predecessorSlot != 0 || + replacement.lineageRootSlot != slot || + coroNativeAtomicLoadV1(&replacement.lineageSlot) != slot || + replacement.handle != domain.handle || + replacement.thread != nil || replacement.self != nil || + replacement.token != 0 || replacement.resume.Detached() || + !replacement.handoff.Idle() || !replacement.deferred.Idle() || + !releasedOK || released != replacement.baton || + replacement.ownerEpoch != released.OwnerEpoch { + _ = parent.deferred.RetryStart(slot) + return false + } + queued, started := coroNativeMRequestPhysicalOwnerV1(replacement, slot) + if !started { + if !parent.deferred.RetryStart(slot) { + coroRuntimeAbort("native deferred replacement retry publication failed") + } + return false + } + if !parent.deferred.PublishStart(slot, queued) { + // A physical owner now owns the exact thread/token/slot record. No + // rollback is safe after this point; fail closed instead of losing + // the strong-return obligation. + coroRuntimeAbort("native deferred replacement start publication failed") + return false + } + return true + default: + return false + } + } +} diff --git a/runtime/internal/runtime/coro_native_fleet.go b/runtime/internal/runtime/coro_native_fleet.go index 69523af4eb..2de9d4e78c 100644 --- a/runtime/internal/runtime/coro_native_fleet.go +++ b/runtime/internal/runtime/coro_native_fleet.go @@ -509,6 +509,24 @@ func coroNativeFleetRequestNeedsRingV1( domain != nil && coroNativeAtomicLoadV1(&domain.borrowedWait) != 0 } +// coroNativeFleetFinishExecutorRequestV1 is the common physical tail after a +// source fact and its executor request are durable. Ring remains the ordinary +// idle/borrowed-wait transport. An accepted request additionally starts a +// prepared syscall replacement when the route's current M released its P +// without pre-existing demand. Non-native profiles provide a no-op target +// adapter, keeping source semantics independent of physical thread policy. +func coroNativeFleetFinishExecutorRequestV1( + domain *coroNativeFleetDomainV1, + result coro.ExecutorRequestResult, +) bool { + if coroNativeFleetRequestNeedsRingV1(domain, result) && + (domain == nil || !domain.doorbell.Ring()) { + return false + } + return !coro.ExecutorRequestAccepted(result) || + coroNativeMActivateDeferredReplacementV1(domain) +} + // coroNativeFleetPostV1 is the complete target-global completion ingress. The // caller supplies only the frozen two-word OperationID (plus a source-specific // POD result/control value). Route selects the stable ingress as the first @@ -575,15 +593,12 @@ func coroNativeFleetPostV1( return coroNativeFleetInvalidIngressV1() } - ringOK := true - if coroNativeFleetRequestNeedsRingV1(domain, result.Executor) { - ringOK = domain.doorbell.Ring() - } + requestOK := coroNativeFleetFinishExecutorRequestV1(domain, result.Executor) _, leaveOK := domain.ingress.Leave() // Leave is the absolute final domain access. The durable source fact is not // rolled back if a physical doorbell fails; fail-stop callers treat the // invalid return as target corruption while the fact remains observable. - if !ringOK || !leaveOK { + if !requestOK || !leaveOK { return coroNativeFleetInvalidIngressV1() } return result diff --git a/runtime/internal/runtime/coro_native_fleet_owner_llgo.go b/runtime/internal/runtime/coro_native_fleet_owner_llgo.go index 2229484649..5fbec94494 100644 --- a/runtime/internal/runtime/coro_native_fleet_owner_llgo.go +++ b/runtime/internal/runtime/coro_native_fleet_owner_llgo.go @@ -299,6 +299,7 @@ func coroNativeFleetPhysicalOwnersStopV1() bool { !coroNativeMJoinPhysicalOwnerV1(owner) || coroNativeMOwnerLifecycleLoadV1(owner) != coroNativeMOwnerReturnedV1 || owner.resume.Detached() || !owner.handoff.Idle() || + !owner.deferred.Idle() || !coroNativeMOwnerLifecycleCASV1( owner, coroNativeMOwnerReturnedV1, @@ -655,6 +656,7 @@ func __llgo_coro_native_fleet_owner_v2(slot uint32) uint32 { route := owner.handle.Route if route < 2 || route > coroNativeFleetV1State.domainCount || route-2 >= state.count || owner.self != nil || + !owner.deferred.Idle() || !coroNativeMOwnerLifecycleCASV1( owner, coroNativeMOwnerPeerPublishedV1, @@ -674,6 +676,7 @@ func __llgo_coro_native_fleet_owner_v2(slot uint32) uint32 { peer.handle != owner.handle || corofleet.OwnerReady(slot) != 0 || !coroNativeFleetRunPhysicalOwnerV1(peer.handle) || + !owner.deferred.Idle() || !coroNativeMOwnerLifecycleCASV1( owner, coroNativeMOwnerPeerActiveV1, diff --git a/runtime/internal/runtime/coro_native_m_owner_llgo.go b/runtime/internal/runtime/coro_native_m_owner_llgo.go index 90d8f9df96..762e08ec93 100644 --- a/runtime/internal/runtime/coro_native_m_owner_llgo.go +++ b/runtime/internal/runtime/coro_native_m_owner_llgo.go @@ -74,8 +74,9 @@ const ( // back at Idle. The physical pthread may then remain in the bounded C standby // cache independently of this logical directory slot. type coroNativeMOwnerV1 struct { - handoff coro.ExecutionDomainHandoff - resume coro.ExecutorResumeHandoff + handoff coro.ExecutionDomainHandoff + deferred coro.DeferredExecutorHandoff + resume coro.ExecutorResumeHandoff thread pthread.Thread self pthread.Thread @@ -314,6 +315,7 @@ func coroNativeMDirectoryStartV1(program coro.ExecutorFleetHandle) bool { owner := &directory.owners[route-1] handle, ok := coroNativeFleetHandleV1(route - 1) if !ok || handle.Route != route || !owner.handoff.Idle() || + !owner.deferred.Idle() || owner.resume.Detached() || owner.thread != nil || owner.self != nil || owner.token != 0 || owner.handle != (coro.ExecutorFleetHandle{}) || @@ -477,7 +479,8 @@ func coroNativeMAllocateSuccessorV1( owner.lineageRootSlot != 0 || coroNativeAtomicLoadV1(&owner.lineageSlot) != 0 || owner.ownerEpoch != 0 || - owner.resume.Detached() || !owner.handoff.Idle() { + owner.resume.Detached() || !owner.handoff.Idle() || + !owner.deferred.Idle() { coroNativeAtomicStoreV1(&owner.lifecycle, uint32(coroNativeMOwnerFailedV1)) return 0, nil, false } @@ -519,7 +522,7 @@ func coroNativeMClaimSuccessorV1( if !ownerOK || coroNativeMOwnerLifecycleLoadV1(owner) != coroNativeMOwnerSuccessorPublishedV1 || owner.predecessorSlot == 0 || owner.ownerEpoch == 0 || !owner.handle.Valid() || - owner.self != nil { + owner.self != nil || !owner.deferred.Idle() { return nil, nil, nil, false } self := pthread.Self() @@ -625,7 +628,8 @@ func coroNativeMAllocateReplacementV1( owner.lineageRootSlot != 0 || coroNativeAtomicLoadV1(&owner.lineageSlot) != 0 || owner.ownerEpoch != 0 || - owner.resume.Detached() || !owner.handoff.Idle() { + owner.resume.Detached() || !owner.handoff.Idle() || + !owner.deferred.Idle() { coroNativeAtomicStoreV1(&owner.lifecycle, uint32(coroNativeMOwnerFailedV1)) return 0, nil, false } @@ -652,7 +656,8 @@ func coroNativeMReleaseUnstartedReplacementV1(slot uint32) bool { coroNativeMOwnerPreparingV1, ) || owner.thread != nil || owner.self != nil || owner.resume.Detached() || owner.token != 0 || - !owner.handoff.Idle() || owner.predecessorSlot != 0 || + !owner.handoff.Idle() || !owner.deferred.Idle() || + owner.predecessorSlot != 0 || owner.lineageRootSlot != slot || coroNativeAtomicLoadV1(&owner.lineageSlot) != slot { return false @@ -668,6 +673,10 @@ func coroNativeMReleaseUnstartedReplacementV1(slot uint32) bool { } func coroNativeMClearReplacementStorageV1(owner *coroNativeMOwnerV1) { + if owner == nil || !owner.deferred.Idle() { + coroRuntimeAbort("native coroutine replacement retained deferred dispatch") + return + } owner.thread = nil owner.self = nil owner.token = 0 @@ -690,7 +699,8 @@ func coroNativeMRecycleReplacementV1(slot uint32) bool { !owner.baton.Valid() || owner.parentSlot == 0 || owner.lineageRootSlot == 0 || owner.ownerEpoch != owner.baton.OwnerEpoch || - owner.resume.Detached() || !owner.handoff.Idle() { + owner.resume.Detached() || !owner.handoff.Idle() || + !owner.deferred.Idle() { return false } switch coroNativeMOwnerLifecycleLoadV1(owner) { @@ -765,6 +775,7 @@ func coroNativeMWaitAndRecycleOSThreadSuspendV1( coroNativeAtomicLoadV1(&owner.lineageSlot) != slot || owner.ownerEpoch != owner.baton.OwnerEpoch || owner.resume.Detached() || !owner.handoff.Idle() || + !owner.deferred.Idle() || parent.handle != owner.handle { return false } @@ -825,7 +836,8 @@ func coroNativeMClaimReplacementV1( owner.parentSlot == 0 || owner.predecessorSlot != 0 || owner.lineageRootSlot != slot || coroNativeAtomicLoadV1(&owner.lineageSlot) != slot || - !owner.handle.Valid() || !owner.baton.Valid() { + !owner.handle.Valid() || !owner.baton.Valid() || + !owner.deferred.Idle() { return nil, nil, nil, false, false } self := pthread.Self() @@ -892,6 +904,7 @@ func coroNativeMFinishReplacementReturnV1( lifecycle := coroNativeMOwnerLifecycleLoadV1(owner) if owner == nil || parent == nil || owner.parentSlot == 0 || owner.baton.OwnerEpoch != owner.ownerEpoch || + !owner.deferred.Idle() || (lifecycle != coroNativeMOwnerReplacementActiveV1 && (lifecycle != coroNativeMOwnerSuccessorActiveV1 || !owner.baton.Valid())) || @@ -997,7 +1010,7 @@ func coroTargetRetirePhysicalOwnerV1( } owner, domain, slot, epoch, ownerOK := coroNativeMCurrentOwnerV1(driver) if !ownerOK || domain == nil || domain.pOwnerV1() != p || - slot == 0 || epoch == 0 { + slot == 0 || epoch == 0 || !owner.deferred.Idle() { _, _ = state.stop.Leave() return false } diff --git a/runtime/internal/runtime/coro_os_thread_foreign_llgo.go b/runtime/internal/runtime/coro_os_thread_foreign_llgo.go index 5d94b69059..9fe72850ba 100644 --- a/runtime/internal/runtime/coro_os_thread_foreign_llgo.go +++ b/runtime/internal/runtime/coro_os_thread_foreign_llgo.go @@ -41,14 +41,14 @@ type coroNativeForeignBoundaryV1 struct { domain *coroNativeFleetDomainV1 replacement *coroNativeMOwnerV1 - parentSlot uint32 - replacementSlot uint32 - ownerEpoch uint32 - baton coro.ExecutionDomainHandoffHandle - replacementQueued bool - replacementSkipped bool - callbackAcquired bool - active bool + parentSlot uint32 + replacementSlot uint32 + ownerEpoch uint32 + baton coro.ExecutionDomainHandoffHandle + replacementQueued bool + replacementDeferred bool + callbackAcquired bool + active bool } var ( @@ -82,7 +82,7 @@ func (boundary *coroNativeForeignBoundaryV1) startReplacementV1( boundary.parent == nil || boundary.domain == nil || boundary.replacement != nil || boundary.replacementSlot != 0 || boundary.baton.Valid() || boundary.replacementQueued || - boundary.replacementSkipped { + boundary.replacementDeferred || !boundary.parent.deferred.Idle() { return false } baton, begun := boundary.parent.handoff.Begin(boundary.ownerEpoch) @@ -131,6 +131,62 @@ func (boundary *coroNativeForeignBoundaryV1) startReplacementV1( return true } +// prepareDeferredReplacementV1 publishes the logical handoff and replacement +// directory slot without consuming a physical M. Arm happens only after the +// managed-execution permit is released. The post-Arm demand recheck closes the +// request-before-Arm window; a later request races Withdraw directly in the +// stable parent owner. +func (boundary *coroNativeForeignBoundaryV1) prepareDeferredReplacementV1() bool { + if boundary == nil || !boundary.active || boundary.driver == nil || + boundary.parent == nil || boundary.domain == nil || + boundary.replacement != nil || boundary.replacementSlot != 0 || + boundary.baton.Valid() || boundary.replacementQueued || + boundary.replacementDeferred || !boundary.parent.deferred.Idle() { + return false + } + baton, begun := boundary.parent.handoff.Begin(boundary.ownerEpoch) + if !begun { + return false + } + slot, replacement, allocated := coroNativeMAllocateReplacementV1( + boundary.parentSlot, + boundary.domain.handle, + baton, + ) + if !allocated { + rolledBack := boundary.parent.handoff.RequestReturn(baton) == + coro.ExecutionDomainHandoffReturnUnclaimed && + boundary.parent.handoff.Complete(baton) + _ = rolledBack + return false + } + if !coroTargetReleaseManagedExecutionV1(boundary.driver) { + // Release may already have published the free permit before a waiter + // doorbell failure. Nothing can safely restore ownership from this result. + coroRuntimeAbort("native deferred foreign execution quota release failed") + return false + } + boundary.replacement = replacement + boundary.replacementSlot = slot + boundary.baton = baton + boundary.replacementDeferred = true + if !boundary.parent.deferred.Arm(slot) { + coroRuntimeAbort("native deferred replacement arm failed") + return false + } + required, demandOK := + coro.ExecutorResumeHandoffCompensationRequired(&boundary.resume) + if !demandOK { + coroRuntimeAbort("native deferred replacement demand recheck failed") + return false + } + if required && !coroNativeMActivateDeferredReplacementV1(boundary.domain) { + coroRuntimeAbort("native deferred replacement demand start failed") + return false + } + return true +} + func (boundary *coroNativeForeignBoundaryV1) beginV1( task *coro.G, mode coro.ExecutorResumeHandoffMode, @@ -141,7 +197,7 @@ func (boundary *coroNativeForeignBoundaryV1) beginV1( boundary.domain != nil || boundary.replacement != nil || boundary.parentSlot != 0 || boundary.replacementSlot != 0 || boundary.ownerEpoch != 0 || boundary.baton.Valid() || - boundary.replacementQueued || boundary.replacementSkipped || + boundary.replacementQueued || boundary.replacementDeferred || boundary.callbackAcquired { return false } @@ -160,30 +216,9 @@ func (boundary *coroNativeForeignBoundaryV1) beginV1( boundary.ownerEpoch = ownerEpoch boundary.active = true if lazyCompensation { - required, demandOK := - coro.ExecutorResumeHandoffCompensationRequired(&boundary.resume) - if demandOK && !required { - if !coroTargetReleaseManagedExecutionV1(boundary.driver) { - // See startReplacementV1: a false result does not prove that the - // quota release itself failed before publication. - coroRuntimeAbort("native direct lazy execution quota release failed") - return false - } - boundary.replacementSkipped = true + if boundary.prepareDeferredReplacementV1() { return true } - if !demandOK || !required { - restored := coro.RestoreExecutorResume(&boundary.resume) - boundary.driver = nil - boundary.task = nil - boundary.parent = nil - boundary.domain = nil - boundary.parentSlot = 0 - boundary.ownerEpoch = 0 - boundary.active = false - _ = restored - return false - } } if boundary.startReplacementV1(true) { return true @@ -204,7 +239,7 @@ func (boundary *coroNativeForeignBoundaryV1) reclaimReplacementV1() bool { if boundary == nil || !boundary.active || boundary.driver == nil || boundary.parent == nil || boundary.domain == nil || boundary.replacement == nil || boundary.replacementSlot == 0 || - !boundary.baton.Valid() || boundary.replacementSkipped { + !boundary.baton.Valid() { return false } if boundary.replacementQueued { @@ -221,10 +256,12 @@ func (boundary *coroNativeForeignBoundaryV1) reclaimReplacementV1() bool { case 0: boundary.replacement.thread = nil boundary.replacement.token = 0 + slot := boundary.replacementSlot withdrawn := boundary.parent.handoff.RequestReturn(boundary.baton) == coro.ExecutionDomainHandoffReturnUnclaimed && boundary.parent.handoff.Complete(boundary.baton) && - coroNativeMReleaseUnstartedReplacementV1(boundary.replacementSlot) + coroNativeMReleaseUnstartedReplacementV1(slot) && + boundary.completeDeferredReplacementV1(slot) if !withdrawn { coroRuntimeAbort("native direct queued replacement withdrawal failed") } @@ -254,8 +291,9 @@ func (boundary *coroNativeForeignBoundaryV1) reclaimReplacementV1() bool { } returnResult := boundary.parent.handoff.RequestReturn(boundary.baton) if returnResult == coro.ExecutionDomainHandoffReturnUnclaimed { + slot := boundary.replacementSlot if !coroNativeMWaitAndRecycleOSThreadSuspendV1( - boundary.replacementSlot, + slot, boundary.replacement, boundary.parent, ) { @@ -264,6 +302,9 @@ func (boundary *coroNativeForeignBoundaryV1) reclaimReplacementV1() bool { if !boundary.parent.handoff.Complete(boundary.baton) { coroRuntimeAbort("native direct revoked handoff completion failed") } + if !boundary.completeDeferredReplacementV1(slot) { + coroRuntimeAbort("native direct revoked deferred dispatch completion failed") + } boundary.replacement = nil boundary.replacementSlot = 0 boundary.baton = coro.ExecutionDomainHandoffHandle{} @@ -323,6 +364,9 @@ func (boundary *coroNativeForeignBoundaryV1) reclaimReplacementV1() bool { if !boundary.parent.handoff.Complete(boundary.baton) { coroRuntimeAbort("native direct claimed handoff completion failed") } + if !boundary.completeDeferredReplacementV1(boundary.replacementSlot) { + coroRuntimeAbort("native direct claimed deferred dispatch completion failed") + } boundary.replacement = nil boundary.replacementSlot = 0 boundary.baton = coro.ExecutionDomainHandoffHandle{} @@ -330,6 +374,69 @@ func (boundary *coroNativeForeignBoundaryV1) reclaimReplacementV1() bool { return true } +func (boundary *coroNativeForeignBoundaryV1) completeDeferredReplacementV1( + slot uint32, +) bool { + if boundary == nil || !boundary.replacementDeferred { + return true + } + if boundary.parent == nil || slot == 0 || + !boundary.parent.deferred.Complete(slot) { + return false + } + boundary.replacementDeferred = false + return true +} + +// resolveDeferredReplacementV1 reconciles the returning syscall owner with +// the durable-request start race. Armed can be withdrawn without ever touching +// a physical thread. Starting is bounded by one request publisher; Queued and +// Started reuse the ordinary generation-bound reclaim path. +func (boundary *coroNativeForeignBoundaryV1) resolveDeferredReplacementV1() bool { + if boundary == nil || !boundary.replacementDeferred || + boundary.parent == nil || boundary.replacement == nil || + boundary.replacementSlot == 0 || !boundary.baton.Valid() { + return false + } + for { + slot, phase, valid := boundary.parent.deferred.Observe() + if !valid || slot != boundary.replacementSlot { + return false + } + switch phase { + case coro.DeferredExecutorHandoffArmed: + if !boundary.parent.deferred.Withdraw(slot) { + continue + } + rolledBack := boundary.parent.handoff.RequestReturn(boundary.baton) == + coro.ExecutionDomainHandoffReturnUnclaimed && + boundary.parent.handoff.Complete(boundary.baton) && + coroNativeMReleaseUnstartedReplacementV1(slot) + if !rolledBack { + return false + } + boundary.replacement = nil + boundary.replacementSlot = 0 + boundary.baton = coro.ExecutionDomainHandoffHandle{} + boundary.replacementQueued = false + boundary.replacementDeferred = false + return true + case coro.DeferredExecutorHandoffStarting: + if corofleet.Yield() != 0 { + return false + } + case coro.DeferredExecutorHandoffQueued: + boundary.replacementQueued = true + return true + case coro.DeferredExecutorHandoffStarted: + boundary.replacementQueued = false + return true + default: + return false + } + } +} + func (boundary *coroNativeForeignBoundaryV1) restartReplacementV1() bool { return boundary != nil && boundary.active && boundary.startReplacementV1(false) @@ -339,10 +446,17 @@ func (boundary *coroNativeForeignBoundaryV1) finishV1() bool { if boundary == nil || boundary.callbackAcquired { return false } - if !boundary.replacementSkipped && !boundary.reclaimReplacementV1() { + if boundary.replacementDeferred && !boundary.resolveDeferredReplacementV1() { + coroRuntimeAbort("native deferred foreign replacement resolution failed") + } + if boundary.replacement != nil && !boundary.reclaimReplacementV1() { coroRuntimeAbort("native direct foreign replacement reclaim failed") } - boundary.replacementSkipped = false + if boundary.replacement != nil || boundary.replacementSlot != 0 || + boundary.baton.Valid() || boundary.replacementQueued || + boundary.replacementDeferred { + return false + } if !coroTargetReenterManagedExecutionV1(boundary.driver) { coroRuntimeAbort("native direct foreign execution quota reentry failed") } @@ -417,7 +531,7 @@ func coroNativeForeignReentryRunV1( if boundary == nil || !boundary.active || !boundary.callbackAcquired || boundary.replacement != nil || boundary.replacementSlot != 0 || boundary.baton.Valid() || boundary.replacementQueued || - boundary.replacementSkipped || child == nil { + boundary.replacementDeferred || child == nil { coroRuntimeAbort("invalid synchronous foreign callback child") } var record coro.ForeignReentryRecord diff --git a/runtime/internal/runtime/coro_ready_distribution_fleet_llgo.go b/runtime/internal/runtime/coro_ready_distribution_fleet_llgo.go index 8b412ceec8..4c6afc673e 100644 --- a/runtime/internal/runtime/coro_ready_distribution_fleet_llgo.go +++ b/runtime/internal/runtime/coro_ready_distribution_fleet_llgo.go @@ -182,8 +182,8 @@ func coroTargetPublishReadyDistributionV1( if !valid { return coroTargetReadyDistributionFailV1("native ready distribution target route mismatch") } - if coroNativeFleetRequestNeedsRingV1(target, distribution.Request) && !target.doorbell.Ring() { - return coroTargetReadyDistributionFailV1("native ready distribution doorbell failed") + if !coroNativeFleetFinishExecutorRequestV1(target, distribution.Request) { + return coroTargetReadyDistributionFailV1("native ready distribution request tail failed") } return true } diff --git a/runtime/internal/runtime/coro_target_native_fleet_llgo.go b/runtime/internal/runtime/coro_target_native_fleet_llgo.go index fc6704df6d..bedefc1540 100644 --- a/runtime/internal/runtime/coro_target_native_fleet_llgo.go +++ b/runtime/internal/runtime/coro_target_native_fleet_llgo.go @@ -197,9 +197,9 @@ func coroTargetRequestExecutorV1(handle coro.ExecutorHandle) bool { result := coroProgramExecutorRegistryV1State.Request(handle) accepted := result == coro.ExecutorRequestPublished || result == coro.ExecutorRequestCoalesced || result == coro.ExecutorRequestIdleWake - ringOK := !coroNativeFleetRequestNeedsRingV1(domain, result) || domain.doorbell.Ring() + requestOK := coroNativeFleetFinishExecutorRequestV1(domain, result) _, leaveOK := domain.ingress.Leave() - return accepted && ringOK && leaveOK + return accepted && requestOK && leaveOK } // coroTargetRequestChannelOperationV1 routes a typed hchan commit to the exact @@ -217,7 +217,7 @@ func coroTargetRequestChannelOperationV1(id coro.OperationID) bool { result := coroNativeFleetV1State.fleet.RequestChannelExecutor(id) accepted := result == coro.ExecutorRequestPublished || result == coro.ExecutorRequestCoalesced || result == coro.ExecutorRequestIdleWake - return accepted && (!coroNativeFleetRequestNeedsRingV1(domain, result) || domain.doorbell.Ring()) + return accepted && coroNativeFleetFinishExecutorRequestV1(domain, result) } func coroTargetPublishDirectChannelCompletionV1( @@ -238,9 +238,9 @@ func coroTargetPublishDirectChannelCompletionV1( result := coroNativeFleetV1State.fleet.RequestExecutor(domain.handle) accepted := result == coro.ExecutorRequestPublished || result == coro.ExecutorRequestCoalesced || result == coro.ExecutorRequestIdleWake - ringOK := !coroNativeFleetRequestNeedsRingV1(domain, result) || domain.doorbell.Ring() + requestOK := coroNativeFleetFinishExecutorRequestV1(domain, result) _, leaveOK := domain.ingress.Leave() - return accepted && ringOK && leaveOK + return accepted && requestOK && leaveOK } // coroTargetRequestControlledTimerV2 requests the exact owner after @@ -263,9 +263,9 @@ func coroTargetRequestControlledTimerV2(route coro.RouteID) bool { result := coroNativeFleetV1State.fleet.RequestTimerExecutor(route) accepted := result == coro.ExecutorRequestPublished || result == coro.ExecutorRequestCoalesced || result == coro.ExecutorRequestIdleWake - ringOK := !coroNativeFleetRequestNeedsRingV1(domain, result) || domain.doorbell.Ring() + requestOK := coroNativeFleetFinishExecutorRequestV1(domain, result) _, leaveOK := domain.ingress.Leave() - return accepted && ringOK && leaveOK + return accepted && requestOK && leaveOK } func coroTargetPostTaskControlV1( diff --git a/runtime/poll_worker_source_test.go b/runtime/poll_worker_source_test.go index 5d1eef9cc6..3a182059a1 100644 --- a/runtime/poll_worker_source_test.go +++ b/runtime/poll_worker_source_test.go @@ -597,9 +597,27 @@ func TestRuntimeCoroWorkerBlockingCallHasOnlyGuardedSameMEntrance(t *testing.T) } } detach := strings.Index(entrance, "coro.DetachExecutorResume(") - start := strings.Index(entrance, "if boundary.startReplacementV1(true)") - leave := strings.Index(entrance, "coroTargetReleaseManagedExecutionV1(boundary.driver)") - create := strings.Index(entrance, "coroNativeMRequestPhysicalOwnerV1(replacement, slot)") + startEntry := strings.Index(entrance, "func (boundary *coroNativeForeignBoundaryV1) startReplacementV1(") + prepareEntry := strings.Index(entrance, "func (boundary *coroNativeForeignBoundaryV1) prepareDeferredReplacementV1()") + beginEntry := strings.Index(entrance, "func (boundary *coroNativeForeignBoundaryV1) beginV1(") + reclaimEntry := strings.Index(entrance, "func (boundary *coroNativeForeignBoundaryV1) reclaimReplacementV1()") + completeEntry := strings.Index(entrance, "func (boundary *coroNativeForeignBoundaryV1) completeDeferredReplacementV1(") + resolveEntry := strings.Index(entrance, "func (boundary *coroNativeForeignBoundaryV1) resolveDeferredReplacementV1()") + finishEntry := strings.Index(entrance, "func (boundary *coroNativeForeignBoundaryV1) finishV1()") + setTLSEntry := strings.Index(entrance, "func coroNativeForeignBoundarySetTLSV1(") + immediate, prepared, reclaim, finishBody := "", "", "", "" + if startEntry >= 0 && prepareEntry > startEntry { + immediate = entrance[startEntry:prepareEntry] + } + if prepareEntry >= 0 && beginEntry > prepareEntry { + prepared = entrance[prepareEntry:beginEntry] + } + if reclaimEntry >= 0 && completeEntry > reclaimEntry { + reclaim = entrance[reclaimEntry:completeEntry] + } + if finishEntry >= 0 && setTLSEntry > finishEntry { + finishBody = entrance[finishEntry:setTLSEntry] + } helper := strings.Index(entrance, "func coroNativeForeignWordCallV1(") begin, call := -1, -1 if helper >= 0 { @@ -612,15 +630,25 @@ func TestRuntimeCoroWorkerBlockingCallHasOnlyGuardedSameMEntrance(t *testing.T) call += helper } } - finish := strings.LastIndex(entrance, "boundary.finishV1()") - cancel := strings.Index(entrance, "corofleet.CancelReuseOwner(") - request := strings.LastIndex(entrance, "boundary.parent.handoff.RequestReturn(boundary.baton)") - recycle := strings.Index(entrance, "coroNativeMRecycleReplacementV1(returnedSlot)") - reenter := strings.LastIndex(entrance, "coroTargetReenterManagedExecutionV1(boundary.driver)") - restore := strings.LastIndex(entrance, "coro.RestoreExecutorResume(&boundary.resume)") - if detach < 0 || start <= detach || create <= leave || cancel <= create || - request < 0 || recycle <= request || reenter <= recycle || restore <= reenter || - helper < 0 || begin < helper || call <= begin || finish <= call { + helperFinish := strings.LastIndex(entrance, "boundary.finishV1()") + immediateRelease := strings.Index(immediate, "coroTargetReleaseManagedExecutionV1(boundary.driver)") + immediateCreate := strings.Index(immediate, "coroNativeMRequestPhysicalOwnerV1(replacement, slot)") + preparedRelease := strings.Index(prepared, "coroTargetReleaseManagedExecutionV1(boundary.driver)") + preparedArm := strings.Index(prepared, "boundary.parent.deferred.Arm(slot)") + preparedRecheck := strings.Index(prepared, "coro.ExecutorResumeHandoffCompensationRequired(&boundary.resume)") + reclaimCancel := strings.Index(reclaim, "corofleet.CancelReuseOwner(") + reclaimRequest := strings.Index(reclaim, "boundary.parent.handoff.RequestReturn(boundary.baton)") + reclaimRecycle := strings.Index(reclaim, "coroNativeMRecycleReplacementV1(returnedSlot)") + finishResolve := strings.Index(finishBody, "boundary.resolveDeferredReplacementV1()") + finishReclaim := strings.Index(finishBody, "boundary.reclaimReplacementV1()") + finishReenter := strings.Index(finishBody, "coroTargetReenterManagedExecutionV1(boundary.driver)") + finishRestore := strings.Index(finishBody, "coro.RestoreExecutorResume(&boundary.resume)") + if detach < beginEntry || immediateRelease < 0 || immediateCreate <= immediateRelease || + preparedRelease < 0 || preparedArm <= preparedRelease || preparedRecheck <= preparedArm || + reclaimCancel < 0 || reclaimRequest <= reclaimCancel || reclaimRecycle <= reclaimRequest || + resolveEntry <= completeEntry || finishResolve < 0 || finishReclaim <= finishResolve || + finishReenter <= finishReclaim || finishRestore <= finishReenter || + helper < 0 || begin < helper || call <= begin || helperFinish <= call { t.Errorf("%s does not bracket same-M C with detach/release/create/return/recycle/restore", runtimeCoroOSThreadForeignSource) } quota := readRuntimePollFile(t, "internal/runtime/coro_execution_quota_native_llgo.go") @@ -647,6 +675,64 @@ func TestRuntimeCoroWorkerBlockingCallHasOnlyGuardedSameMEntrance(t *testing.T) } } +func TestRuntimeNativeSyscallDeferredReplacementUsesStableRequestGate(t *testing.T) { + core := readRuntimePollFile(t, "internal/coro/deferred_executor_handoff.go") + for _, required := range []string{ + "type DeferredExecutorHandoff struct", + "state uint32", + "func (handoff *DeferredExecutorHandoff) Arm(slot uint32) bool", + "func (handoff *DeferredExecutorHandoff) BeginStart()", + "func (handoff *DeferredExecutorHandoff) Withdraw(slot uint32) bool", + "func (handoff *DeferredExecutorHandoff) Complete(slot uint32) bool", + } { + if !strings.Contains(core, required) { + t.Errorf("deferred executor core lacks stable gate marker %q", required) + } + } + + boundary := readRuntimePollFile(t, runtimeCoroOSThreadForeignSource) + prepare := strings.Index(boundary, "func (boundary *coroNativeForeignBoundaryV1) prepareDeferredReplacementV1()") + release, arm, recheck := -1, -1, -1 + if prepare >= 0 { + prepared := boundary[prepare:] + release = strings.Index(prepared, "coroTargetReleaseManagedExecutionV1(boundary.driver)") + arm = strings.Index(prepared, "boundary.parent.deferred.Arm(slot)") + recheck = strings.Index(prepared, "coro.ExecutorResumeHandoffCompensationRequired(&boundary.resume)") + } + if prepare < 0 || release < 0 || arm <= release || recheck <= arm { + t.Error("deferred native syscall does not release, arm, then recheck durable demand") + } + + request := readRuntimePollFile(t, "internal/runtime/coro_native_deferred_replacement_llgo.go") + for _, required := range []string{ + "coroNativeMActiveOwnerV1(domain.driverOwnerV1())", + "parent.deferred.BeginStart()", + "coroNativeMRequestPhysicalOwnerV1(replacement, slot)", + "parent.deferred.PublishStart(slot, queued)", + } { + if !strings.Contains(request, required) { + t.Errorf("deferred replacement request path lacks marker %q", required) + } + } + for _, forbidden := range []string{"unsafe.Pointer", "coroNativeForeignBoundaryV1"} { + if strings.Contains(request, forbidden) { + t.Errorf("deferred replacement request path retained stack/pointer coupling %q", forbidden) + } + } + + fleet := readRuntimePollFile(t, runtimeCoroNativeFleetSource) + tail := strings.Index(fleet, "func coroNativeFleetFinishExecutorRequestV1(") + ring, activate := -1, -1 + if tail >= 0 { + finish := fleet[tail:] + ring = strings.Index(finish, "domain.doorbell.Ring()") + activate = strings.Index(finish, "coroNativeMActivateDeferredReplacementV1(domain)") + } + if tail < 0 || ring < 0 || activate <= ring { + t.Error("durable native request tail does not ring before deferred physical activation") + } +} + func TestRuntimePthreadPrimitivesKeepPhysicalWaitSemantics(t *testing.T) { text := readRuntimePollFile(t, runtimePthreadSyncSource) for _, symbol := range []string{ diff --git a/runtime/time_sleep_source_test.go b/runtime/time_sleep_source_test.go index a33e326189..434bee9a3f 100644 --- a/runtime/time_sleep_source_test.go +++ b/runtime/time_sleep_source_test.go @@ -241,7 +241,7 @@ func TestControlledTimerOwnerUsesUnifiedTimerSource(t *testing.T) { "func coroTargetRequestControlledTimerV2(route coro.RouteID) bool", "domain.ingress.Enter()", "fleet.RequestTimerExecutor(route)", - "domain.doorbell.Ring()", + "coroNativeFleetFinishExecutorRequestV1(domain, result)", "domain.ingress.Leave()", } { if !strings.Contains(targetSource, contract) {