Using scala actor framework as fork-join computation?

Question

Is it possible, in theory, to use the Scala Actor Framework to do a kind of asynchronous divide-and-conquer computation similarly to JDK 7's Fork-Join framework? If so, how could I express an FJ problem with the framework - for example, the tutorial mergesort concept? Code snipplets are welcome.

(I came to the idea based on a resource video I've got in my other FJ related question.)

Beckwith

Answer 1

Scala does have FJ style parallelism. It's call futures and it's part of the actors library

import scala.actors.Future
import scala.actors.Futures._

def mergeSort[A <% Ordered[A]](xs : List[A]) : List[A] =  {
  // merge is not interesting, it's sequential.  The complexity lies in keeping it tail recursive
  def merge[A <% Ordered[A]](accum : List[A], left : List[A], right : List[A]) : List[A] = {
    (left, right) match {
      case (lhead::ltail, rhead::rtail) => 
        if (lhead <= rhead) merge(lhead :: accum, ltail, right)
        else merge(rhead :: accum, left, rtail)
      case (Nil, _) => accum reverse_::: right
      case _ => accum reverse_::: left
    }
  }

    // here's the parallel sort bit
  def sort[A <% Ordered[A]](xs : List[A], length : Int) : List[A] =  {
    if (length <= 1) xs
    else {
      val leftLength = length / 2
      val rightLength = length - leftLength
      val (left, right) = xs splitAt leftLength

      // fork
      val leftFork = future { sort(left, leftLength) }
      val rightFork = future { sort(right, rightLength) }

      // join
      val leftJoin = leftFork()
      val rightJoin = rightFork()

      // merge
      merge(Nil, leftJoin, rightJoin)
    }  
  }

  sort(xs, xs.length)
}

Now, to the heart of the question. If Scala didn't have futures could you write one yourself based on actors. Indeed. It would look more or less like this.

import scala.actors.Actor 
import scala.actors.Actor._

object MyFuture {
  def apply[T](x : => T) : MyFuture[T] = {
    val future = new MyFuture[T]

    val act = actor {
      react {
        case sender : Actor => sender ! (future, x)
      }
    }

    act ! self

    future

  }
}

class MyFuture[A] extends Function0[A] {
  me => 

  lazy val result = receive {
    case (`me`, result) => result.asInstanceOf[A]
  }

  def apply() = result

}

And you would use it like so

scala> val x = MyFuture(28 * 1000)
x: Foo.MyFuture[Int] = <function>

scala> x()
res4: Int = 28000

Exponible

I tried your FJ example with list of size 400 and I don't get almost no cpu load (should be 100% on all cores, right?). Any exaplanation on that? – Tertias

How this can be rewritten with scala 2.10, where they discourage from using blocking future()? – Buie