add answers to exercise 1-5 chapter 7

src/main/scala/org/learningconcurrency/exercises/ch7/ex1.scala

@@ -0,0 +1,72 @@
+package org.learningconcurrency
+package exercises
+package ch7
+
+/**
+  * Implement the transactional pair abstraction, represented with the TPair class:
+  *
+  *   class TPair[P, Q](pinit: P, qinit: Q) {
+  *     def first(implicit txn: InTxn): P = ???
+  *     def first_=(x: P)(implicit txn: InTxn): P = ???
+  *     def second(implicit txn: InTxn): Q = ???
+  *     def second_=(x: Q)(implicit txn: InTxn): Q = ???
+  *     def swap()(implicit e: P =:= Q, txn: InTxn): Unit = ???
+  *   }
+  *
+  * In addition to getters and setters for the two fields,
+  * the transactional pair defines the swap method that swaps the  fields,
+  * and can only be called if the types P and Q are the same.
+  */
+object Ex1 extends App {
+
+  import scala.concurrent._
+  import ExecutionContext.Implicits.global
+  import scala.concurrent.stm._
+
+  class TPair[P, Q](pinit: P, qinit: Q) {
+
+    private val rFirst = Ref[P](pinit)
+    private val rSecond = Ref[Q](qinit)
+
+
+    def first(implicit txn: InTxn): P = rFirst.single()
+
+    def first_=(x: P)(implicit txn: InTxn) = rFirst.single.transform(old => x)
+
+    def second(implicit txn: InTxn): Q = rSecond.single()
+
+    def second_=(x: Q)(implicit txn: InTxn) = rSecond.single.transform(old => x)
+
+    def swap()(implicit e: P =:= Q, txn: InTxn): Unit = {
+      val old = first
+      first = second.asInstanceOf[P]
+      second = e(old)
+    }
+  }
+
+  //test
+  val p = new TPair[String,String]("first value","second value")
+
+  def swapOne = atomic { implicit txn =>
+    p.swap
+
+    val vF = p.first
+    val vS = p.second
+
+    Txn.afterCommit { _ =>
+      assert(vS != vF)
+    }
+  }
+
+  (1 to 1001).map(_ => Future {
+    swapOne
+  })
+
+  Thread.sleep(2000)
+
+  atomic {implicit txn =>
+    log(s"Result: first = '${p.first}' vSecond = '${p.second}'")
+  }
+
+
+}

src/main/scala/org/learningconcurrency/exercises/ch7/ex2.scala

@@ -0,0 +1,118 @@
+package org.learningconcurrency
+package exercises
+package ch7
+
+/**
+  * Use ScalaSTM to implement the mutable location abstraction from Haskell,
+  * represented with the MVar class:
+  *
+  * class MVar[T] {
+  *   def put(x: T)(implicit txn: InTxn): Unit = ???
+  *   def take()(implicit txn: InTxn): T = ???
+  * }
+  *
+  * An MVar object can be either full or empty.
+  * Calling put on a full MVar object blocks until the MVar object becomes empty,
+  * and adds an element.
+  *
+  * Similarly, calling take on an empty MVar object blocks until the MVar object becomes full,
+  * and removes the element.
+  *
+  * Now, implement a method called swap, which takes two MVar objects and swaps their values:
+  *
+  *  def swap[T](a: MVar[T], b: MVar[T])(implicit txn: InTxn) = ???
+  *
+  * Contrast the MVar class with the SyncVar class from Chapter 2,
+  * Concurrency on the JVM and the Java Memory Model.
+  * Is it possible to implement the swap method for SyncVar objects
+  * without modifying the internal implementation of the SyncVar class?
+  *
+  */
+object Ex2 extends App {
+
+  import java.util.concurrent.atomic.AtomicInteger
+  import scala.concurrent._
+  import ExecutionContext.Implicits.global
+  import scala.concurrent.stm._
+
+  class MVar[T] {
+
+    private val rx = Ref[Option[T]](None)
+
+    def put(x: T)(implicit txn: InTxn): Unit = rx() match {
+      case Some(_) => retry
+      case None => rx() = Some(x)
+    }
+
+    def take()(implicit txn: InTxn): T = rx() match {
+      case Some(x) => {
+        rx() = None
+        x
+      }
+      case None => retry
+    }
+  }
+
+  def swap[T](a: MVar[T], b: MVar[T])(implicit txn: InTxn) = {
+    val old = a.take
+    a.put(b.take())
+    b.put(old)
+  }
+
+  //test
+  val mVar = new MVar[Integer]
+
+  val l = 1 to 1001
+
+  l.map(
+    i => Future {
+      atomic {implicit txn =>
+        mVar.put(i)
+      }
+    }
+  )
+
+  val sum = new AtomicInteger(0)
+
+  l.map(
+    i => Future {
+      atomic {implicit txn =>
+        val i = mVar.take
+        Txn.afterCommit(_ => sum.addAndGet(i))
+      }
+    }
+  )
+
+  Thread.sleep(5000)
+
+  if (l.sum != sum.get) log(s"Error !!!! ${l.sum} != $sum")
+
+  log(s"summ = ${sum.get}")
+
+  //test swap
+  log("--- test swap ------------")
+
+  val mva = new MVar[String]
+  val mvb = new MVar[String]
+  atomic {implicit txn =>
+    mva.put("a")
+    mvb.put("b")
+  }
+
+  l.map(i =>
+    Future{
+      atomic {implicit txn =>
+        swap(mva, mvb)
+      }
+    }
+  )
+
+  Thread.sleep(5000)
+
+  atomic {implicit txn =>
+    val a = mva.take
+    val b = mvb.take
+
+    Txn.afterCommit( _ =>  log(s"a= $a, b = $b"))
+  }
+}

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

@@ -0,0 +1,50 @@
+package org.learningconcurrency
+package exercises
+package ch7
+
+/**
+  * Implement the atomicRollbackCount method, which is used to track how many times
+  * a transaction was rolled back before it completed successfully:
+  *
+  *   def atomicRollbackCount[T](block: InTxn => T): (T, Int) = ???
+  */
+object Ex3 extends App {
+
+  import scala.concurrent.ExecutionContext
+  import scala.concurrent.Future
+  import scala.concurrent.stm._
+  import ExecutionContext.Implicits.global
+  import scala.util.Random
+
+  def atomicRollbackCount[T](block: InTxn => T): (T, Int) = {
+    var cnt = 0
+    atomic { implicit txn =>
+      Txn.afterRollback(_ =>
+        cnt += 1
+      )
+      (block(txn), cnt)
+    }
+  }
+
+  //test
+  val r = Ref(10)
+
+  def block(txn: InTxn): Int = {
+    var x: Int = r.get(txn)
+    x = Random.nextInt(10000)
+    Thread.sleep(10)
+    r.set(x)(txn)
+    x
+  }
+
+  (1 to 100).map(i =>
+    Future {
+      atomicRollbackCount[Int](block) match {
+        case (_, cnt) => log(s"Transaction: $i, retries = $cnt")
+        case _ => log("???")
+      }
+    }
+  )
+
+  Thread.sleep(3000)
+}

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

@@ -0,0 +1,60 @@
+package org.learningconcurrency
+package exercises
+package ch7
+
+/**
+  * Implement the atomicWithRetryMax method,
+  * which is used to start a transaction that can be retried at most n times:
+  *
+  * def atomicWithRetryMax[T](n: Int)(block: InTxn => T): T = ???
+  *
+  * Reaching the maximum number of retries throws an exception.
+  */
+object Ex4 extends App {
+
+  import scala.concurrent._
+  import ExecutionContext.Implicits.global
+  import scala.concurrent.stm._
+  import scala.util.Random
+
+  case class ReachMaxNumberException(cntRetries: Int) extends Exception
+
+  def atomicWithRetryMax[T](n: Int)(block: InTxn => T): T = {
+
+    var cntRetries = 0
+
+    atomic{ implicit txn =>
+      Txn.afterRollback(_ => cntRetries += 1)
+
+      if (cntRetries > n) {
+        throw  ReachMaxNumberException(cntRetries)
+      }
+
+      block(txn)
+    }
+  }
+
+  //test
+  val r = Ref(10)
+
+  def block(txn: InTxn): Int = {
+    var x: Int = r.get(txn)
+    Thread.sleep(10)
+    x += Random.nextInt(100)
+    r.set(x)(txn)
+    x
+  }
+
+  (1 to 100).map(i =>
+    Future {
+      try {
+        atomicWithRetryMax[Int](3)(block)
+        log(s"Transaction: $i - ok")
+      } catch {
+        case ReachMaxNumberException(cntRetries) => log(s"Transaction: $i (retries = $cntRetries)")
+      }
+    }
+  )
+
+  Thread.sleep(3000)
+}

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

@@ -0,0 +1,73 @@
+package org.learningconcurrency
+package exercises
+package ch7
+
+/**
+  * Implement a transactional First In First Out (FIFO) queue,
+  * represented with the TQueue class:
+  *
+  * class TQueue[T] {
+  *    def enqueue(x: T)(implicit txn: InTxn): Unit = ???
+  *    def dequeue()(implicit txn: InTxn): T = ???
+  *  }
+  *
+  * The TQueue class has similar semantics as scala.collection.mutable. Queue,
+  * but calling dequeue on an empty queue blocks until a value becomes available.
+  */
+object Ex5 extends App {
+
+  import scala.collection.immutable.Queue
+  import scala.concurrent.stm._
+  import scala.concurrent.{ExecutionContext, Future}
+  import ExecutionContext.Implicits.global
+
+  class TQueue[T] {
+
+    private val r = Ref[Queue[T]](Queue.empty[T])
+
+    def enqueue(x: T)(implicit txn: InTxn): Unit = {
+      r() = r() :+ x
+    }
+
+    def dequeue()(implicit txn: InTxn): T = {
+      r().dequeueOption match {
+        case None => retry
+        case Some((x,q)) => {
+          r() = q
+          x
+        }
+      }
+    }
+  }
+
+  //test
+  val tQueue = new TQueue[Integer]
+
+  val l = 1 to 20
+
+  l.map { i =>
+    Future {
+      atomic {implicit txn =>
+        val x = tQueue.dequeue
+
+        Txn.afterCommit{_ =>
+          log(s"dequeu: $x")
+        }
+      }
+    }
+  }
+
+  l.map { i =>
+    Future {
+      atomic {implicit txn =>
+        tQueue.enqueue(i)
+
+        Txn.afterCommit { _ =>
+          log(s"enque: $i")
+        }
+      }
+    }
+  }
+
+  Thread.sleep(1000)
+}