final release: that's it folks!

Author: Luc Duponcheel

src/main/scala/bescala/Active.scala

@@ -0,0 +1,74 @@
+package bescala
+
+import java.util.concurrent.TimeUnit
+
+import bescala.Util.{Atomic, async}
+
+import scala.language.{higherKinds, implicitConversions}
+
+object Active extends Common {
+
+  override type M[A] = java.util.concurrent.Future[A]
+
+  // actively blocking get
+  override def fromM[A](ma: M[A]): A = ma.get
+
+  // we do not care to much about timeout
+  override def toM[A](a: => A): M[A] = new M[A] {
+    def get = a
+
+    def get(timeout: Long, units: TimeUnit) = get
+
+    def isDone = true
+
+    def isCancelled = false
+
+    def cancel(evenIfRunning: Boolean): Boolean = false
+  }
+
+  override def map[A, B](pa: Par[A])(a2b: A => B): Par[B] =
+    es => {
+      val ma: M[A] = pa(es)
+      toM(a2b(fromM(ma)))
+    }
+
+  // it is *very* important to define `ma` and `mb` outside of `toM(ab2c(fromM(ma), fromM(mb)))`
+  //  override def map2[A, B, C](pa: Par[A], pb: Par[B])(ab2c: (A, B) => C): Par[C] =
+  //  es => {
+  //    val ma = pa(es)
+  //    val mb = pb(es)
+  //    toM(ab2c(fromM(ma), fromM(mb)))
+  //  }
+
+  override def map2[A, B, C](pa: Par[A], pb: Par[B])(ab2c: (A, B) => C): Par[C] =
+    es => {
+      var optionalMa: Option[M[A]] = None
+      var optionalMb: Option[M[B]] = None
+      val atomicC: Atomic[M[C]] = new Atomic[M[C]]
+      val combinerActor = new Actor[Either[M[A], M[B]]](es)({
+        case Left(ma) =>
+          if (optionalMb.isDefined) atomicC.setValue(toM(ab2c(ma.get, optionalMb.get.get)))
+          else optionalMa = Some(ma)
+        case Right(mb) =>
+          if (optionalMa.isDefined) atomicC.setValue(toM(ab2c(optionalMa.get.get, mb.get)))
+          else optionalMb = Some(mb)
+      })
+      combinerActor ! Left(pa(es))
+      combinerActor ! Right(pb(es))
+
+      atomicC.getValue
+    }
+
+  override def flatMap[A, B](pa: Par[A])(a2pb: A => Par[B]): Par[B] =
+    es => {
+      val ma: M[A] = pa(es)
+      a2pb(fromM(ma))(es)
+    }
+
+  override def fork[A](pa: => Par[A]): Par[A] =
+    es => {
+      val ma: M[A] = pa(es)
+      async(es)(fromM(ma))
+    }
+
+}

src/main/scala/bescala/ActiveCommon.scala

@@ -3,10 +3,9 @@ package bescala
 import java.util.concurrent.ExecutorService
 import java.util.concurrent.atomic.{AtomicInteger, AtomicReference}
 
-import scala.annotation.tailrec
 import bescala.Util.async
 
-import scala.util.{Left, Right}
+import scala.annotation.tailrec
 
 
 //              node
@@ -26,7 +25,7 @@ class Actor[M]
 
   val nullM = null.asInstanceOf[M]
 
-  private val maxNumberOfProcessedMessages = 2
+  private val maxNumberOfProcessedMessages = 20
 
   private val suspended = new AtomicInteger(1)
 

src/main/scala/bescala/Actor.scala

@@ -2,9 +2,7 @@ package bescala
 
 import java.util.concurrent.ExecutorService
 
-import language.{higherKinds, implicitConversions}
-
-import Util.async
+import scala.language.{higherKinds, implicitConversions}
 
 trait Common {
 
@@ -24,97 +22,112 @@ trait Common {
   // abstract M[A] related
   //
 
-   def fromM[A](ma: M[A]): A
+  def fromM[A](ma: M[A]): A
 
-   def toM[A](a: => A): M[A]
+  def toM[A](a: => A): M[A]
 
   //
-  // abstract applicative and monad Par[A] features
+  // abstract applicative and monadic Par[A] computational features
   //
 
-  def map[A, B](parA: Par[A])(a2b: A => B): Par[B]
+  def map[A, B](pa: Par[A])(a2b: A => B): Par[B]
 
-  def map2[A, B, C](parA: Par[A], parB: Par[B])(ab2c: (A, B) => C): Par[C]
+  def map2[A, B, C](pa: Par[A], pb: Par[B])(ab2c: (A, B) => C): Par[C]
 
-  def flatMap[A, B](parA: Par[A])(a2pb: A => Par[B]): Par[B]
+  def flatMap[A, B](pa: Par[A])(a2pb: A => Par[B]): Par[B]
 
   //
-  //  only extra abstract Par[A] feature
+  //  only extra abstract Par[A] computational feature
   //
 
-  def fork[A](parA: => Par[A]): Par[A]
+  def fork[A](pa: => Par[A]): Par[A]
 
   //
   // concrete
   //
 
+  // monadic unit in terms of toM
   def unit[A](a: => A): Par[A] =
-    es =>
-      toM(a)
+  es =>
+    toM(a)
+
+  // monadic run in terms of fromM
+  def run[A](es: ExecutorService)(pa: => Par[A]): A =
+  fromM(pa(es))
 
-  def run[A](es: ExecutorService)(parA: => Par[A]): A =
-    fromM(parA(es))
+  //
+  // more concrete stuff
+  //
 
-  def join[A](parParA: Par[Par[A]]): Par[A] =
-    flatMap(parParA)(identity)
+  def join[A](ppa: Par[Par[A]]): Par[A] =
+    flatMap(ppa)(identity)
 
-  def choice[A](parBoolean: Par[Boolean])(trueParA: Par[A], falseParA: Par[A]): Par[A] =
-    flatMap(parBoolean)(b => if (b) trueParA else falseParA)
+  def choice[A](pb: Par[Boolean])(tpa: Par[A], fpa: Par[A]): Par[A] =
+    flatMap(pb)(b => if (b) tpa else fpa)
 
-  def choiceN[A](parInt: Par[Int])(parAs: List[Par[A]]): Par[A] =
-    flatMap(parInt)(parAs(_))
+  def choiceN[A](pi: Par[Int])(pas: List[Par[A]]): Par[A] =
+    flatMap(pi)(pas(_))
 
-  def choiceMap[K,V](parK: Par[K])(choices: Map[K, Par[V]]): Par[V] =
-    flatMap(parK)(choices)
+  def choiceMap[K, V](pk: Par[K])(choices: Map[K, Par[V]]): Par[V] =
+    flatMap(pk)(choices)
 
-  def sequence[A](parAs: List[Par[A]]): Par[List[A]] =
-    parAs.foldRight[Par[List[A]]](unit(List()))(map2(_, _)(_ :: _))
+  // quadratic instead of linear `pas.foldRight[Par[List[A]]](unit(List()))(map2(_, _)(_ :: _))`
+  def sequence[A](pas: List[Par[A]]): Par[List[A]] = {
+    if (pas.isEmpty) unit(List())
+    else if (pas.length == 1) map(pas.head)(List(_))
+    else {
+      val (lpas, rpas) = pas.splitAt(pas.length / 2)
+      val plas = sequence(lpas)
+      val pras = sequence(rpas)
+      map2(plas, pras)(_ ++ _)
+    }
+  }
 
   def mapList[A, B](as: List[A])(a2b: A => B): Par[List[B]] =
     sequence(as.map(forkedFunction(a2b)))
 
-  def filterList[A](as: List[A])(predicateA: A => Boolean): Par[List[A]] =
-    map(sequence(as map (a => unit(if (predicateA(a)) List(a) else List()))))(_.flatten)
+  def filterList[A](as: List[A])(a2b: A => Boolean): Par[List[A]] =
+    map(sequence(as map (a => unit(if (a2b(a)) List(a) else List()))))(_.flatten)
 
   //
-  // forked versions
+  // some forked versions
   //
 
   def forkedUnit[A](a: => A): Par[A] =
     fork(unit(a))
 
-  def forkedMap[A, B](parA: Par[A])(a2b: A => B): Par[B] =
-    map(fork(parA))(a2b)
-//     fork(map(fork(parA))(a2b))
+  def forkedMap[A, B](pa: Par[A])(a2b: A => B): Par[B] =
+    map(fork(pa))(a2b)
 
-  def forkedMap2[A,B,C](parA: Par[A], parB: Par[B])(ab2c: (A,B) => C): Par[C] =
-    map2(fork(parA), fork(parB))(ab2c)
-//     fork(map2(fork(parA), fork(parB))(ab2c))
+  def forkedMap2[A, B, C](pa: Par[A], pb: Par[B])(ab2c: (A, B) => C): Par[C] =
+    map2(fork(pa), fork(pb))(ab2c)
 
-  def forkedSequence[A](parAs: List[Par[A]]): Par[List[A]] = fork {
-    if (parAs.isEmpty) unit(List())
-    else if (parAs.length == 1) forkedMap(parAs.head)(List(_))
+  // also quadratic
+  def forkedSequence[A](pas: List[Par[A]]): Par[List[A]] = {
+    if (pas.isEmpty) unit(List())
+    else if (pas.length == 1) forkedMap(pas.head)(List(_))
     else {
-      val (leftParAs, rightParAs) = parAs.splitAt(parAs.length/2)
-      forkedMap2(forkedSequence(leftParAs), forkedSequence(rightParAs))(_ ++ _)
+      val (lpas, rpas) = pas.splitAt(pas.length / 2)
+      forkedMap2(forkedSequence(lpas), forkedSequence(rpas))(_ ++ _)
     }
   }
 
   def forkedFunction[A, B](a2b: A => B): A => Par[B] =
-    a => forkedUnit(a2b(a))
+    a =>
+      forkedUnit(a2b(a))
 
   def forkedMapList[A, B](as: List[A])(a2b: A => B): Par[List[B]] =
     forkedSequence(as.map(forkedFunction(a2b)))
 
-  def forkedFilterList[A](as: List[A])(predicateA: A => Boolean): Par[List[A]] =
-    forkedMap(forkedSequence(as map (a => forkedUnit(if (predicateA(a)) List(a) else List()))))(_.flatten)
+  def forkedFilterList[A](as: List[A])(a2b: A => Boolean): Par[List[A]] =
+    forkedMap(forkedSequence(as map (a => forkedUnit(if (a2b(a)) List(a) else List()))))(_.flatten)
 
   //
   // example: sorting
   //
 
-  def sort[A : Ordering](parAs: Par[List[A]]) = map(parAs)(_.sorted)
+  def sort[A: Ordering](pas: Par[List[A]]) = map(pas)(_.sorted)
 
-  def forkedSort[A : Ordering](parAs: Par[List[A]]) = forkedMap(parAs)(_.sorted)
+  def forkedSort[A: Ordering](pas: Par[List[A]]) = forkedMap(pas)(_.sorted)
 
 }