.gitlab-ci.yml

@@ -25,7 +25,7 @@ grade:
   tags:
     - cs206
   image:
-    name: smarter3/moocs:reactive-actorbintree-2020-04-15
+    name: smarter3/moocs:parprog1-reductions-2020-02-24-2
     entrypoint: [""]
   allow_failure: true
   before_script:

assignment.sbt

@@ -1,4 +1,9 @@
 // Student tasks (i.e. submit, packageSubmission)
 enablePlugins(StudentTasks)
 
-
+courseraId := ch.epfl.lamp.CourseraId(
+  key = "lUUWddoGEeWPHw6r45-nxw",
+  itemId = "U1eU3",
+  premiumItemId = Some("4rXwX"),
+  partId = "gmSnR"
+)

build.sbt

@@ -1,24 +1,13 @@
-course := "reactive"
-assignment := "actorbintree"
-
-testOptions in Test += Tests.Argument(TestFrameworks.JUnit, "-a", "-v", "-s")
-parallelExecution in Test := false
-
-val akkaVersion = "2.6.0"
+course := "parprog1"
+assignment := "reductions"
 
 scalaVersion := "0.23.0-bin-20200211-5b006fb-NIGHTLY"
-
-scalacOptions ++= Seq(
-  "-feature",
-  "-deprecation",
-  "-encoding", "UTF-8",
-  "-unchecked",
-  "-language:implicitConversions"
-)
-
+scalacOptions ++= Seq("-language:implicitConversions", "-deprecation")
 libraryDependencies ++= Seq(
-  "com.typesafe.akka"        %% "akka-actor"         % akkaVersion,
-  "com.typesafe.akka"        %% "akka-testkit"       % akkaVersion % Test,
-  "com.novocode"             % "junit-interface"     % "0.11"      % Test
+  "com.storm-enroute" %% "scalameter-core" % "0.19",
+  "org.scala-lang.modules" %% "scala-parallel-collections" % "0.2.0",
+  "com.novocode" % "junit-interface" % "0.11" % Test
 ).map(_.withDottyCompat(scalaVersion.value))
-testSuite := "actorbintree.BinaryTreeSuite"
+
+testOptions in Test += Tests.Argument(TestFrameworks.JUnit, "-a", "-v", "-s")
+testSuite := "reductions.ReductionsSuite"

src/main/scala/actorbintree/BinaryTreeSet.scala

@@ -1,189 +0,0 @@
-/**
- * Copyright (C) 2009-2013 Typesafe Inc. <http://www.typesafe.com>
- */
-package actorbintree
-
-import akka.actor._
-import scala.collection.immutable.Queue
-
-object BinaryTreeSet {
-
-  trait Operation {
-    def requester: ActorRef
-    def id: Int
-    def elem: Int
-  }
-
-  trait OperationReply {
-    def id: Int
-  }
-
-  /** Request with identifier `id` to insert an element `elem` into the tree.
-    * The actor at reference `requester` should be notified when this operation
-    * is completed.
-    */
-  case class Insert(requester: ActorRef, id: Int, elem: Int) extends Operation
-
-  /** Request with identifier `id` to check whether an element `elem` is present
-    * in the tree. The actor at reference `requester` should be notified when
-    * this operation is completed.
-    */
-  case class Contains(requester: ActorRef, id: Int, elem: Int) extends Operation
-
-  /** Request with identifier `id` to remove the element `elem` from the tree.
-    * The actor at reference `requester` should be notified when this operation
-    * is completed.
-    */
-  case class Remove(requester: ActorRef, id: Int, elem: Int) extends Operation
-
-  /** Request to perform garbage collection */
-  case object GC
-
-  /** Holds the answer to the Contains request with identifier `id`.
-    * `result` is true if and only if the element is present in the tree.
-    */
-  case class ContainsResult(id: Int, result: Boolean) extends OperationReply
-
-  /** Message to signal successful completion of an insert or remove operation. */
-  case class OperationFinished(id: Int) extends OperationReply
-
-}
-
-
-class BinaryTreeSet extends Actor {
-  import BinaryTreeSet._
-  import BinaryTreeNode._
-
-  def createRoot: ActorRef = context.actorOf(BinaryTreeNode.props(0, initiallyRemoved = true))
-
-  var root = createRoot
-
-  // optional
-  var pendingQueue = Queue.empty[Operation]
-
-  // optional
-  def receive = normal
-
-  // optional
-  /** Accepts `Operation` and `GC` messages. */
-  val normal: Receive = {
-    case op:Operation => root ! op
-    case GC => {
-      val newRoot = createRoot;
-      root ! CopyTo(newRoot)
-      context.become(garbageCollecting(newRoot))
-    }
-   }
-
-  // optional
-  /** Handles messages while garbage collection is performed.
-    * `newRoot` is the root of the new binary tree where we want to copy
-    * all non-removed elements into.
-    */
-  def garbageCollecting(newRoot: ActorRef): Receive = {
-    case op:Operation => pendingQueue = pendingQueue.enqueue(op)
-    case CopyFinished =>
-      pendingQueue.foreach(newRoot ! _) //foreach preserves order of a queue (same as dequeueing)
-      root ! PoisonPill //Will also stop all of its children
-      pendingQueue = Queue.empty
-      root = newRoot;
-      context.become(normal)
-    //Ignore GC messages here
-  }
-}
-
-object BinaryTreeNode {
-  trait Position
-
-  case object Left extends Position
-  case object Right extends Position
-
-  case class CopyTo(treeNode: ActorRef)
-  case object CopyFinished
-
-  def props(elem: Int, initiallyRemoved: Boolean) = Props(classOf[BinaryTreeNode],  elem, initiallyRemoved)
-}
-
-class BinaryTreeNode(val elem: Int, initiallyRemoved: Boolean) extends Actor {
-  import BinaryTreeNode._
-  import BinaryTreeSet._
-
-  var subtrees = Map[Position, ActorRef]()
-  var removed = initiallyRemoved
-
-  // optional
-  def receive = normal
-
-  def goDownTo(elem : Int) : Position = if(elem < this.elem) Left else Right
-  // optional
-  /** Handles `Operation` messages and `CopyTo` requests. */
-  val normal: Receive = {
-    case Insert  (requester, id, elem) =>
-      if(elem == this.elem && !removed){
-        requester ! OperationFinished(id)
-      }else{
-        val nextPos = goDownTo(elem)
-        
-        subtrees get nextPos match{
-          case Some(node) => node ! Insert(requester, id, elem)
-          case None => {
-              val newActorSubtree = (nextPos, context.actorOf(BinaryTreeNode.props(elem, false)))
-              subtrees = subtrees + newActorSubtree
-              requester ! OperationFinished(id);
-          }
-        }
-      }
-    case Contains(requester, id, elem) =>
-      if(elem == this.elem && !removed)
-        requester ! ContainsResult(id, true)
-      else{
-        //Need to search subtrees
-        subtrees get goDownTo(elem) match{
-          case Some(node) => node ! Contains(requester, id, elem)
-          case None => requester ! ContainsResult(id, false)
-        }
-      }
-        
-    case Remove  (requester, id, elem) => 
-      if(elem == this.elem && !removed){
-        removed = true
-        requester ! OperationFinished(id)
-      }else{
-        subtrees get goDownTo(elem) match{
-          case Some(node) => node ! Remove(requester, id, elem)
-          case None => requester ! OperationFinished(id) // (elem isn't in the tree)
-        }
-      }
-
-    case CopyTo(newRoot) =>
-      //We are already done, nothing to do
-      if(removed && subtrees.isEmpty) context.parent ! CopyFinished
-      else{
-        if(!removed) newRoot ! Insert(self, elem, elem)
-        subtrees.values foreach(_ ! CopyTo(newRoot)) //Copy subtrees elems
-        //val insertConfirmed = if(removed) true else false, hence we can simply pass removed
-        context.become(copying(subtrees.values.toSet, removed))
-      }
-  }
-
-  // optional
-  /** `expected` is the set of ActorRefs whose replies we are waiting for,
-    * `insertConfirmed` tracks whether the copy of this node to the new tree has been confirmed.
-    */
-  def copying(expected: Set[ActorRef], insertConfirmed: Boolean): Receive = {
-    //To catch the insert of this node into the new tree beeing finished
-    case OperationFinished(_) => {
-      if(expected.isEmpty) context.parent ! CopyFinished
-      else context.become(copying(expected, true))
-    }
-    
-    case CopyFinished => {
-      val newExp = expected-sender
-      if(insertConfirmed && newExp.isEmpty){
-        context.parent ! CopyFinished
-      }else{
-        context.become(copying(newExp, insertConfirmed))
-      }
-    }
-  }
-}

src/main/scala/reductions/Interfaces.scala

@@ -0,0 +1,26 @@
+package reductions
+
+// Interfaces used by the grading infrastructure. Do not change signatures
+// or your submission will fail with a NoSuchMethodError.
+
+trait LineOfSightInterface {
+  def lineOfSight(input: Array[Float], output: Array[Float]): Unit
+  def upsweepSequential(input: Array[Float], from: Int, until: Int): Float
+  def upsweep(input: Array[Float], from: Int, end: Int, threshold: Int): Tree
+  def downsweepSequential(input: Array[Float], output: Array[Float], startingAngle: Float, from: Int, until: Int): Unit
+  def downsweep(input: Array[Float], output: Array[Float], startingAngle: Float, tree: Tree): Unit
+  def parLineOfSight(input: Array[Float], output: Array[Float], threshold: Int): Unit
+}
+
+trait ParallelCountChangeInterface {
+  def countChange(money: Int, coins: List[Int]): Int
+  def parCountChange(money: Int, coins: List[Int], threshold: (Int, List[Int]) => Boolean): Int
+  def moneyThreshold(startingMoney: Int): (Int, List[Int]) => Boolean
+  def totalCoinsThreshold(totalCoins: Int): (Int, List[Int]) => Boolean
+  def combinedThreshold(startingMoney: Int, allCoins: List[Int]): (Int, List[Int]) => Boolean
+}
+
+trait ParallelParenthesesBalancingInterface {
+  def balance(chars: Array[Char]): Boolean
+  def parBalance(chars: Array[Char], threshold: Int): Boolean
+}

src/main/scala/reductions/LineOfSight.scala

@@ -0,0 +1,126 @@
+package reductions
+
+import org.scalameter._
+
+object LineOfSightRunner {
+
+  val standardConfig = config(
+    Key.exec.minWarmupRuns -> 40,
+    Key.exec.maxWarmupRuns -> 80,
+    Key.exec.benchRuns -> 100,
+    Key.verbose -> false
+  ) withWarmer(new Warmer.Default)
+
+  def main(args: Array[String]): Unit = {
+    val length = 10000000
+    val input = (0 until length).map(_ % 100 * 1.0f).toArray
+    val output = new Array[Float](length + 1)
+    val seqtime = standardConfig measure {
+      LineOfSight.lineOfSight(input, output)
+    }
+    println(s"sequential time: $seqtime")
+
+    val partime = standardConfig measure {
+      LineOfSight.parLineOfSight(input, output, 10000)
+    }
+    println(s"parallel time: $partime")
+    println(s"speedup: ${seqtime.value / partime.value}")
+  }
+}
+
+sealed abstract class Tree {
+  def maxPrevious: Float
+}
+
+case class Node(left: Tree, right: Tree) extends Tree {
+  val maxPrevious = left.maxPrevious.max(right.maxPrevious)
+}
+
+case class Leaf(from: Int, until: Int, maxPrevious: Float) extends Tree
+
+object LineOfSight extends LineOfSightInterface {
+
+  def lineOfSight(input: Array[Float], output: Array[Float]): Unit = {
+    output(0) = 0
+    var i = 1
+    var maxSoFar = input(0);
+    while(i < input.length){
+      val tanOfAngle = input(i) / i;
+      if(tanOfAngle > maxSoFar) maxSoFar = tanOfAngle;
+      output(i) = maxSoFar;
+      i = i + 1
+    }
+  }
+
+  /** Traverses the specified part of the array and returns the maximum angle.
+   */
+  def upsweepSequential(input: Array[Float], from: Int, until: Int): Float = {
+    def helper(i: Int, maxAngle: Float): Float = {
+      if (i < until) {
+          val newMaxAngle = scala.math.max(maxAngle, input(i) / i)
+          helper(i + 1, newMaxAngle)
+      }
+      else maxAngle
+  }
+    helper(from, -1)
+  }
+
+  /** Traverses the part of the array starting at `from` and until `end`, and
+   *  returns the reduction tree for that part of the array.
+   *
+   *  The reduction tree is a `Leaf` if the length of the specified part of the
+   *  array is smaller or equal to `threshold`, and a `Node` otherwise.
+   *  If the specified part of the array is longer than `threshold`, then the
+   *  work is divided and done recursively in parallel.
+   */
+  def upsweep(input: Array[Float], from: Int, end: Int,
+    threshold: Int): Tree = {
+      if(end-from <= threshold) Leaf(from, end, upsweepSequential(input, from , end))
+      else {
+        val mid = (end + from)/2
+        val (lt, rt) = parallel(
+          upsweep(input, from, mid, threshold),
+          upsweep(input, mid, end, threshold))
+
+        Node(lt, rt)
+      }
+  }
+
+  /** Traverses the part of the `input` array starting at `from` and until
+   *  `until`, and computes the maximum angle for each entry of the output array,
+   *  given the `startingAngle`.
+   */
+  def downsweepSequential(input: Array[Float], output: Array[Float],
+    startingAngle: Float, from: Int, until: Int): Unit = {
+      var maxSoFar = startingAngle
+      var i = from;
+      while(i < until){
+        maxSoFar = Math.max(input(i) / i, maxSoFar)
+        output(i) = maxSoFar
+        i = i + 1
+      }
+  }
+
+  /** Pushes the maximum angle in the prefix of the array to each leaf of the
+   *  reduction `tree` in parallel, and then calls `downsweepSequential` to write
+   *  the `output` angles.
+   */
+  def downsweep(input: Array[Float], output: Array[Float], startingAngle: Float,
+    tree: Tree): Unit = {
+      tree match{
+        case Leaf(from, until, maxPrevious) => downsweepSequential(input, output, startingAngle, from, until)
+        case Node(lt, rt) => parallel(
+          downsweep(input, output, startingAngle, lt),
+          downsweep(input, output, Math.max(startingAngle, lt.maxPrevious), rt)
+        )
+      }
+  }
+
+  /** Compute the line-of-sight in parallel. */
+  def parLineOfSight(input: Array[Float], output: Array[Float],
+    threshold: Int): Unit = {
+      val t = upsweep(input, 1, output.length, threshold)
+      downsweep(input, output, 0, t)
+      //output(0) = 0;
+    }
+  }

src/main/scala/reductions/ParallelCountChange.scala

@@ -0,0 +1,96 @@
+package reductions
+
+import org.scalameter._
+
+object ParallelCountChangeRunner {
+
+  @volatile var seqResult = 0
+
+  @volatile var parResult = 0
+
+  val standardConfig = config(
+    Key.exec.minWarmupRuns -> 20,
+    Key.exec.maxWarmupRuns -> 40,
+    Key.exec.benchRuns -> 80,
+    Key.verbose -> false
+  ) withWarmer(new Warmer.Default)
+
+  def main(args: Array[String]): Unit = {
+    val amount = 250
+    val coins = List(1, 2, 5, 10, 20, 50)
+    val seqtime = standardConfig measure {
+      seqResult = ParallelCountChange.countChange(amount, coins)
+    }
+    println(s"sequential result = $seqResult")
+    println(s"sequential count time: $seqtime")
+
+    def measureParallelCountChange(threshold: => ParallelCountChange.Threshold): Unit = try {
+      val fjtime = standardConfig measure {
+        parResult = ParallelCountChange.parCountChange(amount, coins, threshold)
+      }
+      println(s"parallel result = $parResult")
+      println(s"parallel count time: $fjtime")
+      println(s"speedup: ${seqtime.value / fjtime.value}")
+    } catch {
+      case e: NotImplementedError =>
+        println("Not implemented.")
+    }
+
+    println("\n# Using moneyThreshold\n")
+    measureParallelCountChange(ParallelCountChange.moneyThreshold(amount))
+    println("\n# Using totalCoinsThreshold\n")
+    measureParallelCountChange(ParallelCountChange.totalCoinsThreshold(coins.length))
+    println("\n# Using combinedThreshold\n")
+    measureParallelCountChange(ParallelCountChange.combinedThreshold(amount, coins))
+  }
+}
+
+object ParallelCountChange extends ParallelCountChangeInterface {
+
+  /** Returns the number of ways change can be made from the specified list of
+   *  coins for the specified amount of money.
+   */
+  def countChange(money: Int, coins: List[Int]): Int = {
+    if(money < 0) 0
+    else if(money == 0) 1
+    else{
+      coins match{
+        case Nil => 0
+        case c::cs => countChange(money-c, coins) + countChange(money, cs)
+      }
+    }
+  }
+
+  type Threshold = (Int, List[Int]) => Boolean
+
+  /** In parallel, counts the number of ways change can be made from the
+   *  specified list of coins for the specified amount of money.
+   */
+  def parCountChange(money: Int, coins: List[Int], threshold: Threshold): Int = {
+    if(threshold(money, coins) || money <= 0) countChange(money, coins)
+    else {
+      coins match{
+        case Nil => 0
+        case c::cs => {
+          val (a, b) = parallel(parCountChange(money-c, coins, threshold), parCountChange(money, cs, threshold))
+          a+b
+          }
+      }
+    }
+
+  }
+
+  /** Threshold heuristic based on the starting money. */
+  def moneyThreshold(startingMoney: Int): Threshold =
+    (x, _) => x <= (startingMoney * 2) / 3
+
+  /** Threshold heuristic based on the total number of initial coins. */
+  def totalCoinsThreshold(totalCoins: Int): Threshold =
+    (_, cs) => cs.length <= (2 * totalCoins) / 3
+
+
+  /** Threshold heuristic based on the starting money and the initial list of coins. */
+  def combinedThreshold(startingMoney: Int, allCoins: List[Int]): Threshold = {
+    (money, remCoins) => money * remCoins.length <= (startingMoney * allCoins.length)/2
+  }
+}

src/main/scala/reductions/ParallelParenthesesBalancing.scala

@@ -0,0 +1,91 @@
+package reductions
+
+import scala.annotation._
+import org.scalameter._
+
+object ParallelParenthesesBalancingRunner {
+
+  @volatile var seqResult = false
+
+  @volatile var parResult = false
+
+  val standardConfig = config(
+    Key.exec.minWarmupRuns -> 40,
+    Key.exec.maxWarmupRuns -> 80,
+    Key.exec.benchRuns -> 120,
+    Key.verbose -> false
+  ) withWarmer(new Warmer.Default)
+
+  def main(args: Array[String]): Unit = {
+    val length = 100000000
+    val chars = new Array[Char](length)
+    val threshold = 10000
+    val seqtime = standardConfig measure {
+      seqResult = ParallelParenthesesBalancing.balance(chars)
+    }
+    println(s"sequential result = $seqResult")
+    println(s"sequential balancing time: $seqtime")
+
+    val fjtime = standardConfig measure {
+      parResult = ParallelParenthesesBalancing.parBalance(chars, threshold)
+    }
+    println(s"parallel result = $parResult")
+    println(s"parallel balancing time: $fjtime")
+    println(s"speedup: ${seqtime.value / fjtime.value}")
+  }
+}
+
+object ParallelParenthesesBalancing extends ParallelParenthesesBalancingInterface {
+
+  /** Returns `true` iff the parentheses in the input `chars` are balanced.
+   */
+  def balance(chars: Array[Char]): Boolean = {
+    var i = 0;
+    var count = 0;
+    while(i < chars.length){
+      if(chars(i) == '(') count = count + 1
+      else if(chars(i) == ')') count = count - 1
+
+      if(count < 0) return false
+      i = i + 1
+    }
+    if(count != 0) return false else true
+  }
+
+  /** Returns `true` iff the parentheses in the input `chars` are balanced.
+   */
+  def parBalance(chars: Array[Char], threshold: Int): Boolean = {
+
+    def traverse(idx: Int, until: Int, unmatchedOpening: Int, unmatchedClosing: Int) : (Int, Int) = {
+      if(idx >= until) (unmatchedOpening, unmatchedClosing)
+      else chars(idx) match {
+        case '(' => traverse(idx+1, until, unmatchedOpening + 1, unmatchedClosing)
+        case ')' =>
+          if(unmatchedOpening != 0) traverse(idx+1, until, unmatchedOpening - 1, unmatchedClosing)
+          else traverse(idx+1, until, unmatchedOpening, unmatchedClosing+1)
+        case _ => traverse(idx+1, until, unmatchedOpening, unmatchedClosing)
+      }
+    }
+
+    def reduce(from: Int, until: Int) : (Int,Int) = {
+
+      if(until - from <= threshold) traverse(from, until, 0, 0)
+      else{
+        val mid = (from + until) / 2
+        val ((auo, auc), (buo, buc)) = parallel(
+          reduce(from, mid),
+          reduce(mid, until)
+        )
+
+        val canMatch = Math.min(auo, buc)
+        (auo+buo-canMatch, auc + buc - canMatch)
+      }
+    }
+
+    reduce(0, chars.length) == (0, 0)
+  }
+
+  // For those who want more:
+  // Prove that your reduction operator is associative!
+
+}

src/main/scala/reductions/package.scala

@@ -0,0 +1,58 @@
+import java.util.concurrent._
+import scala.util.DynamicVariable
+
+import org.scalameter._
+
+package object reductions {
+  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)
+  }
+
+  def parallel[A, B, C, D](taskA: => A, taskB: => B, taskC: => C, taskD: => D): (A, B, C, D) = {
+    val ta = task { taskA }
+    val tb = task { taskB }
+    val tc = task { taskC }
+    val td = taskD
+    (ta.join(), tb.join(), tc.join(), td)
+  }
+
+  // Workaround Dotty's handling of the existential type KeyValue
+  implicit def keyValueCoerce[T](kv: (Key[T], T)): KeyValue = {
+    kv.asInstanceOf[KeyValue]
+  }
+}

src/test/scala/actorbintree/BinaryTreeSuite.scala

@@ -1,126 +0,0 @@
-/**
- * Copyright (C) 2009-2015 Typesafe Inc. <http://www.typesafe.com>
- */
-package actorbintree
-
-import akka.actor.{ActorRef, ActorSystem, Props, actorRef2Scala, scala2ActorRef}
-import akka.testkit.{ImplicitSender, TestKit, TestProbe}
-import org.junit.Test
-import org.junit.Assert._
-
-import scala.util.Random
-import scala.concurrent.duration._
-
-class BinaryTreeSuite extends TestKit(ActorSystem("BinaryTreeSuite")) with ImplicitSender {
-
-  import actorbintree.BinaryTreeSet._
-
-  def receiveN(requester: TestProbe, ops: Seq[Operation], expectedReplies: Seq[OperationReply]): Unit =
-    requester.within(5.seconds) {
-      val repliesUnsorted = for (i <- 1 to ops.size) yield try {
-        requester.expectMsgType[OperationReply]
-      } catch {
-        case ex: Throwable if ops.size > 10 => sys.error(s"failure to receive confirmation $i/${ops.size}\n$ex")
-        case ex: Throwable                  => sys.error(s"failure to receive confirmation $i/${ops.size}\nRequests:" + ops.mkString("\n    ", "\n     ", "") + s"\n$ex")
-      }
-      val replies = repliesUnsorted.sortBy(_.id)
-      if (replies != expectedReplies) {
-        val pairs = (replies zip expectedReplies).zipWithIndex filter (x => x._1._1 != x._1._2)
-        fail("unexpected replies:" + pairs.map(x => s"at index ${x._2}: got ${x._1._1}, expected ${x._1._2}").mkString("\n    ", "\n    ", ""))
-      }
-    }
-
-  def verify(probe: TestProbe, ops: Seq[Operation], expected: Seq[OperationReply]): Unit = {
-    val topNode = system.actorOf(Props[BinaryTreeSet])
-
-    ops foreach { op =>
-      topNode ! op
-    }
-
-    receiveN(probe, ops, expected)
-    // the grader also verifies that enough actors are created
-  }
-
-  @Test def `proper inserts and lookups (5pts)`(): Unit = {
-    val topNode = system.actorOf(Props[BinaryTreeSet])
-
-    topNode ! Contains(testActor, id = 1, 1)
-    expectMsg(ContainsResult(1, false))
-
-    topNode ! Insert(testActor, id = 2, 1)
-    topNode ! Contains(testActor, id = 3, 1)
-
-    expectMsg(OperationFinished(2))
-    expectMsg(ContainsResult(3, true))
-    ()
-  }
-
-  @Test def `instruction example (5pts)`(): Unit = {
-    val requester = TestProbe()
-    val requesterRef = requester.ref
-    val ops = List(
-      Insert(requesterRef, id=100, 1),
-      Contains(requesterRef, id=50, 2),
-      Remove(requesterRef, id=10, 1),
-      Insert(requesterRef, id=20, 2),
-      Contains(requesterRef, id=80, 1),
-      Contains(requesterRef, id=70, 2)
-    )
-
-    val expectedReplies = List(
-      OperationFinished(id=10),
-      OperationFinished(id=20),
-      ContainsResult(id=50, false),
-      ContainsResult(id=70, true),
-      ContainsResult(id=80, false),
-      OperationFinished(id=100)
-    )
-
-    verify(requester, ops, expectedReplies)
-  }
-
-
-  @Test def `behave identically to built-in set (includes GC) (40pts)`(): Unit = {
-    val rnd = new Random()
-    def randomOperations(requester: ActorRef, count: Int): Seq[Operation] = {
-      def randomElement: Int = rnd.nextInt(100)
-      def randomOperation(requester: ActorRef, id: Int): Operation = rnd.nextInt(4) match {
-        case 0 => Insert(requester, id, randomElement)
-        case 1 => Insert(requester, id, randomElement)
-        case 2 => Contains(requester, id, randomElement)
-        case 3 => Remove(requester, id, randomElement)
-      }
-
-      for (seq <- 0 until count) yield randomOperation(requester, seq)
-    }
-
-    def referenceReplies(operations: Seq[Operation]): Seq[OperationReply] = {
-      var referenceSet = Set.empty[Int]
-      def replyFor(op: Operation): OperationReply = op match {
-        case Insert(_, seq, elem) =>
-          referenceSet = referenceSet + elem
-          OperationFinished(seq)
-        case Remove(_, seq, elem) =>
-          referenceSet = referenceSet - elem
-          OperationFinished(seq)
-        case Contains(_, seq, elem) =>
-          ContainsResult(seq, referenceSet(elem))
-      }
-
-      for (op <- operations) yield replyFor(op)
-    }
-
-    val requester = TestProbe()
-    val topNode = system.actorOf(Props[BinaryTreeSet])
-    val count = 1000
-
-    val ops = randomOperations(requester.ref, count)
-    val expectedReplies = referenceReplies(ops)
-
-    ops foreach { op =>
-      topNode ! op
-      if (rnd.nextDouble() < 0.1) topNode ! GC
-    }
-    receiveN(requester, ops, expectedReplies)
-  }
-}

src/test/scala/reductions/ReductionsSuite.scala

@@ -0,0 +1,124 @@
+package reductions
+
+import java.util.concurrent._
+import scala.collection._
+import java.util.concurrent.ForkJoinPool.ForkJoinWorkerThreadFactory
+import org.junit._
+import org.junit.Assert.assertEquals
+
+class ReductionsSuite {
+  /*****************
+   * LINE OF SIGHT *
+   *****************/
+
+  import LineOfSight._
+  @Test def `lineOfSight should correctly handle an array of size 4`: Unit = {
+    val output = new Array[Float](4)
+    lineOfSight(Array[Float](0f, 1f, 8f, 9f), output)
+    assertEquals(List(0f, 1f, 4f, 4f), output.toList)
+  }
+
+
+
+
+  /*******************************
+   * PARALLEL COUNT CHANGE SUITE *
+   *******************************/
+
+  import ParallelCountChange._
+
+  @Test def `countChange should return 0 for money < 0`: Unit = {
+    def check(money: Int, coins: List[Int]) =
+      assert(countChange(money, coins) == 0,
+        s"countChang($money, _) should be 0")
+
+    check(-1, List())
+    check(-1, List(1, 2, 3))
+    check(-Int.MinValue, List())
+    check(-Int.MinValue, List(1, 2, 3))
+  }
+
+  @Test def `countChange should return 1 when money == 0`: Unit = {
+    def check(coins: List[Int]) =
+      assert(countChange(0, coins) == 1,
+        s"countChang(0, _) should be 1")
+
+    check(List())
+    check(List(1, 2, 3))
+    check(List.range(1, 100))
+  }
+
+  @Test def `countChange should return 0 for money > 0 and coins = List()`: Unit = {
+    def check(money: Int) =
+      assert(countChange(money, List()) == 0,
+        s"countChang($money, List()) should be 0")
+
+    check(1)
+    check(Int.MaxValue)
+  }
+
+  @Test def `countChange should work when there is only one coin`: Unit = {
+    def check(money: Int, coins: List[Int], expected: Int) =
+      assert(countChange(money, coins) == expected,
+        s"countChange($money, $coins) should be $expected")
+
+    check(1, List(1), 1)
+    check(2, List(1), 1)
+    check(1, List(2), 0)
+    check(Int.MaxValue, List(Int.MaxValue), 1)
+    check(Int.MaxValue - 1, List(Int.MaxValue), 0)
+  }
+
+  @Test def `countChange should work for multi-coins`: Unit = {
+    def check(money: Int, coins: List[Int], expected: Int) =
+      assert(countChange(money, coins) == expected,
+        s"countChange($money, $coins) should be $expected")
+
+    check(50, List(1, 2, 5, 10), 341)
+    check(250, List(1, 2, 5, 10, 20, 50), 177863)
+  }
+
+
+  /**********************************
+   * PARALLEL PARENTHESES BALANCING *
+   **********************************/
+
+  import ParallelParenthesesBalancing._
+
+  @Test def `balance should work for empty string`: Unit = {
+    def check(input: String, expected: Boolean) =
+      assert(balance(input.toArray) == expected,
+        s"balance($input) should be $expected")
+
+    check("", true)
+  }
+
+  @Test def `balance should work for string of length 1`: Unit = {
+    def check(input: String, expected: Boolean) =
+      assert(balance(input.toArray) == expected,
+        s"balance($input) should be $expected")
+
+    check("(", false)
+    check(")", false)
+    check(".", true)
+  }
+
+  @Test def `balance should work for string of length 2`: Unit = {
+    def check(input: String, expected: Boolean) =
+      assert(balance(input.toArray) == expected,
+        s"balance($input) should be $expected")
+
+    check("()", true)
+    check(")(", false)
+    check("((", false)
+    check("))", false)
+    check(".)", false)
+    check(".(", false)
+    check("(.", false)
+    check(").", false)
+  }
+
+
+  @Rule def individualTestTimeout = new org.junit.rules.Timeout(10 * 1000)
+}
+