fpinscala 第 7 章练习 3 深度解析：为 `Par.map2` 实现尊重超时的 Future 版本

示例工程【免费下载链接】fpinscalaCode, exercises, answers, and hints to go along with the book Functional Programming in Scala项目地址https://gitcode.com/gh_mirrors/fp/fpinscala点击查看免费下载导读本文围绕《Functional Programming in Scala》配套仓库 fpinscala 中answerkey/parallelism/03.hint.md给出的练习提示深入讲解如何为并行抽象Par的map2实现一个真正尊重超时timeout的版本。你将理解朴素实现为何会失效、提示中记录耗时并扣减剩余预算这一核心思想如何落地为可运行代码并结合仓库源码exercises/parallelism/Par.scala逐行剖析完整答案掌握Future.get(timeout, units)时间记账的工程细节。练习背景这是第 7 章并行性模块中的哪道题本仓库按书本章节组织练习、提示与答案见 README.md 中的章节对照表Chapter 7: parallelism。在answerkey/parallelism/目录下01.answer.md给出了map2的扩展方法签名并注明map2不会在独立逻辑线程中求值f控制并行的唯一入口是fork02.answer.md只有一句话Keep reading - we choose a representation forParlater in the chapter.继续读——稍后章节会确定Par的表示03.hint.md就是本次的主题提示它针对练习 3修复map2使其尊重调用方传入的超时限制。仓库中对应练习骨架位于 src/main/scala/fpinscala/exercises/parallelism/Par.scala该文件已给出基于UnitFuture的朴素map2第 19–23 行并在注释中明确写道这个实现不尊重超时需要一个新的Future实现来修复。练习者要做的正是根据提示写出map2Timeouts这样的超时感知版本。提示原文解读一句话概括了整个方案03.hint.md的全文只有一句话却是整道题的灵魂In order to respect timeouts, wed need a newFutureimplementation that records the amount of time spent evaluating one future, then subtracts that time from the available time allocated for evaluating the other future.拆解这句话可以得到实现的三条纲领不能沿用现成的UnitFuture需要一个新的Future实现来承载时间记账逻辑评估第一个 future 时要记录真实耗时而不是只记录它有没有完成把第一个 future 消耗的时间从总预算中扣除把剩余时间作为第二个 future 的get超时上限从而保证两个子计算的总等待时间不会超过调用方给出的整体预算。朴素map2为什么做不到尊重超时先看仓库练习文件里已有的实现exercises/parallelism/Par.scalaprivate case class UnitFutureA extends Future[A]: def isDone true def get(timeout: Long, units: TimeUnit) get def isCancelled false def cancel(evenIfRunning: Boolean): Boolean false extension A def map2B, C(f: (A, B) C): Par[C] es val af pa(es) val bf pb(es) UnitFuture(f(af.get, bf.get))问题一目了然UnitFuture.get(timeout, units)完全忽略 timeout 参数直接返回已经持有的常量值——它只能包装已经算完的结果外层map2用无参af.get、bf.get无限期阻塞等待两个子任务调用方即使传入get(100, MILLISECONDS)也无法对整体等待时间做任何限制朴素版本是先等a等到底再等b等到底两者耗时是串行相加的任何一方卡死都会让整体无限挂起。因此练习 3 的核心矛盾是map2返回的Future必须自己管理剩余时间预算并在内部对两个子 future 分别施加递减的超时而不是各给一个满额超时。完整答案实现解析map2Timeouts配套答案文件 answerkey/parallelism/03.answer.md 给出了完整实现该实现同样收录在答案模块 src/main/scala/fpinscala/answers/parallelism/Par.scala。代码全文如下/* This version respects timeouts. Note: this implementation will not prevent repeated evaluation if multiple threads call get in parallel. We could prevent this using synchronization, but it isnt needed for our purposes here (also, repeated evaluation of pure values wont affect results). */ extension A def map2TimeoutsB, C(f: (A, B) C): Par[C] es new Future[C]: private val futureA pa(es) private val futureB pb(es) volatile private var cache: Option[C] None def isDone cache.isDefined def get() get(Long.MaxValue, TimeUnit.NANOSECONDS) def get(timeout: Long, units: TimeUnit) val timeoutNanos TimeUnit.NANOSECONDS.convert(timeout, units) val started System.nanoTime val a futureA.get(timeoutNanos, TimeUnit.NANOSECONDS) val elapsed System.nanoTime - started val b futureB.get(timeoutNanos - elapsed, TimeUnit.NANOSECONDS) val c f(a, b) cache Some(c) c def isCancelled futureA.isCancelled || futureB.isCancelled def cancel(evenIfRunning: Boolean) futureA.cancel(evenIfRunning) || futureB.cancel(evenIfRunning)步骤 1构造期只求值不阻塞es new Future[C]表明调用map2Timeouts(pb)(f)得到Par[C]后只有在真正run时才把两个子Par提交给ExecutorService得到底层Future[A]与Future[B]futureA、futureB整个过程不阻塞、不等待结果把等待行为全部推迟到get方法里。步骤 2结果缓存与完成状态volatile private var cache: Option[C] None def isDone cache.isDefinedcache保存f(a, b)的结果isDone通过它是否为Some来判断声明为volatile保证一个线程写入后其他线程读取isDone能看到最新状态答案注释同时提醒如果多个线程并发调用get本实现不阻止重复求值。要彻底避免需要引入同步如只允许第一个线程计算结果、其余等待但对纯函数求值而言重复计算不会改变结果所以练习目的下可以接受。步骤 3时间记账是核心def get(timeout: Long, units: TimeUnit) val timeoutNanos TimeUnit.NANOSECONDS.convert(timeout, units) val started System.nanoTime val a futureA.get(timeoutNanos, TimeUnit.NANOSECONDS) val elapsed System.nanoTime - started val b futureB.get(timeoutNanos - elapsed, TimeUnit.NANOSECONDS) val c f(a, b) cache Some(c) c这正是提示记录耗时、扣减剩余预算的逐行落地统一单位TimeUnit.NANOSECONDS.convert(timeout, units)把调用方传入的任意TimeUnit秒、毫秒等统一换算成纳秒后续所有记账都在纳秒尺度上进行记录起点System.nanoTime是单调时钟monotonic clock只适合测量时间间隔不会受系统时间调整影响正适合做耗时统计用满额预算等第一个结果futureA.get(timeoutNanos, TimeUnit.NANOSECONDS)——a可以吃光整个预算计算实际消耗elapsed System.nanoTime - started精确记录等待a花了多久用剩余预算等第二个结果futureB.get(timeoutNanos - elapsed, TimeUnit.NANOSECONDS)——b只能拿到总预算减去a已消耗的剩余时间。这样无论a多快多慢a b的总等待上限始终约为调用方给出的timeoutNanos组合并缓存得到a、b后调用f写入cache再返回。步骤 4无参get的语义def get() get(Long.MaxValue, TimeUnit.NANOSECONDS)无参get代表无限期等待这里用Long.MaxValue纳秒约 292 年来近似没有超时。同时它复用带超时版本保证两条路径的时间记账逻辑一致。步骤 5取消与完成状态委托def isCancelled futureA.isCancelled || futureB.isCancelled def cancel(evenIfRunning: Boolean) futureA.cancel(evenIfRunning) || futureB.cancel(evenIfRunning)把取消操作委托给两个底层 future任一被取消即视为整体被取消cancel依次尝试取消两者只要有一个取消成功就返回true。边界情况与并发语义分析从代码结构可以推断并总结以下行为特征剩余预算为负数若a耗时超过总预算timeoutNanos - elapsed为负。底层java.util.concurrent.Future.get对负数超时通常会立即抛出TimeoutException即整体超时由第二个get表现出来符合总等待不超过预算的承诺f的求值不计入预算f(a, b)在两次get之后执行其计算时间不在超时保护范围内。若需要把f的求值也纳入超时控制可以进一步把整段逻辑放进带超时保护的求值中但这不是本练习的要求单位转换的溢出风险调用方若传入极大的timeout如Long.MaxValue毫秒TimeUnit.NANOSECONDS.convert可能溢出为负值。因此无参get才选择直接使用Long.MaxValue纳秒从源头避免转换溢出线程安全cache为volatileisDone读取安全但并发get可能重复求值答案注释已明示这是有意简化。若需要单次求值语义可引入同步或原子状态机属于工程加固而非练习要求。对比Nonblocking 版本中的map2仓库还提供了非阻塞 CPS 风格的并行实现 src/main/scala/fpinscala/exercises/parallelism/Nonblocking.scala。在那里opaque type Future[A] (A Unit) Unit opaque type Par[A] ExecutorService Future[A]map2用 Actor.scala 中的Actor[Either[A,B]]合并两个结果谁先到达就暂存到Option变量双方都到齐后再通过eval(es)异步执行回调cb(f(a, b))。这种实现天然不阻塞线程但也完全没有超时概念——03.hint.md讨论的新Future实现记录耗时方案针对的是基于java.util.concurrent.Future的阻塞版本Par.scala。理解这两者的差异有助于体会以fork为唯一并行控制原语的设计取舍。验证与运行仓库基于 Scala CLI 构建见 README.md 的 Building 一节你可以按如下方式验证# 编译全部练习与答案 scala-cli compile . # 打开 REPL 手工验证 scala-cli console .在 REPL 中可构造一个极端的实验让futureB对应的任务休眠很久然后用很小的超时调用map2Timeouts的get观察其是否在预算内抛出TimeoutException从而证明超时被尊重import fpinscala.exercises.parallelism.Par.* import java.util.concurrent.* val es Executors.newFixedThreadPool(4) val slowB: Par[Int] lazyUnit { Thread.sleep(10_000); 2 } val pa: Par[Int] unit(1) val combined pa.map2Timeouts(slowB)(_ _) // 预期100 毫秒内因 TimeoutException 失败而不是挂起 10 秒 combined.run(es).get(100, TimeUnit.MILLISECONDS)注练习文件 exercises/parallelism/Par.scala 中并未预置map2Timeouts需要读者依据提示自行实现上述代码引用的是答案模块中的现成实现也可对照 answers/parallelism/Par.scala 查看完整形态。小结03.hint.md用一句话点破了并行map2支持超时的全部要点换一个会记账的Future先花时间等第一个结果再拿剩余预算等第二个结果。配套答案map2Timeouts在源码中逐行兑现了这一思路TimeUnit统一换算纳秒、System.nanoTime记录单调耗时、timeoutNanos - elapsed完成预算扣减、volatile cache支撑完成状态。掌握这一模式你就同时理解了java.util.concurrent.Future超时语义的正确用法以及函数式并行库中把时间当作显式预算来传递的常见工程手法——它也是后续练习中理解更复杂并行组合子的基础。赞分享示例工程【免费下载链接】fpinscalaCode, exercises, answers, and hints to go along with the book Functional Programming in Scala项目地址https://gitcode.com/gh_mirrors/fp/fpinscala点击查看免费下载相关推荐fpinscala 第 3 章练习题 21 深度解析用 flatMap 实现 filterfilterViaFlatMapfpinscala 第 3 章练习题 21 深度解析用 flatMap 实现 filterfilterViaFlatMap 导读 本文围绕《Scala 函示例工程fpinscala 第 3 章练习 28 详解为 Tree 数据类型实现 map 函数fpinscala 第 3 章练习 28 详解为 Tree 数据类型实现 map 函数 本文基于 fpinscala 开源仓库 answerkey/datas示例工程fpinscala 第 13 章练习 3 详解实现 Free Monad 的通用解释器 runfpinscala 第 13 章练习 3 详解实现 Free Monad 的通用解释器 run 导读 本文围绕 fpinscala 仓库中 answerkey示例工程创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考