.gitlab-ci.yml

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

assignment.sbt

@@ -1,9 +1,4 @@
 // 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,13 +1,24 @@
-course := "parprog1"
+course := "reactive"
-assignment := "reductions"
+assignment := "actorbintree"
-
-scalaVersion := "0.23.0-bin-20200211-5b006fb-NIGHTLY"
-scalacOptions ++= Seq("-language:implicitConversions", "-deprecation")
-libraryDependencies ++= Seq(
-  "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))
 
 testOptions in Test += Tests.Argument(TestFrameworks.JUnit, "-a", "-v", "-s")
-testSuite := "reductions.ReductionsSuite"
+parallelExecution in Test := false
+
+val akkaVersion = "2.6.0"
+
+scalaVersion := "0.23.0-bin-20200211-5b006fb-NIGHTLY"
+
+scalacOptions ++= Seq(
+  "-feature",
+  "-deprecation",
+  "-encoding", "UTF-8",
+  "-unchecked",
+  "-language:implicitConversions"
+)
+
+libraryDependencies ++= Seq(
+  "com.typesafe.akka"        %% "akka-actor"         % akkaVersion,
+  "com.typesafe.akka"        %% "akka-testkit"       % akkaVersion % Test,
+  "com.novocode"             % "junit-interface"     % "0.11"      % Test
+).map(_.withDottyCompat(scalaVersion.value))
+testSuite := "actorbintree.BinaryTreeSuite"

src/main/scala/actorbintree/BinaryTreeSet.scala

@@ -0,0 +1,189 @@
+/**
+ * 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

@@ -1,26 +0,0 @@
-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

@@ -1,126 +0,0 @@
-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

@@ -1,96 +0,0 @@
-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

@@ -1,91 +0,0 @@
-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

@@ -1,58 +0,0 @@
-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

@@ -0,0 +1,126 @@
+/**
+ * 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

@@ -1,124 +0,0 @@
-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)
-}
-