纯函数式的并行计算（2）

打破法则：一个微妙的bug
在上篇博文的实现中，实际上会有一个相当微妙的问题出现在大多数folk的实现上，当使用固定大小的线程池作为ExecutorService时， 是很容易出现死锁的。
那我们可以修复这个folk实现使其能够在固定大小的线程池中工作吗？我们来看看不同的实现：

  def folk1[A](pa: => Par[A]): Par[A] = es => pa(es)

这无疑是避免的死锁，唯一的问题是，我们实际上并没有派生一个单独的逻辑线程来对pa进行求值。不过这仍然是一个有用的组合子，因为它让我们做到延时计算的实例。让我们来给它一个更加适合的名字，delay：

  def delay[A](pa: => Par[A]): Par[A] = es => pa(es)

但是我们很像能够在固定大小的线程池中运行任意的计算。为了做到这一点，我们需要为Par挑选另一个表示类型。
用Actor来实现一个完全无阻塞的Par
怎样才可以实现一个非阻塞的Par表示类型呢？我们可以尝试自己写一个Future类型替换java.until.concurrent.Future。

  sealed trait Future[A] {
    private[chp7] def apply(f: A => Unit): Unit
  }

  type Par[+A] = ExecutorService => Future[A]

基于新的Par类型，让我们重新实现API：

  sealed trait Future[A] {
    private[chp7] def apply(f: A => Unit): Unit
  }

  type Par[+A] = ExecutorService => Future[A]


  //接受一个已求值的A，返回结果将会在另一个线程中执行
  def unit[A](a: A): Par[A] = es => new Future[A] {
    def apply(f: (A) => Unit): Unit = f(a)
  }

  //接受一个未求值的A，返回结果将会在另一个线程中执行
  def lazyUnit[A](a: => A): Par[A] = folk(unit(a))

  //从并行计算中抽取结果
  def run[A](es: ExecutorService)(pa: Par[A]): A = {
    val ref = new AtomicReference[A]
    val latch = new CountDownLatch(1)
    pa(es) {a => ref.set(a); latch.countDown()}
    latch.await()
    ref.get()
  }

  //将par[A]分配另一个独立的线程中去运行
  def folk[A](pa: => Par[A]): Par[A] = es => new Future[A] {
    def apply(f: (A) => Unit): Unit = eval(es)(pa(es)(f))
  }

  def eval(es: ExecutorService)(f: => Unit): Unit =
    es.submit(new Callable[Unit] {
      override def call(): Unit = f
    })

  def map2[A, B, C](pa: Par[A], pb: Par[B])(f: (A, B) => C): Par[C] =
    es => new Future[C] {
      def apply(cb: (C) => Unit): Unit = {
        var ar: Option[A] = None
        var br: Option[B] = None
        val combiner = Actor[Either[A, B]](es) {
          case Left(a) => br match {
            case None => ar = Some(a)
            case Some(b) => eval(es)(cb(f(a, b)))
          }
          case Right(b) => ar match {
            case None => br = Some(b)
            case Some(a) => eval(es)(cb(f(a, b)))
          }
        }
        pa(es)(a => combiner ! Left(a))
        pb(es)(b => combiner ! Right(b))
      }
    }

完善组合子为更加通用的形式
练习 7.11
实现choiceN，根据一个结果多多个计算进行选择。

  def choiceN[A](n: Par[Int])(choice: List[Par[A]]): Par[A] =
    es => {
      val i = n(es).get()
      choice(i)(es)
    }

练习 7.13
实现一个更加通用的chooser

  def chooser[A, B](pa: Par[A])(choices: A => Par[B]): Par[B] =
    es => {
      val a = pa(es).get()
      choices(a)(es)
    }