Answer to exercise 3-10 ch.2

src/main/scala/org/learningconcurrency/exercises/ch2/Ex10.scala

@@ -0,0 +1,62 @@
+package org.learningconcurrency
+package exercises
+package ch2
+
+import scala.collection.mutable
+
+object Ex10 extends App {
+
+  class PriorityTaskPool(val p:Int, val important: Int) {
+
+    implicit val ord: Ordering[(Int,() => Unit)] = Ordering.by(_._1)
+
+    private val tasks = mutable.PriorityQueue[(Int,() => Unit)]()
+
+    @volatile
+    private var terminated = false
+
+    def asynchronous(priority: Int)(task: => Unit):Unit = tasks synchronized {
+      tasks.enqueue((priority,() => task))
+      tasks.notify()
+    }
+
+    class Worker extends Thread {
+
+      def poll() = tasks.synchronized {
+        while (tasks.isEmpty) {
+          tasks.wait()
+        }
+        log(s"queue: " + tasks.foldLeft("")((s,t)=>s"$s${t._1},"))
+        tasks.dequeue()
+      }
+
+      override def run() = {
+        while (true) {
+          poll() match {
+            case (p, task) if (p > important) || (!terminated) => task()
+            case _  =>
+          }
+        }
+      }
+    }
+
+    def shutdown() = tasks.synchronized {
+      terminated = true
+      tasks.notify()
+    }
+
+    (1 to p).map((i) => new Worker()).map(_.start)
+
+  }
+
+  val tasks = new PriorityTaskPool(10, 300)
+
+  (1 to 1000).foreach((i) => {
+    val a = (Math.random*1000).toInt
+    tasks.asynchronous(a)({log(s"<- $a")})
+  })
+
+  Thread.sleep(1)
+  tasks.shutdown()
+
+}

src/main/scala/org/learningconcurrency/exercises/ch2/ex3.scala

@@ -0,0 +1,33 @@
+package org.learningconcurrency
+package exercises
+package ch2
+
+object Ex3 extends App {
+
+  class SyncVar[T] {
+
+    private var empty:Boolean = true
+
+    private var x:T = null.asInstanceOf[T]
+
+    def get(): T = this.synchronized {
+      if (empty) throw new Exception("must be non-empty")
+      else {
+        empty = true
+        val v = x
+        x = null.asInstanceOf[T]
+        v
+      }
+    }
+
+    def put(x: T):Unit = this.synchronized {
+      if (!empty) throw new Exception("must be empty")
+      else {
+        empty = false
+        this.x = x
+      }
+    }
+
+  }
+
+}

src/main/scala/org/learningconcurrency/exercises/ch2/ex4.scala

@@ -0,0 +1,69 @@
+package org.learningconcurrency
+package exercises
+package ch2
+
+object Ex4 extends App {
+
+  class SyncVar[T] {
+
+    private var empty: Boolean = true
+
+    private var x: T = null.asInstanceOf[T]
+
+    def get(): T = this.synchronized {
+      if (empty) throw new Exception("must be non-empty")
+      else {
+        empty = true
+        x
+      }
+    }
+
+    def put(x: T): Unit = this.synchronized {
+      if (!empty) throw new Exception("must be empty")
+      else {
+        empty = false
+        this.x = x
+      }
+    }
+
+    def isEmpty = synchronized {
+      empty
+    }
+
+    def nonEmpty = synchronized {
+      !empty
+    }
+
+  }
+
+  import org.learningconcurrency.ch2.thread
+
+  val syncVar = new SyncVar[Int]
+
+  val producer = thread {
+    var x = 0
+    while (x < 15) {
+      if (syncVar.isEmpty) {
+        syncVar.put(x)
+        x = x + 1
+      }
+
+    }
+  }
+
+  val consumer = thread {
+    var x = 0
+    while (x != 15) {
+      if (syncVar.nonEmpty) {
+        log(s"get = ${syncVar.get}")
+        x = x + 1
+      }
+    }
+  }
+
+  producer.join()
+  consumer.join()
+
+
+
+}

src/main/scala/org/learningconcurrency/exercises/ch2/ex5.scala

@@ -0,0 +1,65 @@
+package org.learningconcurrency
+package exercises
+package ch2
+
+object Ex5 extends App {
+
+  class SyncVar[T] {
+
+    private var empty: Boolean = true
+
+    private var x: T = null.asInstanceOf[T]
+
+    def isEmpty = synchronized {
+      empty
+    }
+
+    def nonEmpty = synchronized {
+      !empty
+    }
+
+    def getWait():T = this.synchronized {
+      while (empty)
+        this.wait
+
+      empty = true
+      this.notify
+      x
+    }
+
+    def putWait(x: T): Unit = this.synchronized {
+      while (!empty)
+        this.wait()
+
+      empty = false
+      this.x = x
+      this.notify()
+    }
+
+
+  }
+
+  import org.learningconcurrency.ch2.thread
+
+  val syncVar = new SyncVar[Int]
+
+  val producer = thread {
+    var x = 0
+    while(x < 15) {
+      syncVar.putWait(x)
+      x = x + 1
+    }
+  }
+
+  val consumer = thread {
+    var x = -1
+    while(x < 14) {
+      x = syncVar.getWait
+      log(s"get: $x")
+    }
+  }
+
+  producer.join()
+  consumer.join()
+
+}

src/main/scala/org/learningconcurrency/exercises/ch2/ex6.scala

@@ -0,0 +1,58 @@
+package org.learningconcurrency
+package exercises
+package ch2
+
+import scala.collection.mutable._
+
+
+object Ex6 extends App {
+
+  class SyncQueue[T](val n:Int) {
+
+    private var syncQueue = Queue[T]()
+
+    def getWait():T = this.synchronized {
+      while (syncQueue.isEmpty) {
+        this.notify
+        this.wait
+      }
+
+      val x = syncQueue.dequeue
+      this.notify()
+      x
+    }
+
+    def putWait(x: T): Unit = this.synchronized {
+      while (syncQueue.length == n)
+        this.wait()
+
+      syncQueue += x
+      this.notify()
+    }
+
+  }
+
+  import org.learningconcurrency.ch2.thread
+
+  val syncVar = new SyncQueue[Int](10)
+
+  val producer = thread {
+    var x = 0
+    while(x < 15) {
+      syncVar.putWait(x)
+      x = x + 1
+    }
+  }
+
+  val consumer = thread {
+    var x = -1
+    while(x < 14) {
+      x = syncVar.getWait
+      log(s"get: $x")
+    }
+  }
+
+  producer.join()
+  consumer.join()
+
+}

src/main/scala/org/learningconcurrency/exercises/ch2/ex7.scala

@@ -0,0 +1,66 @@
+package org.learningconcurrency
+package exercises
+package ch2
+
+import org.learningconcurrency.ch2._
+
+object Ex7 extends App {
+
+  import SynchronizedProtectedUid._
+
+  class Account(val name: String, var money: Int) {
+    val uid = getUniqueId()
+  }
+
+  def send(a1: Account, a2: Account, n: Int) {
+    def adjust() {
+      a1.money -= n
+      a2.money += n
+    }
+
+    if (a1.uid < a2.uid) {
+      a1.synchronized {
+        a2.synchronized {
+          adjust()
+        }
+      }
+    } else {
+      a2.synchronized {
+        a1.synchronized {
+          adjust()
+        }
+      }
+    }
+  }
+
+
+  def sendAll(accounts: Set[Account], target: Account): Unit = {
+
+    def adjust() = {
+      target.money = accounts.foldLeft(0)((s, a) => {
+        val money = a.money
+        a.money = 0
+        s + money
+      }
+      )
+    }
+
+    def sendAllWithSynchronize(la: List[Account]): Unit = la match {
+      case h :: t => h synchronized {
+        sendAllWithSynchronize(t)
+      }
+      case _ => adjust()
+    }
+
+    sendAllWithSynchronize((target :: accounts.toList).sortBy(_.uid))
+  }
+
+  val accounts = (1 to 100).map((i) => new Account(s"Account: $i",i*10)).toSet
+  val target = new Account("Target account", 0)
+
+  sendAll(accounts,target)
+
+  accounts.foreach((a) => log(s"${a.name}, money = ${a.money}"))
+  log(s"${target.name} - money = ${target.money}")
+
+}