diff --git a/doc/coro-async-core-contract.md b/doc/coro-async-core-contract.md index c15dff1e25..ba8367d033 100644 --- a/doc/coro-async-core-contract.md +++ b/doc/coro-async-core-contract.md @@ -293,7 +293,7 @@ Source-specific submit保留在各自模块，但成功后必须返回统一 `Op 不设置中心化completion fact容量。`OperationRecord.completionPublished`本身是durable fact；所有source完成一个有界publication epoch后，再由各source枚举本轮affected operation并调用同一个park resolver。多个candidate指向同一ParkState时，第一次扫描本epoch完整sticky snapshot完成决策，后续重复项看到已进入detaching phase即可跳过。因此epoch开始前已durable的winner不依赖source顺序，也不需要每P固定大数组、batch overflow或全局transaction rollback；epoch进行期间并发到达的事实允许本轮或下一轮处理。 -高并发promotion使用直接park物理协程frame内的临时`WaitSetRecord`，不为所有G常驻增加`prevWait`或affected link。record只包含owner G、exact ParkTicket、active-wait双链和affected work link/state；64-bit为48 bytes、32-bit/WASM为28 bytes。active双链允许ready wait-set在O(1)内从P移除，per-P affected FIFO在每个published epoch结束时切成线性batch；同一wait-set的多个source fact通过`clean/queued/processing/dirty`状态合并。bootstrap或无法由compiler提供frame slot的入口使用调用方提供的静态pool，且必须在任何producer admission前reserve；native profile可选择可增长pool，baremetal/RTOS必须显式声明静态容量和同步失败。 +高并发promotion使用直接park物理协程frame内的临时`WaitSetRecord`，不为所有G常驻增加`prevWait`或affected link。record包含owner G、exact ParkTicket、active-wait双链、affected work link/state，以及一个只在同一frame内解释的`unsafe.Pointer + resumeBindingKind`联合槽；64-bit为56 bytes、32-bit/WASM为32 bytes。联合槽只能是none、单source packet、固定枚举typed cleanup plan或已物化packet，不保存interface/function value。active双链允许ready wait-set在O(1)内从P移除，per-P affected FIFO在每个published epoch结束时切成线性batch；同一wait-set的多个source fact通过`clean/queued/processing/dirty`状态合并。bootstrap或无法由compiler提供frame slot的入口使用调用方提供的静态pool，且必须在任何producer admission前reserve；native profile可选择可增长pool，baremetal/RTOS必须显式声明静态容量和同步失败。 该结构只让同时parked的任务付费，并保持producer/ISR仍只处理两字POD `OperationID`。迁移阶段legacy与V2各保留一对active head/tail，P另有affected head/tail，`Frame`暂存一个record pointer；legacy删除后应让`G.nextWait`原位承担当前record入口并合并这些队列字段。V2 fact mark、affected pop、promotion以及record-aware attach/detach已经只做O(1) header/邻接preflight，完整审计保留在测试、debug和terminal边界；`ParkLink.previous`由同时parked的source-owned operation支付。目标复杂度是`O(F + A + C)`：本轮source fact数F、受影响wait-set数A以及这些wait-set的candidate数C，与其余parked G无关。 @@ -333,9 +333,11 @@ Running和Waiting G不可迁移，completion必须投递原owner；只有已经 当前native fleet profile已经把这条core接入真实程序：route 1原位收养command线程上的program P，route 2由一个固定pthread M拥有；两个domain各有独立P/driver/source catalog、pipe doorbell和POSIX poll set，并并行执行同一个物理reducer。固定8槽`RunnableTransferMailbox`接受never-run root、普通`SuspendYield`以及已进入`parkMaterialized`的`SuspendPark` continuation；slot以generation和FIFO exact import持有唯一GC root。G复用既有对齐padding中的一字节`transferState`，Published期间关闭pointer-only preempt gate并拒绝普通`Enqueue`，因此不会被第三个P重复取得，且native/wasm32的G大小不变。当前分配策略只机会性转移每次physical resume产生的第一个新spawn；mailbox争用或满时保留本地FIFO。它证明双M/双P并行、启动/停止/join、初始任务迁移和受限parked-result迁移可运行，但尚不是动态GOMAXPROCS、通用global run queue或任意runnable work stealing。 -2026-07-26的第一阶段物化只接受零或一个Timer、Manual、Poll、Worker source。`WaitSetRecord`可绑定compiler/runtime提供的52-byte、align-4、pointer-free frame-local `ResumePacket`；原owner在`parkReady` promotion事务中先Consume winner lease，再按source类型Take/Discard payload、确认quiescence并回收exact generation，最后把`ParkState`压缩为不含winner/source ownership的`parkMaterialized`。Worker标量和Poll枚举在这里复制，Timer/Manual不保留payload；迁移后的resume只消费packet，prompt task cancellation也只抑制packet，不回访旧source。packet-bound wait不能走缺少`ExecutorSourceSet`的legacy promotion入口，旧`parkReady/parkConsumed/parkDelivered`仍被迁移gate拒绝。 +2026-07-26的第一阶段物化接受零或一个Timer、Manual、Poll、Worker source。`WaitSetRecord`可绑定compiler/runtime提供的52-byte、align-4、pointer-free frame-local `ResumePacket`；原owner在`parkReady` promotion事务中先Consume winner lease，再按source类型Take/Discard payload、确认quiescence并回收exact generation，最后把`ParkState`压缩为不含winner/source ownership的`parkMaterialized`。Worker标量和Poll枚举在这里复制，Timer/Manual不保留payload；迁移后的resume只消费packet，prompt task cancellation也只抑制packet，不回访旧source。packet-bound wait不能走缺少`ExecutorSourceSet`的legacy promotion入口，旧`parkReady/parkConsumed/parkDelivered`仍被迁移gate拒绝。 -该阶段有意不接受Channel/多case select、HostOp deadline和带runtime-private cleanup的keyed/host adapter。Channel必须先从所有hchan队列删除winner/loser节点并完成buffer/claim reconciliation；双source deadline必须同时退休Worker与Timer；runtime-private registry/transport也必须先完成typed cleanup。它们将在统一的固定枚举materialization plan中完成，不能以callback/interface或迁移后回访旧P来绕过。完成该plan前，这些park仍保持owner-affine。 +第二阶段已把direct Channel和multi-case channel `select`接到同一机制。compiler-spilled `ResumeCleanupPlan`只保存固定`ResumeCleanupKind`、typed runtime context、`{entries,count,stride,idOffset}` ID range、source/claim和阶段游标；没有callback、interface、函数地址反查或永久G字段。common resolver在旧P上Consume exact decision后停在`ExecutorRunStepMaterialize`，runtime direct switch每次处理一个logical case：从hchan队列摘除typed waiter、完成必要的buffer/close reconciliation、复制winner的封闭`waitStatus`，并清除frame中的hchan/source/claim指针，只暂留exact ID。随后core逐ID执行`ConfirmQuiesced`，统一reset共享claim、Take/Discard winner lease，再逐ID Recycle并清零；最后才写入P-neutral packet、转成`parkMaterialized`并进入ready queue。direct、两case winner/loser、close、task cancel、native race/shuffle、JS/WASM、旧P source为空后跨P迁移，以及迁移后prompt task cancel均已覆盖。真正发生的Yield/Park会确认已经由本次suspension满足的G-local preempt request；executor/source request仍有自己的sticky gate，因此不会因残留`preemptRequested`误拒合法迁移，也不会丢服务义务。 + +当前typed plan仍未接受HostOp deadline和带runtime-private cleanup的keyed/host adapter。双source deadline必须同时退休Worker与Timer；runtime-private registry/transport也必须先完成对应固定枚举cleanup。它们复用同一个plan与runner边界，不能以callback/interface或迁移后回访旧P来绕过。另一个严格cost gate是把现有hchan buffer/closed-queue reconciliation内部可能处理多个peer的循环继续拆成显式子游标；当前materialization已经把select的N个frame case分成N个runner reduction，但不能据此宣称整个hchan critical section已完成常数wall-work认证。 两字`OperationID`已经用稳定`RouteID`解决多P source namespace：两个P的同类source可以都从local slot 1开始，但callback必须携带完整`{source, route, local, generation}`，通过route admission精确发布source fact并请求对应executor。P teardown先seal route并strong-join producer，旧route留下永久tombstone；该路由约束不允许重新引入Go pointer callback ABI或按fd/函数地址反查owner。 @@ -447,7 +449,7 @@ Linux共享`RawSyscall` carrier还要求与函数地址证书正交的trap证书 这些是core-first证据，不是标准库兼容性证明。native compiler command entry仍是固定机器栈V2循环，但target start现已把program P/driver/source/registry作为route 1原位收养，同时创建route 2并启动其pthread owner；program return会停止并join peer与共享worker pool，再按route、backend、driver顺序强关闭。Timer catalog、Poll V2 callback与Worker transport均使用exact route：Timer由各domain owner的单调时钟扫描；Poll和Worker的两字`OperationID`在route admission内完成source publication、exact executor request、duplicate/stale分类和关闭强汇合；Worker复用一个进程级固定物理池。单source result物化/迁移已经越过，尚未完成的是multi-source typed materialization、动态P数量、普通global injection/steal和thread-affinity策略。target-neutral host-owned pull adapter已完成core/race与JS/WASM、WASIp1、baremetal、显式embedded交叉编译验证：`more`只发布later-turn action，idle只暴露POD executor/generation/epoch/deadline，alarm/notification取消后才可复用epoch，shutdown seal两个ingress并strong-join callback tail。compiler的host-target V2 entry也已完成：它只执行一次有界initial slice，保留`next_action/profile/next_deadline/publish_time/ack_cancel/continue/post_wait`等callback reference roots，仅接受canonical `Complete`或executor slot/generation/epoch/flags/deadline均精确的`Yielded`/`Suspended` tuple，后两者detached返回host，不在entry内继续调度或递归re-entry。仓库Node runner现已作为显式embedding真实消费这些action，并在`wasm-unknown`/`wasip2`运行file与TCP/UDP标准库fixture；普通JS/WASM或WASI `_start`仍没有内建reactor，因此不能把该受控embedding外推为通用platform E2E。JS command pump、WASI `poll_oneoff`、RTOS HAL与baremetal IRQ/WFI glue、完整cleanup/defer/recover/Goexit lowering以及precise/moving GC仍是明确缺口。 -执行器已收口到一套物理run-step reducer；fleet外层只保留exact domain owner、显式单调时钟和budget，不存在第二套resume/destroy/commit语义。空fleet P使用公共timer-aware idle事务完成standby admission；提交后先释放owner epoch，route-local固定poll set执行fault-containment有界物理pass，doorbell/fd/deadline唤醒后再取得新epoch并执行公共`WakeExecutorAt`全source重扫。普通domain最后一个G销毁只完成该G并保持executor可接收未来routed work；command route的main-return语义仍由program coordinator独占。实际peer M循环、共享worker pool lifecycle、分布式shutdown和强关闭引用清理现均已接入并通过production-island、source/race及TCP E2E验证。 +执行器已收口到一套物理run-step reducer；fleet外层只保留exact domain owner、显式单调时钟和budget，不存在第二套resume/destroy/commit语义。空fleet P使用公共timer-aware idle事务完成standby admission；提交后先释放owner epoch，route-local固定poll set执行fault-containment有界物理pass，doorbell/fd/deadline唤醒后再取得新epoch并执行公共`WakeExecutorAt`，只完成idle→active并设置`sourceMore`，所有source解析和typed materialization继续由同一run-step reducer完成。idle prepare/commit中若request或事实源赢得竞争，也执行相同的“leave idle + sourceMore”转换，不再嵌套调用legacy全量poll。普通domain最后一个G销毁只完成该G并保持executor可接收未来routed work；command route的main-return语义仍由program coordinator独占。实际peer M循环、共享worker pool lifecycle、分布式shutdown和强关闭引用清理现均已接入并通过production-island、source/race及TCP E2E验证。 Worker transport的route-aware核心现已完成：仍只有一个进程级固定4线程/1024-job物理池，C11 sequence ring允许多个exact P owner并发预留不同cell；每个reservation在compiler-enforced no-suspend hook内submit或以tombstone cancel。Job不增加指针或全局operation目录，直接使用既有两字`OperationID.SourceSlot`中的route；完成端按compiler-reserved target profile静态选择legacy program或fleet ingress，fleet再完成exact source/request/doorbell投递。owner在把完成发布为可resume winner前seal generation并等待所有已admit producer退出，避免回调release尾与result recycle竞态。C11环已通过route 1/2并发producer、交错wrap、四consumer和stop/join测试；program-level fleet coordinator实际start/stop该池，当前fleet标准库profile不再退回single-P worker transport。 diff --git a/doc/coro-ir-design.md b/doc/coro-ir-design.md index e15fa243d5..41db0b2ed8 100644 --- a/doc/coro-ir-design.md +++ b/doc/coro-ir-design.md @@ -1268,7 +1268,7 @@ site只接受ProgramIR projection。旧callback/classifier在同一提交删除 4. dynamic coroutine descriptor、closure/method/interface； 5. syscall/netpoll/worker/host sources； 6. precise GC/debug metadata； -7. 在已完成的零/单source packet上补typed multi-source materialization，再完成global injection、多P/affinity； +7. 在已完成的零/单source packet和Channel/select typed materialization上补HostOp deadline、keyed/transport composite cleanup，再完成global injection、多P/affinity； 8. reflect和完整平台adapter。 Phase B–F严格保持plan、runtime ABI和可观察行为不变；Phase G才是新功能开发，并且必须等待下面Phase R的四个hard-cutover gate全部通过，不能混为一个巨大PR。PrimitiveCatalog生成器重构和Runtime V3也分别立项，不塞入等价迁移。 @@ -1453,7 +1453,7 @@ channel/select physical operation choice、panic/outcome/cleanup choice及remain 5. runtime Phase R已经完成：fleet唯一target、Park/Operation唯一logical wait、统一source dispatcher和 mandatory stackless架构均以旧production符号/配置入口为零，并由hard-cutover gate持续约束。 6. 本轮合并门运行runtime race、LLVM 19–22、native/wasm32结构验证和六项fresh stdlib E2E。后续compiler - whole-function emitter仍按完整replacement cohort推进；single-source P-neutral result已独立落地，typed multi-source result、dynamic P、GC、panic/Goexit和更多平台adapter + whole-function emitter仍按完整replacement cohort推进；single-source P-neutral result和Channel/select typed multi-source materialization已独立落地，HostOp/keyed composite result、dynamic P、GC、panic/Goexit和更多平台adapter 属于功能阶段，不能借机恢复旧runtime轨道。 迁移过程可以用独立test invocation比较旧/新输出，但production永远只有一个被选择的consumer；临时双轨 diff --git a/doc/llvm-coro-runtime-design.md b/doc/llvm-coro-runtime-design.md index 923facc124..7f87ff4a2d 100644 --- a/doc/llvm-coro-runtime-design.md +++ b/doc/llvm-coro-runtime-design.md @@ -1350,7 +1350,7 @@ Platform completion 还需要一个稳定的 executor request gate，不能在 c - Phase 17 的 `ExecutorDriver` 已把一个 exact-zero `ExecutorRegistry` generation 和一个空 `WaitRegistrationTable` 绑定到单个空闲 `P`。绑定按 scheduler-owned pointer first、原子 `executorMode` last 发布；绑定前调用者必须 strong-quiesce 所有仍可能持有 `*P` 的 legacy `RequestSchedule` source，因为 mode 只是 capability guard，不是旧 ABI 的 admission/refcount barrier。绑定后 legacy `RequestSchedule` fail closed，producer 必须使用稳定的 POD executor handle。 - 绑定后的 registration table 只接受同一 `P` owner，standalone `Drain` 被拒绝。平台 `Post` 仍只发布 durable slot；scheduler-owned driver 才能解引用 waiter、promote `G`，而 drain/cancel 不再写旧 `P.schedule`。 - 运行中 G 的 `PollPreempt` 只 observe stable Requested 并 yield，同一个请求可被多个 poll 重复观察；只有重新取得 `P` 的 `PollExecutor` 可以 ack。driver 每轮对当前已接入的事实源执行“registration table 全表扫描和 ready promotion → ack → 无条件全表重扫”，若第二次扫描产生工作、pending 仍在或新 Requested 可见则继续，覆盖 producer 在 drain 与 ack 之间合并的请求；timer/channel/syscall source set 接入后必须加入同一 transaction。 -- `PrepareExecutorSleep` 先完成上述 transaction，只在无 runnable 且仍有 parked G 时 `ArmIdle`；随后不依赖 advisory pending，而是无条件扫描 `Posted` 事实源并检查 request/schedule，再选择 `LeaveIdle` 后继续 poll 或 exact `CommitSleep`。成功 commit 只授权 target 进入 retained wait；真实或 spurious wake 必须先 `WakeExecutor/LeaveIdle` 再 drain/ack/recheck。 +- `PrepareExecutorSleep` 只在无 runnable 且仍有 parked G 时 `ArmIdle`；随后不依赖 advisory pending，而是无条件扫描 `Posted` 事实源并检查 request/schedule，再选择 exact `CommitSleep`。若 request/事实源赢得 prepare 或 commit 竞争，只执行 `LeaveIdle` 并设置统一runner的`sourceMore`，不会在idle helper中嵌套legacy全量poll。成功 commit 只授权 target 进入 retained wait；真实或 spurious wake 必须先以`WakeExecutor[/At]`完成`LeaveIdle + sourceMore`，再由统一run-step reducer完成drain/ack、typed materialization与recheck。 - 常规 driver close 只允许 scheduler idle、无 parked G、无 live registration；ready G 可留给后续 command cancellation。`BeginExecutorClose` seal gate 后，target 必须 strong unregister/join 整个 ingress shim（包括 pre-lease entry 和 Request-to-doorbell tail），之后 `ConfirmExecutorClose` 才做 final source scan、retire generation、解绑 table 和 `P`。 - Phase 18 已为“root frame 已 destroy、当前 G 是队列中最后一个、executor 仍 bound”增加显式 terminal-close handoff。core 先执行 durable-source drain→executor ack→无条件重扫，再以 exact gate close 与 Request 竞争并 seal producer admission；成功后只在 driver 保留 `terminalKind`，清除已释放 frame 的 `g.root`，并把 `P.action` 切换为不携带 handle 的 `ActionTerminalExecutorClose`。已 destroy 的 LLVM handle 不进入持久 scheduler 状态，stale `ActionDestroy`/`ActionPanicDestroy` 也无法再通过 `expectedAction`。 - target 完成 strong unregister/join 后由 scheduler owner 调用 `ConfirmTerminalExecutorClose(driver)`。Confirm 不依赖原 caller stack，而是从稳定 driver/P 状态恢复 G、close marker 和 `terminalKind`；它在 join 后无条件执行 final source scan，随后 `ConfirmQuiesced`、`Retire`、解绑 registration table，按 `p.executor=nil`、driver zero、内部 commit token 恢复、`executorMode=Unbound` last 的顺序发布。最终 normal 或 panic terminal commit 只在 core 内调用 `Destroyed`/`PanicDestroyed` 重试，并只能向 adapter 返回 `ActionComplete` 或 `ActionPanicComplete`；该路径没有第二次 `llvm.coro.destroy` 操作。这个稳定状态交接允许 WASM/embedded 在异步 join 期间返回 host，不保留 managed continuation 或 native scheduler caller stack。 @@ -1860,11 +1860,11 @@ Pure sync library/archive不需要链接scheduler。Executable一旦选择 `-sch - terminal-only ExplicitStatus runtime core 已有 task-local 两字 `PanicRecord`、原子 once publication和无 TLS 的 `__llgo_coro_panic_prepare_v1(g, handle, header, typeWord, dataWord)`。compiler 对精确 cleanup-free PhysicalABIV1 body 生成 `SuspendPanic`/`FinalSuspended`，panic 与 normal return branch 到同一个 LLVM final suspend；active panic frame 先经过 `coro.done` 验证并 destroy，之后 suspended-await ancestor 不再 resume，而是从深到 root 直接 destroy，最终保留 record 并返回独立 `PanicComplete`。后续hard-cutover已把implicit language fault接入同一显式结果：无操作数的nil、bounds、channel和unsafe fault使用allocation-free V1静态payload；需要现场值的slice-to-array失败使用V2的两个target-width operand，由异常路径分配可跨LLVM frame destroy存活的GC可见`boundsError`，从而保持Go 1.26的动态类型、`runtime.Error`和逐值文本。V1/V2 hook均为required raw/plain runtime root并进入bootstrap/physical ABI identity；尚未定义的parameterized kind或不匹配的operand必须fail closed，不能静默退回无参数错误。更宽的dynamic explicit-panic payload、所有cleanup/recover/Goexit组合和最终进程报告仍分别由其专门验收矩阵约束，不能由本条fault ABI证据外推。 - park/wake handshake 已落地 32-bit 原子 `WaitToken`、generation ticket、early/late completion、park 前后 cancellation、唯一 waiter claim、ABA 范围校验及 terminal gate。完成/取消 outcome 在 scheduler consume 后仍保持到下一次 Arm，恢复后的同步风格 continuation 可用 exact ticket 查询赢家；精确 intrinsic `llgo.coroPark(token, ticket)` 被 Effect 分析识别为 `MayPark`，并在调用者当前 LLVM frame 中生成 park prepare、stateID、`coro.suspend` 和恢复路径，没有隐藏在普通同步 helper 中。channel/timer/syscall 的 submit/retry producer 尚未接入。 - 固定容量 `WaitRegistrationTable` 已实现 POD `{slot,generation}` handle、producer admission seal/refcount、`Active→Posting→Posted→Draining→Delivered` one-shot mailbox、scheduler-side Drain、strong unregister/quiescence handoff、cancel-vs-complete winner、silent late callback、generation reuse 和 capacity fail-closed。平台 Post 不接触 P/G/token/LLVM handle；Phase 17 绑定后 table 固定归一个 `P`，只有 driver drain 可解引用 waiter，standalone drain 和错误 owner registration 均 fail closed。race+shuffle 已覆盖 concurrent post、post-vs-close、旧 producer pin/reuse、pre-park cancel 和 exactly-once promotion。真实 backend 的 strong unregister/join 与 result payload 仍需各 target adapter 证明。 -- 固定容量 `ExecutorRegistry` 已实现 POD `{slot,generation}` handle、`Requested|IdleArmed|Closed` gate、producer admission seal/refcount、exact idle commit、request/close 线性化、strong join/quiescence、generation reuse 和 capacity fail-closed。Phase 17 的 target-neutral `ExecutorDriver` 已把 gate 绑定单 P：`PollPreempt` observe-only，scheduler 独占 drain→ack→无条件重扫，idle 执行 poll→ArmIdle→完整事实源重扫→exact CommitSleep，wake 执行 LeaveIdle→poll，close 执行 seal→外部 strong join→confirm/retire/unbind。旧 `P.schedule` 只保留给 unbound internal path，wait drain/cancel 不再发布 legacy request。 +- 固定容量 `ExecutorRegistry` 已实现 POD `{slot,generation}` handle、`Requested|IdleArmed|Closed` gate、producer admission seal/refcount、exact idle commit、request/close 线性化、strong join/quiescence、generation reuse 和 capacity fail-closed。Phase 17 的 target-neutral `ExecutorDriver` 已把 gate 绑定单 P；当前统一语义是`PollPreempt` observe-only，scheduler run-step独占drain→ack→无条件重扫，idle执行ArmIdle→完整事实源重扫→exact CommitSleep，prepare/commit竞争失败与wake都只执行LeaveIdle→`sourceMore`，再返回同一run-step reducer，close执行seal→外部strong join→confirm/retire/unbind。旧 `P.schedule` 只保留给 unbound internal path，wait drain/cancel 不再发布 legacy request。 - Phase 17 host 验证已通过 `runtime/internal/coro` unit、`-race -shuffle=on -count=30`、focused `ExecutorDriver -race -count=100` 和 `go vet`；package cross-build 覆盖 `js/wasm`、`wasip1/wasm`、`linux/arm`、`linux/riscv64`，current-source LLGo package build 也通过。确定性交错包括 running poll 重复 observe 到 scheduler ack、Post-before-delayed-Request、300 次 Post×PrepareSleep race、wake-before-physical-block retained doorbell、spurious wake、错误 owner/direct drain、premature close 和 bound terminal fail-closed。这些验证只覆盖 target-neutral scheduler core，不代表 production `coroRun` 或真实 backend 已接线。 - Phase 18 已实现 handle-free `ActionTerminalExecutorClose` 和 `executorDriverTerminalClosing`。last-G root 在物理 destroy 后会先 settle request 并 seal executor，driver 只保留 normal/initial-panic `ActionDestroy` 或 ancestor-panic `ActionPanicDestroy` 的 `terminalKind`，清除 `g.root` 后发布 close marker。`ConfirmTerminalExecutorClose(driver)` 完全从稳定 driver/P 恢复必要状态，在 target 完成外部 join 后执行 final scan、confirm/retire/unbind、mode-last 发布和 core-only terminal commit；adapter 不会再收到已销毁 handle 或新的 destroy action。 - Phase 18 host 验证已通过 `runtime/internal/coro` unit、`-race -shuffle=on -count=30`、focused terminal executor `-race -count=100` 和 `go vet`；package cross-build 通过 `js/wasm`、`wasip1/wasm`、`linux/arm`、`linux/riscv64`。已覆盖 normal terminal、单帧 panic（`ActionDestroy`）、多帧 panic 的 root ancestor（`ActionPanicDestroy`）、stale destroy action 拒绝、错误 G/generic close 拒绝、executor request settle 和 producer lease 在 strong join 前阻止 Confirm。这些测试不能代替真实 target 对 pre-lease entry 及 Request-to-doorbell tail 的 join 证明。 -- Phase 19 已把 production program runner 绑定到上述静态 driver，并把handle-free terminal close交给静态target dispatcher；其whole-G action循环已由Phase 31的bounded physical-action dispatcher替代。runner以显式drive status区分main return、retained sleep、terminal close和panic；`__llgo_coro_program_continue_v1(epoch)`通过`DriveAdmission`单owner重入，不保留caller stack、G/Action或LLVM handle。last-G normal/panic执行terminal strong join；main正常返回且仍有ready child时先执行generic executor close/join，再进入command cancellation。parked root的wait registration也已贯通Post/IdleWake→continue→WakeExecutor→resume→consume/retire→terminal。 +- Phase 19 已把 production program runner 绑定到上述静态 driver，并把handle-free terminal close交给静态target dispatcher；其whole-G action循环已由Phase 31的bounded physical-action dispatcher替代。runner以显式drive status区分main return、retained sleep、terminal close和panic；`__llgo_coro_program_continue_v1(epoch)`通过`DriveAdmission`单owner重入，不保留caller stack、G/Action或LLVM handle。last-G normal/panic执行terminal strong join；main正常返回且仍有ready child时先执行generic executor close/join，再进入command cancellation。parked root的wait registration也已贯通Post/IdleWake→continue→`WakeExecutor`→统一source/materialization reducer→resume→consume/retire→terminal。 - Phase 19 host验证已通过DriveAdmission定向竞态、`runtime/internal/coro -race -shuffle`、program adapter `-race -shuffle`、`js/wasm`实际运行、native+nogc spawn/panic E2E、完整coroutine build integration与named-source vet；cross compile覆盖`js/wasm`、`wasip1/wasm`、`linux/arm`、`linux/riscv64`和cortexm。测试target覆盖同步/异步join、Begin返回前completion、并发/stale/duplicate continue和executor wake。production `coro_target_none`没有ingress，只能同步确认空executor，不能充当真实retained-doorbell backend。 - Phase 20 已加入只在 `llgo && llgo_coro && llgo_coro_native_pipe && (linux || darwin) && !baremetal` 选择的 production native pipe/poll backend；`llgo_coro_native_pipe` 是compiler-reserved capability，只由编译器对默认POSIX Linux/Darwin配置下发，`Config.Tags`、`GoBuildFlags`和named-target `BuildTags`均不能伪造。不能仅凭 `GOOS=linux` 推断该能力，因为部分embedded named target会借用Linux源码选择却没有process pipe/poll；普通host Go test、named target、WASM、baremetal和`coro_runtime_adapter_test`继续选择各自的非native target，避免runtime与planner/root/hash/anchor错配。验证覆盖pipe提前wake、并发coalesce、满管EAGAIN、TargetIngress Enter/Seal竞态与strong join、2048次同步wake迭代深度、真实runtime required-plain planner，以及Linux arm/arm64/riscv64和Darwin amd64/arm64静态交叉编译/target选择。native+nogc spawn/panic E2E按运行测试的host做最终链接执行：当前focused CI提供Linux执行覆盖，Darwin同一E2E仅在Darwin host运行时执行，尚无Darwin CI runner。该结果仍不表示timer/syscall source、blocking worker compensation或多P已经完成。 - Phase 21 已通过真实 `nogc` pthread producer E2E覆盖 `prepare -> publish POD -> llgo.coroPark -> CommitSleep -> pending-clear/poll窗口 post -> pipe wake -> scheduler drain/consume -> 原frame恢复 -> pthread_join -> registration retire -> terminal target close`。unit/race覆盖transactional prepare在nil owner与满64槽时回滚到新generation、pre-park rollback、只有当前resume owner可prepare/retire、以及永久retired ingress诊断；planner把三个owner ABI作为精确DirectPlain runtime roots并把完整签名纳入bootstrap hash。hook和终态audit只在compiler-reserved测试capability下存在，production默认IR常量消除hook调用。 @@ -1941,15 +1941,16 @@ Pure sync library/archive不需要链接scheduler。Executable一旦选择 `-sch physical operation selector预算没有上升。使用本次重构后新构建的compiler，两个WASM目标均 重新通过直接网络`16/2/15`、文件`5/10`、Sleep `4/1`及TCP/DNS `594/17/18/651`纵向账本。 -- Phase 36之后的主要缺口仍是工程闭环而非新的coroutine可行性障碍：P-neutral - `ResumePacket`、动态P/steal和blocking compensation；paged/dynamic channel、timer与 +- Phase 36之后的主要缺口仍是工程闭环而非新的coroutine可行性障碍：零/单source + `ResumePacket`及Channel/select typed multi-source materialization已经完成，下一步是HostOp deadline、 + keyed/transport composite cleanup、动态P/steal和blocking compensation；paged/dynamic channel、timer与 worker capacity；完整defer/recover/Goexit和precise suspended-frame GC；外部DNS server/cgo resolver、Unix/raw socket与ancillary OOB、process/signal；logical stack/tooling；以及 WASM/WASI/RTOS/baremetal各自内建 event/host/HAL adapter。native退出实验模式仍必须满足35.1与35.2，尤其仓库 `test/*`和GOROOT不得存在unexpected failure。 - compiler的所有现有initial、child-await、yield和legacy-park resume边已接入terminating dispatch gate。zero-ticket路径调用scalar `__llgo_coro_run_decision_take_zero_v1(g) uint32`，正常值进入唯一normal continuation，Abort/Shutdown在cleanup lowering完成前进入共享trap而不会误执行用户continuation；full ticket/lease ABI继续供bootstrap与未来park-site reconciliation使用。同一LLVM/target的gate开关对照证明scalar gate不会增加stackless coroutine frame，CoroSplit ramp/destroy也没有可达gate。 -- 两字Operation identity已冻结为`source:8/route:9/local:15 + generation:32`，保持size 8、align 4。route按runtime instance单调分配且永不复用，关闭后保留永久tombstone；Manual/TaskControl ingress的producer lease覆盖`source.Post -> executor.Request`完整tail，strong join后才允许清除source/executor pointer；Timer V2 reserve、publish、Apply和result lease也验证exact route/local/generation。该机制只解决多executor寻址与ABA前置条件；P-neutral ResumePacket、global injection与work stealing仍未完成。 +- 两字Operation identity已冻结为`source:8/route:9/local:15 + generation:32`，保持size 8、align 4。route按runtime instance单调分配且永不复用，关闭后保留永久tombstone；Manual/TaskControl ingress的producer lease覆盖`source.Post -> executor.Request`完整tail，strong join后才允许清除source/executor pointer；Timer V2 reserve、publish、Apply和result lease也验证exact route/local/generation。52-byte pointer-free `ResumePacket`已覆盖零/单Timer、Manual、Poll、Worker以及经typed plan物化的direct Channel/multi-case select；global injection与通用work stealing仍未完成。 - 第一个标准库同步风格原型已以GOROOT source patch实现`time.Sleep`：普通`time.Sleep(d)`被Effect分析自动传播为`DirectCoro/AwaitStructured`，不修改public signature，不依赖libuv、BDWGC、pthread producer或用户goroutine。真实linked native+nogc E2E已编译production runtime island，实际等待30ms并恢复原frame；timer/wake路径由monotonic clock与pipe/poll/fcntl实现，符号审计确认不依赖libuv、BDWGC或pthread producer。另一focused production-overlay测试直接读取真实注入的`time.Sleep`源，不用测试effect seed，验证跨包同步caller染色、frame证书和CoroSplit，但不声称链接执行标准库`time.Sleep`。LLVM 19–22都跑该契约，Go 1.24跑真实linked E2E，Go 1.26也跑production overlay分析/codegen。 - Phase 22 仍是有界prototype，不是完整`time`runtime：第65个同时live timer会按fail-stop ABI终止，尚需dynamic/sharded table和heap；`Timer`/`Ticker`/`AfterFunc`仍使用legacy libuv路径；`f := time.Sleep`、interface/reflect和dynamic dispatch还没有end-to-end callable coroutine descriptor；`Sleep(0)`/负值在Sleep体内不注册timer，但value-insensitive caller仍会创建并await child frame，尚需conditional effect或call-site fast path才能避免可观测的多余handoff。完整`Do`标准库构建现在先被`sync.Pool` TLS destructor的捕获闭包挡住：exact同步C callback ABI没有closure context slot，不能直接放宽。后续需改成显式`owner/local` TLS state，并同时为`tls.Handle[T]`经`Pool.local`的unsafe transport建立字段级whole-program证书。WASM、WASI、RTOS和baremetal也尚未有对应timer source。 - wait/preempt core 要求目标提供可靠的 32-bit atomic load/store/CAS。WASM 可直接满足；带 A 扩展的 RISC-V 可满足；ESP32-C3 RV32IMC 当前会在链接时缺少 `__atomic_*_4`，直到平台用 IRQ critical section 提供单核适配。这里故意不使用非原子 fallback。 @@ -1960,7 +1961,7 @@ Pure sync library/archive不需要链接scheduler。Executable一旦选择 `-sch - terminal panic 的独立 native+nogc scheduler-island 已真实编译并运行 `panic(&GlobalPayload)`。production internal runner返回精确`DrivePanic`状态，导出的void program-run ABI随后执行fatal abort；bootstrap、main、panicChild三个不同LLVM handle各destroy一次，两个祖先均不resume，task-local record在三层frame销毁后仍保持exact type/data word，且G为Dead/non-Reclaimable。最终二进制要求production `PreparePanic`/`PanicDestroyed`/`LoadPanicRecord`并禁止legacy panic/print链；测试report只观察internal drive-panic与record，不代替production printer/exit owner。 - 完整真实 `entry → allocator → v2 factory → runtime/package init → main → scheduler` linked smoke 仍受上述 runtime/Panic/foreign blockers 限制；scheduler-island、runtime adapter 和 freestanding wasm CLI fixture 各自证明的边界不能合并表述为完整 Go runtime 已经端到端运行。 - 当前 cache digest 只解决同一完整程序计划下的内部 package cache；未知未来 caller 可复用的预编译 archive/标准库仍需 producer summary、canonical boundary Dispatch 和 linker ABI 校验。 -- 后续依赖顺序先切分或认证Phase 31 physical resume内部的frame/ancestry/link scan与select排序，再把idle prepare/wake、terminal/command close、shutdown、frame scan/Zero与source-specific隐藏工作纳入同一账本，并把Phase 31b POD扩展为WASM/JS、WASI、RTOS和baremetal的queued/blocked/deadline adapter，完成post-LLVM cost certificate；同时在Phase 32 C0 Channel claim/core上完成C1 typed hchan、payload与compiler lowering，并为Poll/Host接入对应`TryCommit`，再在已冻结的result ownership/lease协议上完成typed payload materialization、`CompletionRecord`和可挂起cleanup。其后才把当前64槽native timer升级为dynamic/sharded heap，补齐`Sleep(0)` fast path、Timer/Ticker/AfterFunc和dynamic callable descriptor，并实现有界blocking worker、registration unregister和异步syscall source。各target复用同一core并分别证明完整ingress join边界。多P开放前还必须先物化P-neutral `ResumePacket`和parkable capacity permit；未物化packet的G不可steal。随后补suspended-frame GC、完整defer/recover/Goexit、dynamic/closure/method `go`及平台tooling。所有阶段保持无栈、单primary、静态source catalog和未证明即fail closed，不引入其他语言的Task/Future对象层。 +- 后续依赖顺序先把hchan buffer/closed reconciliation的内部peer循环拆成显式runtime materialization子游标，并继续认证Phase 31 physical resume内部的frame/ancestry/link scan与select排序；再把idle prepare/wake、terminal/command close、shutdown、frame scan/Zero与source-specific隐藏工作纳入同一账本，并把Phase 31b POD扩展为WASM/JS、WASI、RTOS和baremetal的queued/blocked/deadline adapter，完成post-LLVM cost certificate。现有固定枚举typed plan继续接入HostOp deadline、keyed registry和transport cleanup，不增加callback/interface；其后把当前64槽native timer升级为dynamic/sharded heap，补齐`Sleep(0)` fast path、Timer/Ticker/AfterFunc和dynamic callable descriptor，并实现有界blocking worker、registration unregister和异步syscall source。各target复用同一core并分别证明完整ingress join边界。多P开放前还必须为每一种composite park完成P-neutral packet和parkable capacity permit；未物化packet的G不可steal。随后补suspended-frame GC、完整defer/recover/Goexit、dynamic/closure/method `go`及平台tooling。所有阶段保持无栈、单primary、静态source catalog和未证明即fail closed，不引入其他语言的Task/Future对象层。 - 标准host-Go下`self` pointer导致的逃逸只是Phase 32 C0 test artifact，不是production allocation contract。C1需由compiler提供caller-owned pair storage，并用noescape/frame certificate证明它位于不移动coroutine frame且无heap allocation，同时单独证明`hchan` critical section NoSuspend/NoPanic。 ### Phase 1：单 P deterministic scheduler diff --git a/internal/build/coro_native_fleet_e2e_test.go b/internal/build/coro_native_fleet_e2e_test.go index 5e5e503db6..b8bf392e71 100644 --- a/internal/build/coro_native_fleet_e2e_test.go +++ b/internal/build/coro_native_fleet_e2e_test.go @@ -23,9 +23,11 @@ import ( "fmt" goimporter "go/importer" "go/types" + "os" "os/exec" "path/filepath" "runtime" + "strings" "testing" "time" @@ -373,6 +375,11 @@ func runCoroNativeFleetE2E(t *testing.T, source, name string, enableChannel bool } executable := filepath.Join(temp, "coro-native-fleet-"+name+"-e2e") + // Keep an opt-in stable output path for inspecting a failed native fleet + // executable with the platform debugger. Ordinary CI never sets it. + if diagnostic := strings.TrimSpace(os.Getenv("LLGO_CORO_NATIVE_FLEET_OUTPUT")); diagnostic != "" { + executable = diagnostic + } linkArgs := []string{driverObject, entryObject, userObject, runtimeArchive, "-pthread", "-o", executable} if runtime.GOOS == "darwin" { linkArgs = append(linkArgs, "-Wl,-dead_strip") diff --git a/runtime/internal/coro/channel_park_owner.go b/runtime/internal/coro/channel_park_owner.go index 83047766b2..76b51ebfb4 100644 --- a/runtime/internal/coro/channel_park_owner.go +++ b/runtime/internal/coro/channel_park_owner.go @@ -20,8 +20,9 @@ import "unsafe" // CurrentExecutorChannelDriver resolves the exact channel source owner during // the compiler's narrow SuspendPark/FrameSuspended transition window. Channel -// queue nodes may later rendezvous across executor routes, but preparation and -// resume cleanup are always owned by the current G's exact P/source pair. +// queue nodes may later rendezvous across executor routes, but preparation is +// owned by the current G's exact P/source pair. Typed resume cleanup now runs +// on the scheduler stack before runnable publication. func CurrentExecutorChannelDriver(g *G) (*ExecutorDriver, ExecutorHandle, RouteID, bool) { driver, handle, route, ok := currentExecutorParkDriver(g) if !ok || driver.sources.channel == nil || @@ -51,10 +52,10 @@ func CurrentExecutorChannelParkOwner( return driver.p, &g.park, driver.sources.channel, true } -// ActiveChannelParkOwner returns the scheduler-owner context used by the -// trusted typed-channel adapter while a compiler resume gate is active. The -// returned ParkState pointer is frame-independent G storage and must not be -// retained after the adapter returns or passed to a producer. +// ActiveChannelParkOwner is retained for target-neutral lifecycle tests and +// compatibility adapters which have not opted into ResumeCleanupPlan. New +// typed channel code must materialize before promotion instead of consulting +// its old source from a compiler resume gate. func ActiveChannelParkOwner(g *G, source *ChannelOperationSource) (*P, *ParkState, bool) { if !ValidG(g) || !resumeGateTaken(g) || g.runP == nil || !validChannelOperationOwner(source, g.runP) { return nil, nil, false @@ -154,9 +155,8 @@ func CanReserveChannelOperations(p *P, source *ChannelOperationSource, needed ui return false } -// FinishSingleChannelPark releases one detached channel operation after the -// compiler's exact-ticket resume gate has consumed its RunDecision and the -// typed hchan layer has removed the frame node from its queue. A valid lease +// FinishSingleChannelPark is the compatibility release boundary for adapters +// which have not opted into ResumeCleanupPlan. A valid lease // is taken for a selected continuation or discarded when task cancellation // suppresses an already committed result. A canceled operation with no // physical winner carries a zero lease. diff --git a/runtime/internal/coro/executor_driver.go b/runtime/internal/coro/executor_driver.go index 483fb11b23..b1bb584850 100644 --- a/runtime/internal/coro/executor_driver.go +++ b/runtime/internal/coro/executor_driver.go @@ -453,42 +453,38 @@ func leaveExecutorIdle(driver *ExecutorDriver) bool { return true } -func leaveExecutorIdleAndPoll(driver *ExecutorDriver) (drained, promoted int, ok bool) { +func leaveExecutorIdleForRun(driver *ExecutorDriver) bool { if !leaveExecutorIdle(driver) { - return 0, 0, false - } - return pollExecutor(driver) -} - -func leaveExecutorIdleAndPollAt(driver *ExecutorDriver, now int64) (scan executorSourceScan, ok bool) { - if !leaveExecutorIdle(driver) { - return executorSourceScan{}, false + return false } - return pollExecutorSourcesAt(driver, now, true) + // Facts discovered during the idle transaction may require a typed runtime + // materialization step. Retain only a source-work hint here; the unified + // reducer owns every later source-neutral and direct-runtime reduction. + driver.run.sourceMore = true + return validExecutorDriver(driver) } -// PrepareExecutorSleep services current work and, only when parked Gs remain -// with no runnable work, executes ArmIdle, an unconditional final source scan, -// and exact CommitSleep. A true sleep result authorizes the target to enter its -// retained wait. false,true means work or a racing request won and the -// scheduler should continue without blocking. +// PrepareExecutorSleep executes ArmIdle, an unconditional source fact scan, +// and exact CommitSleep only when no runnable exists and parked Gs remain. +// Source resolution stays in the unified runner. A true sleep result authorizes +// the target to enter its retained wait. false,true means work or a racing +// request won and the scheduler should continue without blocking. func PrepareExecutorSleep(driver *ExecutorDriver) (sleep bool, ok bool) { if !validExecutorDriver(driver) || driver.sources.usesMonotonicTime() || driver.state != executorDriverActive || !emptyExecutorRunCursor(driver) || !idleExecutorScheduler(driver.p) { return false, false } - if _, _, ok = pollExecutor(driver); !ok { - return false, false - } if runnableForOSThreadOwner(driver.p) || !HasWaiting(driver.p) { return false, true } if !driver.registry.ArmIdle(driver.handle) { - // Request won the exact zero-gate race. Service it while still active. - if _, _, ok = pollExecutor(driver); !ok { + // Request won the exact zero-gate race. Its fact remains durable and the + // unified runner will service it after this compatibility boundary. + if !driver.registry.ObserveRequested(driver.handle) { return false, false } - return false, true + driver.run.sourceMore = true + return false, validExecutorDriver(driver) } // Scan facts, not just pending, after publishing IdleArmed. This closes a @@ -505,13 +501,13 @@ func PrepareExecutorSleep(driver *ExecutorDriver) (sleep bool, ok bool) { hasWork := drained != 0 || runnableForOSThreadOwner(driver.p) || driver.sources.pending(driver.p) || driver.registry.ObserveRequested(driver.handle) || preemptLoad(&driver.p.schedule) != scheduleIdle if hasWork { - if _, _, ok = leaveExecutorIdleAndPoll(driver); !ok { + if !leaveExecutorIdleForRun(driver) { return false, false } return false, true } if !driver.registry.CommitSleep(driver.handle) { - if _, _, ok = leaveExecutorIdleAndPoll(driver); !ok { + if !leaveExecutorIdleForRun(driver) { return false, false } return false, true @@ -521,11 +517,12 @@ func PrepareExecutorSleep(driver *ExecutorDriver) (sleep bool, ok bool) { } // PrepareExecutorSleepAt performs the first half of timer-aware retained-wait -// admission. It services the complete source set at now, publishes IdleArmed, -// and scans the complete set once more. true,true leaves the driver in an explicit -// idle-preparing state and requires the caller to take a fresh monotonic sample -// and call CommitExecutorSleepAt. false,true means work won and the driver is -// active. A failure never leaves a newly armed idle gate behind. +// admission. It publishes IdleArmed and scans every durable source fact at now +// without resolving the resulting epoch. true,true leaves the driver in an +// explicit idle-preparing state and requires the caller to take a fresh +// monotonic sample and call CommitExecutorSleepAt. false,true means work won +// and the unified runner must continue. A failure never leaves a newly armed +// idle gate behind. func prepareExecutorSleepAt( driver *ExecutorDriver, now int64, @@ -535,18 +532,17 @@ func prepareExecutorSleepAt( !emptyExecutorRunCursor(driver) || !idleExecutorScheduler(driver.p) || now < 0 { return false, false } - if _, ok = pollExecutorSourcesAt(driver, now, true); !ok { - return false, false - } if runnableForOSThreadOwner(driver.p) || !allowEmpty && !HasWaiting(driver.p) { return false, true } if !driver.registry.ArmIdle(driver.handle) { - // Request won the exact zero-gate race. Service it while still active. - if _, ok = pollExecutorSourcesAt(driver, now, true); !ok { + // Request won the exact zero-gate race. Defer source service to the + // unified runner so typed materialization stays resumable. + if !driver.registry.ObserveRequested(driver.handle) { return false, false } - return false, true + driver.run.sourceMore = true + return false, validExecutorDriver(driver) } // Scan facts, not just pending, after publishing IdleArmed. Commit performs @@ -560,7 +556,7 @@ func prepareExecutorSleepAt( driver.sources.pending(driver.p) || driver.registry.ObserveRequested(driver.handle) || preemptLoad(&driver.p.schedule) != scheduleIdle if hasWork { - if _, ok = leaveExecutorIdleAndPollAt(driver, now); !ok { + if !leaveExecutorIdleForRun(driver) { return false, false } return false, true @@ -610,7 +606,7 @@ func CommitExecutorSleepAt(driver *ExecutorDriver, now int64) (sleep bool, deadl driver.sources.pending(driver.p) || driver.registry.ObserveRequested(driver.handle) || preemptLoad(&driver.p.schedule) != scheduleIdle if hasWork { - if _, ok = leaveExecutorIdleAndPollAt(driver, now); !ok { + if !leaveExecutorIdleForRun(driver) { return false, 0, false, false } return false, 0, false, true @@ -620,7 +616,7 @@ func CommitExecutorSleepAt(driver *ExecutorDriver, now int64) (sleep bool, deadl return false, 0, false, false } if !driver.registry.CommitSleep(driver.handle) { - if _, ok = leaveExecutorIdleAndPollAt(driver, now); !ok { + if !leaveExecutorIdleForRun(driver) { return false, 0, false, false } return false, 0, false, true @@ -631,26 +627,28 @@ func CommitExecutorSleepAt(driver *ExecutorDriver, now int64) (sleep bool, deadl return true, scan.deadline, scan.hasDeadline, true } -// WakeExecutor leaves a committed retained wait and immediately services all -// durable sources. It also accepts a spurious target wake while the gate still -// contains exact IdleArmed. -func WakeExecutor(driver *ExecutorDriver) (drained, promoted int, ok bool) { - if !validExecutorDriver(driver) || driver.sources.usesMonotonicTime() || driver.state != executorDriverSleeping || - !emptyExecutorRunCursor(driver) || !idleExecutorScheduler(driver.p) { - return 0, 0, false +func wakeExecutorRun(driver *ExecutorDriver, now int64, withDeadline bool) bool { + if !validExecutorDriver(driver) || driver.sources.usesMonotonicTime() != withDeadline || + driver.state != executorDriverSleeping || !emptyExecutorRunCursor(driver) || + !idleExecutorScheduler(driver.p) || withDeadline && now < 0 { + return false } - return leaveExecutorIdleAndPoll(driver) + return leaveExecutorIdleForRun(driver) } -// WakeExecutorAt leaves a committed timer-aware retained wait and services both -// source set using the target's fresh post-wake monotonic sample. -func WakeExecutorAt(driver *ExecutorDriver, now int64) (timers, promoted int, ok bool) { - if !validExecutorDriver(driver) || !driver.sources.usesMonotonicTime() || driver.state != executorDriverSleeping || - !emptyExecutorRunCursor(driver) || !idleExecutorScheduler(driver.p) || now < 0 { - return 0, 0, false - } - scan, ok := leaveExecutorIdleAndPollAt(driver, now) - return scan.timers, scan.promoted, ok +// WakeExecutor leaves a retained no-deadline wait and defers all source +// service to NextExecutorRunStep. Production runtimes use this boundary because +// typed resume materialization may require a direct runtime reduction between +// source-neutral steps. +func WakeExecutor(driver *ExecutorDriver) bool { + return wakeExecutorRun(driver, 0, false) +} + +// WakeExecutorAt is the monotonic-time counterpart to WakeExecutor. The +// timestamp validates the target wake sample; each later bounded source +// reduction receives its own fresh sample through NextExecutorRunStepAt. +func WakeExecutorAt(driver *ExecutorDriver, now int64) bool { + return wakeExecutorRun(driver, now, true) } func canBeginExecutorClose(driver *ExecutorDriver) bool { diff --git a/runtime/internal/coro/executor_driver_test.go b/runtime/internal/coro/executor_driver_test.go index 77226da99e..6ec213df1b 100644 --- a/runtime/internal/coro/executor_driver_test.go +++ b/runtime/internal/coro/executor_driver_test.go @@ -110,3 +110,62 @@ func TestExecutorDriverBindCloseLifecycle(t *testing.T) { t.Fatal("closed driver retained stable ownership") } } + +func drainTimerAwareExecutorRunSources(t *testing.T, driver *ExecutorDriver, now int64) { + t.Helper() + for reduction := 0; reduction < 4096; reduction++ { + step, ok := NextExecutorRunStepAt(driver, now) + if !ok { + t.Fatalf("drain timer-aware executor source reduction %d", reduction) + } + switch step.Kind { + case ExecutorRunStepSource: + continue + case ExecutorRunStepIdle: + return + default: + t.Fatalf("timer-aware source drain reduction %d = %d", reduction, step.Kind) + } + } + t.Fatal("timer-aware executor source drain exceeded reduction bound") +} + +func TestExecutorRunWakeDefersSourceService(t *testing.T) { + p := new(P) + driver, _, _, _ := bindTestExecutorDriverWithTimers(t, p) + prepared, ok := PrepareExecutorStandbyAt(driver, 10) + if !ok || !prepared { + t.Fatalf("prepare empty timer-aware standby = (%t, %t)", prepared, ok) + } + sleep, deadline, hasDeadline, ok := CommitExecutorSleepAt(driver, 11) + if !ok || !sleep || hasDeadline || deadline != 0 { + t.Fatalf("commit empty timer-aware standby = (%t, %d, %t, %t)", sleep, deadline, hasDeadline, ok) + } + if !WakeExecutorAt(driver, 12) || driver.state != executorDriverActive || + !driver.run.sourceMore || driver.poll.phase != executorPollIdle { + t.Fatal("run wake did not retain source work for the unified reducer") + } + drainTimerAwareExecutorRunSources(t, driver, 13) + if driver.run != (executorRunCursor{}) { + t.Fatalf("drained run cursor = %+v", driver.run) + } + closeTestExecutorDriver(t, driver) +} + +func TestPrepareExecutorStandbyDefersRacingRequest(t *testing.T) { + p := new(P) + driver, registry, _, handle := bindTestExecutorDriverWithTimers(t, p) + if result := registry.Request(handle); result != ExecutorRequestPublished { + t.Fatalf("publish pre-standby executor request = %d", result) + } + prepared, ok := PrepareExecutorStandbyAt(driver, 20) + if !ok || prepared || driver.state != executorDriverActive || !driver.run.sourceMore { + t.Fatalf("prepare over racing request = (%t, %t), state=%d run=%+v", + prepared, ok, driver.state, driver.run) + } + drainTimerAwareExecutorRunSources(t, driver, 21) + if registry.ObserveRequested(handle) { + t.Fatal("unified source reducer retained the racing executor request") + } + closeTestExecutorDriver(t, driver) +} diff --git a/runtime/internal/coro/poll_operation_source.go b/runtime/internal/coro/poll_operation_source.go index 65e6753ba6..e31a3ee5ba 100644 --- a/runtime/internal/coro/poll_operation_source.go +++ b/runtime/internal/coro/poll_operation_source.go @@ -82,8 +82,9 @@ type PollOperationSnapshot struct { // PollOperationPostResult classifies owner-side import of a readiness event. // The first native slice has no reactor callback thread: target wait returns on -// the executor owner and imports an exact handle before WakeExecutorAt scans -// this source. Keeping a generation-classified result now preserves the same +// the executor owner and imports an exact handle before WakeExecutorAt returns +// source service to the unified reducer. Keeping a generation-classified +// result now preserves the same // ABI if a future multi-P target moves import behind a stable ingress shim. type PollOperationPostResult uint8 diff --git a/runtime/internal/coro/published_epoch_resolution.go b/runtime/internal/coro/published_epoch_resolution.go index 0a04b2b728..b6f87c40eb 100644 --- a/runtime/internal/coro/published_epoch_resolution.go +++ b/runtime/internal/coro/published_epoch_resolution.go @@ -124,7 +124,10 @@ func validPublishedEpochResolveCursor(cursor *publishedEpochResolveCursor, p *P) case publishedEpochResolveFinish, publishedEpochResolvePromote: return cursor.park == (parkResolutionCursor{}) && cursor.link == nil && validActiveWaitSetRecordFast(p, cursor.wait) && - (cursor.wait.g.park.phase == parkDetaching || cursor.wait.g.park.phase == parkReady) + (cursor.wait.g.park.phase == parkDetaching || cursor.wait.g.park.phase == parkReady || + cursor.phase == publishedEpochResolvePromote && + cursor.wait.resumeKind == resumeBindingCleanup && + cursor.wait.g.park.phase == parkConsumed) default: return false } @@ -530,6 +533,34 @@ func resolvePublishedEpochPromoteStep(sources *ExecutorSourceSet, p *P, cursor * if wait == nil || wait.workNext != cursor.nextWait { return false } + if wait.resumeKind == resumeBindingCleanup { + plan := (*ResumeCleanupPlan)(wait.resume) + if !validResumeCleanupPlan(wait, plan) { + return false + } + switch plan.phase { + case resumeCleanupBound: + return beginResumeCleanup(wait, plan) + case resumeCleanupRuntime: + // The unified runner must return ExecutorRunStepMaterialize so the + // direct runtime switch can remove exactly one typed queue node. + return false + default: + finalized, ok := advanceResumeCleanupCore(sources, p, wait, plan) + if !ok { + return false + } + if !finalized { + return true + } + if !promoteReadyWaitSet(sources, p, wait) { + return false + } + wait.workNext = nil + step.promoted = 1 + return advancePublishedEpochWaitAfterCleared(sources, cursor, p, step) + } + } wait.workNext = nil if wait.work == waitSetWorkAwaitingExternal { if wait.g.park.phase != parkDetaching { diff --git a/runtime/internal/coro/resume_cleanup.go b/runtime/internal/coro/resume_cleanup.go new file mode 100644 index 0000000000..5b1edd024b --- /dev/null +++ b/runtime/internal/coro/resume_cleanup.go @@ -0,0 +1,421 @@ +/* + * 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 "unsafe" + +// ResumeCleanupKind is the closed runtime materialization switch. It describes +// only the typed frame nodes which the source-neutral core cannot inspect. +// Source confirmation, result ownership, generation recycle, and runnable +// promotion remain common core phases. +type ResumeCleanupKind uint8 + +const ( + ResumeCleanupInvalid ResumeCleanupKind = iota + ResumeCleanupChannelDirect + ResumeCleanupChannelSelect +) + +type resumeCleanupPhase uint8 + +const ( + resumeCleanupIdle resumeCleanupPhase = iota + resumeCleanupBound + resumeCleanupRuntime + resumeCleanupConfirm + resumeCleanupClaim + resumeCleanupResult + resumeCleanupRecycle + resumeCleanupFinalize +) + +// ResumeCleanupBinding is a trusted compiler/runtime description of one +// frame-local typed cleanup. Entries is a logical-case array; an OperationID at +// Entries+i*Stride+IDOffset is zero for a disabled case. Context is interpreted +// only by the direct runtime switch selected by Kind. +// +// The descriptor is consumed at bind time and is never producer-visible. +type ResumeCleanupBinding struct { + Kind ResumeCleanupKind + Context unsafe.Pointer + Entries unsafe.Pointer + Source *ChannelOperationSource + Claim *SelectClaim + Count uint32 + Stride uintptr + IDOffset uintptr +} + +// ResumeCleanupPlan is compiler-spilled continuation state for bounded +// owner-side materialization. It is not a callback, interface, Future, heap +// task, or permanent G field. Runtime cleanup advances one logical case per +// ExecutorRunStepMaterialize; every later core call confirms or recycles at +// most one exact source generation. +type ResumeCleanupPlan struct { + packet *ResumePacket + source *ChannelOperationSource + claim *SelectClaim + context unsafe.Pointer + entries unsafe.Pointer + lease OperationResultLease + ticket ParkTicket + stride uintptr + idOffset uintptr + count uint32 + index uint32 + caseID uint32 + outcome ParkOutcome + kind ResumeCleanupKind + phase resumeCleanupPhase + small uint8 + _ [1]byte +} + +func validResumeCleanupKind(kind ResumeCleanupKind) bool { + return kind == ResumeCleanupChannelDirect || kind == ResumeCleanupChannelSelect +} + +func validResumeCleanupRange(entries unsafe.Pointer, count uint32, stride, idOffset uintptr) bool { + if entries == nil || count == 0 || count > MaxSelectOperationCases || + stride < unsafe.Sizeof(OperationID{}) || idOffset > stride-unsafe.Sizeof(OperationID{}) { + return false + } + last := uintptr(count - 1) + return last == 0 || stride <= (^uintptr(0)-idOffset)/last +} + +func resumeCleanupIDAt(plan *ResumeCleanupPlan, index uint32) *OperationID { + if plan == nil || index >= plan.count || + !validResumeCleanupRange(plan.entries, plan.count, plan.stride, plan.idOffset) { + return nil + } + return (*OperationID)(unsafe.Add(plan.entries, uintptr(index)*plan.stride+plan.idOffset)) +} + +func validResumeCleanupBindingForWait( + record *WaitSetRecord, + packet *ResumePacket, + plan *ResumeCleanupPlan, + binding ResumeCleanupBinding, +) bool { + if record == nil || packet == nil || plan == nil || *packet != (ResumePacket{}) || + *plan != (ResumeCleanupPlan{}) || !validResumeCleanupKind(binding.Kind) || + binding.Context == nil || binding.Source == nil || binding.Claim == nil || + !validResumeCleanupRange(binding.Entries, binding.Count, binding.Stride, binding.IDOffset) || + record.g == nil || record.g.runP == nil { + return false + } + p, state := record.g.runP, &record.g.park + if !validChannelOperationOwner(binding.Source, p) || p.channelSource != binding.Source || + selectClaimLoad(binding.Claim) != selectClaimOpen || state.expected == 0 || + state.expected > binding.Count { + return false + } + physical := uint32(0) + for index := uint32(0); index < binding.Count; index++ { + id := (*OperationID)(unsafe.Add( + binding.Entries, + uintptr(index)*binding.Stride+binding.IDOffset, + )) + if *id == (OperationID{}) { + continue + } + slot, ok := channelOperationSlotFor(binding.Source, *id) + if !ok || preemptLoad(&slot.generation) != id.Generation || + preemptLoad(&slot.state) != uint32(producerSourceActive) || + preemptLoad(&slot.external) != uint32(channelExternalExposed) || + slot.claim != binding.Claim || slot.record.id != *id || + slot.record.phase != operationActive || slot.record.link.operation != &slot.record || + slot.record.link.park != state || slot.record.link.wait != record || + slot.record.link.ticket != record.ticket || slot.record.link.caseID != index+1 { + return false + } + physical++ + } + if physical != state.expected { + return false + } + return binding.Kind != ResumeCleanupChannelDirect || + binding.Count == 1 && physical == 1 +} + +// BindWaitSetResumeCleanup opts a typed channel park into owner-side +// materialization after all exact source endpoints are Exposed and before any +// hchan queue node is made reachable. The complete range is audited once here; +// later reductions use only the retained fixed descriptor and exact IDs. +func BindWaitSetResumeCleanup( + record *WaitSetRecord, + packet *ResumePacket, + plan *ResumeCleanupPlan, + binding ResumeCleanupBinding, +) bool { + if record == nil || record.resume != nil || record.resumeKind != resumeBindingNone || + record.state != waitSetRecordCommitted || record.work != waitSetWorkIdle || + record.activePrev != nil || record.activeNext != nil || record.workNext != nil || + record.g == nil || !ValidG(record.g) || !validParkTicket(record.ticket) || + record.g.park.ticket != record.ticket || record.g.park.phase != parkParked || + record.g.pending.kind != pendingParkSet || record.g.pending.from != record.g.active || + record.g.active == nil || record.g.active.parkWait != record { + return false + } + p := record.g.runP + if p == nil || !validExecutorDriverForP(p.executor, p) || p.current != record.g || + !validResumeCleanupBindingForWait(record, packet, plan, binding) { + return false + } + *packet = ResumePacket{ + ticket: record.ticket, + state: resumePacketBound, + } + *plan = ResumeCleanupPlan{ + packet: packet, + source: binding.Source, + claim: binding.Claim, + context: binding.Context, + entries: binding.Entries, + ticket: record.ticket, + stride: binding.Stride, + idOffset: binding.IDOffset, + count: binding.Count, + kind: binding.Kind, + phase: resumeCleanupBound, + } + record.resume = unsafe.Pointer(plan) + record.resumeKind = resumeBindingCleanup + return validResumeCleanupPlan(record, plan) +} + +func validResumeCleanupPlan(record *WaitSetRecord, plan *ResumeCleanupPlan) bool { + if record == nil || plan == nil || record.resume != unsafe.Pointer(plan) || + record.resumeKind != resumeBindingCleanup || plan.packet == nil || + !validBoundResumePacket(plan.packet, record.ticket) || + plan.source == nil || plan.claim == nil || plan.context == nil || + plan.ticket != record.ticket || !validResumeCleanupKind(plan.kind) || + !validResumeCleanupRange(plan.entries, plan.count, plan.stride, plan.idOffset) || + plan.index > plan.count || plan.outcome > ParkOutcomeDefault { + return false + } + switch plan.phase { + case resumeCleanupBound: + return plan.index == 0 && plan.caseID == 0 && plan.outcome == ParkOutcomePending && + plan.lease == (OperationResultLease{}) && plan.small == ResumeSmallInvalid && + record.g != nil && + (record.g.park.phase == parkParked || record.g.park.phase == parkDetaching || + record.g.park.phase == parkReady) + case resumeCleanupRuntime: + return plan.index < plan.count && plan.outcome != ParkOutcomePending && + record.g != nil && record.g.park.phase == parkConsumed + case resumeCleanupConfirm, resumeCleanupRecycle: + return plan.index < plan.count && plan.outcome != ParkOutcomePending && + record.g != nil && record.g.park.phase == parkConsumed + case resumeCleanupClaim, resumeCleanupResult, resumeCleanupFinalize: + return plan.index == 0 && plan.outcome != ParkOutcomePending && + record.g != nil && record.g.park.phase == parkConsumed + default: + return false + } +} + +func beginResumeCleanup(record *WaitSetRecord, plan *ResumeCleanupPlan) bool { + if !validResumeCleanupPlan(record, plan) || plan.phase != resumeCleanupBound { + return false + } + state := &record.g.park + physicalOutcome := state.outcome + outcome, caseID, lease, ok := ConsumeParkSet(state, record.ticket) + if !ok || outcome != physicalOutcome { + return false + } + switch outcome { + case ParkOutcomeCompleted: + if caseID == 0 || caseID > plan.count || !lease.Valid() { + return false + } + id := resumeCleanupIDAt(plan, caseID-1) + leaseID, leaseOK := lease.ID() + if id == nil || !leaseOK || *id != leaseID { + return false + } + case ParkOutcomeCanceled: + if caseID != 0 || lease.Valid() { + return false + } + default: + return false + } + plan.outcome, plan.caseID, plan.lease = outcome, caseID, lease + plan.phase = resumeCleanupRuntime + return validResumeCleanupPlan(record, plan) +} + +// ResumeCleanupStep is one direct-runtime typed cleanup reduction. The +// unexported plan token binds CommitResumeCleanupStep to the exact outstanding +// frame/index; callers can inspect only the closed kind, context, index, and +// logical decision needed by their direct switch. +type ResumeCleanupStep struct { + Kind ResumeCleanupKind + Context unsafe.Pointer + Index uint32 + WinnerCase uint32 + Outcome ParkOutcome + plan *ResumeCleanupPlan +} + +func pendingResumeCleanupStep(driver *ExecutorDriver) (ResumeCleanupStep, bool) { + if driver == nil || driver.poll.resolve.phase != publishedEpochResolvePromote { + return ResumeCleanupStep{}, false + } + record := driver.poll.resolve.wait + if record == nil || record.resumeKind != resumeBindingCleanup { + return ResumeCleanupStep{}, false + } + plan := (*ResumeCleanupPlan)(record.resume) + if !validResumeCleanupPlan(record, plan) || plan.phase != resumeCleanupRuntime { + return ResumeCleanupStep{}, false + } + return ResumeCleanupStep{ + Kind: plan.kind, + Context: plan.context, + Index: plan.index, + WinnerCase: plan.caseID, + Outcome: plan.outcome, + plan: plan, + }, true +} + +// CommitResumeCleanupStep completes the exact outstanding typed runtime case. +// small is zero for every loser/canceled case and is the closed runtime status +// for the physical winner. +func CommitResumeCleanupStep(step ResumeCleanupStep, small uint8) bool { + plan := step.plan + if plan == nil || plan.phase != resumeCleanupRuntime || plan.index != step.Index || + plan.kind != step.Kind || plan.context != step.Context || + plan.caseID != step.WinnerCase || plan.outcome != step.Outcome || + step.Index >= plan.count { + return false + } + selected := plan.outcome == ParkOutcomeCompleted && plan.caseID == plan.index+1 + if selected { + if small == ResumeSmallInvalid || plan.small != ResumeSmallInvalid { + return false + } + plan.small = small + } else if small != ResumeSmallInvalid { + return false + } + plan.index++ + if plan.index == plan.count { + plan.index = 0 + plan.phase = resumeCleanupConfirm + } + return true +} + +// advanceResumeCleanupCore performs one source-neutral bounded reduction after +// the runtime has removed every typed queue node. +func advanceResumeCleanupCore( + sources *ExecutorSourceSet, + p *P, + record *WaitSetRecord, + plan *ResumeCleanupPlan, +) (finalized bool, ok bool) { + if sources == nil || p == nil || !validExecutorSourceSet(sources, p) || + sources.channel != plan.source || !validResumeCleanupPlan(record, plan) { + return false, false + } + switch plan.phase { + case resumeCleanupConfirm: + if plan.index < plan.count { + id := resumeCleanupIDAt(plan, plan.index) + if id == nil || *id != (OperationID{}) && !plan.source.ConfirmQuiesced(p, *id) { + return false, false + } + plan.index++ + if plan.index == plan.count { + plan.index = 0 + plan.phase = resumeCleanupClaim + } + return false, true + } + case resumeCleanupClaim: + if !plan.source.ResetSelectClaim(p, plan.claim) { + return false, false + } + plan.phase = resumeCleanupResult + return false, true + case resumeCleanupResult: + if plan.lease.Valid() { + var released bool + if plan.outcome == ParkOutcomeCompleted { + released = plan.source.TakeResult(p, plan.lease) + } else { + released = plan.source.DiscardResult(p, plan.lease) + } + if !released { + return false, false + } + } + plan.lease = OperationResultLease{} + plan.phase = resumeCleanupRecycle + return false, true + case resumeCleanupRecycle: + if plan.index < plan.count { + id := resumeCleanupIDAt(plan, plan.index) + if id == nil { + return false, false + } + if *id != (OperationID{}) { + if !plan.source.Recycle(p, *id) { + return false, false + } + *id = OperationID{} + } + plan.index++ + if plan.index == plan.count { + plan.index = 0 + plan.phase = resumeCleanupFinalize + } + return false, true + } + case resumeCleanupFinalize: + if plan.outcome == ParkOutcomeCompleted && plan.small == ResumeSmallInvalid || + plan.outcome != ParkOutcomeCompleted && plan.small != ResumeSmallInvalid { + return false, false + } + packet, ticket, outcome, caseID, small := plan.packet, plan.ticket, plan.outcome, plan.caseID, plan.small + if !materializedParkState(&record.g.park, ticket, outcome, caseID) { + return false, false + } + result := ResumeResultNone + if outcome == ParkOutcomeCompleted { + result = ResumeResultChannel + } + *packet = ResumePacket{ + ticket: ticket, + caseID: caseID, + outcome: outcome, + result: result, + small: small, + state: resumePacketMaterialized, + } + record.resume = unsafe.Pointer(packet) + record.resumeKind = resumeBindingMaterialized + *plan = ResumeCleanupPlan{} + return true, validMaterializedResumePacket(packet) + } + return false, false +} diff --git a/runtime/internal/coro/resume_cleanup_test.go b/runtime/internal/coro/resume_cleanup_test.go new file mode 100644 index 0000000000..f960c1767b --- /dev/null +++ b/runtime/internal/coro/resume_cleanup_test.go @@ -0,0 +1,172 @@ +/* + * 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 ( + "reflect" + "runtime" + "testing" + "unsafe" +) + +func TestResumeCleanupPlanHasNoDynamicGoPayload(t *testing.T) { + var check func(string, reflect.Type) + check = func(path string, typ reflect.Type) { + switch typ.Kind() { + case reflect.Array: + check(path, typ.Elem()) + case reflect.Struct: + for index := 0; index < typ.NumField(); index++ { + field := typ.Field(index) + check(path+"."+field.Name, field.Type) + } + case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, + reflect.Slice, reflect.String: + t.Fatalf("%s has dynamic Go payload type %s", path, typ) + } + } + for _, typ := range []reflect.Type{ + reflect.TypeOf(ResumeCleanupPlan{}), + reflect.TypeOf(ResumeCleanupBinding{}), + reflect.TypeOf(ResumeCleanupStep{}), + } { + check(typ.Name(), typ) + } +} + +func TestTypedResumeCleanupIsBoundedAndPNeutral(t *testing.T) { + var ( + packet ResumePacket + plan ResumeCleanupPlan + token byte + ) + fixture := newChannelClaimCoreFixtureBeforeResume( + t, + "typed-resume-cleanup", + []uint32{1, 2}, + true, + 0, + func(fixture *channelClaimCoreFixture) { + binding := ResumeCleanupBinding{ + Kind: ResumeCleanupChannelSelect, + Context: unsafe.Pointer(&token), + Entries: unsafe.Pointer(&fixture.ids[0]), + Source: fixture.source, + Claim: fixture.claim, + Count: uint32(len(fixture.ids)), + Stride: unsafe.Sizeof(OperationID{}), + } + malformed := binding + malformed.Count = 1 + if BindWaitSetResumeCleanup(&fixture.wait, &packet, &plan, malformed) || + packet != (ResumePacket{}) || plan != (ResumeCleanupPlan{}) || + fixture.wait.resume != nil || fixture.wait.resumeKind != resumeBindingNone { + t.Fatal("cleanup binding accepted a truncated physical range") + } + if !BindWaitSetResumeCleanup(&fixture.wait, &packet, &plan, binding) { + t.Fatal("bind typed resume cleanup") + } + }, + ) + externallyCommitChannelCandidate(t, fixture, 1) + requestChannelClaimCoreFixture(t, fixture) + + materialized := uint32(0) + for reduction := 0; reduction < 10000; reduction++ { + step, ok := NextExecutorRunStep(fixture.driver) + if !ok { + t.Fatalf("select cleanup reduction %d", reduction) + } + switch step.Kind { + case ExecutorRunStepSource: + if step.Poll.Complete { + if materialized != uint32(len(fixture.ids)) { + t.Fatalf("typed cleanup runtime reductions = %d, want %d", materialized, len(fixture.ids)) + } + goto complete + } + case ExecutorRunStepMaterialize: + if step.Cleanup.Kind != ResumeCleanupChannelSelect || + step.Cleanup.Context != unsafe.Pointer(&token) || + step.Cleanup.Index != materialized || + step.Cleanup.WinnerCase != 2 || + step.Cleanup.Outcome != ParkOutcomeCompleted { + t.Fatalf("typed cleanup step %d = %+v", materialized, step.Cleanup) + } + small := ResumeSmallInvalid + if step.Cleanup.Index == 1 { + small = 3 + } + if !CommitResumeCleanupStep(step.Cleanup, small) { + t.Fatalf("commit typed cleanup step %d", materialized) + } + materialized++ + default: + t.Fatalf("unexpected typed cleanup runner step %d", step.Kind) + } + } + t.Fatal("typed resume cleanup did not complete") + +complete: + if !EnterExecutorRunCompatibility(fixture.driver) || + packet.state != resumePacketMaterialized || + packet.source != (OperationID{}) || + packet.outcome != ParkOutcomeCompleted || + packet.caseID != 2 || + packet.result != ResumeResultChannel || + packet.small != 3 || + plan != (ResumeCleanupPlan{}) || + fixture.wait != (WaitSetRecord{}) || + fixture.ids[0] != (OperationID{}) || + fixture.ids[1] != (OperationID{}) || + !channelOperationSourceEmpty(fixture.source, fixture.p) { + t.Fatalf("typed cleanup retained owner state: packet=%+v plan=%+v wait=%+v ids=%+v empty=%t", + packet, plan, fixture.wait, fixture.ids, + channelOperationSourceEmpty(fixture.source, fixture.p)) + } + + target := new(P) + var mailbox RunnableTransferMailbox + if !BindRunnableTransferMailbox(&mailbox, target) { + t.Fatal("bind typed cleanup transfer mailbox") + } + transfer, published := PublishPNeutralRunnable(&mailbox, fixture.p, fixture.task.g) + if !published || !transfer.Valid() || !ImportPNeutralRunnable(&mailbox, target, transfer) { + t.Fatalf("transfer typed cleanup packet = (%+v, %t)", transfer, published) + } + if g, ok := NextRunnable(target); !ok || g != fixture.task.g { + t.Fatal("dequeue transferred typed cleanup task") + } + action := beginWaitTestResume(t, target, fixture.task) + outcome, caseID, cancel, result, small, taken := TakeResumePacket( + fixture.task.g, + fixture.ticket, + &packet, + nil, + ) + if !taken || outcome != ParkOutcomeCompleted || caseID != 2 || + cancel != TaskCancelNone || result != ResumeResultChannel || small != 3 { + t.Fatalf("take typed cleanup packet = outcome:%d case:%d cancel:%d result:%d small:%d taken:%t", + outcome, caseID, cancel, result, small, taken) + } + finishWaitTestTask(t, target, fixture.task, action) + closeTestExecutorDriver(t, fixture.driver) + if !fixture.source.CanRelease() || !fixture.registry.CanRelease() { + t.Fatal("typed cleanup retained source/registry") + } + runtime.KeepAlive(fixture.task.frame.memory) +} diff --git a/runtime/internal/coro/resume_packet.go b/runtime/internal/coro/resume_packet.go index 12d11cba6a..d26e5b275e 100644 --- a/runtime/internal/coro/resume_packet.go +++ b/runtime/internal/coro/resume_packet.go @@ -19,16 +19,19 @@ package coro import "unsafe" // ResumeResultKind is the closed pointer-free result union retained in a -// compiler-provided ResumePacket. Typed channel data already lives in its -// frame slots and therefore uses ResumeResultNone. +// compiler-provided ResumePacket. Typed channel element data already lives in +// frame slots; ResumeResultChannel carries only the selected operation status. type ResumeResultKind uint8 const ( ResumeResultNone ResumeResultKind = iota ResumeResultScalar ResumeResultPoll + ResumeResultChannel ) +const ResumeSmallInvalid uint8 = 0 + type resumePacketState uint8 const ( @@ -37,6 +40,15 @@ const ( resumePacketMaterialized ) +type resumeBindingKind uint8 + +const ( + resumeBindingNone resumeBindingKind = iota + resumeBindingSingle + resumeBindingCleanup + resumeBindingMaterialized +) + // ResumePacket is stable storage in the direct-parking LLVM coroutine frame. // Bound temporarily identifies one exact old-route source generation. // Materialized is P-neutral: source is zero, any payload has been copied by the @@ -52,7 +64,7 @@ type ResumePacket struct { caseID uint32 outcome ParkOutcome result ResumeResultKind - poll PollOperationResult + small uint8 state resumePacketState } @@ -107,7 +119,7 @@ func validBoundResumePacket(packet *ResumePacket, ticket ParkTicket) bool { validParkTicket(ticket) && supportedSingleResumeSource(packet.source) && packet.scalar == (ScalarResultPayloadV1{}) && packet.caseID == 0 && packet.outcome == ParkOutcomePending && packet.result == ResumeResultNone && - packet.poll == PollOperationResultInvalid + packet.small == ResumeSmallInvalid } func validMaterializedResumePacket(packet *ResumePacket) bool { @@ -129,23 +141,40 @@ func validMaterializedResumePacket(packet *ResumePacket) bool { } if packet.outcome != ParkOutcomeCompleted { return packet.result == ResumeResultNone && packet.scalar == (ScalarResultPayloadV1{}) && - packet.poll == PollOperationResultInvalid + packet.small == ResumeSmallInvalid } switch packet.result { case ResumeResultNone: - return packet.scalar == (ScalarResultPayloadV1{}) && packet.poll == PollOperationResultInvalid + return packet.scalar == (ScalarResultPayloadV1{}) && packet.small == ResumeSmallInvalid case ResumeResultScalar: - return packet.scalar.Valid() && packet.poll == PollOperationResultInvalid + return packet.scalar.Valid() && packet.small == ResumeSmallInvalid case ResumeResultPoll: return packet.scalar == (ScalarResultPayloadV1{}) && - packet.poll >= PollOperationReady && packet.poll <= PollOperationTimeout + PollOperationResult(packet.small) >= PollOperationReady && + PollOperationResult(packet.small) <= PollOperationTimeout + case ResumeResultChannel: + return packet.scalar == (ScalarResultPayloadV1{}) && packet.small != ResumeSmallInvalid default: return false } } func validWaitSetResumeBinding(record *WaitSetRecord) bool { - return record != nil && (record.resume == nil || validBoundResumePacket(record.resume, record.ticket)) + if record == nil { + return false + } + switch record.resumeKind { + case resumeBindingNone: + return record.resume == nil + case resumeBindingSingle: + return validBoundResumePacket((*ResumePacket)(record.resume), record.ticket) + case resumeBindingCleanup: + return validResumeCleanupPlan(record, (*ResumeCleanupPlan)(record.resume)) + case resumeBindingMaterialized: + return validMaterializedResumePacket((*ResumePacket)(record.resume)) + default: + return false + } } // BindSingleWaitSetResumePacket opts one zero/one-source direct park into @@ -156,7 +185,8 @@ func validWaitSetResumeBinding(record *WaitSetRecord) bool { // and typed hchan queue node must be retired before transfer; accepting it here // would make a partially neutral G look stealable. func BindSingleWaitSetResumePacket(record *WaitSetRecord, packet *ResumePacket, source OperationID) bool { - if record == nil || packet == nil || *packet != (ResumePacket{}) || record.resume != nil || + if record == nil || packet == nil || *packet != (ResumePacket{}) || + record.resume != nil || record.resumeKind != resumeBindingNone || record.state != waitSetRecordCommitted || record.work != waitSetWorkIdle || record.activePrev != nil || record.activeNext != nil || record.workNext != nil || record.g == nil || !ValidG(record.g) || !validParkTicket(record.ticket) || @@ -185,7 +215,8 @@ func BindSingleWaitSetResumePacket(record *WaitSetRecord, packet *ResumePacket, source: source, state: resumePacketBound, } - record.resume = packet + record.resume = unsafe.Pointer(packet) + record.resumeKind = resumeBindingSingle return true } @@ -305,10 +336,11 @@ func materializePollResume( func materializeSingleResumePacket(sources *ExecutorSourceSet, p *P, record *WaitSetRecord) bool { if sources == nil || !validExecutorSourceSet(sources, p) || record == nil || !validActiveWaitSetRecordFast(p, record) || record.work != waitSetWorkResolving || - record.g.park.phase != parkReady || !validBoundResumePacket(record.resume, record.ticket) { + record.g.park.phase != parkReady || record.resumeKind != resumeBindingSingle || + !validBoundResumePacket((*ResumePacket)(record.resume), record.ticket) { return false } - packet, state := record.resume, &record.g.park + packet, state := (*ResumePacket)(record.resume), &record.g.park sourceID := packet.source physicalOutcome := state.outcome if physicalOutcome == ParkOutcomeCompleted { @@ -340,7 +372,7 @@ func materializeSingleResumePacket(sources *ExecutorSourceSet, p *P, record *Wai var ( result ResumeResultKind scalar ScalarResultPayloadV1 - poll PollOperationResult + small uint8 ) if sourceID == (OperationID{}) { if (physicalOutcome != ParkOutcomeCanceled && physicalOutcome != ParkOutcomeDefault) || lease.Valid() { @@ -368,7 +400,10 @@ func materializeSingleResumePacket(sources *ExecutorSourceSet, p *P, record *Wai return false } case OperationSourcePoll: - var out *PollOperationResult + var ( + poll PollOperationResult + out *PollOperationResult + ) if !discard { out = &poll } @@ -377,6 +412,7 @@ func materializeSingleResumePacket(sources *ExecutorSourceSet, p *P, record *Wai } if !discard { result = ResumeResultPoll + small = uint8(poll) } default: return false @@ -394,9 +430,10 @@ func materializeSingleResumePacket(sources *ExecutorSourceSet, p *P, record *Wai caseID: caseID, outcome: outcome, result: result, - poll: poll, + small: small, state: resumePacketMaterialized, } + record.resumeKind = resumeBindingMaterialized return validMaterializedResumePacket(packet) } @@ -414,46 +451,46 @@ func TakeResumePacket( caseID uint32, task TaskCancelKind, result ResumeResultKind, - poll PollOperationResult, + small uint8, ok bool, ) { if !ValidG(g) || g.runP == nil || !validMaterializedResumePacket(packet) || packet.ticket != expected { - return ParkOutcomePending, 0, TaskCancelNone, ResumeResultNone, PollOperationResultInvalid, false + return ParkOutcomePending, 0, TaskCancelNone, ResumeResultNone, ResumeSmallInvalid, false } p := g.runP if p.current != g || !p.inResume || g.state != GRunning || p.runDecisionTaken || !expectedAction(p, g, p.action, ActionResume) || !validRunDecision(p.runDecision) { - return ParkOutcomePending, 0, TaskCancelNone, ResumeResultNone, PollOperationResultInvalid, false + return ParkOutcomePending, 0, TaskCancelNone, ResumeResultNone, ResumeSmallInvalid, false } decision := &p.runDecision if !decision.materialized || decision.g != g || decision.ticket != expected || g.park.phase != parkMaterialized || g.park.ticket != expected || packet.outcome != g.park.outcome || packet.caseID != g.park.winnerCase { - return ParkOutcomePending, 0, TaskCancelNone, ResumeResultNone, PollOperationResultInvalid, false + return ParkOutcomePending, 0, TaskCancelNone, ResumeResultNone, ResumeSmallInvalid, false } var scalar ScalarResultPayloadV1 if decision.outcome == ParkOutcomeCompleted { if decision.caseID != packet.caseID { - return ParkOutcomePending, 0, TaskCancelNone, ResumeResultNone, PollOperationResultInvalid, false + return ParkOutcomePending, 0, TaskCancelNone, ResumeResultNone, ResumeSmallInvalid, false } switch packet.result { case ResumeResultScalar: if scalarOut == nil { - return ParkOutcomePending, 0, TaskCancelNone, ResumeResultNone, PollOperationResultInvalid, false + return ParkOutcomePending, 0, TaskCancelNone, ResumeResultNone, ResumeSmallInvalid, false } scalar = packet.scalar - case ResumeResultPoll, ResumeResultNone: + case ResumeResultPoll, ResumeResultChannel, ResumeResultNone: if scalarOut != nil { - return ParkOutcomePending, 0, TaskCancelNone, ResumeResultNone, PollOperationResultInvalid, false + return ParkOutcomePending, 0, TaskCancelNone, ResumeResultNone, ResumeSmallInvalid, false } default: - return ParkOutcomePending, 0, TaskCancelNone, ResumeResultNone, PollOperationResultInvalid, false + return ParkOutcomePending, 0, TaskCancelNone, ResumeResultNone, ResumeSmallInvalid, false } - result, poll = packet.result, packet.poll + result, small = packet.result, packet.small } if !deliverMaterializedParkResume(&g.park, expected) { - return ParkOutcomePending, 0, TaskCancelNone, ResumeResultNone, PollOperationResultInvalid, false + return ParkOutcomePending, 0, TaskCancelNone, ResumeResultNone, ResumeSmallInvalid, false } if scalarOut != nil { *scalarOut = scalar @@ -462,5 +499,5 @@ func TakeResumePacket( *packet = ResumePacket{} p.runDecision = RunDecision{} p.runDecisionTaken = true - return outcome, caseID, task, result, poll, true + return outcome, caseID, task, result, small, true } diff --git a/runtime/internal/coro/resume_packet_test.go b/runtime/internal/coro/resume_packet_test.go index 64e41a92b9..0c58aed2d5 100644 --- a/runtime/internal/coro/resume_packet_test.go +++ b/runtime/internal/coro/resume_packet_test.go @@ -116,11 +116,11 @@ func TestZeroSourceResumePacketMaterializesDefault(t *testing.T) { t.Fatal("dequeue zero-source packet resume") } action = beginWaitTestResume(t, targetP, task) - outcome, caseID, cancel, result, poll, taken := TakeResumePacket(task.g, ticket, packet, nil) + outcome, caseID, cancel, result, small, taken := TakeResumePacket(task.g, ticket, packet, nil) if !taken || outcome != ParkOutcomeDefault || caseID != 31 || cancel != TaskCancelNone || - result != ResumeResultNone || poll != PollOperationResultInvalid { - t.Fatalf("take zero-source packet = outcome:%d case:%d cancel:%d result:%d poll:%d taken:%t", - outcome, caseID, cancel, result, poll, taken) + result != ResumeResultNone || small != ResumeSmallInvalid { + t.Fatalf("take zero-source packet = outcome:%d case:%d cancel:%d result:%d small:%d taken:%t", + outcome, caseID, cancel, result, small, taken) } finishWaitTestTask(t, targetP, task, action) closeTestExecutorDriver(t, driver) @@ -163,14 +163,14 @@ func TestManualResumePacketRetiresOldRouteBeforePTransfer(t *testing.T) { t.Fatal("dequeue transferred packet task") } action := beginWaitTestResume(t, targetP, task) - outcome, caseID, cancel, result, poll, taken := TakeResumePacket(task.g, ticket, packet, nil) + outcome, caseID, cancel, result, small, taken := TakeResumePacket(task.g, ticket, packet, nil) current, executor, route, currentOK := CurrentExecutorDriver(task.g) if !taken || outcome != ParkOutcomeCompleted || caseID != 17 || cancel != TaskCancelNone || - result != ResumeResultNone || poll != PollOperationResultInvalid || + result != ResumeResultNone || small != ResumeSmallInvalid || *packet != (ResumePacket{}) || task.g.park.phase != parkDelivered || current != nil || executor != (ExecutorHandle{}) || route != 0 || currentOK { - t.Fatalf("take transferred packet = outcome:%d case:%d cancel:%d result:%d poll:%d taken:%t packet:%+v phase:%d", - outcome, caseID, cancel, result, poll, taken, *packet, task.g.park.phase) + t.Fatalf("take transferred packet = outcome:%d case:%d cancel:%d result:%d small:%d taken:%t packet:%+v phase:%d", + outcome, caseID, cancel, result, small, taken, *packet, task.g.park.phase) } finishWaitTestTask(t, targetP, task, action) closeTestExecutorDriver(t, driver) @@ -204,12 +204,12 @@ func TestTaskCancelAfterResumePacketMigrationNeverRevisitsOldSource(t *testing.T t.Fatal("dequeue late-cancel packet task") } action := beginWaitTestResume(t, targetP, task) - outcome, caseID, cancel, result, poll, taken := TakeResumePacket(task.g, ticket, packet, nil) + outcome, caseID, cancel, result, small, taken := TakeResumePacket(task.g, ticket, packet, nil) if !taken || outcome != ParkOutcomeCanceled || caseID != 0 || cancel != TaskCancelAbort || - result != ResumeResultNone || poll != PollOperationResultInvalid || + result != ResumeResultNone || small != ResumeSmallInvalid || *packet != (ResumePacket{}) || task.g.park.taskCancelPhase != taskCancelCleanup { - t.Fatalf("take late-cancel packet = outcome:%d case:%d cancel:%d result:%d poll:%d taken:%t phase:%d", - outcome, caseID, cancel, result, poll, taken, task.g.park.taskCancelPhase) + t.Fatalf("take late-cancel packet = outcome:%d case:%d cancel:%d result:%d small:%d taken:%t phase:%d", + outcome, caseID, cancel, result, small, taken, task.g.park.taskCancelPhase) } finishWaitTestTask(t, targetP, task, action) if !AcknowledgeTaskCancellation(task.g, TaskCancelAbort) || !TerminalG(targetP, task.g) { @@ -279,11 +279,11 @@ func TestWorkerResumePacketCopiesScalarBeforePTransfer(t *testing.T) { } action = beginWaitTestResume(t, targetP, task) var got ScalarResultPayloadV1 - outcome, caseID, cancel, result, poll, taken := TakeResumePacket(task.g, ticket, packet, &got) + outcome, caseID, cancel, result, small, taken := TakeResumePacket(task.g, ticket, packet, &got) if !taken || outcome != ParkOutcomeCompleted || caseID != 23 || cancel != TaskCancelNone || - result != ResumeResultScalar || poll != PollOperationResultInvalid || got != payload { - t.Fatalf("take packet worker = outcome:%d case:%d cancel:%d result:%d poll:%d taken:%t payload:%+v", - outcome, caseID, cancel, result, poll, taken, got) + result != ResumeResultScalar || small != ResumeSmallInvalid || got != payload { + t.Fatalf("take packet worker = outcome:%d case:%d cancel:%d result:%d small:%d taken:%t payload:%+v", + outcome, caseID, cancel, result, small, taken, got) } finishWaitTestTask(t, targetP, task, action) closeTestExecutorDriver(t, driver) diff --git a/runtime/internal/coro/run_slice.go b/runtime/internal/coro/run_slice.go index 1e103cbce1..a51eb79f44 100644 --- a/runtime/internal/coro/run_slice.go +++ b/runtime/internal/coro/run_slice.go @@ -78,6 +78,7 @@ type ExecutorRunStepKind uint8 const ( ExecutorRunStepInvalid ExecutorRunStepKind = iota ExecutorRunStepSource + ExecutorRunStepMaterialize ExecutorRunStepDispatch ExecutorRunStepAction ExecutorRunStepDestroyCommit @@ -87,10 +88,11 @@ const ( // ExecutorRunStep carries no callback or interface value. Action handles are // live only for Dispatch/Action. DestroyCommit is always handle-free. type ExecutorRunStep struct { - Kind ExecutorRunStepKind - G *G - Action Action - Poll ExecutorPollProgress + Kind ExecutorRunStepKind + G *G + Action Action + Poll ExecutorPollProgress + Cleanup ResumeCleanupStep } func executorRunExternalSourceRequested(driver *ExecutorDriver) bool { @@ -178,6 +180,9 @@ 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 { + return ExecutorRunStep{Kind: ExecutorRunStepMaterialize, Cleanup: cleanup}, true + } return serviceExecutorRunSource(driver, now, withDeadline) } if driver.run.readyDebt { diff --git a/runtime/internal/coro/scheduler.go b/runtime/internal/coro/scheduler.go index 8c64c403a3..333c890968 100644 --- a/runtime/internal/coro/scheduler.go +++ b/runtime/internal/coro/scheduler.go @@ -319,6 +319,36 @@ func PollPreempt(g *G) bool { return valid && requested } +// acknowledgeSuspendedGPreempt consumes a request which was already satisfied +// by an actual Yield/Park suspension. Keeping it on a stable runnable/waiting G +// would spuriously prevent P-neutral transfer even though the scheduler has +// regained ownership and every executor/source request remains independently +// sticky. No producer is allowed to retain a dynamic G as an asynchronous +// handle, so only the scheduler can race this depth-zero transition. +func acknowledgeSuspendedGPreempt(g *G) bool { + if g == nil { + return false + } + gate := preemptAddress(g) + for { + word := preemptLoad(gate) + if preemptWordDepth(word) != 0 { + return false + } + switch preemptWordState(word) { + case preemptIdle: + return true + case preemptRequested: + idle := word&^preemptStateMask | preemptIdle + if preemptCompareAndSwap(gate, word, idle) { + return true + } + default: + return false + } + } +} + // RequestSchedule is the legacy/internal P request gate. It coalesces one // request without reading scheduler-owned queue or current-G fields, but it has // no retained platform doorbell and is therefore rejected after BindExecutor. @@ -957,6 +987,9 @@ func Resumed(p *P, g *G, action Action) (Action, bool) { (p.readyTail != nil && p.readyTail.nextReady != nil) { return Action{}, false } + if !acknowledgeSuspendedGPreempt(g) { + return Action{}, false + } g.state = GRunnable g.runP = nil p.current = nil @@ -971,6 +1004,9 @@ func Resumed(p *P, g *G, action Action) (Action, bool) { if g.queued || g.nextReady != nil || !validParkWaitQueueHeader(p) || !validAffectedWaitQueueHeader(p) { return Action{}, false } + if !acknowledgeSuspendedGPreempt(g) { + return Action{}, false + } g.state = GWaiting g.runP = nil p.current = nil diff --git a/runtime/internal/coro/wait_set_record.go b/runtime/internal/coro/wait_set_record.go index 94da5dcff5..1d576583ec 100644 --- a/runtime/internal/coro/wait_set_record.go +++ b/runtime/internal/coro/wait_set_record.go @@ -16,6 +16,8 @@ package coro +import "unsafe" + // waitSetRecordState is owner-P-only. A WaitSetRecord is caller storage which // is live only across one direct V2 park; in production the compiler spills it // into the direct-parking LLVM coroutine frame. Tests and bootstrap adapters @@ -59,11 +61,12 @@ type WaitSetRecord struct { activePrev *WaitSetRecord activeNext *WaitSetRecord workNext *WaitSetRecord - resume *ResumePacket + resume unsafe.Pointer ticket ParkTicket state waitSetRecordState work waitSetWorkState - _ [2]byte + resumeKind resumeBindingKind + _ [1]byte } // PrepareWaitSetRecord binds zero caller storage to one preparing logical @@ -83,7 +86,8 @@ func PrepareWaitSetRecord(record *WaitSetRecord, g *G, ticket ParkTicket) bool { func validPreparingWaitSetRecord(record *WaitSetRecord, state *ParkState, ticket ParkTicket) bool { return record != nil && record.g != nil && &record.g.park == state && record.ticket == ticket && record.state == waitSetRecordPreparing && record.work == waitSetWorkIdle && - record.activePrev == nil && record.activeNext == nil && record.workNext == nil && record.resume == nil + record.activePrev == nil && record.activeNext == nil && record.workNext == nil && + record.resume == nil && record.resumeKind == resumeBindingNone } func validCommittedWaitSetRecord(record *WaitSetRecord, g *G, frame *Frame) bool { @@ -125,6 +129,8 @@ func validActiveParkStateHeader(state *ParkState, ticket ParkTicket) bool { if state.attached != 0 || state.head != nil { return false } + case parkConsumed, parkMaterialized: + return validParkState(state) default: return false } @@ -418,7 +424,8 @@ func appendRunnableUnchecked(p *P, g *G) { func promoteReadyWaitSet(sources *ExecutorSourceSet, p *P, record *WaitSetRecord) bool { if !validActiveWaitSetRecordFast(p, record) || record.work != waitSetWorkResolving || - record.g.park.phase != parkReady || !validReadyQueueHeader(p) { + (record.g.park.phase != parkReady && record.g.park.phase != parkMaterialized) || + !validReadyQueueHeader(p) { return false } g := record.g @@ -428,7 +435,11 @@ func promoteReadyWaitSet(sources *ExecutorSourceSet, p *P, record *WaitSetRecord (schedule != scheduleIdle && schedule != scheduleRequested) { return false } - if record.resume != nil && !materializeSingleResumePacket(sources, p, record) { + if record.resumeKind == resumeBindingSingle && !materializeSingleResumePacket(sources, p, record) { + return false + } + if record.resumeKind == resumeBindingCleanup || + record.g.park.phase == parkMaterialized && record.resumeKind != resumeBindingMaterialized { return false } previous, next := record.activePrev, record.activeNext @@ -507,7 +518,7 @@ func promoteResolvedWaitSets(p *P, batch *WaitSetRecord) (promoted int, ok bool) func ReleasePreparedWaitSetRecord(record *WaitSetRecord) bool { if record == nil || record.state != waitSetRecordPreparing || record.work != waitSetWorkIdle || record.g == nil || record.activePrev != nil || record.activeNext != nil || record.workNext != nil || - record.resume != nil || + record.resume != nil || record.resumeKind != resumeBindingNone || !validParkState(&record.g.park) || record.g.park.ticket != record.ticket || record.g.park.attached != 0 || record.g.park.head != nil || (record.g.park.phase != parkReady && record.g.park.phase != parkConsumed) { return false diff --git a/runtime/internal/runtime/coro_channel_adapter_test.go b/runtime/internal/runtime/coro_channel_adapter_test.go index 3f4e4ed5e7..aaf2623de3 100644 --- a/runtime/internal/runtime/coro_channel_adapter_test.go +++ b/runtime/internal/runtime/coro_channel_adapter_test.go @@ -257,12 +257,24 @@ func yieldCoroChannelAdapterFrame(t *testing.T, p *coro.P, frame *coroChannelAda func pollCoroChannelAdapterExecutor(t *testing.T, driver *coro.ExecutorDriver) { t.Helper() for step := 0; ; step++ { - progress, ok := coro.PollExecutorSlice(driver, 1) + runStep, ok := coro.NextExecutorRunStep(driver) if !ok { t.Fatalf("poll channel adapter executor at step %d", step) } - if progress.Complete { - return + switch runStep.Kind { + case coro.ExecutorRunStepSource: + if runStep.Poll.Complete { + if !coro.EnterExecutorRunCompatibility(driver) { + t.Fatalf("leave bounded channel adapter executor at step %d", step) + } + return + } + case coro.ExecutorRunStepMaterialize: + if !coroMaterializeResumeCleanupStepV1(runStep.Cleanup) { + t.Fatalf("materialize channel adapter executor at step %d", step) + } + default: + t.Fatalf("unexpected channel adapter executor step %d at %d", runStep.Kind, step) } if step == 10000 { t.Fatal("channel adapter executor did not complete") @@ -647,18 +659,42 @@ func TestCoroChannelAdapterPairCommitAndResume(t *testing.T) { if !deferredCanceledOK || deferredCanceledG == nil || deferredCanceledG == canceledFrame.g { t.Fatalf("dequeue unrelated G before select cancellation = (%p, %t)", deferredCanceledG, deferredCanceledOK) } - if !coro.RequestTaskCancellation(p, canceledFrame.g, coro.TaskCancelAbort) { - t.Fatal("request channel select task cancellation") + lateValue := uint32(0x1a2b3c4d) + if !ChanTrySend( + canceledChannels[1], + unsafe.Pointer(&lateValue), + int(unsafe.Sizeof(lateValue)), + ) { + t.Fatal("complete channel selector before P-neutral transfer") } pollCoroChannelAdapterExecutor(t, driver) - if next, ok := coro.NextRunnable(p); !ok || next != canceledFrame.g { - t.Fatalf("dequeue canceled channel selector = (%p, %t), want %p", next, ok, canceledFrame.g) + for index := range canceledCases { + if canceledCases[index] != (CoroChanSelectCaseV1{}) { + t.Fatalf("materialized channel select case %d retained frame state: %+v", + index, canceledCases[index]) + } + } + targetP := new(coro.P) + var transfer coro.RunnableTransferMailbox + if !coro.BindRunnableTransferMailbox(&transfer, targetP) { + t.Fatal("bind channel select transfer mailbox") + } + transferID, transferred := coro.PublishPNeutralRunnable(&transfer, p, canceledFrame.g) + if !transferred || !transferID.Valid() || + !coro.ImportPNeutralRunnable(&transfer, targetP, transferID) { + t.Fatalf("transfer materialized channel selector = (%+v, %t)", transferID, transferred) + } + if !coro.RequestTaskCancellation(targetP, canceledFrame.g, coro.TaskCancelAbort) { + t.Fatal("request channel select cancellation after migration") + } + if next, ok := coro.NextRunnable(targetP); !ok || next != canceledFrame.g { + t.Fatalf("dequeue transferred channel selector = (%p, %t), want %p", next, ok, canceledFrame.g) } - canceledAction, ok = coro.BeginRunG(p, canceledFrame.g) + canceledAction, ok = coro.BeginRunG(targetP, canceledFrame.g) if !ok || canceledAction.Kind != coro.ActionCheckResume { t.Fatalf("begin canceled channel selector = (%+v, %t)", canceledAction, ok) } - canceledAction, ok = coro.Checked(p, canceledFrame.g, canceledAction, false) + canceledAction, ok = coro.Checked(targetP, canceledFrame.g, canceledAction, false) if !ok || canceledAction.Kind != coro.ActionResume { t.Fatalf("activate canceled channel selector = (%+v, %t)", canceledAction, ok) } diff --git a/runtime/internal/runtime/coro_executor_driver_host_llgo.go b/runtime/internal/runtime/coro_executor_driver_host_llgo.go index 76563b5740..0a017e9cbb 100644 --- a/runtime/internal/runtime/coro_executor_driver_host_llgo.go +++ b/runtime/internal/runtime/coro_executor_driver_host_llgo.go @@ -89,6 +89,5 @@ func coroProgramWakeExecutorV1(driver *coro.ExecutorDriver) bool { if !clockOK { return false } - _, _, ok := coro.WakeExecutorAt(driver, now) - return ok && coroProgramSyncHostOperationCancelsV1(driver) + return coro.WakeExecutorAt(driver, now) && coroProgramSyncHostOperationCancelsV1(driver) } diff --git a/runtime/internal/runtime/coro_executor_driver_legacy.go b/runtime/internal/runtime/coro_executor_driver_legacy.go index c706a3f189..0697f18c16 100644 --- a/runtime/internal/runtime/coro_executor_driver_legacy.go +++ b/runtime/internal/runtime/coro_executor_driver_legacy.go @@ -41,6 +41,5 @@ func coroProgramPollExecutorV1(driver *coro.ExecutorDriver) bool { } func coroProgramWakeExecutorV1(driver *coro.ExecutorDriver) bool { - _, _, ok := coro.WakeExecutor(driver) - return ok + return coro.WakeExecutor(driver) } diff --git a/runtime/internal/runtime/coro_executor_driver_timer_llgo.go b/runtime/internal/runtime/coro_executor_driver_timer_llgo.go index bcfc86f19d..ac1565948b 100644 --- a/runtime/internal/runtime/coro_executor_driver_timer_llgo.go +++ b/runtime/internal/runtime/coro_executor_driver_timer_llgo.go @@ -116,6 +116,5 @@ func coroProgramWakeExecutorV1(driver *coro.ExecutorDriver) bool { if !clockOK { return false } - _, _, ok := coro.WakeExecutorAt(driver, now) - return ok + return coro.WakeExecutorAt(driver, now) } diff --git a/runtime/internal/runtime/coro_executor_driver_worker_llgo.go b/runtime/internal/runtime/coro_executor_driver_worker_llgo.go index 3b279bca0e..e19938fff9 100644 --- a/runtime/internal/runtime/coro_executor_driver_worker_llgo.go +++ b/runtime/internal/runtime/coro_executor_driver_worker_llgo.go @@ -53,6 +53,5 @@ func coroProgramPollExecutorV1(driver *coro.ExecutorDriver) bool { } func coroProgramWakeExecutorV1(driver *coro.ExecutorDriver) bool { - _, _, ok := coro.WakeExecutor(driver) - return ok + return coro.WakeExecutor(driver) } diff --git a/runtime/internal/runtime/coro_native_fleet.go b/runtime/internal/runtime/coro_native_fleet.go index 98e7279c75..4d35913232 100644 --- a/runtime/internal/runtime/coro_native_fleet.go +++ b/runtime/internal/runtime/coro_native_fleet.go @@ -894,13 +894,13 @@ func coroNativeFleetPrepareOwnerWaitAtV1( func coroNativeFleetWakeOwnerAtV1( handle coro.ExecutorFleetHandle, now int64, -) (epoch uint32, timers, promoted int, ok bool) { +) (epoch uint32, ok bool) { if now < 0 { - return 0, 0, 0, false + return 0, false } epoch, acquired := coroNativeFleetBeginOwnerEpochV1(handle) if !acquired { - return 0, 0, 0, false + return 0, false } domain, valid := coroNativeFleetDomainForHandleV1( &coroNativeFleetV1State, @@ -908,17 +908,16 @@ func coroNativeFleetWakeOwnerAtV1( coroNativeFleetDomainActiveV1, ) if !valid || domain.ownerEpoch != epoch { - return 0, 0, 0, false + return 0, false } driver := domain.driverOwnerV1() if driver == nil { - return 0, 0, 0, false + return 0, false } - timers, promoted, ok = coro.WakeExecutorAt(driver, now) - if !ok { - return 0, 0, 0, false + if !coro.WakeExecutorAt(driver, now) { + return 0, false } - return epoch, timers, promoted, true + return epoch, true } func coroNativeFleetFinishOwnerEpochV1(handle coro.ExecutorFleetHandle, epoch uint32) bool { diff --git a/runtime/internal/runtime/coro_native_fleet_owner_llgo.go b/runtime/internal/runtime/coro_native_fleet_owner_llgo.go index 98af2c4363..894995b422 100644 --- a/runtime/internal/runtime/coro_native_fleet_owner_llgo.go +++ b/runtime/internal/runtime/coro_native_fleet_owner_llgo.go @@ -297,7 +297,7 @@ func coroNativeFleetRunPhysicalOwnerPassV1( if !wakeClockOK { return coroNativeFleetPhysicalOwnerFailV1("native fleet peer wake clock failed") } - next, _, _, wakeOK := coroNativeFleetWakeOwnerAtV1(handle, wakeNow) + next, wakeOK := coroNativeFleetWakeOwnerAtV1(handle, wakeNow) if !wakeOK { return coroNativeFleetPhysicalOwnerFailV1("native fleet peer wake transition failed") } diff --git a/runtime/internal/runtime/coro_native_fleet_reactor.go b/runtime/internal/runtime/coro_native_fleet_reactor.go index 366d849d52..0eb9eb3a6d 100644 --- a/runtime/internal/runtime/coro_native_fleet_reactor.go +++ b/runtime/internal/runtime/coro_native_fleet_reactor.go @@ -156,7 +156,8 @@ func coroNativeFleetPollPostAcceptedV1(result coro.OperationRouteIngressResult) // bounded poll call. Retry requires the physical owner to take a fresh // monotonic sample and re-enter, giving the coordinator a stop-check boundary // after EINTR and ordinary timeouts. Wake means it may reacquire a new logical -// owner epoch and run the common WakeExecutorAt transaction. +// owner epoch and run the common WakeExecutorAt transition. Source service +// then resumes through the same bounded reducer used before sleep. func coroNativeFleetWaitOwnerPassAtV1( wait coroNativeFleetArmedWaitV1, now int64, diff --git a/runtime/internal/runtime/coro_native_fleet_test.go b/runtime/internal/runtime/coro_native_fleet_test.go index 5524f78d0d..bb63296662 100644 --- a/runtime/internal/runtime/coro_native_fleet_test.go +++ b/runtime/internal/runtime/coro_native_fleet_test.go @@ -29,6 +29,10 @@ import ( "github.com/goplus/llgo/runtime/internal/coro" ) +// The source-island fleet test does not compile the typed hchan adapter. +// Reaching materialization without that adapter is therefore invalid. +func coroMaterializeResumeCleanupStepV1(coro.ResumeCleanupStep) bool { return false } + type coroNativeFleetTestTask struct { g *coro.G handle unsafe.Pointer @@ -401,11 +405,13 @@ func TestCoroNativeFleetProductionIslandsV1(t *testing.T) { standby.HasDeadline || standby.Deadline != 0 || second.ownerEpoch != 0 { t.Fatalf("empty fleet standby = (%+v, %t), owner=%d", standby, standbyOK, second.ownerEpoch) } - wakeEpoch, timers, promoted, wakeOK := coroNativeFleetWakeOwnerAtV1(secondHandle, 12) - if !wakeOK || wakeEpoch == 0 || timers != 0 || promoted != 0 || + wakeEpoch, wakeOK := coroNativeFleetWakeOwnerAtV1(secondHandle, 12) + if !wakeOK || wakeEpoch == 0 { + t.Fatalf("spurious fleet standby wake = (%d, %t)", wakeEpoch, wakeOK) + } + if wakeRun := coroNativeFleetRunOwnerEpochV1(secondHandle, wakeEpoch, 13, 8); wakeRun.stop != coroRunIdleV1 || !coroNativeFleetFinishOwnerEpochV1(secondHandle, wakeEpoch) { - t.Fatalf("spurious fleet standby wake = (%d, %d, %d, %t)", - wakeEpoch, timers, promoted, wakeOK) + t.Fatalf("spurious fleet standby reduction = %+v", wakeRun) } // Drive one real retained fd wait through the ordinary-domain reducer, @@ -462,10 +468,9 @@ func TestCoroNativeFleetProductionIslandsV1(t *testing.T) { if pass := coroNativeFleetWaitOwnerPassAtV1(armedPoll, 23); pass != coroNativeFleetWaitPassWakeV1 { t.Fatalf("routed poll physical wait = %d", pass) } - pollEpoch, timers, promoted, pollWakeOK := coroNativeFleetWakeOwnerAtV1(secondHandle, 24) - if !pollWakeOK || pollEpoch == 0 || timers != 0 || promoted != 1 { - t.Fatalf("wake routed poll owner = (%d, %d, %d, %t)", - pollEpoch, timers, promoted, pollWakeOK) + pollEpoch, pollWakeOK := coroNativeFleetWakeOwnerAtV1(secondHandle, 24) + if !pollWakeOK || pollEpoch == 0 { + t.Fatalf("wake routed poll owner = (%d, %t)", pollEpoch, pollWakeOK) } pollComplete := false for attempt := 0; attempt < 64; attempt++ { diff --git a/runtime/internal/runtime/coro_poll_owner_llgo.go b/runtime/internal/runtime/coro_poll_owner_llgo.go index 0cd5d26c01..f22bdd3553 100644 --- a/runtime/internal/runtime/coro_poll_owner_llgo.go +++ b/runtime/internal/runtime/coro_poll_owner_llgo.go @@ -163,7 +163,7 @@ func __llgo_coro_poll_resume_v2(g, storage unsafe.Pointer) uint32 { return 0 } task := (*coro.G)(g) - outcome, caseID, cancel, resultKind, result, taken := coro.TakeResumePacket( + outcome, caseID, cancel, resultKind, small, taken := coro.TakeResumePacket( task, state.ticket, &state.packet, @@ -174,17 +174,18 @@ func __llgo_coro_poll_resume_v2(g, storage unsafe.Pointer) uint32 { return 0 } status := uint32(0) + result := coro.PollOperationResult(small) switch { case outcome == coro.ParkOutcomeCompleted && caseID == 1 && cancel == coro.TaskCancelNone && resultKind == coro.ResumeResultPoll && result != coro.PollOperationResultInvalid: case outcome == coro.ParkOutcomeCanceled && caseID == 0 && cancel == coro.TaskCancelNone && - resultKind == coro.ResumeResultNone && result == coro.PollOperationResultInvalid: + resultKind == coro.ResumeResultNone && small == coro.ResumeSmallInvalid: status = coroPollResumeOperationCanceledV2 case outcome == coro.ParkOutcomeCanceled && caseID == 0 && cancel == coro.TaskCancelAbort && - resultKind == coro.ResumeResultNone && result == coro.PollOperationResultInvalid: + resultKind == coro.ResumeResultNone && small == coro.ResumeSmallInvalid: status = coroPollResumeTaskAbortV2 case outcome == coro.ParkOutcomeCanceled && caseID == 0 && cancel == coro.TaskCancelShutdown && - resultKind == coro.ResumeResultNone && result == coro.PollOperationResultInvalid: + resultKind == coro.ResumeResultNone && small == coro.ResumeSmallInvalid: status = coroPollResumeShutdownV2 default: coroPollAbortV2("unsupported coroutine Poll V2 run decision") diff --git a/runtime/internal/runtime/coro_program_test.go b/runtime/internal/runtime/coro_program_test.go index cd6782e4b7..dea8ac5a0f 100644 --- a/runtime/internal/runtime/coro_program_test.go +++ b/runtime/internal/runtime/coro_program_test.go @@ -26,6 +26,10 @@ import ( "github.com/goplus/llgo/runtime/internal/coro" ) +// The source-island program test does not compile the typed hchan adapter. +// Reaching materialization without that adapter is therefore invalid. +func coroMaterializeResumeCleanupStepV1(coro.ResumeCleanupStep) bool { return false } + // The named runtime-adapter source island includes coro_panic_payload.go but // intentionally does not load the complete LLGo runtime package into the host // Go runtime. Supply only the current Go panic(nil) interface contract that diff --git a/runtime/internal/runtime/coro_run_slice.go b/runtime/internal/runtime/coro_run_slice.go index 187eb6d0bd..2ed421ae9b 100644 --- a/runtime/internal/runtime/coro_run_slice.go +++ b/runtime/internal/runtime/coro_run_slice.go @@ -166,6 +166,12 @@ func coroReduceExecutorRunStepV1( result.used++ result.sources++ return false, true + case coro.ExecutorRunStepMaterialize: + if !coroMaterializeResumeCleanupStepV1(step.Cleanup) { + return false, false + } + result.used++ + return false, true case coro.ExecutorRunStepDispatch: if step.G == nil || step.Action.Handle == nil { return false, false diff --git a/runtime/internal/runtime/coro_target_wait_timer_llgo.go b/runtime/internal/runtime/coro_target_wait_timer_llgo.go index 5cb3ad4682..f056616324 100644 --- a/runtime/internal/runtime/coro_target_wait_timer_llgo.go +++ b/runtime/internal/runtime/coro_target_wait_timer_llgo.go @@ -110,8 +110,9 @@ func coroTargetWaitExecutorV1(pipe *corodoorbell.Pipe, deadline int64, hasDeadli return false } if reached { - // A fresh WakeExecutorAt sample publishes all due timer and poll - // deadlines; a physical timeout is not itself a completion. + // A fresh WakeExecutorAt sample returns all due timer and + // poll deadlines to the unified reducer; a physical timeout is + // not itself a completion. return true } } diff --git a/runtime/internal/runtime/coro_timer_owner_llgo.go b/runtime/internal/runtime/coro_timer_owner_llgo.go index de180be842..8eeb8b2468 100644 --- a/runtime/internal/runtime/coro_timer_owner_llgo.go +++ b/runtime/internal/runtime/coro_timer_owner_llgo.go @@ -181,7 +181,7 @@ func __llgo_coro_timer_resume_v2(g, storage unsafe.Pointer) uint32 { return 0 } task := (*coro.G)(g) - outcome, caseID, cancel, result, poll, taken := coro.TakeResumePacket( + outcome, caseID, cancel, result, small, taken := coro.TakeResumePacket( task, state.ticket, &state.packet, @@ -194,16 +194,16 @@ func __llgo_coro_timer_resume_v2(g, storage unsafe.Pointer) uint32 { status := uint32(0) switch { case outcome == coro.ParkOutcomeCompleted && caseID == 1 && cancel == coro.TaskCancelNone && - result == coro.ResumeResultNone && poll == coro.PollOperationResultInvalid: + result == coro.ResumeResultNone && small == coro.ResumeSmallInvalid: status = coroTimerResumeSuccessV2 case outcome == coro.ParkOutcomeCanceled && caseID == 0 && cancel == coro.TaskCancelNone && - result == coro.ResumeResultNone && poll == coro.PollOperationResultInvalid: + result == coro.ResumeResultNone && small == coro.ResumeSmallInvalid: status = coroTimerResumeOperationCanceledV2 case outcome == coro.ParkOutcomeCanceled && caseID == 0 && cancel == coro.TaskCancelAbort && - result == coro.ResumeResultNone && poll == coro.PollOperationResultInvalid: + result == coro.ResumeResultNone && small == coro.ResumeSmallInvalid: status = coroTimerResumeTaskAbortV2 case outcome == coro.ParkOutcomeCanceled && caseID == 0 && cancel == coro.TaskCancelShutdown && - result == coro.ResumeResultNone && poll == coro.PollOperationResultInvalid: + result == coro.ResumeResultNone && small == coro.ResumeSmallInvalid: status = coroTimerResumeShutdownV2 default: coroTimerAbortV2("unsupported coroutine Timer V2 run decision") diff --git a/runtime/internal/runtime/coro_worker_owner_llgo.go b/runtime/internal/runtime/coro_worker_owner_llgo.go index 0dbdcfadd5..f438deecba 100644 --- a/runtime/internal/runtime/coro_worker_owner_llgo.go +++ b/runtime/internal/runtime/coro_worker_owner_llgo.go @@ -155,7 +155,7 @@ func __llgo_coro_worker_resume_v1( task := (*coro.G)(g) var payload coro.ScalarResultPayloadV1 - outcome, caseID, cancel, result, poll, ok := coro.TakeResumePacket( + outcome, caseID, cancel, result, small, ok := coro.TakeResumePacket( task, state.ticket, &state.packet, @@ -168,13 +168,13 @@ func __llgo_coro_worker_resume_v1( discard := outcome == coro.ParkOutcomeCanceled if outcome == coro.ParkOutcomeCompleted { if caseID != 1 || cancel != coro.TaskCancelNone || - result != coro.ResumeResultScalar || poll != coro.PollOperationResultInvalid { + result != coro.ResumeResultScalar || small != coro.ResumeSmallInvalid { coroWorkerAbortV1('D', "invalid completed coroutine worker decision") return 0 } } else if !discard || caseID != 0 || cancel != coro.TaskCancelAbort && cancel != coro.TaskCancelShutdown || - result != coro.ResumeResultNone || poll != coro.PollOperationResultInvalid { + result != coro.ResumeResultNone || small != coro.ResumeSmallInvalid { coroWorkerAbortV1('E', "invalid canceled coroutine worker decision") return 0 } diff --git a/runtime/internal/runtime/z_chan_coro.go b/runtime/internal/runtime/z_chan_coro.go index cb29cdf02f..88ad96dd4b 100644 --- a/runtime/internal/runtime/z_chan_coro.go +++ b/runtime/internal/runtime/z_chan_coro.go @@ -52,6 +52,8 @@ type coroChanOperationV1 struct { type CoroChanParkV1 struct { wait coro.WaitSetRecord claim coro.SelectClaim + packet coro.ResumePacket + cleanup coro.ResumeCleanupPlan ticket coro.ParkTicket operation coroChanOperationV1 waiter chanWaiter @@ -74,6 +76,8 @@ type CoroChanSelectCaseV1 struct { type CoroChanSelectV1 struct { wait coro.WaitSetRecord claim coro.SelectClaim + packet coro.ResumePacket + cleanup coro.ResumeCleanupPlan ticket coro.ParkTicket candidates unsafe.Pointer count uintptr @@ -111,11 +115,17 @@ func validCoroChanOperationV1(operation *coroChanOperationV1, waiter *chanWaiter } func validCoroChanParkV1(state *CoroChanParkV1) bool { - if state == nil || state.magic != coroChanParkMagicV1 || state.waiter.coro != &state.operation || - state.operation.waiter != &state.waiter || state.operation.claim != &state.claim || + if state == nil || state.magic != coroChanParkMagicV1 || state.waiter.status > waitSendClosed || state.waiter.size < 0 { return false } + if state.operation == (coroChanOperationV1{}) && state.waiter == (chanWaiter{}) { + return state.ticket != (coro.ParkTicket{}) + } + if state.waiter.coro != &state.operation || state.operation.waiter != &state.waiter || + state.operation.claim != &state.claim { + return false + } if state.waiter.ch == nil { return state.operation.id == (coro.OperationID{}) && state.operation.magic == 0 } @@ -197,9 +207,9 @@ func unlockCoroChanSelectChannels(candidates unsafe.Pointer, ops []ChanOp) { } } -func validCoroChanSelectV1(state *CoroChanSelectV1, candidates unsafe.Pointer, count uintptr) bool { +func validCoroChanSelectV1(state *CoroChanSelectV1, candidates unsafe.Pointer, ops []ChanOp) bool { return state != nil && state.magic == coroChanSelectMagicV1 && state.candidates == candidates && - state.count == count && (count == 0 || candidates != nil) + state.count == uintptr(len(ops)) && (len(ops) == 0 || candidates != nil) } func classifyCoroChanSingleBegin(result coro.ChannelExternalCommitBeginResult) coroChanMatchResult { @@ -737,7 +747,6 @@ func prepareCoroChanSelectV1( } physical++ } - sortCoroChanSelectOrder(candidates, ops) state := (*CoroChanSelectV1)(storage) *state = CoroChanSelectV1{ candidates: candidates, @@ -753,8 +762,12 @@ func prepareCoroChanSelectV1( return } state.ticket = ticket + if !coro.BindSingleWaitSetResumePacket(&state.wait, &state.packet, coro.OperationID{}) { + coroRuntimeAbort("cannot bind empty coroutine channel select resume") + } return } + sortCoroChanSelectOrder(candidates, ops) driver, _, route, current := coro.CurrentExecutorChannelDriver(task) p, park, source, ownerOK := coro.CurrentExecutorChannelParkOwner(driver, task) if !current || !ownerOK || !coro.CanReserveChannelOperations(p, source, physical) { @@ -815,6 +828,23 @@ func prepareCoroChanSelectV1( } } state.ticket = ticket + if !coro.BindWaitSetResumeCleanup( + &state.wait, + &state.packet, + &state.cleanup, + coro.ResumeCleanupBinding{ + Kind: coro.ResumeCleanupChannelSelect, + Context: unsafe.Pointer(state), + Entries: candidates, + Source: source, + Claim: &state.claim, + Count: uint32(len(ops)), + Stride: unsafe.Sizeof(CoroChanSelectCaseV1{}), + }, + ) { + coroRuntimeAbort("cannot bind coroutine channel select cleanup") + return + } lockCoroChanSelectChannels(candidates, ops) start := selectStart(len(ops)) for offset := 0; offset < len(ops); offset++ { @@ -863,186 +893,52 @@ func CoroChanSelectPark(g, handle, header, candidates, storage unsafe.Pointer, o prepareCoroChanSelectV1(g, handle, header, candidates, storage, ops) } -func cleanupCoroChanSelectWaiters(candidates unsafe.Pointer, ops []ChanOp) bool { - for index := range ops { - candidate := coroChanSelectCaseAt(candidates, uintptr(index)) - if candidate.operation.magic == 0 { - if candidate != nil && (candidate.operation != (coroChanOperationV1{}) || - candidate.waiter != (chanWaiter{})) { - coroRuntimeAbort("nil coroutine channel select case retained physical state") - return false - } - continue - } - if candidate.operation.source == nil { - coroRuntimeAbort("coroutine channel select waiter lost its source") - return false - } - route, routed := candidate.operation.source.Route() - if !routed || route != candidate.operation.id.Route() { - coroRuntimeAbort("coroutine channel select waiter source route changed") - return false - } - if !validCoroChanOperationV1(&candidate.operation, &candidate.waiter) { - coroRuntimeAbort("coroutine channel select waiter lifecycle is invalid") - return false - } - if candidate.operation.claim == nil { - coroRuntimeAbort("coroutine channel select waiter lost its claim") - return false - } - ch := candidate.waiter.ch - ch.mutex.Lock() - if candidate.waiter.send { - ch.sendq.remove(&candidate.waiter) - } else { - ch.recvq.remove(&candidate.waiter) - } - if !reconcileBufferedChanLocked(ch, !ch.closed) || ch.closed && !drainClosedChanWaitersLocked(ch) { - ch.mutex.Unlock() - return false - } - ch.mutex.Unlock() - } - return true -} - -func finishCoroChanSelectOperations( - g *coro.G, - state *CoroChanSelectV1, - candidates unsafe.Pointer, - ops []ChanOp, - lease coro.OperationResultLease, - discard bool, -) bool { - driver, _, route, current := coro.CurrentExecutorChannelDriver(g) - p, _, source, ownerOK := coro.CurrentExecutorChannelParkOwner(driver, g) - if !current || !ownerOK { - return false - } - for index := range ops { - operation := &coroChanSelectCaseAt(candidates, uintptr(index)).operation - if operation.magic == 0 { - continue - } - id := operation.id - if operation.source != source || id.Route() != route || !source.ConfirmQuiesced(p, id) { - return false - } - } - if !source.ResetSelectClaim(p, &state.claim) { - return false - } - if lease.Valid() { - var released bool - if discard { - released = source.DiscardResult(p, lease) - } else { - released = source.TakeResult(p, lease) - } - if !released { - return false - } - } - for index := range ops { - operation := &coroChanSelectCaseAt(candidates, uintptr(index)).operation - if operation.magic == 0 { - continue - } - id := operation.id - if !source.Recycle(p, id) { - return false - } - } - return true -} - -// CoroChanSelectResume consumes the exact ParkTicket decision, detaches every -// queue node before releasing frame storage, and returns the selected SSA -// tuple prefix plus the same typed status used by direct channel lowering. +// CoroChanSelectResume consumes only the P-neutral packet produced before +// runnable publication. Every queue node, source generation, result lease, and +// old-P pointer has already been retired by typed materialization. func CoroChanSelectResume( g, candidates, storage unsafe.Pointer, ops ...ChanOp, ) (isel int, recvOK bool, status uint32) { isel = -1 state := (*CoroChanSelectV1)(storage) - if g == nil || !validCoroChanSelectV1(state, candidates, uintptr(len(ops))) { + if g == nil || !validCoroChanSelectV1(state, candidates, ops) { coroRuntimeAbort("invalid coroutine channel select resume ABI") return -1, false, coroChanResumeInvalid } task := (*coro.G)(g) - outcome, caseID, lease, cancel, ok := coro.TakeRunDecision(task, state.ticket) + outcome, caseID, cancel, result, small, ok := coro.TakeResumePacket( + task, + state.ticket, + &state.packet, + nil, + ) if !ok { - coroRuntimeAbort("invalid coroutine channel select run decision") + coroRuntimeAbort("invalid coroutine channel select resume packet") return -1, false, coroChanResumeInvalid } - physical := 0 - for index := range ops { - if coroChanSelectCaseAt(candidates, uintptr(index)).operation.magic != 0 { - physical++ - } - } - if physical == 0 { - if outcome != coro.ParkOutcomeCanceled || caseID != 0 || lease.Valid() || + if result == coro.ResumeResultNone { + if outcome != coro.ParkOutcomeCanceled || caseID != 0 || small != coro.ResumeSmallInvalid || cancel != coro.TaskCancelAbort && cancel != coro.TaskCancelShutdown { coroRuntimeAbort("invalid empty coroutine channel select decision") return -1, false, coroChanResumeInvalid } - for index := range ops { - *coroChanSelectCaseAt(candidates, uintptr(index)) = CoroChanSelectCaseV1{} - } *state = CoroChanSelectV1{} if cancel == coro.TaskCancelShutdown { return -1, false, coroChanResumeShutdown } return -1, false, coroChanResumeTaskAbort } - if !cleanupCoroChanSelectWaiters(candidates, ops) { - coroRuntimeAbort("cannot clean coroutine channel select waiters") - return -1, false, coroChanResumeInvalid - } - discard := outcome == coro.ParkOutcomeCanceled - var selected *CoroChanSelectCaseV1 - if outcome == coro.ParkOutcomeCompleted { - if caseID == 0 || int(caseID) > len(ops) || - cancel != coro.TaskCancelNone || !lease.Valid() { - coroRuntimeAbort("invalid completed coroutine channel select decision") - return -1, false, coroChanResumeInvalid - } - selected = coroChanSelectCaseAt(candidates, uintptr(caseID-1)) - if selected.operation.magic == 0 { - coroRuntimeAbort("completed coroutine channel select chose a nil case") - return -1, false, coroChanResumeInvalid - } - leaseID, validLease := lease.ID() - if !validLease || leaseID != selected.operation.id || !selected.waiter.status.done() { - coroRuntimeAbort("invalid coroutine channel select winner") - return -1, false, coroChanResumeInvalid - } - } else if outcome != coro.ParkOutcomeCanceled || caseID != 0 || - cancel != coro.TaskCancelAbort && cancel != coro.TaskCancelShutdown { - coroRuntimeAbort("invalid canceled coroutine channel select decision") - return -1, false, coroChanResumeInvalid - } - if !finishCoroChanSelectOperations(task, state, candidates, ops, lease, discard) { - coroRuntimeAbort("cannot finish coroutine channel select operations") + if result != coro.ResumeResultChannel || outcome != coro.ParkOutcomeCompleted || + caseID == 0 || int(caseID) > len(ops) || cancel != coro.TaskCancelNone || + small == coro.ResumeSmallInvalid { + coroRuntimeAbort("invalid materialized coroutine channel select decision") return -1, false, coroChanResumeInvalid } - if selected != nil { - isel = int(caseID - 1) - status = uint32(selected.waiter.status) - recvOK = selected.waiter.status.recvOK() - } - for index := range ops { - *coroChanSelectCaseAt(candidates, uintptr(index)) = CoroChanSelectCaseV1{} - } + isel = int(caseID - 1) + status = uint32(small) + recvOK = waitStatus(small).recvOK() *state = CoroChanSelectV1{} - if discard { - if cancel == coro.TaskCancelShutdown { - return -1, false, coroChanResumeShutdown - } - return -1, false, coroChanResumeTaskAbort - } switch waitStatus(status) { case waitSendOK: return isel, recvOK, coroChanResumeSendOK @@ -1084,6 +980,9 @@ func prepareCoroChanParkV1( return } state.ticket = ticket + if !coro.BindSingleWaitSetResumePacket(&state.wait, &state.packet, coro.OperationID{}) { + coroRuntimeAbort("cannot bind nil coroutine channel resume") + } return } if size != ch.elemsize { @@ -1115,6 +1014,23 @@ func prepareCoroChanParkV1( state.operation.id = id state.operation.source = source state.operation.magic = coroChanOperationMagicV1 + if !coro.BindWaitSetResumeCleanup( + &state.wait, + &state.packet, + &state.cleanup, + coro.ResumeCleanupBinding{ + Kind: coro.ResumeCleanupChannelDirect, + Context: unsafe.Pointer(state), + Entries: unsafe.Pointer(&state.operation.id), + Source: source, + Claim: &state.claim, + Count: 1, + Stride: unsafe.Sizeof(coro.OperationID{}), + }, + ) { + coroRuntimeAbort("cannot bind coroutine channel cleanup") + return + } ch.mutex.Lock() var ready bool if send { @@ -1160,13 +1076,18 @@ func __llgo_coro_chan_resume_v1(g, storage unsafe.Pointer) uint32 { coroRuntimeAbort("invalid coroutine channel resume ABI") return coroChanResumeInvalid } - outcome, caseID, lease, task, ok := coro.TakeRunDecision((*coro.G)(g), state.ticket) + outcome, caseID, task, result, small, ok := coro.TakeResumePacket( + (*coro.G)(g), + state.ticket, + &state.packet, + nil, + ) if !ok { - coroRuntimeAbort("invalid coroutine channel run decision") + coroRuntimeAbort("invalid coroutine channel resume packet") return coroChanResumeInvalid } - if state.waiter.ch == nil { - if outcome != coro.ParkOutcomeCanceled || caseID != 0 || lease.Valid() { + if result == coro.ResumeResultNone { + if outcome != coro.ParkOutcomeCanceled || caseID != 0 || small != coro.ResumeSmallInvalid { coroRuntimeAbort("invalid nil-channel run decision") return coroChanResumeInvalid } @@ -1181,57 +1102,13 @@ func __llgo_coro_chan_resume_v1(g, storage unsafe.Pointer) uint32 { return coroChanResumeInvalid } } - ch := state.waiter.ch - ch.mutex.Lock() - if state.waiter.send { - ch.sendq.remove(&state.waiter) - } else { - ch.recvq.remove(&state.waiter) - } - if !reconcileBufferedChanLocked(ch, !ch.closed) { - ch.mutex.Unlock() - coroRuntimeAbort("cannot reconcile coroutine buffered channel") - return coroChanResumeInvalid - } - if ch.closed { - if !drainClosedChanWaitersLocked(ch) { - ch.mutex.Unlock() - coroRuntimeAbort("cannot finish closed coroutine channel drain") - return coroChanResumeInvalid - } - } - ch.mutex.Unlock() - discard := outcome == coro.ParkOutcomeCanceled - if outcome == coro.ParkOutcomeCompleted { - if caseID != 1 || task != coro.TaskCancelNone || !lease.Valid() || !state.waiter.status.done() { - coroRuntimeAbort("invalid completed coroutine channel decision") - return coroChanResumeInvalid - } - } else if outcome != coro.ParkOutcomeCanceled || caseID != 0 || - task != coro.TaskCancelAbort && task != coro.TaskCancelShutdown { - coroRuntimeAbort("invalid canceled coroutine channel decision") + if result != coro.ResumeResultChannel || outcome != coro.ParkOutcomeCompleted || + caseID != 1 || task != coro.TaskCancelNone || small == coro.ResumeSmallInvalid { + coroRuntimeAbort("invalid materialized coroutine channel decision") return coroChanResumeInvalid } - if !coro.FinishSingleChannelPark( - (*coro.G)(g), - state.operation.source, - state.operation.id, - state.operation.claim, - lease, - discard, - ) { - coroRuntimeAbort("cannot finish coroutine channel park") - return coroChanResumeInvalid - } - status := state.waiter.status *state = CoroChanParkV1{} - if discard { - if task == coro.TaskCancelShutdown { - return coroChanResumeShutdown - } - return coroChanResumeTaskAbort - } - switch status { + switch waitStatus(small) { case waitSendOK: return coroChanResumeSendOK case waitRecvOK: @@ -1245,3 +1122,81 @@ func __llgo_coro_chan_resume_v1(g, storage unsafe.Pointer) uint32 { return coroChanResumeInvalid } } + +func materializeCoroChanOperationV1( + operation *coroChanOperationV1, + waiter *chanWaiter, + selected bool, +) (uint8, bool) { + if operation == nil || waiter == nil { + return 0, false + } + if *operation == (coroChanOperationV1{}) && *waiter == (chanWaiter{}) { + if selected { + return 0, false + } + return coro.ResumeSmallInvalid, true + } + if !validCoroChanOperationV1(operation, waiter) { + return 0, false + } + status := waiter.status + if selected != status.done() { + return 0, false + } + ch := waiter.ch + ch.mutex.Lock() + if waiter.send { + ch.sendq.remove(waiter) + } else { + ch.recvq.remove(waiter) + } + ok := reconcileBufferedChanLocked(ch, !ch.closed) + if ok && ch.closed { + ok = drainClosedChanWaitersLocked(ch) + } + ch.mutex.Unlock() + if !ok { + return 0, false + } + id := operation.id + *waiter = chanWaiter{} + *operation = coroChanOperationV1{id: id} + if selected { + return uint8(status), true + } + return coro.ResumeSmallInvalid, true +} + +func coroMaterializeResumeCleanupStepV1(step coro.ResumeCleanupStep) bool { + selected := step.Outcome == coro.ParkOutcomeCompleted && step.WinnerCase == step.Index+1 + var ( + small uint8 + ok bool + ) + switch step.Kind { + case coro.ResumeCleanupChannelDirect: + if step.Index != 0 { + return false + } + state := (*CoroChanParkV1)(step.Context) + if state == nil || state.magic != coroChanParkMagicV1 { + return false + } + small, ok = materializeCoroChanOperationV1(&state.operation, &state.waiter, selected) + case coro.ResumeCleanupChannelSelect: + state := (*CoroChanSelectV1)(step.Context) + if state == nil || state.magic != coroChanSelectMagicV1 || step.Index >= uint32(state.count) || + state.candidates == nil { + return false + } + candidate := coroChanSelectCaseAt(state.candidates, uintptr(step.Index)) + small, ok = materializeCoroChanOperationV1(&candidate.operation, &candidate.waiter, selected) + if ok { + candidate.order = 0 + } + default: + return false + } + return ok && coro.CommitResumeCleanupStep(step, small) +}