Why is using replicate much slower than serial execution?

Question

Hi, I've got a bit of a problem. I wanted to use scala.concurrent.ops.replicate to parallelize my program. But I found out, that the algorithm actually becomes much slower. So I wrote a little test and still got the same result. So here they are.

Serial Code: Takes about 63 seconds to finish

object SerTest {
  def main(args: Array[String]) {
      for(x <- 1 to 10){
        for(i <- 1 to 4) {
          for(j <- 1 to 100000) {
            val a = BigInt(j).isProbablePrime(1000)
            if(!a && j == 100000) println(i + " is ready")}}}}}

Concurrent Code: Takes about 161 seconds to finish

object ParTest {
  def main(args: Array[String]) {
      for(x <- 1 to 10){
        replicate(1,5) { i =>
          for(j <- 1 to 100000) {
            val a = BigInt(j).isProbablePrime(1000)
            if(!a && j == 100000) println(i + " is ready")}}}}}

So where is the completely obvious and embarrassing error I made? :)

Edit: Ohh, and I am running this on a Quadcore-CPU. So it should actually be faster :)

Edit2: Because of the answer of Kevin Wright I changed the programs slightly to have a longer time to run.

Answer 1

Looking at your sample code, I'd guess that you're jumping straight into the main method from the command line. This is the absolute worst way that you can go about microprofiling in Java!

You should first run your test for a handful of times (within the same VM invocation), at least enough so that the JVM has been properly warmed up and running for a good 30 seconds before you even think about starting to measure anything. This will ensure that it's running compiled (and not interpreted) code, and that it's been fully optimised.

You also need to be aware of the cost of starting up threads. For short-running loops, this will be a prohibitive overhead, and will consume more time than the loop itself!

update

The following definitions come from ops.scala:

val defaultRunner: FutureTaskRunner = TaskRunners.threadRunner
def spawn(p: => Unit)(implicit runner: TaskRunner = defaultRunner): Unit = {...}
def replicate(start: Int, end: Int)(p: Int => Unit) {...}

So the actual runner used is injected as an implicit, or defaults to TaskRunners.threadRunner

You can try changing this to use a thread pool by prefixing your code with:

implicit val runner = TaskRunners.threadPoolRunner

Or I believe the following will also work:

import concurrent.TaskRunners.threadPoolRunner

See if that makes any difference

---

On second thoughts...

I don't think that parameter is actually going to get passed through to the nested call to spawn, probably better if you just duplicate the method yourself (I currently have a query about this posted on the mailing lists).

For your convenience, here's the method in its full, terrifying, glory:

def replicate(start: Int, end: Int)(p: Int => Unit) {
  if (start == end) 
    ()
  else if (start + 1 == end)
    p(start)
  else {
    val mid = (start + end) / 2
    spawn { replicate(start, mid)(p) }
    replicate(mid, end)(p)
  }
}

(you still need to define the implicit runner...)

Answer 2

Take a look at the source for BigInteger.isProbablePrime (BigInt delegates to the java library). It's doing a serious amount of new BigInteger() since that's an immutable class.

My guess is that the memory allocation is causing too much contention to benefit from parallelization. You can probably confirm by substituting a simple calculation (like multiplying a 100MM numbers together) for your prime test. Or, rewrite the prime test using var longs instead of BigInt.

Also, ops.replicate spawns operations into new threads rather than utilizing some sort of thread pool. Thread creation has a certain amount of overhead, but not enough to be a problem in this case. I personally prefer to stick with the more robust java.util.concurrent libraries.