CODEWITHSUNDEEP
scalaparallel-processing
Will I Am
Will I Am

Reputation: 2672

Parallel processing pattern in Scala

I hope this is not a stupid question or I'm missing something obvious. I'm following the Coursera parallel programming class, and in week 1 they have the following code to run tasks in parallel (may differ slightly, since I typed mine in):

object parallelism {
  val forkJoinPool = new ForkJoinPool

  abstract class TaskScheduler {
    def schedule[T](body: => T): ForkJoinTask[T]
    def parallel[A, B](taskA: => A, taskB: => B): (A, B) = {
      val right = task {
        taskB
      }
      val left = taskA
      (left, right.join())
    }
  }

  class DefaultTaskScheduler extends TaskScheduler {
    def schedule[T](body: => T): ForkJoinTask[T] = {
      val t = new RecursiveTask[T] {
        def compute = body
      }
      Thread.currentThread match {
        case wt: ForkJoinWorkerThread =>  t.fork()
        case _ => forkJoinPool.execute(t)
      }
      t
    }
  }

  val scheduler =
    new DynamicVariable[TaskScheduler](new DefaultTaskScheduler)

  def task[T](body: => T): ForkJoinTask[T] = {
    scheduler.value.schedule(body)
  }

  def parallel[A, B](taskA: => A, taskB: => B): (A, B) = {
    scheduler.value.parallel(taskA, taskB)
  }
}

I wrote a unit test that goes soemthing like this:

test("Test two task parallelizer") {
    val (r1, t1) = timed {
      ( sieveOfEratosthenes(100000),
        sieveOfEratosthenes(100000))
    }

    val (r2, t2) = timed {
      parallel (
        sieveOfEratosthenes(100000),
        sieveOfEratosthenes(100000)
      )
    }    
    assert(t2 < t1)
  }

  test("Test four task parallelizer") {
    val (r1, t1) = timed {
      (sieveOfEratosthenes(100000),
      sieveOfEratosthenes(100000),
      sieveOfEratosthenes(100000),
      sieveOfEratosthenes(100000))
    }

    val (r2, t2) = timed {
      parallel (
        parallel (
          sieveOfEratosthenes(100000),
          sieveOfEratosthenes(100000)
        ),
        parallel (
          sieveOfEratosthenes(100000),
          sieveOfEratosthenes(100000)
        )

      )
    }
    assert(t2 < t1)
 }

On the first test, I get good savings (300ms down to 50ms) savings, but on the second test, I only get about 20ms savings, and if I run it often enough the time may actually increase and fail my test. (the second value in the tuple returned by "timed" is the time in milliseconds)

The test method is first version from here: https://rosettacode.org/wiki/Sieve_of_Eratosthenes#Scala

Can someone teach me what is going on in the second test? If it matters, I'm running on a single cpu, quad core i5. The number of threads I create doesn't seem to make a lot of difference for this particular test.

Upvotes: 2

Views: 593

Answers (1)

Victor Moroz
Victor Moroz

Reputation: 9225

The implementation of sieveOfEratosthenes you chose is already parallel (it's using ParSet), so parallelizing it further won't help.

The speedup you see in the first test is probably JIT warmup.

Upvotes: 1

Related Questions