.gitlab-ci.yml

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

build.sbt

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

@@ -1,32 +0,0 @@
-package lockfree
-
-import instrumentation.LockFreeMonitor
-import scala.collection._
-import java.util.concurrent.atomic._
-
-
-abstract class AbstractSortedList extends LockFreeMonitor {
-  
-  def createNode(value: Int, tail: Option[Node], isHead: Boolean = false) = new Node(value, tail) {
-    override def toString = if(isHead) "HEAD" else super.toString
-  }
-  
-  def firstNode: Option[Node] 
-  
-  def findNodeWithPrev(pred: Int => Boolean): (Node, Option[Node])
-  
-  def insert(e: Int): Unit
-  
-  def delete(e: Int): Boolean
-    
-  def toList: List[Int] = {
-    var curr = firstNode
-    var list = List[Int]()
-    while(curr.nonEmpty) {
-      if(!curr.get.deleted) list :+= curr.get.value
-      curr = curr.get.next 
-    }
-    list
-  }
-}
-

src/main/scala/lockfree/AtomicVariable.scala

@@ -1,25 +0,0 @@
-package lockfree
-
-import java.util.concurrent.atomic._
-
-abstract class AbstractAtomicVariable[T] {
-  def get: T
-  def compareAndSet(expect: T, newval: T) : Boolean
-}
-
-class AtomicVariable[T](initial: T) extends AbstractAtomicVariable[T] {
-
-  private val atomic = new AtomicReference[T](initial)
-
-  override def get: T = atomic.get()
-
-  override def compareAndSet(expected: T, value: T): Boolean = {
-    val current = atomic.get
-    if (current == expected) {
-      atomic.compareAndSet(current, value)
-    }
-    else {
-      false
-    }
-  }
-}

src/main/scala/lockfree/Node.scala

@@ -1,23 +0,0 @@
- 
-package lockfree
-
-abstract class Node(val value: Int, initTail: Option[Node]) {
-
-  // The type of mutable state held by node.
-  type State = (Option[Node], Boolean) //Boolean flag indicates whether THIS has been deleted or not !! not next
-
-  // The initial mutable state of the node.
-  def initialState: State = (initTail, false);
-
-  // The atomic variable that holds the state.
-  val atomicState: AbstractAtomicVariable[State] = new AtomicVariable[State](initialState)
-
-  // Function to read the next node from the state.
-  def next: Option[Node] = atomicState.get._1
-
-  // Should return true if the node was marked as deleted.
-  def deleted: Boolean = atomicState.get._2
-
-  // Should mark the node as deleted.
-  def mark: Boolean = atomicState.compareAndSet((next, false), (next, true))
-}

src/main/scala/lockfree/SortedList.scala

@@ -1,54 +0,0 @@
-package lockfree
-
-import scala.annotation.tailrec
-
-class SortedList extends AbstractSortedList {
-
-  // The sentinel node at the head.
-  private val _head: Node = createNode(0, None, isHead = true)
-
-  // The first logical node is referenced by the head.
-  def firstNode: Option[Node] = _head.next
-
-  // Finds the first node whose value satisfies the predicate.
-  // Returns the predecessor of the node and the node.
-  def findNodeWithPrev(pred: Int => Boolean): (Node, Option[Node]) = {
-    def findRec(prev : Node, curr : Option[Node]) : (Node, Option[Node]) = {
-      (prev, curr) match {
-        case (last, None) => (last, None);
-        case (_, Some(node)) =>
-          if(node.deleted) {
-            prev.atomicState.compareAndSet((curr, false), (node.next, false))
-            findNodeWithPrev(pred)
-          }else if(pred(node.value)) (prev, curr)
-          else findRec(node, node.next)
-      }
-    }
-    findRec(_head, _head.next)
-  }
-
-  // Insert an element in the list.
-  def insert(e: Int): Unit = {
-    val (pred, next) = findNodeWithPrev(_ >= e);
-    val n = createNode(e, next)
-    if(!pred.atomicState.compareAndSet((next, false), (Some(n), false))) insert(e)
-  }
-
-  // Checks if the list contains an element.
-  def contains(e: Int): Boolean = {
-    findNodeWithPrev(_ == e)._2 match {
-      case None => false
-      case _ => true;
-    }
-  }
-
-  // Delete an element from the list.
-  // Should only delete one element when multiple occurences are present.
-  def delete(e: Int): Boolean = {
-    val (pred, next) = findNodeWithPrev(_ == e)
-    next match{
-      case None => false
-      case Some(x) => if(x.mark) true else delete(e)
-    }
-  }
-}

src/main/scala/lockfree/SortingBenchmarks.scala

@@ -1,83 +0,0 @@
-package lockfree
-
-import scala.util.Random
-import instrumentation.Stats._
-
-object SortingBenchmarks extends App {
-
-  val items = 2500
-  val threads = 12
-
-  val insertions = List.fill(items)(Random.nextInt)
-  val deletions = Random.shuffle(insertions)
-  val parInsertions = parBuckets(threads, insertions)_
-  val parDeletions = parBuckets(threads, deletions)_
-
-  private class SynchronizedSortedList extends SortedList {
-    val lock = new Object
-    def syncTraverse(stop: Int => Boolean) = lock.synchronized(super.findNodeWithPrev(stop))
-    def syncInsert(e: Int) = lock.synchronized(super.insert(e))
-    def syncDelete(e: Int) = lock.synchronized(super.delete(e))
-  }
-
-  val syncSeqTime = {
-    val l = new SynchronizedSortedList
-    withTime {
-      insertions.foreach(l.syncInsert)
-      deletions.foreach(l.syncDelete)
-    }._2
-  }
-
-  val syncParTime = {
-    val l = new SynchronizedSortedList
-    withTime {
-      val ops = parInsertions(l.syncInsert) ++ parDeletions(l.syncDelete)
-      ops.foreach(_.start)
-      ops.foreach(_.join)
-    }._2
-  }
-
-  val fineSeqTime = {
-    val l = new SortedList
-    withTime {
-      insertions.foreach(l.insert)
-      deletions.foreach(l.delete)
-    }._2
-  }
-
-  val fineParTime = {
-    val l = new SortedList
-    withTime {
-      val ops = parInsertions(l.insert) ++ parDeletions(l.delete)
-      ops.foreach(_.start)
-      ops.foreach(_.join)
-    }._2
-  }
-
-  println(
-    s"""
-      |***** Results *****
-      |
-      |synchronized:
-      |  - sequential:  $syncSeqTime
-      |  - parallel:    $syncParTime
-      |  - speedup:     ${syncSeqTime.toFloat / syncParTime}
-      |
-      |fine-grained:
-      |  - sequential:  $fineSeqTime
-      |  - parallel:    $fineParTime
-      |  - speedup:     ${fineSeqTime.toFloat / fineParTime}
-      |
-      |synchronized/fine-grained:
-      |  - speedup:     ${syncParTime.toFloat / fineParTime}
-      |
-    """.stripMargin)
-
-  def parBuckets(threadCount: Int, items: List[Int])(op: Int => Any): List[Thread] =
-    items.grouped(items.size / threadCount).toList.map { bucket =>
-      new Thread {
-        override def run(): Unit =
-          bucket.foreach(op)
-      }
-    }
-}

src/main/scala/lockfree/instrumentation/MockedMonitor.scala

@@ -1,72 +0,0 @@
-package lockfree.instrumentation
-
-trait MockedMonitor extends Monitor {
-  def scheduler: Scheduler
-  
-  // Can be overriden.
-  override def waitDefault() = {
-    scheduler.log("wait")
-    scheduler updateThreadState Wait(this, scheduler.threadLocks.tail)
-  }
-  override def synchronizedDefault[T](toExecute: =>T): T = {
-    scheduler.log("synchronized check") 
-    val prevLocks = scheduler.threadLocks
-    scheduler updateThreadState Sync(this, prevLocks) // If this belongs to prevLocks, should just continue.
-    scheduler.log("synchronized -> enter")
-    try {
-      toExecute
-    } finally {
-      scheduler updateThreadState Running(prevLocks)
-      scheduler.log("synchronized -> out")
-    }    
-  }
-  override def notifyDefault() = {
-    scheduler mapOtherStates {
-      state => state match {
-        case Wait(lockToAquire, locks) if lockToAquire == this => SyncUnique(this, state.locks)
-        case e => e
-      }
-    }
-    scheduler.log("notify")
-  }
-  override def notifyAllDefault() = {
-    scheduler mapOtherStates {
-      state => state match {
-        case Wait(lockToAquire, locks) if lockToAquire == this => Sync(this, state.locks)
-        case SyncUnique(lockToAquire, locks) if lockToAquire == this => Sync(this, state.locks)
-        case e => e
-      }
-    }
-    scheduler.log("notifyAll")
-  }
-}
-
-trait LockFreeMonitor extends Monitor {
-  override def waitDefault() = {
-    throw new Exception("Please use lock-free structures and do not use wait()")
-  }
-  override def synchronizedDefault[T](toExecute: =>T): T = {
-    throw new Exception("Please use lock-free structures and do not use synchronized()")
-  }
-  override def notifyDefault() = {
-    throw new Exception("Please use lock-free structures and do not use notify()")
-  }
-  override def notifyAllDefault() = {
-    throw new Exception("Please use lock-free structures and do not use notifyAll()")
-  }
-}
-
-
-abstract class ThreadState {
-  def locks: Seq[AnyRef]
-}
-trait CanContinueIfAcquiresLock extends ThreadState {
-  def lockToAquire: AnyRef
-}
-case object Start extends ThreadState { def locks: Seq[AnyRef] = Seq.empty }
-case object End extends ThreadState { def locks: Seq[AnyRef] = Seq.empty }
-case class Wait(lockToAquire: AnyRef, locks: Seq[AnyRef]) extends ThreadState
-case class SyncUnique(lockToAquire: AnyRef, locks: Seq[AnyRef]) extends ThreadState with CanContinueIfAcquiresLock
-case class Sync(lockToAquire: AnyRef, locks: Seq[AnyRef]) extends ThreadState with CanContinueIfAcquiresLock
-case class Running(locks: Seq[AnyRef]) extends ThreadState
-case class VariableReadWrite(locks: Seq[AnyRef]) extends ThreadState

src/main/scala/lockfree/instrumentation/Monitor.scala

@@ -1,23 +0,0 @@
-package lockfree.instrumentation
-
-class Dummy
-
-trait Monitor {
-  implicit val dummy: Dummy = new Dummy
-
-  def wait()(implicit i: Dummy) = waitDefault()
-
-  def synchronized[T](e: => T)(implicit i: Dummy) = synchronizedDefault(e)
-
-  def notify()(implicit i: Dummy) = notifyDefault()
-
-  def notifyAll()(implicit i: Dummy) = notifyAllDefault()
-
-  private val lock = new AnyRef
-
-  // Can be overriden.
-  def waitDefault(): Unit = lock.wait()
-  def synchronizedDefault[T](toExecute: =>T): T = lock.synchronized(toExecute)
-  def notifyDefault(): Unit = lock.notify()
-  def notifyAllDefault(): Unit = lock.notifyAll()
-}

src/main/scala/lockfree/instrumentation/SchedulableSortedList.scala

@@ -1,35 +0,0 @@
-package lockfree.instrumentation
-
-import scala.annotation.tailrec
-import lockfree._
-import java.util.concurrent.atomic._
-
-class SchedulableAtomicVariable[T](initial: T, scheduler: Scheduler, self: Node) extends AbstractAtomicVariable[T] {
-  private val proxied: AtomicVariable[T] = new AtomicVariable[T](initial)
-
-  override def get: T = scheduler.exec {
-    proxied.get
-  } ( s"", Some(res => s"$self: get $res") )
-
-  override def compareAndSet(expect: T, newval: T): Boolean = {
-    scheduler.exec {
-      proxied.compareAndSet(expect, newval)
-    } (s"$self: compareAndSet expect = $expect, newval = $newval", Some(res => s"$self: Did it set? $res") )
-  }
-}
-
-class SchedulableNode(value: Int, initTail: Option[Node], val scheduler: Scheduler) extends Node(value, initTail) with LockFreeMonitor { self =>
-
-  override val atomicState: AbstractAtomicVariable[State] = new SchedulableAtomicVariable[State](initialState, scheduler, this)
-
-  override def toString: String =
-    String.format("Node(%s)#%02d", value.toString, new Integer(## % 100))
-}
-
-class SchedulableSortedList(val scheduler: Scheduler) extends SortedList with LockFreeMonitor {
-
-  override def createNode(value: Int, tail: Option[Node], isHead: Boolean) = new SchedulableNode(value, tail, scheduler) {
-    override def toString = if(isHead) "HEAD" else super.toString
-  }
-}
-

src/main/scala/lockfree/instrumentation/Scheduler.scala

@@ -1,304 +0,0 @@
-package lockfree.instrumentation
-
-import java.util.concurrent._;
-import scala.concurrent.duration._
-import scala.collection.mutable._
-import Stats._
-
-import java.util.concurrent.atomic.AtomicInteger
-
-sealed abstract class Result
-case class RetVal(rets: List[Any]) extends Result
-case class Except(msg: String, stackTrace: Array[StackTraceElement]) extends Result
-case class Timeout(msg: String) extends Result
-
-/**
- * A class that maintains schedule and a set of thread ids.
- * The schedules are advanced after an operation of a SchedulableBuffer is performed.
- * Note: the real schedule that is executed may deviate from the input schedule
- * due to the adjustments that had to be made for locks
- */
-class Scheduler(sched: List[Int]) {
-  val maxOps = 500 // a limit on the maximum number of operations the code is allowed to perform
-
-  private var schedule = sched
-  private var numThreads = 0
-  private val realToFakeThreadId = Map[Long, Int]()
-  private val opLog = ListBuffer[String]() // a mutable list (used for efficient concat)
-  private val threadStates = Map[Int, ThreadState]()
-
-  /**
-   * Runs a set of operations in parallel as per the schedule.
-   * Each operation may consist of many primitive operations like reads or writes
-   * to shared data structure each of which should be executed using the function `exec`.
-   * @timeout in milliseconds
-   * @return true - all threads completed on time,  false -some tests timed out.
-   */
-  def runInParallel(timeout: Long, ops: List[() => Any]): Result = {
-    numThreads = ops.length
-    val threadRes = Array.fill(numThreads) { None: Any }
-    var exception: Option[Except] = None
-    val syncObject = new Object()
-    var completed = new AtomicInteger(0)
-    // create threads
-    val threads = ops.zipWithIndex.map {
-      case (op, i) =>
-        new Thread(new Runnable() {
-          def run(): Unit = {
-            val fakeId = i + 1
-            setThreadId(fakeId)
-            try {
-              updateThreadState(Start)
-              val res = op()
-              updateThreadState(End)
-              threadRes(i) = res
-              // notify the master thread if all threads have completed
-              if (completed.incrementAndGet() == ops.length) {
-                syncObject.synchronized { syncObject.notifyAll() }
-              }
-            } catch {
-              case e: Throwable if exception != None => // do nothing here and silently fail
-              case e: Throwable =>
-                log(s"throw ${e.toString}")
-                exception = Some(Except(s"Thread $fakeId crashed on the following schedule: \n" + opLog.mkString("\n"),
-                    e.getStackTrace))
-                syncObject.synchronized { syncObject.notifyAll() }
-              //println(s"$fakeId: ${e.toString}")
-              //Runtime.getRuntime().halt(0) //exit the JVM and all running threads (no other way to kill other threads)
-            }
-          }
-        })
-    }
-    // start all threads
-    threads.foreach(_.start())
-    // wait for all threads to complete, or for an exception to be thrown, or for the time out to expire
-    var remTime = timeout
-    syncObject.synchronized {
-      timed { if(completed.get() != ops.length) syncObject.wait(timeout) } { time => remTime -= time }
-    }
-    if (exception.isDefined) {
-      exception.get
-    } else if (remTime <= 1) { // timeout ? using 1 instead of zero to allow for some errors
-      Timeout(opLog.mkString("\n"))
-    } else {
-      // every thing executed normally
-      RetVal(threadRes.toList)
-    }
-  }
-
-  // Updates the state of the current thread
-  def updateThreadState(state: ThreadState): Unit = {
-    val tid = threadId
-    synchronized {
-      threadStates(tid) = state
-    }
-    state match {
-      case Sync(lockToAquire, locks) =>
-        if (locks.indexOf(lockToAquire) < 0) waitForTurn else {
-          // Re-aqcuiring the same lock
-          updateThreadState(Running(lockToAquire +: locks))
-        }
-      case Start      => waitStart()
-      case End        => removeFromSchedule(tid)
-      case Running(_) =>
-      case _          => waitForTurn // Wait, SyncUnique, VariableReadWrite
-    }
-  }
-
-  def waitStart(): Unit = {
-    //while (threadStates.size < numThreads) {
-    //Thread.sleep(1)
-    //}
-    synchronized {
-      if (threadStates.size < numThreads) {
-        wait()
-      } else {
-        notifyAll()
-      }
-    }
-  }
-
-  def threadLocks = {
-    synchronized {
-      threadStates(threadId).locks
-    }
-  }
-
-  def threadState = {
-    synchronized {
-      threadStates(threadId)
-    }
-  }
-
-  def mapOtherStates(f: ThreadState => ThreadState) = {
-    val exception = threadId
-    synchronized {
-      for (k <- threadStates.keys if k != exception) {
-        threadStates(k) = f(threadStates(k))
-      }
-    }
-  }
-
-  def log(str: String) = {
-    if((realToFakeThreadId contains Thread.currentThread().getId())) {
-      val space = (" " * ((threadId - 1) * 2))
-      val s = space + threadId + ":" + "\n".r.replaceAllIn(str, "\n" + space + "  ")
-      opLog += s
-    }
-  }
-
-  /**
-   * Executes a read or write operation to a global data structure as per the given schedule
-   * @param msg a message corresponding to the operation that will be logged
-   */
-  def exec[T](primop: => T)(msg: => String, postMsg: => Option[T => String] = None): T = {
-    if(! (realToFakeThreadId contains Thread.currentThread().getId())) {
-      primop
-    } else {
-      updateThreadState(VariableReadWrite(threadLocks))
-      val m = msg
-      if(m != "") log(m)
-      if (opLog.size > maxOps)
-        throw new Exception(s"Total number of reads/writes performed by threads exceed $maxOps. A possible deadlock!")
-      val res = primop
-      postMsg match {
-        case Some(m) => log(m(res))
-        case None =>
-      }
-      res
-    }
-  }
-
-  private def setThreadId(fakeId: Int) = synchronized {
-    realToFakeThreadId(Thread.currentThread.getId) = fakeId
-  }
-
-  def threadId =
-    try {
-      realToFakeThreadId(Thread.currentThread().getId())
-    } catch {
-    case e: NoSuchElementException =>
-      throw new Exception("You are accessing shared variables in the constructor. This is not allowed. The variables are already initialized!")
-    }
-
-  private def isTurn(tid: Int) = synchronized {
-    (!schedule.isEmpty && schedule.head != tid)
-  }
-
-  def canProceed(): Boolean = {
-    val tid = threadId
-    canContinue match {
-      case Some((i, state)) if i == tid =>
-        //println(s"$tid: Runs ! Was in state $state")
-        canContinue = None
-        state match {
-          case Sync(lockToAquire, locks) => updateThreadState(Running(lockToAquire +: locks))
-          case SyncUnique(lockToAquire, locks) =>
-            mapOtherStates {
-              _ match {
-                case SyncUnique(lockToAquire2, locks2) if lockToAquire2 == lockToAquire => Wait(lockToAquire2, locks2)
-                case e => e
-              }
-            }
-            updateThreadState(Running(lockToAquire +: locks))
-          case VariableReadWrite(locks) => updateThreadState(Running(locks))
-        }
-        true
-      case Some((i, state)) =>
-        //println(s"$tid: not my turn but $i !")
-        false
-      case None =>
-        false
-    }
-  }
-
-  var threadPreference = 0 // In the case the schedule is over, which thread should have the preference to execute.
-
-  /** returns true if the thread can continue to execute, and false otherwise */
-  def decide(): Option[(Int, ThreadState)] = {
-    if (!threadStates.isEmpty) { // The last thread who enters the decision loop takes the decision.
-      //println(s"$threadId: I'm taking a decision")
-      if (threadStates.values.forall { case e: Wait => true case _ => false }) {
-        val waiting = threadStates.keys.map(_.toString).mkString(", ")
-        val s = if (threadStates.size > 1) "s" else ""
-        val are = if (threadStates.size > 1) "are" else "is"
-        throw new Exception(s"Deadlock: Thread$s $waiting $are waiting but all others have ended and cannot notify them.")
-      } else {
-        // Threads can be in Wait, Sync, SyncUnique, and VariableReadWrite mode.
-        // Let's determine which ones can continue.
-        val notFree = threadStates.collect { case (id, state) => state.locks }.flatten.toSet
-        val threadsNotBlocked = threadStates.toSeq.filter {
-          case (id, v: VariableReadWrite)         => true
-          case (id, v: CanContinueIfAcquiresLock) => !notFree(v.lockToAquire) || (v.locks contains v.lockToAquire)
-          case _                                  => false
-        }
-        if (threadsNotBlocked.isEmpty) {
-          val waiting = threadStates.keys.map(_.toString).mkString(", ")
-          val s = if (threadStates.size > 1) "s" else ""
-          val are = if (threadStates.size > 1) "are" else "is"
-          val whoHasLock = threadStates.toSeq.flatMap { case (id, state) => state.locks.map(lock => (lock, id)) }.toMap
-          val reason = threadStates.collect {
-            case (id, state: CanContinueIfAcquiresLock) if !notFree(state.lockToAquire) =>
-              s"Thread $id is waiting on lock ${state.lockToAquire} held by thread ${whoHasLock(state.lockToAquire)}"
-          }.mkString("\n")
-          throw new Exception(s"Deadlock: Thread$s $waiting are interlocked. Indeed:\n$reason")
-        } else if (threadsNotBlocked.size == 1) { // Do not consume the schedule if only one thread can execute.
-          Some(threadsNotBlocked(0))
-        } else {
-          val next = schedule.indexWhere(t => threadsNotBlocked.exists { case (id, state) => id == t })
-          if (next != -1) {
-            //println(s"$threadId: schedule is $schedule, next chosen is ${schedule(next)}")
-            val chosenOne = schedule(next) // TODO: Make schedule a mutable list.
-            schedule = schedule.take(next) ++ schedule.drop(next + 1)
-            Some((chosenOne, threadStates(chosenOne)))
-          } else {
-            threadPreference = (threadPreference + 1) % threadsNotBlocked.size
-            val chosenOne = threadsNotBlocked(threadPreference) // Maybe another strategy
-            Some(chosenOne)
-            //threadsNotBlocked.indexOf(threadId) >= 0
-            /*
-            val tnb = threadsNotBlocked.map(_._1).mkString(",")
-            val s = if (schedule.isEmpty) "empty" else schedule.mkString(",")
-            val only = if (schedule.isEmpty) "" else " only"
-            throw new Exception(s"The schedule is $s but$only threads ${tnb} can continue")*/
-          }
-        }
-      }
-    } else canContinue
-  }
-
-  /**
-   * This will be called before a schedulable operation begins.
-   * This should not use synchronized
-   */
-  var numThreadsWaiting = new AtomicInteger(0)
-  //var waitingForDecision = Map[Int, Option[Int]]() // Mapping from thread ids to a number indicating who is going to make the choice.
-  var canContinue: Option[(Int, ThreadState)] = None // The result of the decision thread Id of the thread authorized to continue.
-  private def waitForTurn = {
-    synchronized {
-      if (numThreadsWaiting.incrementAndGet() == threadStates.size) {
-        canContinue = decide()
-        notifyAll()
-      }
-      //waitingForDecision(threadId) = Some(numThreadsWaiting)
-      //println(s"$threadId Entering waiting with ticket number $numThreadsWaiting/${waitingForDecision.size}")
-      while (!canProceed()) wait()
-    }
-    numThreadsWaiting.decrementAndGet()
-  }
-
-  /**
-   * To be invoked when a thread is about to complete
-   */
-  private def removeFromSchedule(fakeid: Int) = synchronized {
-    //println(s"$fakeid: I'm taking a decision because I finished")
-    schedule = schedule.filterNot(_ == fakeid)
-    threadStates -= fakeid
-    if (numThreadsWaiting.get() == threadStates.size) {
-      canContinue = decide()
-      notifyAll()
-    }
-  }
-
-  def getOperationLog() = opLog
-}

src/main/scala/lockfree/instrumentation/Stats.scala

@@ -1,23 +0,0 @@
-/* Copyright 2009-2015 EPFL, Lausanne */
-package lockfree.instrumentation
-
-import java.lang.management._
-
-/**
- * A collection of methods that can be used to collect run-time statistics about Leon programs.
- * This is mostly used to test the resources properties of Leon programs
- */
-object Stats {
-  def timed[T](code: => T)(cont: Long => Unit): T = {
-    var t1 = System.currentTimeMillis()
-    val r = code
-    cont((System.currentTimeMillis() - t1))
-    r
-  }
-
-  def withTime[T](code: => T): (T, Long) = {
-    var t1 = System.currentTimeMillis()
-    val r = code
-    (r, (System.currentTimeMillis() - t1))
-  }
-}

src/main/scala/lockfree/instrumentation/TestHelper.scala

@@ -1,124 +0,0 @@
-package lockfree.instrumentation
-
-import scala.util.Random
-import scala.collection.mutable.{Map => MutableMap}
-
-import Stats._
-
-object TestHelper {
-  val noOfSchedules = 10000 // set this to 100k during deployment
-  val readWritesPerThread = 20 // maximum number of read/writes possible in one thread
-  val contextSwitchBound = 10
-  val testTimeout = 240 // the total time out for a test in seconds
-  val schedTimeout = 15 // the total time out for execution of a schedule in secs
-
-  // Helpers
-  /*def testManySchedules(op1: => Any): Unit = testManySchedules(List(() => op1))
-  def testManySchedules(op1: => Any, op2: => Any): Unit = testManySchedules(List(() => op1, () => op2))
-  def testManySchedules(op1: => Any, op2: => Any, op3: => Any): Unit = testManySchedules(List(() => op1, () => op2, () => op3))
-  def testManySchedules(op1: => Any, op2: => Any, op3: => Any, op4: => Any): Unit = testManySchedules(List(() => op1, () => op2, () => op3, () => op4))*/
-
-  def testSequential[T](ops: Scheduler => Any)(assertions: T => (Boolean, String)) =
-    testManySchedules(1,
-      (sched: Scheduler) => {
-        (List(() => ops(sched)),
-         (res: List[Any]) => assertions(res.head.asInstanceOf[T]))
-      })
-
-  /**
-   * @numThreads number of threads
-   * @ops operations to be executed, one per thread
-   * @assertion as condition that will executed after all threads have completed (without exceptions)
-   * 					 the arguments are the results of the threads
-   */
-  def testManySchedules(numThreads: Int,
-      ops: Scheduler =>
-        (List[() => Any], // Threads
-         List[Any] => (Boolean, String)) // Assertion
-      ) = {
-    var timeout = testTimeout * 1000L
-    val threadIds = (1 to numThreads)
-    //(1 to scheduleLength).flatMap(_ => threadIds).toList.permutations.take(noOfSchedules).foreach {
-    val schedules = (new ScheduleGenerator(numThreads)).schedules()
-    var schedsExplored = 0
-    schedules.takeWhile(_ => schedsExplored <= noOfSchedules && timeout > 0).foreach {
-      //case _ if timeout <= 0 => // break
-      case schedule =>
-        schedsExplored += 1
-        val schedr = new Scheduler(schedule)
-        //println("Exploring Sched: "+schedule)
-        val (threadOps, assertion) = ops(schedr)
-        if (threadOps.size != numThreads)
-          throw new IllegalStateException(s"Number of threads: $numThreads, do not match operations of threads: $threadOps")
-        timed { schedr.runInParallel(schedTimeout * 1000, threadOps) } { t => timeout -= t } match {
-          case Timeout(msg) =>
-            throw new java.lang.AssertionError("assertion failed\n"+"The schedule took too long to complete. A possible deadlock! \n"+msg)
-          case Except(msg, stkTrace) =>
-            val traceStr = "Thread Stack trace: \n"+stkTrace.map(" at "+_.toString).mkString("\n")
-            throw new java.lang.AssertionError("assertion failed\n"+msg+"\n"+traceStr)
-          case RetVal(threadRes) =>
-            // check the assertion
-            val (success, custom_msg) = assertion(threadRes)
-            if (!success) {
-              val msg = "The following schedule resulted in wrong results: \n" + custom_msg + "\n" + schedr.getOperationLog().mkString("\n")
-              throw new java.lang.AssertionError("Assertion failed: "+msg)
-            }
-        }
-    }
-    if (timeout <= 0) {
-      throw new java.lang.AssertionError("Test took too long to complete! Cannot check all schedules as your code is too slow!")
-    }
-  }
-
-  /**
-   * A schedule generator that is based on the context bound
-   */
-  class ScheduleGenerator(numThreads: Int) {
-    val scheduleLength = readWritesPerThread * numThreads
-    val rands = (1 to scheduleLength).map(i => new Random(0xcafe * i)) // random numbers for choosing a thread at each position
-    def schedules(): LazyList[List[Int]] = {
-      var contextSwitches = 0
-      var contexts = List[Int]() // a stack of thread ids in the order of context-switches
-      val remainingOps = MutableMap[Int, Int]()
-      remainingOps ++= (1 to numThreads).map(i => (i, readWritesPerThread)) // num ops remaining in each thread
-      val liveThreads = (1 to numThreads).toSeq.toBuffer
-
-      /**
-       * Updates remainingOps and liveThreads once a thread is chosen for a position in the schedule
-       */
-      def updateState(tid: Int): Unit = {
-        val remOps = remainingOps(tid)
-        if (remOps == 0) {
-          liveThreads -= tid
-        } else {
-          remainingOps += (tid -> (remOps - 1))
-        }
-      }
-      val schedule = rands.foldLeft(List[Int]()) {
-        case (acc, r) if contextSwitches < contextSwitchBound =>
-          val tid = liveThreads(r.nextInt(liveThreads.size))
-          contexts match {
-            case prev :: tail if prev != tid => // we have a new context switch here
-              contexts +:= tid
-              contextSwitches += 1
-            case prev :: tail =>
-            case _ => // init case
-              contexts +:= tid
-          }
-          updateState(tid)
-          acc :+ tid
-        case (acc, _) => // here context-bound has been reached so complete the schedule without any more context switches
-          if (!contexts.isEmpty) {
-            contexts = contexts.dropWhile(remainingOps(_) == 0)
-          }
-          val tid = contexts match {
-            case top :: tail => top
-            case _ => liveThreads(0)  // here, there has to be threads that have not even started
-          }
-          updateState(tid)
-          acc :+ tid
-      }
-      schedule #:: schedules()
-    }
-  }
-}

src/main/scala/lockfree/instrumentation/TestUtils.scala

@@ -1,19 +0,0 @@
-package lockfree.instrumentation
-
-import scala.concurrent._
-import scala.concurrent.duration._
-import scala.concurrent.ExecutionContext.Implicits.global
-
-object TestUtils {
-  def failsOrTimesOut[T](action: => T): Boolean = {
-    val asyncAction = Future {
-      action
-    }
-    try {
-      Await.result(asyncAction, 2000.millisecond)
-    } catch {
-      case _: Throwable => return true
-    }
-    return false
-  }
-}

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/lockfree/SortedListSuite.scala

@@ -1,96 +0,0 @@
-package lockfree
-
-import scala.concurrent._
-import scala.concurrent.duration._
-import scala.collection.mutable.HashMap
-import scala.util.Random
-import org.junit._
-import org.junit.Assert.assertEquals
-import instrumentation.SchedulableSortedList
-import instrumentation.TestHelper._
-import instrumentation.TestUtils._
-
-class SortedListSuite {
-
-  @Test def `Should work when 1, 2, and 3 are inserted sequentially`: Unit = {
-    val l = new SortedList
-    l.insert(1)
-    l.insert(2)
-    l.insert(3)
-    assertEquals(List(1, 2, 3), l.toList)
-  }
-
-  @Test def `Should work when 3, 2, and 1 are inserted sequentially`: Unit = {
-    val l = new SortedList
-    l.insert(3)
-    l.insert(2)
-    l.insert(1)
-    assertEquals(List(1, 2, 3), l.toList)
-  }
-
-  @Test def `Should work when duplicate elements are inserted sequentially`: Unit = {
-    val l = new SortedList
-    l.insert(0)
-    l.insert(0)
-    l.insert(2)
-    l.insert(2)
-    assertEquals(List(0, 0, 2, 2), l.toList)
-  }
-
-  @Test def `Should return [1,4,5] when from [1,2,3,5], a thread removes 3, 2 and then inserts 4`: Unit = {
-    val l = new SortedList
-    l.insert(1)
-    l.insert(2)
-    l.insert(3)
-    l.insert(5)
-    l.delete(3)
-    l.delete(2)
-    l.insert(4)
-    assertEquals(List(1, 4, 5), l.toList)
-  }
-
-  @Test def `Should work when a random list of 100 elements are inserted sequentially`: Unit = {
-    val rand = new Random()
-    val randvals = for(i <- 1 to 100) yield rand.nextInt()
-    val l = new SortedList
-    randvals.foreach { l.insert }
-    assertEquals(randvals.sorted, l.toList)
-  }
-
-  @Test def `Should insert in parallel 1, 2 and 3 in the list (0, 4)`: Unit = {
-    testManySchedules(3, sched => {
-      val sortedList = new SchedulableSortedList(sched)
-      sortedList.insert(0)
-      sortedList.insert(4)
-      ((for(i <- 1 to 3) yield () => sortedList.insert(i)).toList,
-       results => {
-        val res = sortedList.toList
-        (res == List(0, 1, 2, 3, 4),
-        s"expected List(0, 1, 2, 3, 4), got $res")
-      })
-    })
-  }
-
-   @Test def `Should return List(true, false) when the first thread deletes 2 and the sec`: Unit = {
-    testManySchedules(2, sched => {
-      val sortedList = new SchedulableSortedList(sched)
-      sortedList.insert(1)
-      sortedList.insert(2)
-      sortedList.insert(3)
-      (List(() => sortedList.delete(2),
-            () => sortedList.delete(4)),
-      results => {
-        val res = sortedList.toList
-        val expected = List(1, 3)
-        if(res != expected) {
-           (false, s"expected the final list to be $expected, your 'delete' implementation returned ${res}")
-        } else if(results != List(true, false)) {
-          (false, s"expected threads to return List(true, false), your 'delete' implementation returned ${results}")
-        } else (true, "")
-      })
-    })
-  }
-
-
-  @Rule def individualTestTimeout = new org.junit.rules.Timeout(200 * 1000)
-}