chapter07_par.scala

package Chapter07.Parallelism.Ex02

import java.util.concurrent._

object Par {

  private case class UnitFuture[A](get: A) extends Future[A] {
    def isDone = true
    def get(timeout: Long, units: TimeUnit) = get
    def isCancelled = false
    def cancel(evenIfRunning: Boolean): Boolean = false
  }

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

  def unit[A](a: A): Par[A] = (_) => UnitFuture(a)

  def map2[A, B, C](a: Par[A], b: Par[B])(f: (A, B) => C): Par[C] =
    (es) => {
      val af = a(es)
      val bf = b(es)
      UnitFuture(f(af.get, bf.get))
    }

  def fork[A](a: => Par[A]): Par[A] =
    es => es.submit(new Callable[A] {
      def call = {
        println("tid:" + Thread.currentThread().getId)
        a(es).get
      }
    })

  def lazyUnit[A](a: => A): Par[A] =
    fork(unit(a))

  def run[A](es: ExecutorService)(a: Par[A]): Future[A] =
    a(es)

  def sum(ints: IndexedSeq[Int]): Par[Int] =
    if (ints.length <= 1) {
      lazyUnit(ints.headOption getOrElse 0)
    } else {
      val (l, r) = ints.splitAt(ints.length / 2)
      Par.map2(Par.fork(sum(l)), Par.fork(sum(r)))((a, b) => a + b)
    }

  def main(args: Array[String]): Unit = {
    val t0 = System.currentTimeMillis()
    val es = Executors.newFixedThreadPool(20)
    val par = sum(IndexedSeq(1,2,3)) // "서술"을 만든다.
    val result = run(es)(par) // "서술"을 "실행" 한다.
    println(result.get())
    val t1 = System.currentTimeMillis()
    println("Elapsed time: " + (t1 - t0) + "ms")
  }
}

/*
map2(
  fork(map2(
      fork(unit(1)),
      fork(unit(2))), (_ + _)
    ),map2(
      fork(unit(3)),
      fork(unit(4))), (_ + _)
    )
  )
), (_ + _)
*/