Effect 中的 PartitionedSemaphore 中断安全修复解析部分获取许可证泄漏问题【免费下载链接】effectBuild production-ready applications in TypeScript项目地址: https://gitcode.com/GitHub_Trending/ef/effect导读本文围绕effect仓库中的一条变更记录.changeset/pre/eff-51-partitioned-semaphore-interruption.md展开PartitionedSemaphore.take在被中断时曾存在部分已获取许可证泄漏的缺陷现已修复。文章将以PartitionedSemaphore模块为核心讲解其设计模型、核心 API 与源码实现并深入剖析该中断泄漏缺陷的成因、修复策略以及配套的回归测试帮助你理解在 Effect 中编写中断安全interruption-safe并发原语的关键要点。背景为什么需要PartitionedSemaphorePartitionedSemaphore是effect库中一个用于限制并发度的原语。与普通Semaphore的区别在于它在共享的许可证池之上按分区键partition key对等待者进行分组。从 PartitionedSemaphore.ts 的模块注释可以看出其设计动机APartitionedSemaphoreKis useful when many independent groups of work compete for the same bounded resource and each group should make progress without one busy group monopolizing released permits.即当许多相互独立的工作组竞争同一个有界资源时每个组都应能取得进展而不允许某个繁忙的组垄断释放出来的许可证。典型的应用场景包括多租户/多用户的 API 限流以userId或tenantId作为分区键避免某个大客户占满全部并发配额按目标下游服务分组的连接池防止一个慢服务阻塞其他服务的请求多队列任务处理按任务来源分区保证各分区公平获得处理额度。核心模型与 APIPartitionedSemaphoreK的接口定义在 PartitionedSemaphore.ts核心成员如下成员签名说明capacitynumber固定总容量创建后不再变化availableEffectnumber读取当前可用许可证数量的快照take(key, permits) Effectvoid为指定分区键获取permits个许可证可能等待release(permits) Effectnumber释放许可证返回当前可用数withPermits(key, permits) effect effect包裹一个 effect自动获取并在退出时释放withPermit(key) effect effectwithPermits的单个许可证变体withPermitsIfAvailable(permits) effect EffectOptionA仅在许可证立即可用时才运行否则返回None等待中的许可证按分区以轮询round-robin顺序分配这是保证各分区公平进展的核心机制。创建import { Effect, PartitionedSemaphore } from effect // 在 Effect 内部创建推荐 const program Effect.gen(function*() { const sem yield* PartitionedSemaphore.makestring({ permits: 4 }) // ... }) // 在 Effect 外部同步创建 const sem PartitionedSemaphore.makeUnsafestring({ permits: 4 })创建时的两个边界行为见 makeUnsafe 实现负数许可证数被钳制为 0Math.max(0, options.permits)非有限数Infinity/NaN创建无界信号量take/release立即完成withPermits直接透传原始 effect。手动获取与释放// 为分区 a 获取 1 个许可证可能等待使用后手动释放 yield* PartitionedSemaphore.take(sem, a, 1) yield* Effect.sleep(10 millis) yield* PartitionedSemaphore.release(sem, 1)take的语义模块注释有足够许可证时立即完成否则等待释放的许可证被轮询分配到本分区请求数超过总容量时永不完成返回Effect.never请求 0 或负数许可证时立即完成、不获取任何东西。自动获取/释放withPermits与withPermit会在包裹的 effect成功、失败或中断时均释放许可证通过Effect.ensuring保证是日常使用最安全的形式// 权重模式为分区 b 获取 2 个许可证后运行 effect yield* PartitionedSemaphore.withPermits(sem, b, 2, Effect.succeed(done)) // 单许可证模式支持管道式调用 const out yield* Effect.succeed(3).pipe( PartitionedSemaphore.withPermit(sem, c) )立即可用检测withPermitsIfAvailable不等待许可证不可用则直接返回Option.none()可用则执行 effect 并返回Option.some(result)实现const result yield* PartitionedSemaphore.withPermitsIfAvailable(sem, 1, Effect.succeed(ok)) // result: Optionstring不可用时为 none()源码实现分区等待队列与轮询分配在makeUnsafe内部PartitionedSemaphore.ts状态由三部分组成totalPermits当前可用许可证数上限为maxPermitswaitingPermits所有分区累计的、仍欠缺的许可证总数partitionsMutableHashMapK, SetWaiter按分区键存放等待者Waiter记录了所需许可证数和resume回调。releaseUnsafe是轮询分配的核心L144-L176释放的许可证优先喂给等待中的分区逐个数地递减选中等待者的permits只有当某个等待者的欠缺数归零时才调用其resume()只有当waitingPermits 0无人等待时多余的许可证才回流到totalPermits封顶为容量。模块维护一个持久化的分区迭代器iterator每轮释放从上次的位置继续从而实现轮询公平。take的实现L178-L232分为三步立即满足totalPermits permits时直接扣减并成功返回部分满足 入队totalPermits 0时先把剩余可用数全部扣光把欠缺的needed permits - totalPermits计入waitingPermits再向对应分区的等待集合加入一个Waiter条目中断清理回调返回一个Effect.sync清理器在 effect 被中断时执行cleanup()从分区集合删除条目集合为空则移除该分区键并回滚waitingPermits。缺陷剖析中断时泄漏部分已获取的许可证变更记录指出eff-51-partitioned-semaphore-interruption.mdFixPartitionedSemaphore.takeleaking partially acquired permits when interrupted.问题发生在部分满足 入队的中间态take(sem, b, 3)在池中只剩 1 个许可证时会立即扣掉这 1 个许可证totalPermits归零并为欠缺的 2 个许可证登记一个Waiter。如果此时请求被中断中断清理逻辑执行cleanup() // 从分区集合中移除等待条目 waitingPermits - entry.permits // 回滚欠缺的许可证数 releaseUnsafe(permits - entry.permits) // 把已经拿到的部分许可证释放回池中即已经被当前请求消费的permits - entry.permits个许可证会被重新归还到可用池而不是永久泄漏。这正是修复后的正确行为中断的请求既不会卡住队列也不会白白吞掉资源。修复前该路径的处理不完整中断的请求会保留已获取的部分许可证而不归还导致池容量逐步萎缩——例如容量 4 的池子经过反复中断后实际可用额度可能永久少于 4。配套回归测试位于 PartitionedSemaphore.test.ts精确复现了该场景it.effect(interrupting a partially satisfied waiter releases all acquired permits, () Effect.gen(function*() { const sem yield* PartitionedSemaphore.makestring({ permits: 4 }) yield* PartitionedSemaphore.take(sem, a, 3) // 池剩 1 const waiter yield* PartitionedSemaphore.take(sem, b, 3) .pipe(Effect.forkChild) // 部分满足 等待 yield* Effect.yieldNow yield* PartitionedSemaphore.release(sem, 1) // 把 1 个许可证喂给 b assert.strictEqual(yield* PartitionedSemaphore.available(sem), 0) yield* Fiber.interrupt(waiter) // 中断 b assert.strictEqual(yield* PartitionedSemaphore.available(sem), 2) // 部分获取的 2 个许可证被归还 yield* PartitionedSemaphore.release(sem, 2) assert.strictEqual(yield* PartitionedSemaphore.available(sem), 4) // 容量恢复 yield* PartitionedSemaphore.take(sem, c, 4) }))测试断言中断部分满足的等待者后available立即恢复为 2即被该等待者提前消费的部分许可证已归还随后释放剩余 2 个即可恢复到完整容量 4证明没有许可证泄漏。关联边界缺陷已唤醒但未完成获取的等待者被中断同一模块还有一条紧密相关的修复partitioned-semaphore-stale-cleanup.mdFixPartitionedSemaphoreleaving a new waiter suspended when a previously resumed waiter for the same partition is interrupted before its acquisition completes.它处理的是另一种竞态某个等待者已通过resume()被唤醒其条目已从分区集合中删除但它的takeeffect 尚未真正完成执行。此时若它被中断会触发上文的中断清理逻辑——而清理逻辑会重新执行一次cleanup()并再次回滚waitingPermits。如果此时同分区恰好已有新的等待者入队这种重复清理会造成新的等待者被错误悬挂suspended或计数错乱。对应的回归测试PartitionedSemaphore.test.ts通过自定义调度器人为制造唤醒后、完成前的中断窗口分别验证了在下一个等待者使用相同分区键与不同分区键两种情况下中断已唤醒的等待者都不会影响后续等待者正常获得许可证next.pollUnsafe()返回Exit.void。模块演化与迁移提示PartitionedSemaphore相关 API 近期仍在演进仓库内其他 changeset 记录了其去向shiny-trains-hug.md为Semaphore、Latch以及抽离出的PartitionedSemaphore操作新增模块级辅助函数extract-semaphore-latch.md计划将PartitionedSemaphore合并进Semaphore以Semaphore.Partitioned、Semaphore.makePartitioned、Semaphore.makePartitionedUnsafe的形式暴露同时Semaphore.make/makeUnsafe将取代旧的Effect.makeSemaphore系列。如果你正在基于effect编写依赖PartitionedSemaphore的代码建议关注上述命名调整以便在新版本发布后平滑迁移。小结PartitionedSemaphore通过共享许可证池 分区等待队列 轮询分配三要素为多组竞争同一有界资源的场景提供了公平的并发控制。而中断安全是并发原语正确性的底线部分获取必须可回滚中断的take要将已消费的部分许可证归还到可用池releaseUnsafe(permits - entry.permits)清理必须幂等已被唤醒的等待者再次被中断时其清理逻辑不能误伤同分区的新等待者回归测试覆盖竞态窗口仓库测试通过forkChild、自定义Scheduler等手段精确复现中断时机确保修复可验证。理解这些底层细节能帮助你在使用 Effect 编写高并发、可中断的代码时写出真正健壮且可预期的资源管理逻辑。相关实现与测试可进一步阅读 PartitionedSemaphore.ts 与 PartitionedSemaphore.test.ts。【免费下载链接】effectBuild production-ready applications in TypeScript项目地址: https://gitcode.com/GitHub_Trending/ef/effect创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Effect 中的 PartitionedSemaphore 中断安全修复：解析部分获取许可证泄漏问题
Effect 中的 PartitionedSemaphore 中断安全修复解析部分获取许可证泄漏问题【免费下载链接】effectBuild production-ready applications in TypeScript项目地址: https://gitcode.com/GitHub_Trending/ef/effect导读本文围绕effect仓库中的一条变更记录.changeset/pre/eff-51-partitioned-semaphore-interruption.md展开PartitionedSemaphore.take在被中断时曾存在部分已获取许可证泄漏的缺陷现已修复。文章将以PartitionedSemaphore模块为核心讲解其设计模型、核心 API 与源码实现并深入剖析该中断泄漏缺陷的成因、修复策略以及配套的回归测试帮助你理解在 Effect 中编写中断安全interruption-safe并发原语的关键要点。背景为什么需要PartitionedSemaphorePartitionedSemaphore是effect库中一个用于限制并发度的原语。与普通Semaphore的区别在于它在共享的许可证池之上按分区键partition key对等待者进行分组。从 PartitionedSemaphore.ts 的模块注释可以看出其设计动机APartitionedSemaphoreKis useful when many independent groups of work compete for the same bounded resource and each group should make progress without one busy group monopolizing released permits.即当许多相互独立的工作组竞争同一个有界资源时每个组都应能取得进展而不允许某个繁忙的组垄断释放出来的许可证。典型的应用场景包括多租户/多用户的 API 限流以userId或tenantId作为分区键避免某个大客户占满全部并发配额按目标下游服务分组的连接池防止一个慢服务阻塞其他服务的请求多队列任务处理按任务来源分区保证各分区公平获得处理额度。核心模型与 APIPartitionedSemaphoreK的接口定义在 PartitionedSemaphore.ts核心成员如下成员签名说明capacitynumber固定总容量创建后不再变化availableEffectnumber读取当前可用许可证数量的快照take(key, permits) Effectvoid为指定分区键获取permits个许可证可能等待release(permits) Effectnumber释放许可证返回当前可用数withPermits(key, permits) effect effect包裹一个 effect自动获取并在退出时释放withPermit(key) effect effectwithPermits的单个许可证变体withPermitsIfAvailable(permits) effect EffectOptionA仅在许可证立即可用时才运行否则返回None等待中的许可证按分区以轮询round-robin顺序分配这是保证各分区公平进展的核心机制。创建import { Effect, PartitionedSemaphore } from effect // 在 Effect 内部创建推荐 const program Effect.gen(function*() { const sem yield* PartitionedSemaphore.makestring({ permits: 4 }) // ... }) // 在 Effect 外部同步创建 const sem PartitionedSemaphore.makeUnsafestring({ permits: 4 })创建时的两个边界行为见 makeUnsafe 实现负数许可证数被钳制为 0Math.max(0, options.permits)非有限数Infinity/NaN创建无界信号量take/release立即完成withPermits直接透传原始 effect。手动获取与释放// 为分区 a 获取 1 个许可证可能等待使用后手动释放 yield* PartitionedSemaphore.take(sem, a, 1) yield* Effect.sleep(10 millis) yield* PartitionedSemaphore.release(sem, 1)take的语义模块注释有足够许可证时立即完成否则等待释放的许可证被轮询分配到本分区请求数超过总容量时永不完成返回Effect.never请求 0 或负数许可证时立即完成、不获取任何东西。自动获取/释放withPermits与withPermit会在包裹的 effect成功、失败或中断时均释放许可证通过Effect.ensuring保证是日常使用最安全的形式// 权重模式为分区 b 获取 2 个许可证后运行 effect yield* PartitionedSemaphore.withPermits(sem, b, 2, Effect.succeed(done)) // 单许可证模式支持管道式调用 const out yield* Effect.succeed(3).pipe( PartitionedSemaphore.withPermit(sem, c) )立即可用检测withPermitsIfAvailable不等待许可证不可用则直接返回Option.none()可用则执行 effect 并返回Option.some(result)实现const result yield* PartitionedSemaphore.withPermitsIfAvailable(sem, 1, Effect.succeed(ok)) // result: Optionstring不可用时为 none()源码实现分区等待队列与轮询分配在makeUnsafe内部PartitionedSemaphore.ts状态由三部分组成totalPermits当前可用许可证数上限为maxPermitswaitingPermits所有分区累计的、仍欠缺的许可证总数partitionsMutableHashMapK, SetWaiter按分区键存放等待者Waiter记录了所需许可证数和resume回调。releaseUnsafe是轮询分配的核心L144-L176释放的许可证优先喂给等待中的分区逐个数地递减选中等待者的permits只有当某个等待者的欠缺数归零时才调用其resume()只有当waitingPermits 0无人等待时多余的许可证才回流到totalPermits封顶为容量。模块维护一个持久化的分区迭代器iterator每轮释放从上次的位置继续从而实现轮询公平。take的实现L178-L232分为三步立即满足totalPermits permits时直接扣减并成功返回部分满足 入队totalPermits 0时先把剩余可用数全部扣光把欠缺的needed permits - totalPermits计入waitingPermits再向对应分区的等待集合加入一个Waiter条目中断清理回调返回一个Effect.sync清理器在 effect 被中断时执行cleanup()从分区集合删除条目集合为空则移除该分区键并回滚waitingPermits。缺陷剖析中断时泄漏部分已获取的许可证变更记录指出eff-51-partitioned-semaphore-interruption.mdFixPartitionedSemaphore.takeleaking partially acquired permits when interrupted.问题发生在部分满足 入队的中间态take(sem, b, 3)在池中只剩 1 个许可证时会立即扣掉这 1 个许可证totalPermits归零并为欠缺的 2 个许可证登记一个Waiter。如果此时请求被中断中断清理逻辑执行cleanup() // 从分区集合中移除等待条目 waitingPermits - entry.permits // 回滚欠缺的许可证数 releaseUnsafe(permits - entry.permits) // 把已经拿到的部分许可证释放回池中即已经被当前请求消费的permits - entry.permits个许可证会被重新归还到可用池而不是永久泄漏。这正是修复后的正确行为中断的请求既不会卡住队列也不会白白吞掉资源。修复前该路径的处理不完整中断的请求会保留已获取的部分许可证而不归还导致池容量逐步萎缩——例如容量 4 的池子经过反复中断后实际可用额度可能永久少于 4。配套回归测试位于 PartitionedSemaphore.test.ts精确复现了该场景it.effect(interrupting a partially satisfied waiter releases all acquired permits, () Effect.gen(function*() { const sem yield* PartitionedSemaphore.makestring({ permits: 4 }) yield* PartitionedSemaphore.take(sem, a, 3) // 池剩 1 const waiter yield* PartitionedSemaphore.take(sem, b, 3) .pipe(Effect.forkChild) // 部分满足 等待 yield* Effect.yieldNow yield* PartitionedSemaphore.release(sem, 1) // 把 1 个许可证喂给 b assert.strictEqual(yield* PartitionedSemaphore.available(sem), 0) yield* Fiber.interrupt(waiter) // 中断 b assert.strictEqual(yield* PartitionedSemaphore.available(sem), 2) // 部分获取的 2 个许可证被归还 yield* PartitionedSemaphore.release(sem, 2) assert.strictEqual(yield* PartitionedSemaphore.available(sem), 4) // 容量恢复 yield* PartitionedSemaphore.take(sem, c, 4) }))测试断言中断部分满足的等待者后available立即恢复为 2即被该等待者提前消费的部分许可证已归还随后释放剩余 2 个即可恢复到完整容量 4证明没有许可证泄漏。关联边界缺陷已唤醒但未完成获取的等待者被中断同一模块还有一条紧密相关的修复partitioned-semaphore-stale-cleanup.mdFixPartitionedSemaphoreleaving a new waiter suspended when a previously resumed waiter for the same partition is interrupted before its acquisition completes.它处理的是另一种竞态某个等待者已通过resume()被唤醒其条目已从分区集合中删除但它的takeeffect 尚未真正完成执行。此时若它被中断会触发上文的中断清理逻辑——而清理逻辑会重新执行一次cleanup()并再次回滚waitingPermits。如果此时同分区恰好已有新的等待者入队这种重复清理会造成新的等待者被错误悬挂suspended或计数错乱。对应的回归测试PartitionedSemaphore.test.ts通过自定义调度器人为制造唤醒后、完成前的中断窗口分别验证了在下一个等待者使用相同分区键与不同分区键两种情况下中断已唤醒的等待者都不会影响后续等待者正常获得许可证next.pollUnsafe()返回Exit.void。模块演化与迁移提示PartitionedSemaphore相关 API 近期仍在演进仓库内其他 changeset 记录了其去向shiny-trains-hug.md为Semaphore、Latch以及抽离出的PartitionedSemaphore操作新增模块级辅助函数extract-semaphore-latch.md计划将PartitionedSemaphore合并进Semaphore以Semaphore.Partitioned、Semaphore.makePartitioned、Semaphore.makePartitionedUnsafe的形式暴露同时Semaphore.make/makeUnsafe将取代旧的Effect.makeSemaphore系列。如果你正在基于effect编写依赖PartitionedSemaphore的代码建议关注上述命名调整以便在新版本发布后平滑迁移。小结PartitionedSemaphore通过共享许可证池 分区等待队列 轮询分配三要素为多组竞争同一有界资源的场景提供了公平的并发控制。而中断安全是并发原语正确性的底线部分获取必须可回滚中断的take要将已消费的部分许可证归还到可用池releaseUnsafe(permits - entry.permits)清理必须幂等已被唤醒的等待者再次被中断时其清理逻辑不能误伤同分区的新等待者回归测试覆盖竞态窗口仓库测试通过forkChild、自定义Scheduler等手段精确复现中断时机确保修复可验证。理解这些底层细节能帮助你在使用 Effect 编写高并发、可中断的代码时写出真正健壮且可预期的资源管理逻辑。相关实现与测试可进一步阅读 PartitionedSemaphore.ts 与 PartitionedSemaphore.test.ts。【免费下载链接】effectBuild production-ready applications in TypeScript项目地址: https://gitcode.com/GitHub_Trending/ef/effect创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考