Par.scala

Par.scala

class ExecutorService {
    def submit[A](a: Callable[A]): Future[A]
}

trait Callable[A] { def call: A }

trait Future[A] {
    def get: A
    def get(timeout: Long, unit: TimeUnit): A
    def cancel(evenIfRunning: Boolean): Boolean
    def isDone: Boolean
    def isCancelled: Boolean
}

object Par {
    def unit[A](a: A): Par[A] = (es: ExecutionContext) => UnitFuture(a)
    
    private case class UnitFuture[A](get: A) extends Future[A] {
        def isDone: true
        def get(timeout: Long, unit: TimeUnit): get
        def isCancelled: false
        def cancel(evenIfRunning: Boolean): Boolean = false
    }
    
    def map2[A, B, C](a: Par[A], b: Par[B])(f: (A, B) => C): Par[C] = (es: ExecutionContext) => {
        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 Callbale[A] {
        def call = a(es).get
    })
}

// 7.4
def asyncF[A, B](f: A => B): A => Par[B] = a => lazyUnit(f(a))

// 7.5
def sequenceBalanced[A](as: IndexedSeq[Par[A]]): Par[IndexedSeq[A]] = fork {
    if (as.isEmpty) unit(Vector())
    else if (as.length == 1) map(as.head)(a => Vector(a))
    else {
        val (l, r) = as.splitAt(as.length / 2)
        map2(sequenceBalanced(l), sequenceBalanced(r))(_ ++ _)
    }
}
def sequence[A](as: List[Par[A]]): Par[List[A]] = map(sequenceBalanced(as.toIndexSeq))(_.toList)

def parMap[A, B](ps: List[A])(f: A => B): Par[List[B]] = fork {
    val fbs: List[Par[B]] = ps.map(asyncF(f))
    sequence(fbs)
}

// 7.6
def parFilter[A](as: List[A])(f: A => Boolean): Par[List[A]] = {
    val pars: List[Par[List[A]]] = as map (asyncF((a: A) => if (f(a)) List(a) else List()))
    map(sequence(pars))(_.flatten)
}

// 7.11
def choicdN[A](n: Par[Int])(choices: List[Par[A]]): Par[A] = 
    es => {
        val ind = run(es)(n).get
        run(es)(choices(ind))
    }

// 7.12
def choiceMap[K, V](key: Par[K])(choices: Map[K, Par[V]]): Par[V]
    es => {
        val k = run(es)(key).get
        run(es)(choices(k))
    }

// 7.13
def chooser[A, B](pa: Par[A])(choices: A => Par[B]): Par[B]
    es => {
        val k = run(es)(p).get
        run(es)(choices(k))
    }