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,31 @@
+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
+        x
+      }
+    }
+
+    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,72 @@
+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 = empty
+
+    def nonEmpty = !empty
+
+  }
+
+  import org.learningconcurrency.ch2.thread
+
+  val lock = new AnyRef
+
+  val syncVar = new SyncVar[Int]
+
+  val producer = thread {
+    var x = 0
+    while(x < 15) {
+
+      lock.synchronized {
+        if (syncVar.isEmpty) {
+          syncVar.put(x)
+          x = x + 1
+        }
+        lock.wait
+      }
+
+    }
+  }
+
+  val consumer = thread {
+    var x = -1
+    while(x < 14) {
+      lock.synchronized {
+        if (syncVar.nonEmpty) {
+          x = syncVar.get
+          log(s"get = $x")
+        }
+        lock.notify
+      }
+    }
+  }
+
+  producer.join
+  consumer.join
+
+}

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

@@ -0,0 +1,61 @@
+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 = empty
+
+    def nonEmpty = !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,56 @@
+package org.learningconcurrency
+package exercises
+package ch2
+
+import scala.collection.mutable._
+
+
+object Ex6 extends App {
+
+  class SyncQueue[T](val n:Int) {
+
+    private var qX = Queue[T]()
+
+    def getWait():T = this.synchronized {
+      while (qX.isEmpty) {
+        this.notify
+        this.wait
+      }
+
+      qX.dequeue
+    }
+
+    def putWait(x: T): Unit = this.synchronized {
+      while (qX.length == n)
+        this.wait()
+
+      qX += 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,46 @@
+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
+    }
+
+    a1.synchronized {
+      a2.synchronized {
+        adjust()
+      }
+    }
+
+  }
+
+  def sendAll(accounts: Set[Account], target: Account):Unit = {
+    accounts.map((a) => thread {
+      send(a,target,a.money)
+    } ).foreach(_.join)
+
+    //accounts.foreach((a) => send(a,target,a.money))
+  }
+
+  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}")
+
+
+}

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

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