src/main/scala/concpar22final02/instrumentation/Stats.scala

+/* Copyright 2009-2015 EPFL, Lausanne */
+package concpar22final02.instrumentation
+
+import java.lang.management.*
+
+/** A collection of methods that can be used to collect run-time statistics */
+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/concpar22final03/Economics.scala

+package concpar22final03
+
+import scala.concurrent.Future
+
+import scala.concurrent.ExecutionContext.Implicits.global
+import scala.concurrent.Future
+import scala.util.Random
+
+trait Economics:
+
+  /** A trading card from the game Scala: The Programming. We can own a card,
+    * but once don't anymore.
+    */
+  final class Card(val name: String)
+  def isMine(c: Card): Boolean
+
+  /** This function uses the best available database to return the sell value of
+    * a card on the market.
+    */
+  def valueOf(cardName: String): Int = List(1, cardName.length).max
+
+  /** This method represents an exact amount of money that can be hold, spent,
+    * or put in the bank
+    */
+  final class MoneyBag()
+  def moneyIn(m: MoneyBag): Int
+
+  /** If you sell a card, at some point in the future you will get some money
+    * (in a bag).
+    */
+  def sellCard(c: Card): Future[MoneyBag]
+
+  /** You can buy any "Scala: The Programming" card by providing a bag of money
+    * with the appropriate amount and waiting for the transaction to take place.
+    * You will own the returned card.
+    */
+  def buyCard(money: MoneyBag, name: String): Future[Card]
+
+  /** This simple bank account holds money for you. You can bring a money bag to
+    * increase your account's balance, or withdraw a money bag of any size not
+    * greater than your account's balance.
+    */
+  def balance: Int
+  def withdraw(amount: Int): Future[MoneyBag]
+  def deposit(bag: MoneyBag): Future[Unit]
+
+  class NotEnoughMoneyException
+      extends Exception("Not enough money provided to buy those cards")

src/main/scala/concpar22final03/Problem3.scala

+package concpar22final03
+
+import scala.concurrent.Future
+import concurrent.ExecutionContext.Implicits.global
+
+trait Problem3:
+  val economics: Economics
+  import economics.*
+
+  /** The objective is to propose a service of deck building. People come to you
+    * with some money and some cards they want to sell, and you need to return
+    * them a complete deck of the cards they want.
+    */
+  def orderDeck(
+      bag: MoneyBag,
+      cardsToSell: List[Card],
+      wantedDeck: List[String]
+  ): Future[List[Card]] =
+    Future {
+      val totalGivenMoney =
+        cardsToSell.foldLeft(moneyIn(bag))((sum, c) => sum + valueOf(c.name))
+      val totalNeededMoney =
+        wantedDeck.foldLeft(0)((sum, n) => sum + valueOf(n))
+      if totalGivenMoney < totalNeededMoney then
+        throw new NotEnoughMoneyException()
+      val soldCards: Future[Unit] =
+        if moneyIn(bag) != 0 then
+          sellListOfCards(cardsToSell).zip(deposit(bag)).map(_ => ())
+        else sellListOfCards(cardsToSell).map(_ => ())
+      soldCards.flatMap { _ => buyListOfCards(wantedDeck) }
+    }.flatten
+
+  /** This helper function will sell the provided list of cards and put the
+    * money on your personal bank account. It returns a Future of Unit, which
+    * indicates when all sales are completed.
+    */
+  def sellListOfCards(cardsToSell: List[Card]): Future[Unit] =
+    val moneyFromSales: List[Future[Unit]] = cardsToSell.map { c =>
+      sellCard(c).flatMap(m => deposit(m).map { _ => })
+    }
+    Future
+      .sequence(moneyFromSales)
+      .map(_ =>
+        ()
+      ) // Future.sequence transforms a List[Future[A]] into a Future[List[A]]
+
+  /** This helper function, given a list of wanted card names and assuming there
+    * is enough money in the bank account, will buy (in the future) those cards,
+    * and return them.
+    */
+  def buyListOfCards(wantedDeck: List[String]): Future[List[Card]] =
+
+    val boughtCards: List[Future[Card]] = wantedDeck.map { name =>
+      withdraw(valueOf(name)).flatMap(mb => buyCard(mb, name))
+    }
+    Future.sequence(boughtCards)

src/main/scala/concpar22final04/Problem4.scala

+package concpar22final04
+
+import akka.actor.*
+import akka.testkit.*
+import java.util.Date
+import akka.event.LoggingReceive
+import akka.pattern.*
+import akka.util.Timeout
+import concurrent.duration.*
+import scala.concurrent.Future
+import scala.concurrent.ExecutionContext
+
+given timeout: Timeout = Timeout(200.millis)
+
+/** Data associated with a song: a unique `id`, a `title` and an `artist`.
+  */
+case class Song(id: Int, title: String, artist: String)
+
+/** An activity in a user's activity feed, representing that `userRef` is
+  * listening to `songId`.
+  */
+case class Activity(userId: String, userName: String, songId: Int)
+
+/** Companion object of the `User` class.
+  */
+object User:
+  /** Messages that can be sent to User actors.
+    */
+  enum Protocol:
+    /** Asks for a user name and id. Should be answered by a Response.Info.
+      */
+    case GetInfo
+
+    /** Asks home page data. Should be answered by a Response.HomepageData.
+      */
+    case GetHomepageData
+
+    /** Like song with id `songId`.
+      */
+    case Like(songId: Int)
+
+    /** Unlike song with id `songId`.
+      */
+    case Unlike(songId: Int)
+
+    /** Adds `subscriber` to the list of subscribers.
+      */
+    case Subscribe(subscriber: ActorRef)
+
+    /** Remove `subscriber` from the list of subscribers.
+      */
+    case Unsubscribe(subscriber: ActorRef)
+
+    /** Adds the activity `activity` to the activity feed. This message will be
+      * sent by the users this user has subscribed to.
+      */
+    case AddActivity(activity: Activity)
+
+    /** Sent when a user starts playing a song with id `songId`. The recipient
+      * should notify all its subscribers to update their activity feeds by
+      * sending them `AddActivity(Activity(...))` messages. No answer is
+      * expected. This message is sent by external actors.
+      */
+    case Play(songId: Int)
+
+    /** Asks for home page text. Should be answered by a Response.HomepageText.
+      */
+    case GetHomepageText
+
+  /** Responses that can be sent back from User actors.
+    */
+  enum Responses:
+    /** Answer to a Protocol.GetInfo message
+      */
+    case Info(id: String, name: String)
+
+    /** Answer to a Protocol.GetHomepageData message
+      */
+    case HomepageData(songIds: List[Int], activities: List[Activity])
+
+    /** Answer to a Protocol.GetHomepageText message
+      */
+    case HomepageText(result: String)
+
+/** The `User` actor, responsible to handle `User.Protocol` messages.
+  */
+class User(id: String, name: String, songsStore: ActorRef) extends Actor:
+  import User.*
+  import User.Protocol.*
+  import User.Responses.*
+  import SongsStore.Protocol.*
+  import SongsStore.Responses.*
+
+  given ExecutionContext = context.system.dispatcher
+
+  /** Liked songs, by reverse date of liking time (the last liked song must be
+    * the first must be the first element of the list). Elements of this list
+    * must be unique: a song can only be liked once. Liking a song twice should
+    * not change the order.
+    */
+  var likedSongs: List[Int] = List()
+
+  /** Users who have subscribed to this users.
+    */
+  var subscribers: Set[ActorRef] = Set()
+
+  /** Activity feed, by reverse date of activity time (the last added activity
+    * must be the first element of the list). Items in this list should be
+    * unique by `userRef`. If a new activity with a `userRef` already in the
+    * list is added, the former should be removed, so that we always see the
+    * latest activity for each user we have subscribed to.
+    */
+  var activityFeed: List[Activity] = List()
+
+  /** This actor's behavior. */
+
+  override def receive: Receive = LoggingReceive {
+    case GetInfo =>
+      sender() ! Info(id, name)
+    case GetHomepageData =>
+      sender() ! HomepageData(likedSongs, activityFeed)
+    case Like(songId) if !likedSongs.contains(songId) =>
+      likedSongs = songId :: likedSongs
+    case Unlike(songId) =>
+      likedSongs = likedSongs.filter(_ != songId)
+    case Subscribe(ref: ActorRef) =>
+      subscribers = subscribers + ref
+    case Unsubscribe(ref: ActorRef) =>
+      subscribers = subscribers - ref
+    case AddActivity(activity: Activity) =>
+      activityFeed =
+        activity :: activityFeed.filter(_.userId != activity.userId)
+    case Play(songId) =>
+      subscribers.foreach(_ ! AddActivity(Activity(id, name, songId)))
+    case GetHomepageText =>
+      val likedSongsFuture: Future[Songs] =
+        (songsStore ? GetSongs(likedSongs)).mapTo[Songs]
+      val activitySongsFuture: Future[Songs] =
+        (songsStore ? GetSongs(activityFeed.map(_.songId))).mapTo[Songs]
+      val response: Future[HomepageText] =
+        for
+          likedSongs <- likedSongsFuture;
+          activitySongs <- activitySongsFuture
+        yield HomepageText(
+          f"""
+          |Howdy ${name}!
+          |
+          |Liked Songs:
+          |${likedSongs.songs
+              .map(song => f"* ${song.title} by ${song.artist}")
+              .mkString("\n")}
+          |
+          |Activity Feed:
+          |${activityFeed
+              .zip(activitySongs.songs)
+              .map((activity, song) =>
+                f"* ${activity.userName} is listening to ${song.title} by ${song.artist}"
+              )
+              .mkString("\n")}""".stripMargin.trim
+        )
+      response.pipeTo(sender())
+  }
+
+/** Objects containing the messages a songs store should handle.
+  */
+object SongsStore:
+  /** Ask information about a list of songs by their ids.
+    */
+  enum Protocol:
+    case GetSongs(songIds: List[Int])
+
+  /** List of `Song` corresponding to the list of IDs given to `GetSongs`.
+    */
+  enum Responses:
+    case Songs(songs: List[Song])
+
+/** A mock implementation of a songs store.
+  */
+class MockSongsStore extends Actor:
+  import SongsStore.Protocol.*
+  import SongsStore.Responses.*
+  import SongsStore.*
+
+  val songsDB = Map(
+    1 -> Song(1, "High Hopes", "Pink Floyd"),
+    2 -> Song(2, "Sunny", "Boney M."),
+    3 -> Song(3, "J'irai où tu iras", "Céline Dion & Jean-Jacques Goldman"),
+    4 -> Song(4, "Ce monde est cruel", "Vald"),
+    5 -> Song(5, "Strobe", "deadmau5"),
+    6 -> Song(6, "Désenchantée", "Mylène Farmer"),
+    7 -> Song(7, "Straight Edge", "Minor Threat"),
+    8 -> Song(8, "Hold the line", "TOTO"),
+    9 -> Song(9, "Anarchy in the UK", "Sex Pistols"),
+    10 -> Song(10, "Breakfast in America", "Supertramp")
+  )
+
+  override def receive: Receive = LoggingReceive { case GetSongs(songsIds) =>
+    sender() ! Songs(songsIds.map(songsDB))
+  }
+
+/////////////////////////
+//        DEBUG        //
+/////////////////////////
+
+/** Infrastructure to help debugging. In sbt use `run` to execute this code. The
+  * TestKit is an actor that can send messages and check the messages it
+  * receives (or not).
+  */
+@main def debug() = new TestKit(ActorSystem("DebugSystem")) with ImplicitSender:
+  import User.*
+  import User.Protocol.*
+  import User.Responses.*
+  import SongsStore.Protocol.*
+  import SongsStore.Responses.*
+
+  try
+    val songsStore = system.actorOf(Props(MockSongsStore()), "songsStore")
+    val anita = system.actorOf(Props(User("100", "Anita", songsStore)))
+
+    anita ! Like(6)
+    expectNoMessage() // expects no message is received
+
+    anita ! GetHomepageData
+    expectMsg(HomepageData(List(6), List()))
+  finally shutdown(system)

src/main/scala/ed/39368.scala

+package ed
+
+// To demonstrate different ways of pattern matching, let's consider the
+// following example case class and instance:
+case class Person(name: String, age: Int)
+val ada = Person("Ada", 36)
+
+// Run using `sbt "runMain ed.patternMatching"`.
+@main def patternMatching =
+  // There are several ways to pattern match on the `ada` instance:
+  // 1. Pattern matching on the case class constructor using `Person(name, age)`.
+  //    If the pattern matches, the value of `n` is bound `ada.name` field, and
+  //    `a` is bound to the `ada.age` field.
+  ada match
+    case Person(n, a) => println(f"$n is $a years old")
+
+  // 2. We can also check only that `ada` is of type `Person` without binding its
+  //    fields by using a `:` pattern. The `:` pattern is used to check if a value is
+  //    of a certain type.
+  ada match
+    case p: Person => println(f"${p.name} is ${p.age} years old")
+
+  // 3. If we want to both bind the fields and bind a value to the whole instance,
+  //    we can use an `@` pattern.
+  ada match
+    case p @ Person(n, a) => println(f"${p.name} is ${a} years old")

src/main/scala/ed/40056.scala

+package ed
+
+import concurrent.{Await, Future}
+import concurrent.duration.Duration
+import concurrent.ExecutionContext.Implicits.global
+
+// Pro Tip: you can print the code generated by the Scala compiler after the
+// translation of for-comprehensions by running:
+//
+// scala -Xprint:firstTransform src/main/scala/ed/40056.scala
+@main def futuresForTranslation =
+  val f1 = Future { 451 }
+  val f2 = Future { 1984 }
+  val f3 = for v1 <- f1; v2 <- f2 yield v1 + v2
+  println(Await.result(f3, Duration.Inf))
+
+  val f4 = Future { 451 }
+  val f5 = Future { 1984 }
+  val f6 = f4.flatMap(v4 => f5.map(v5 => v4 + v5))
+  println(Await.result(f6, Duration.Inf))

src/main/scala/lecture13/AskPattern.scala

+package lecture13
+
+import concurrent.{ExecutionContext, Future, Promise}
+import concurrent.duration.DurationInt
+
+import akka.actor.{Actor, ActorContext, ActorRef, ActorSystem, Props}
+import akka.event.LoggingReceive
+import akka.pattern.pipe
+import akka.testkit.{ImplicitSender, TestKit}
+import akka.util.Timeout
+
+/** This demonstrates a simplified implementation of the ask pattern.
+  *
+  * @param promise
+  *   the promise to be completed when a message is received
+  */
+class AskMiniActor(promise: Promise[Any]) extends Actor:
+  def receive = LoggingReceive {
+    case msg => promise.success(msg)
+  }
+
+extension (receiver: ActorRef)
+  def ?(msg: Any)(using
+      // In this simplified implementation, we don't use the timeout.
+      timeout: Timeout,
+      // This only used for logging purposes (and to get the actor system in the
+      // real implementation), but is not used otherwise in the implementation.
+      sender: ActorRef,
+      // In the real implementation, the actor system is retrieved differently,
+      // but here we pass it as an additional implicit parameter for simplicity.
+      context: ActorContext
+  ): Future[Any] =
+    context.system.log.debug(
+      s"Create mini actor to ask $msg from $sender to $receiver"
+    )
+
+    // Create a `Promise` that will be completed when a message is received.
+    val promise = Promise[Any]()
+
+    // Create a mini actor that will complete the `Promise` when it receives a
+    // message.
+    val miniActor = context.system.actorOf(Props(AskMiniActor(promise)))
+
+    // Send the message to the mini actor, *using the mini actor as the sender*.
+    // This part is important as it is the mini actor that needs to receive the
+    // response.
+    receiver.tell(msg, miniActor)
+
+    // Using the `Promise` from the Scala standard library, the corresponding
+    // `Future` can be retrieved with the method `future`.
+    promise.future
+
+object Person:
+  enum Protocol:
+    case GetName
+    case GetAge
+  enum Response:
+    case Name(name: String)
+    case Age(age: Int)
+    case UnknownMessage
+
+class Person(name: String, age: Int) extends Actor:
+  import Person.Protocol.*
+  import Person.Response.*
+
+  def receive = LoggingReceive {
+    case GetName => sender() ! Name(name)
+    case GetAge  => sender() ! Age(age)
+    case _       => sender() ! UnknownMessage
+  }
+
+object PersonsDatabase:
+  enum Protocol:
+    case CreatePerson(name: String, age: Int)
+    case GetPersonNames(ids: List[Int])
+  enum Response:
+    case PersonCreated(id: Int)
+    case PersonNames(names: List[String])
+
+class PersonsDatabase extends Actor:
+  import Person.Protocol.*
+  import Person.Response.*
+  import PersonsDatabase.Protocol.*
+  import PersonsDatabase.Response.*
+
+  var persons: Map[Int, ActorRef] = Map.empty
+  var maxId = 0
+
+  given ExecutionContext = context.system.dispatcher
+  given Timeout = Timeout(200.millis)
+
+  def receive = LoggingReceive {
+    case CreatePerson(name, age) =>
+      val personRef = context.actorOf(Props(Person(name, age)))
+      persons = persons + (maxId -> personRef)
+      sender() ! PersonCreated(maxId)
+      maxId += 1
+    case GetPersonNames(ids) =>
+      // We ask every person for their name using the ask pattern. This
+      // gives us a list of `Future`s that will each be completed with a
+      // `Name` message.
+      val rawResponsesFutures: List[Future[Any]] =
+        ids.map(id => (persons(id) ? GetName))
+
+      // We first map each `Future[Any]` to a `Future[Name]` using the
+      // `mapTo` method. Then, we map each `Future[Name]` to a
+      // `Future[String]` using the `map` method.
+      val namesFutures: List[Future[String]] =
+        rawResponsesFutures.map(_.mapTo[Name].map(_.name))
+
+      // We use the `Future.sequence` method to convert a
+      // `List[Future[String]]` to a `Future[List[String]]`. The resulting
+      // future will be completed once all the futures in the input list
+      // are completed.
+      val futureOfNames: Future[List[String]] = Future.sequence(namesFutures)
+
+      // Finally, map the `Future[List[String]]` to a `Future[PersonNames]` and
+      // pipe it to the sender of the `GetPersonNames` message.
+      futureOfNames.map(PersonNames.apply).pipeTo(sender())
+  }
+
+@main def askPatternDemo() =
+  new TestKit(ActorSystem("DebugSystem")) with ImplicitSender:
+    import Person.Protocol.*
+    import Person.Response.*
+    import PersonsDatabase.Protocol.*
+    import PersonsDatabase.Response.*
+
+    try
+      val personsDb = system.actorOf(Props(PersonsDatabase()))
+
+      personsDb ! CreatePerson("Anita", 20)
+      expectMsg(PersonCreated(0))
+
+      personsDb ! CreatePerson("Berta", 30)
+      expectMsg(PersonCreated(1))
+
+      personsDb ! CreatePerson("Cecilia", 40)
+      expectMsg(PersonCreated(2))
+
+      personsDb ! GetPersonNames(List(0, 1, 2))
+      expectMsg(PersonNames(List("Anita", "Berta", "Cecilia")))
+    finally shutdown(system)

src/main/scala/lecture13/Become.scala

+package lecture13
+
+import akka.actor.{Actor, ActorContext, ActorRef, ActorSystem, Props}
+import akka.event.LoggingReceive
+import akka.testkit.{ImplicitSender, TestKit}
+
+class CounterRight extends Actor:
+  def counter(n: Int): Receive = {
+    case "increment" => context.become(counter(n + 1), discardOld = false)
+    //                                                 ^^^^^^^^^^^^^^^^^^
+    //                                                 This is needed.
+    case "get"       => sender() ! n
+    case "decrement" => if n != 0 then context.unbecome()
+  }
+  def receive = counter(0)
+
+class CounterWrong extends Actor:
+  def counter(n: Int): Receive = {
+    case "increment" => context.become(counter(n + 1))
+    case "get"       => sender() ! n
+    case "decrement" => if n != 0 then context.unbecome()
+  }
+  def receive = counter(0)
+
+@main def becomeDemo() =
+  new TestKit(ActorSystem("DebugSystem")) with ImplicitSender:
+    try
+      val counter = system.actorOf(Props(CounterRight()), "counter")
+
+      counter ! "increment"
+      counter ! "get"
+      expectMsg(1)
+
+      counter ! "increment"
+      counter ! "get"
+      expectMsg(2)
+
+      counter ! "decrement"
+      counter ! "get"
+      expectMsg(1)
+      // This is wrong if we use CounterWrong because it doesn't set the
+      // discardOld parameter to false. Therefore, it will not remember the
+      // previous state and reset the behavior to the initial one, which is
+      // `counter(0)`.
+
+      counter ! "decrement"
+      counter ! "get"
+      expectMsg(0)
+    finally shutdown(system)

src/main/scala/lecture6/05-DiningPhilosophers.scala

+package lecture6
+
+class Fork(var index: Int)
+
+def philosopherTurn(i: Int, l: Fork, r: Fork): Unit =
+  Thread.sleep(10)
+
+  var (left, right) =
+    if i % 2 == 0 then (l, r)
+    else (r, l)
+    // if l.index < r.index then (l, r) else (r, l)
+
+  left.synchronized {
+    log(f"Philosopher $i picks up left fork ${left.index}")
+    right.synchronized {
+      log(f"Philosopher $i picks up right fork ${right.index} - eating")
+      Thread.sleep(100)
+      log(f"Philosopher $i puts down right fork ${right.index}")
+    }
+    log(f"Philosopher $i puts down left fork ${left.index}")
+  }
+
+@main def run() =
+  val n = 5
+  val k: Int = 3
+  val forks = new Array[Fork](n)
+  val philosophers = new Array[Thread](n)
+  for p <- 0 to n - 1 do
+    forks(p) = Fork(p)
+
+  for p <- 0 to n - 1 do
+    philosophers(p) = new Thread:
+      override def run() =
+        for _ <- 0 until k do
+          philosopherTurn(p, forks(p % n), forks((p + 1) % n))
+    philosophers(p).start
+
+  for p <- 0 to n - 1 do
+    philosophers(p).join()

src/main/scala/midterm23/BarberShopSolution.scala

+package midterm23
+
+import instrumentation.Monitor
+import scala.annotation.tailrec
+
+/** Runs one of the three provided solutions to the barber shop problem.
+  *
+  * Run using `sbt "runMain midterm23.barberSolution 2 3"` for example.
+  *
+  * @param solutionNumber
+  *   One of the 3 provided solutions to run: 1, 2, or 3.
+  * @param nCustomers
+  *   The number of customers to simulate.
+  */
+@main def barberSolution(solutionNumber: Int, nCustomers: Int): Unit =
+  val barberShop: AbstractBarberShopSolution =
+    solutionNumber match
+      case 1 => BarberShopSolution1(nCustomers)
+      case 2 => BarberShopSolution2(nCustomers)
+      case 3 => BarberShopSolution3(nCustomers)
+
+  barberSolutionMain(barberShop, solutionNumber, nCustomers)
+
+def barberSolutionMain(
+    barberShop: AbstractBarberShopSolution,
+    solutionNumber: Int,
+    nCustomers: Int
+): Unit =
+  val bthread = new Thread:
+    override def run() =
+      barberShop.barber()
+  bthread.start()
+
+  val customerThreads =
+    for i <- 1 to nCustomers yield
+      val cthread = new Thread:
+        override def run() =
+          barberShop.customer(i)
+      cthread.start()
+      cthread
+
+  bthread.join()
+  customerThreads.foreach(_.join())
+
+abstract class AbstractBarberShopSolution:
+  // var notifyBarber: Boolean = false
+  // var notifyCustomer: Boolean = false
+  // val lockObj = Object()
+
+  // In order to be able to mock the three attributes above in the tests, we
+  // replace them with three private attributes and public getters and setters.
+  // Without overrides, this is equivalent to the definitions above.
+
+  private var _notifyBarber: Boolean = false
+  private var _notifyCustomer: Boolean = false
+  private val _lockObject: Monitor = new Monitor {}
+  def notifyBarber = _notifyBarber
+  def notifyBarber_=(v: Boolean) = _notifyBarber = v
+  def notifyCustomer = _notifyCustomer
+  def notifyCustomer_=(v: Boolean) = _notifyCustomer = v
+  def lockObj: Monitor = _lockObject
+
+  def hairCut(): Unit =
+    log("start hair cut")
+    Thread.sleep(1000)
+    log("finish hair cut")
+
+  def customer(n: Int): Unit
+  def barber(): Unit
+
+  /** Logs a message to the console.
+    *
+    * Overridden in the tests to avoid printing gigabytes of output.
+    *
+    * @param s
+    */
+  def log(s: String) = println(s)
+
+/** This is the solution we expected for the `customer` method.
+  *
+  * However, now that you have learned about `@volatile`, you can see that the
+  * `customer` method is not thread-safe. Can you spot a problem with this
+  * solution?
+  *
+  * Note that the `log` calls were added to help you understand what is going
+  * on. They are not part of the expected solution.
+  */
+trait BarberShopCustomerSolution1 extends AbstractBarberShopSolution:
+  override def customer(n: Int) =
+    lockObj.synchronized {
+      log(f"customer ${n} sits on the chair and waits for the barber")
+      notifyBarber = true
+      while !notifyCustomer do {}
+      notifyCustomer = false
+      log(f"customer ${n} leaves the chair")
+    }
+
+/** This is the solution we expected for the `barber` method.
+  *
+  * Same question as above: can you spot a problem with this solution?
+  *
+  * @param customersNumber
+  *   The number of customers to simulate.
+  */
+trait BarberShopBarberSolution1(customersNumber: Int)
+    extends AbstractBarberShopSolution:
+  override def barber() =
+    // while true do
+
+    // We replace the infinite loop from the instructions with a loop doing a
+    // limited number of iterations so that we can test the code.
+    for _ <- 1 to customersNumber do
+      log("barber waits for a customer")
+      while !notifyBarber do {}
+      notifyBarber = false
+      hairCut()
+      notifyCustomer = true
+
+    log("barber stops working")
+
+class BarberShopSolution1(customersNumber: Int)
+    extends BarberShopCustomerSolution1,
+      BarberShopBarberSolution1(customersNumber)
+
+/** The problem with BarberShopSolution1 is that the `notifyBarber` and
+  * `notifyCustomer` are accessed from multiple threads without any
+  * synchronization. This means that the compiler is free to reorder the
+  * instructions in the `customer` and `barber` methods, which can lead to
+  * unexpected behaviors like infinite loops!
+  *
+  * To solve this problem, the solution below redefines the two attributes as
+  * `@volatile`.
+  *
+  * @param customersNumber
+  *   The number of customers to simulate.
+  */
+class BarberShopSolution2(customersNumber: Int)
+    extends BarberShopSolution1(customersNumber):
+  @volatile private var _notifyBarber: Boolean = false
+  @volatile private var _notifyCustomer: Boolean = false
+
+  override def notifyBarber = _notifyBarber
+  override def notifyBarber_=(v: Boolean) = _notifyBarber = v
+  override def notifyCustomer = _notifyCustomer
+  override def notifyCustomer_=(v: Boolean) = _notifyCustomer = v
+
+/** The solution below uses a different approach to solve the problem of
+  * BarberShopSolution1. Instead of using `@volatile`, it uses the lock object
+  * to synchronize the accesses to the two attributes. Note how each access to
+  * the attributes is now wrapped in a `lockObj.synchronized` block.
+  *
+  * @param customersNumber
+  *   The number of customers to simulate.
+  */
+class BarberShopSolution3(customersNumber: Int)
+    extends AbstractBarberShopSolution:
+  override def customer(n: Int) =
+    getHairCut(n)
+
+  @tailrec
+  private def getHairCut(n: Int): Unit =
+    val sited = lockObj.synchronized {
+      if notifyBarber then
+        false
+      else if notifyCustomer then
+        false
+      else
+        log(f"customer ${n} sits on the chair and waits for the barber")
+        notifyBarber = true
+        true
+    }
+    if sited then
+      while !lockObj.synchronized { notifyCustomer } do {}
+      lockObj.synchronized {
+        log(f"customer ${n} leaves the chair")
+        notifyCustomer = false
+      }
+    else
+      getHairCut(n)
+
+  override def barber() =
+    for _ <- 1 to customersNumber do
+      log("barber waits for a customer")
+
+      while !lockObj.synchronized { notifyBarber } do {}
+      hairCut()
+      lockObj.synchronized {
+        notifyCustomer = true
+        notifyBarber = false
+      }
+
+    log("barber stops working")

src/main/scala/midterm23/BarberShopStudentAnswer333.scala

+package midterm23
+
+import java.util.concurrent.atomic.AtomicInteger
+
+/** This is an example `BarberShopStudentAnswer`, like the one we made for your
+  * own solution and that you can find on Moodle. The solution below is boringly
+  * correct as it use the same implementation as the expected solution. See
+  * BarberShopSolution.scala for explanations and comments.
+  *
+  * @param customersNumber
+  *   The number of customers to simulate.
+  */
+class BarberShopStudentAnswer333(customersNumber: Int)
+    extends AbstractBarberShopSolution:
+
+  /** This attribute is used to count the number of hair cuts. It is needed for
+    * testing because we do not want to loop forever in the `barber` method!
+    */
+  val hairCutsCounter = AtomicInteger(customersNumber)
+
+  override def customer(n: Int) =
+    lockObj.synchronized {
+      notifyBarber = true
+      while !notifyCustomer do {}
+      notifyCustomer = false
+    }
+
+  override def barber() =
+    // while true
+
+    // We replace the infinite loop from the instructions with a loop doing a
+    // limited number of iterations so that we can test the code.
+    while hairCutsCounter.get() != 0 do
+      while !notifyBarber do {}
+      notifyBarber = false
+      hairCut()
+      hairCutsCounter.decrementAndGet() // You can ignore this line. It has been added for testing.
+      notifyCustomer = true
+
+/** Version of `BarberShopStudentAnswer333` with the methods `customer`
+  * overridden to use the expected implementation.
+  *
+  * @param n
+  *   The number of customers to simulate.
+  */
+class BarberShopStudentAnswer333WithCorrectCustomer(n: Int)
+    extends BarberShopStudentAnswer333(n)
+    with BarberShopCustomerSolution1
+
+/** Version of `BarberShopStudentAnswer333` with the methods `barber` overridden
+  * to use the expected implementation.
+  *
+  * @param n
+  *   The number of customers to simulate.
+  */
+class BarberShopStudentAnswer333WithCorrectBarber(n: Int)
+    extends BarberShopStudentAnswer333(n)
+    with BarberShopBarberSolution1(n)
+
+// To grade, we ran tests with each of these classes.

src/main/scala/midterm23/convolution.worksheet.sc

+import lecture1.parallel
+
+def numContributeTo(input: Array[Int], kernel: Array[Int], output: Array[Int], i: Int): Int =
+  if i < 0 || i >= output.length then
+    0
+  else if (i - kernel.length < 0) then
+    i+1
+  else if (i >= input.length) then
+    output.length - i
+  else kernel.length
+
+def sequentialConvolve(
+  input: Array[Int],
+  kernel: Array[Int],
+  output: Array[Int],
+  from: Int,
+  until: Int
+  ): Unit = {
+    // Approach A, as described in the clarification announcement for this
+    // exercise, where we treat `from` and `until` as indices on `output`
+    // instead of `input` as in the given code.
+    var iOutput = from
+    while iOutput < until do
+      var iKernel = math.max(0, iOutput - input.length + 1)
+      while iKernel < kernel.length && iKernel <= iOutput do
+        // `output` is only ever written to between the indices `from` and
+        // `until`, the range of `iOutput`. The indices for `input` and
+        // `kernel` are computed accordingly.
+        output(iOutput) += input(iOutput - iKernel) * kernel(iKernel)
+        iKernel += 1
+      iOutput += 1
+
+    // ALTERNATE SOLUTION: Approach B, as described in the clarification
+    // announcement for this exercise, which is unchanged from the given
+    // code, i.e. we treat `from` and `until` as indices on `input`.
+    var iInput = from
+    while iInput < until do
+      var iKernel = 0
+      while iKernel < kernel.length do
+        output(iInput + iKernel) += input(iInput) * kernel(iKernel)
+        iKernel += 1
+      iInput += 1
+  }
+
+def parallelConvolve(
+    input: Array[Int],
+    kernel: Array[Int],
+    output: Array[Int],
+    from: Int,
+    until: Int,
+    threshold: Int
+): Unit =
+  // Approach A, as described in the clarification announcement for this
+  // exercise, where we treat `from` and `until` as indices on `output`
+  // instead of `input` as in the given code. This does not require us to
+  // change anything in this function.  Only receives full credit if used
+  // together with Approach A for `sequentialConvolve`.
+  if (until - from) <= threshold then
+    sequentialConvolve(input, kernel, output, from, until)
+  else
+    val mid = from + (until - from) / 2
+    parallel(
+      parallelConvolve(input, kernel, output, from, mid, threshold),
+      parallelConvolve(input, kernel, output, mid, until, threshold)
+    )
+
+  // ALTERNATE SOLUTION: Approach B, as described in the clarification
+  // announcement for this exercise, where we treat `from` and `until` as
+  // indices on `input` as in the given code. This requires up to leave a
+  // gap in the parallel calls to be filled in sequentially as described
+  // below. Only receives full credit if used together with Approach B for
+  // `sequentialConvolve`.
+  if (until - from) <= threshold then
+    sequentialConvolve(input, kernel, output, from, until)
+  else
+    val mid = from + (until - from) / 2
+    val gap = numContributeTo(input, kernel, output, mid)
+    // Leave a gap large enough that accesses to `output` from the first
+    // parallel call will not overlap with accesses from the second. This
+    // gap is of size equal to the number of elements of `input` that will
+    // contribute to the computation of the result at the lowest index of
+    // `output` in the second parallel call, namely, at index `mid`.
+    // However, we are careful not to access indices lower than `from`.
+    val gapBeforeMid = Math.max(from, mid - gap + 1)
+    parallel(
+      parallelConvolve(input, kernel, output, from, gapBeforeMid, threshold),
+      parallelConvolve(input, kernel, output, mid, until, threshold)
+    )
+    // Fill in the gap sequentially.
+    sequentialConvolve(input, kernel, output, gapBeforeMid, mid)
+
+object Original_WithOutputRace:
+
+  def sequentialConvolve(
+      input: Array[Int],
+      kernel: Array[Int],
+      output: Array[Int],
+      from: Int,
+      until: Int): Unit =
+    var iInput = from
+    while iInput < until do
+      var iKernel = 0
+      while iKernel < kernel.length do
+        output(iInput + iKernel) += input(iInput) * kernel(iKernel)
+        iKernel += 1
+      iInput += 1
+
+  def parallelConvolve(
+      input: Array[Int],
+      kernel: Array[Int],
+      output: Array[Int],
+      from: Int,
+      until: Int,
+      threshold: Int): Unit =
+    if (until - from) <= threshold then
+      sequentialConvolve(input, kernel, output, from, until)
+    else
+      val mid = from + (until - from)/2
+      parallel(
+        parallelConvolve(input, kernel, output, from, mid, threshold),
+        parallelConvolve(input, kernel, output, mid, until, threshold))

src/main/scala/midterm23/groupBy.worksheet.sc

+import lecture1.parallel
+import collection.concurrent.TrieMap
+import collection.parallel.CollectionConverters.*
+
+// Example run of `groupBy`:
+case class Course(semester: String, id: String)
+val cs210 = Course("BA3", "CS210")
+val cs250 = Course("BA3", "CS250")
+val cs206 = Course("BA4", "CS206")
+val courses = Vector(cs210, cs250, cs206)
+assertEquals(
+  courses.groupBy(_.semester),
+  Map(
+    "BA3" -> Vector(cs210, cs250),
+    "BA4" -> Vector(cs206)
+  )
+)
+
+/** Asymptotic run time: O(1) */
+def appendAtKey[K, V](m: Map[K, Vector[V]], k: K, v: V): Map[K, Vector[V]] =
+  val prev = m.getOrElse(k, Vector.empty)
+  m.updated(k, prev :+ v)
+
+// Example runs of `appendAtKey`:
+assertEquals(
+  appendAtKey(Map(), 0, 1),
+  Map(0 -> Vector(1))
+)
+assertEquals(
+  appendAtKey(Map("BA3" -> Vector(cs210)), "BA3", cs250),
+  Map("BA3" -> List(cs210, cs250))
+)
+
+/** Asymptotic run time: O(1) */
+def mergeMaps[K, V](a: Map[K, Vector[V]], b: Map[K, Vector[V]]) =
+  var res = a
+  for (k, vs) <- b do // O(1) iterations (we assume the number of keys to be O(1))
+    val prev = res.getOrElse(k, Vector.empty) // O(1) lookup
+    res = res.updated(k, prev ++ vs) // O(1) update, O(1) concatenation
+  res
+
+// Example run of `mergeMaps`:
+val m1 = Map(1 -> Vector(5, 7))
+val m2 = Map(1 -> Vector(6), 2 -> Vector(3))
+assertEquals(
+  mergeMaps(m1, m2),
+  Map(
+    1 -> Vector(5, 7, 6),
+    2 -> Vector(3)
+  )
+)
+
+def parGroupBy1[K, V](vs: Vector[V], key: V => K): Map[K, Vector[V]] =
+  if vs.size == 1 then
+    Map(key(vs.head) -> Vector(vs.head))
+  else
+    val mid = vs.size / 2
+    val (m1, m2) = parallel(
+      parGroupBy1(vs.slice(0, mid), key),
+      parGroupBy1(vs.slice(mid, vs.size), key)
+    )
+    mergeMaps(m1, m2)
+
+assertEquals(
+  parGroupBy1(courses, c => c.semester),
+  Map(
+    "BA3" -> List(cs210, cs250),
+    "BA4" -> List(cs206)
+  )
+)
+
+assertEquals(
+  parGroupBy1(courses, c => c.semester),
+  Map(
+    "BA3" -> List(cs210, cs250),
+    "BA4" -> List(cs206)
+  )
+)
+
+def parGroupBy2[K, V](vs: Vector[V], key: V => K): Map[K, Vector[V]] =
+  if vs.size == 1 then
+    Map(key(vs.head) -> Vector(vs.head))
+  else
+    val (m1, m2) = parallel(
+      parGroupBy2(Vector(vs.head), key),
+      parGroupBy2(vs.tail, key)
+    )
+    mergeMaps(m1, m2)
+
+assertEquals(
+  parGroupBy2(courses, c => c.semester),
+  Map(
+    "BA3" -> List(cs210, cs250),
+    "BA4" -> List(cs206)
+  )
+)
+
+assertEquals(
+  parGroupBy2(courses, c => c.semester),
+  Map(
+    "BA3" -> List(cs210, cs250),
+    "BA4" -> List(cs206)
+  )
+)
+
+def parGroupBy3[K, V](vs: Vector[V], key: V => K): Map[K, Vector[V]] =
+  val lock = Object()
+  var res = Map[K, Vector[V]]()
+  for v <- vs.par do
+    lock.synchronized { res = appendAtKey(res, key(v), v) }
+  res
+
+var values3 = Set[Map[String, Vector[Course]]]()
+for _ <- 1 to 100000 do
+  values3 += parGroupBy3(courses, _.semester)
+assertEquals(values3.size, 2)
+values3
+
+def parGroupBy4[K, V](vs: Vector[V], key: V => K): Map[K, Vector[V]] =
+  val res = TrieMap[K, Vector[V]]()
+  for v <- vs.par do
+    val k = key(v)
+    val prev = res.getOrElse(k, Vector.empty)
+    res.update(k, prev :+ v)
+  res.toMap
+
+var values4 = Set[Map[String, Vector[Course]]]()
+for _ <- 1 to 100000 do
+  values4 += parGroupBy4(courses, _.semester)
+assertEquals(values4.size, 4)
+values4
+
+def parGroupBySolution[K, V](vs: Vector[V], key: V => K): Map[K, Vector[V]] =
+  vs.par.aggregate(Map.empty)(
+    (m, v) => appendAtKey(m, key(v), v),
+    mergeMaps
+  )
+
+var valuesSol = Set[Map[String, Vector[Course]]]()
+for _ <- 1 to 100000 do
+  valuesSol += parGroupBySolution(courses, _.semester)
+assertEquals(valuesSol.size, 1)
+valuesSol
+
+def assertEquals(a: Any, b: Any) = assert(a == b)

src/test/scala/concpar21final01/Problem1Suite.scala

+package concpar21final01
+
+import scala.concurrent.duration.*
+import scala.concurrent.Future
+import scala.concurrent.ExecutionContext.Implicits.global
+
+class Problem1Suite extends munit.FunSuite:
+  test(
+    "Retrieves grades at the end of the exam (everyone pushed something) (10pts)"
+  ) {
+    class Problem1Done extends Problem1:
+      override def getGrade(sciper: Int): Future[Option[Grade]] =
+        Future {
+          Thread.sleep(100)
+          Some(Grade(sciper, sciper))
+        }
+      override def getScipers(): Future[List[Int]] =
+        Future {
+          Thread.sleep(100)
+          List(1, 2, 3, 4)
+        }
+
+    val expected: List[Grade] =
+      List(Grade(1, 1.0), Grade(2, 2.0), Grade(3, 3.0), Grade(4, 4.0))
+
+    (new Problem1Done).leaderboard().map { grades =>
+      assertEquals(grades.toSet, expected.toSet)
+    }
+  }
+
+  test("Retrieves grades mid exam (some students didn't push yet) (10pts)") {
+    class Problem1Partial extends Problem1:
+      override def getGrade(sciper: Int): Future[Option[Grade]] =
+        Future {
+          Thread.sleep(100)
+          if sciper % 2 == 0 then None
+          else Some(Grade(sciper, sciper))
+        }
+      override def getScipers(): Future[List[Int]] =
+        Future {
+          Thread.sleep(100)
+          List(1, 2, 3, 4)
+        }
+
+    val expected: List[Grade] =
+      List(Grade(1, 1.0), Grade(3, 3.0))
+
+    (new Problem1Partial).leaderboard().map { grades =>
+      assertEquals(grades.toSet, expected.toSet)
+    }
+  }
+
+  test("The output list is sorted by grade (10pts)") {
+    (new Problem1MockData).leaderboard().map { grades =>
+      assert(grades.size >= 176)
+      assert(grades.zipWithIndex.forall { case (g, i) =>
+        grades.drop(i).forall(x => g.grade >= x.grade)
+      })
+    }
+  }
+
+  test("GitLab API calls are done in parallel (2pts)") {
+    var inParallel: Boolean = false
+
+    class Problem1Par extends Problem1MockData:
+      var in: Boolean = false
+
+      override def getGrade(sciper: Int): Future[Option[Grade]] =
+        Future {
+          if in then inParallel = true
+          in = true
+          val out = super.getGrade(sciper)
+          in = false
+          concurrent.Await.result(out, Duration(10, SECONDS))
+        }
+
+    (new Problem1Par).leaderboard().map { grades =>
+      assert(grades.size >= 176)
+      assert(inParallel)
+    }
+  }
+
+  test("The IS-academia API is called exactly once (2pts)") {
+    var called: Int = 0
+
+    class Problem1Once extends Problem1MockData:
+      override def getScipers(): Future[List[Int]] =
+        called += 1
+        super.getScipers()
+
+    (new Problem1Once).leaderboard().map { grades =>
+      assert(grades.size >= 176)
+      assert(called == 1)
+    }
+  }

src/test/scala/concpar21final02/Problem2Suite.scala

+package concpar21final02
+
+import akka.actor.*
+import akka.testkit.*
+import scala.collection.mutable
+import concurrent.duration.*
+
+import Problem2.*
+
+class Problem2Suite extends munit.FunSuite:
+  import NotificationService.Protocol.*
+  import NotificationService.Responses.*
+  import DiscordChannel.Protocol.*
+  import DiscordChannel.Responses.*
+
+  test("Notification register (1pts)") {
+    new MyTestKit:
+      def tests() =
+        val actor = system.actorOf(Props[NotificationService]())
+        actor ! Register
+        expectMsg(Registered(true))
+  }
+
+  test("Notification register and un-register (1pts)") {
+    new MyTestKit:
+      def tests() =
+        val actor = system.actorOf(Props[NotificationService]())
+        actor ! Register
+        expectMsg(Registered(true))
+        actor ! UnRegister
+        expectMsg(Registered(false))
+        actor ! UnRegister
+        expectMsg(Registered(false))
+        actor ! Register
+        expectMsg(Registered(true))
+        actor ! UnRegister
+        expectMsg(Registered(false))
+  }
+
+  test("Notification notify (1pts)") {
+    new MyTestKit:
+      def tests() =
+        val actor = system.actorOf(Props[NotificationService]())
+        actor ! Register
+        expectMsg(Registered(true))
+        actor ! NotifyAll
+        expectMsg(Notification)
+        actor ! NotifyAll
+        expectMsg(Notification)
+        actor ! UnRegister
+        expectMsg(Registered(false))
+        actor ! NotifyAll
+        expectNoMessage()
+        actor ! Register
+        expectMsg(Registered(true))
+        actor ! NotifyAll
+        expectMsg(Notification)
+        actor ! UnRegister
+        expectMsg(Registered(false))
+        actor ! NotifyAll
+        expectNoMessage()
+  }
+
+  test("NotifyAll from other actor (1pts)") {
+    new MyTestKit:
+      def tests() =
+        val actor = system.actorOf(Props[NotificationService]())
+        val otherActor = system.actorOf(Props[DummyActor]())
+
+        def notifyFormAllFromOtherActor() =
+          given ActorRef = otherActor
+          actor ! NotifyAll
+
+        expectNoMessage()
+
+        actor ! Register
+        expectMsg(Registered(true))
+
+        notifyFormAllFromOtherActor()
+        expectMsg(Notification)
+  }
+
+  test("Channel init (1pts)") {
+    new MyTestKit:
+      def tests() =
+        val notificationService = system.actorOf(Props[NotificationService]())
+        val channel = system.actorOf(Props[DiscordChannel]())
+        channel ! Init(notificationService)
+        expectMsg(Active)
+  }
+
+  test("Channel post and get post (1pts)") {
+    new MyTestKit:
+      def tests() =
+        val notificationService = system.actorOf(Props[NotificationService]())
+        val channel = system.actorOf(Props[DiscordChannel]())
+        channel ! Init(notificationService)
+        expectMsg(Active)
+        channel ! Post("hello")
+        channel ! GetLastPosts(1)
+        expectMsg(Posts(List("hello")))
+        channel ! GetLastPosts(10)
+        expectMsg(Posts(List("hello")))
+        channel ! GetLastPosts(0)
+        expectMsg(Posts(Nil))
+  }
+
+  test("Channel multiple posts (1pts)") {
+    new MyTestKit:
+      def tests() =
+        val notificationService = system.actorOf(Props[NotificationService]())
+        val channel = system.actorOf(Props[DiscordChannel]())
+        channel ! Init(notificationService)
+        expectMsg(Active)
+        channel ! Post("hello")
+        channel ! Post("world")
+        channel ! GetLastPosts(2)
+        channel ! GetLastPosts(1)
+        channel ! Post("!")
+        channel ! GetLastPosts(3)
+        expectMsg(Posts(List("world", "hello")))
+        expectMsg(Posts(List("world")))
+        expectMsg(Posts(List("!", "world", "hello")))
+  }
+
+  test("Channel posts and notify (1pts)") {
+    new MyTestKit:
+      def tests() =
+        val notificationService = system.actorOf(Props[NotificationService]())
+        val channel = system.actorOf(Props[DiscordChannel]())
+        channel ! Init(notificationService)
+        expectMsg(Active)
+        notificationService ! Register
+        expectMsg(Registered(true))
+        channel ! Post("hello")
+        channel ! Post("world")
+        expectMsg(Notification)
+        expectMsg(Notification)
+  }
+
+  test("Channel init twice (1pts)") {
+    new MyTestKit:
+      def tests() =
+        val notificationService = system.actorOf(Props[NotificationService]())
+        val channel = system.actorOf(Props[DiscordChannel]())
+        channel ! Init(notificationService)
+        expectMsg(Active)
+        channel ! Init(notificationService)
+        expectMsg(AlreadyActive)
+        channel ! Init(notificationService)
+        expectMsg(AlreadyActive)
+  }
+
+  test("Channel not active (1pts)") {
+    new MyTestKit:
+      def tests() =
+        val channel1 = system.actorOf(Props[DiscordChannel]())
+        channel1 ! Post("hello")
+        expectMsg(NotActive)
+
+        val channel2 = system.actorOf(Props[DiscordChannel]())
+        channel2 ! GetLastPosts(0)
+        expectMsg(NotActive)
+  }
+
+  abstract class MyTestKit
+      extends TestKit(ActorSystem("TestSystem"))
+      with ImplicitSender:
+    def tests(): Unit
+    try tests()
+    finally shutdown(system)
+
+class DummyActor extends Actor:
+  def receive: Receive = { case _ =>
+    ()
+  }

src/test/scala/concpar21final03/Problem3Suite.scala

+package concpar21final03
+
+import scala.annotation.tailrec
+import scala.concurrent.*
+import scala.concurrent.duration.*
+import scala.collection.mutable.HashMap
+import scala.util.Random
+import instrumentation.*
+import instrumentation.TestHelper.*
+import instrumentation.TestUtils.*
+
+class Problem3Suite extends munit.FunSuite:
+
+  test("Part 1: ThreadMap (3pts)") {
+    testManySchedules(
+      4,
+      sched =>
+        val tmap = new SchedulableThreadMap[Int](sched)
+
+        def writeThread(): Unit =
+          tmap.setCurrentThreadValue(0)
+          tmap.setCurrentThreadValue(-1)
+          val readBack = tmap.currentThreadValue
+          assertEquals(readBack, Some(-1))
+
+        def writeAndDeleteThread(): Unit =
+          tmap.setCurrentThreadValue(42)
+          tmap.deleteCurrentThreadValue()
+
+        @tailrec
+        def waitThread(): Unit =
+          tmap.waitForall(_ < 0)
+          val all = tmap.allValues
+          if all != List(-1) then waitThread()
+
+        val threads = List(
+          () => writeThread(),
+          () => writeAndDeleteThread(),
+          () => waitThread(),
+          () => waitThread()
+        )
+
+        (threads, _ => (true, ""))
+    )
+  }
+
+  test("Part 2: RCU (5pts)") {
+    testManySchedules(
+      3,
+      sched =>
+        val rcu = new SchedulableRCU(sched)
+
+        case class State(
+            value: Int,
+            isDeleted: AtomicLong = SchedulableAtomicLong(0, sched, "isDeleted")
+        )
+
+        val sharedState =
+          SchedulableAtomicReference(State(0), sched, "sharedState")
+
+        def readThread(): Unit =
+          rcu.startRead()
+          val state = sharedState.get
+          val stateWasDeleted = state.isDeleted.get != 0
+          assert(
+            !stateWasDeleted,
+            "RCU shared state deleted in the middle of a read."
+          )
+          rcu.stopRead()
+
+        def writeThread(): Unit =
+          val oldState = sharedState.get
+          sharedState.set(State(oldState.value + 1))
+          rcu.waitForOldReads()
+          oldState.isDeleted.set(1)
+
+        val threads = List(
+          () => readThread(),
+          () => readThread(),
+          () => writeThread()
+        )
+
+        (threads, _ => (true, ""))
+    )
+  }
+
+  test("Part 3: UpdateServer (2pts)") {
+    testManySchedules(
+      3,
+      sched =>
+        val fs = SchedulableInMemoryFileSystem(sched)
+        val server = new SchedulableUpdateServer(sched, fs)
+
+        def writeThread(): Unit =
+          server.newUpdate("update1.bin", "Update 1")
+          server.newUpdate("update2.bin", "Update 2")
+          assertEquals(fs.fsMap.toSet, Set("update2.bin" -> "Update 2"))
+
+        def fetchThread(): Unit =
+          val res = server.fetchUpdate()
+          assert(
+            List(None, Some("Update 1"), Some("Update 2")).contains(res),
+            s"fetchUpdate returned unexpected value $res"
+          )
+
+        val threads = List(
+          () => writeThread(),
+          () => fetchThread(),
+          () => fetchThread()
+        )
+
+        (threads, _ => (true, ""))
+    )
+  }
+
+  import scala.concurrent.duration.*
+  override val munitTimeout = 200.seconds
+end Problem3Suite

src/test/scala/concpar21final03/instrumentation/AtomicReference.scala

+package concpar21final03.instrumentation
+
+class AtomicReference[T](initial: T):
+
+  private val atomic =
+    new java.util.concurrent.atomic.AtomicReference[T](initial)
+
+  def get: T = atomic.get()
+
+  def set(value: T): Unit = atomic.set(value)

src/test/scala/concpar21final03/instrumentation/MockedMonitor.scala

+package concpar21final03.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/test/scala/concpar21final03/instrumentation/Scheduler.scala

+package concpar21final03.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 then
+                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 then syncObject.wait(timeout) } {
+        time => remTime -= time
+      }
+    }
+    if exception.isDefined then exception.get
+    else if remTime <= 1
+    then // 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 then 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 then 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 do
+        threadStates(k) = f(threadStates(k))
+    }
+
+  def log(str: String) =
+    if (realToFakeThreadId contains Thread.currentThread().getId()) then
+      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()) then primop
+    else
+      updateThreadState(VariableReadWrite(threadLocks))
+      val m = msg
+      if m != "" then log(m)
+      if opLog.size > maxOps then
+        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
+    then // 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
+        }
+      then
+        val waiting = threadStates.keys.map(_.toString).mkString(", ")
+        val s = if threadStates.size > 1 then "s" else ""
+        val are = if threadStates.size > 1 then "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 then
+          val waiting = threadStates.keys.map(_.toString).mkString(", ")
+          val s = if threadStates.size > 1 then "s" else ""
+          val are = if threadStates.size > 1 then "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
+        then // 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 then
+            // 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 then
+        canContinue = decide()
+        notifyAll()
+      // waitingForDecision(threadId) = Some(numThreadsWaiting)
+      // println(s"$threadId Entering waiting with ticket number $numThreadsWaiting/${waitingForDecision.size}")
+      while !canProceed() do 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 then
+      canContinue = decide()
+      notifyAll()
+  }
+
+  def getOperationLog() = opLog

src/test/scala/concpar21final03/instrumentation/Stats.scala

+/* Copyright 2009-2015 EPFL, Lausanne */
+package concpar21final03.instrumentation
+
+import java.lang.management.*
+
+/** A collection of methods that can be used to collect run-time statistics */
+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))