Lock & STM

thread-test.scala

#!/usr/bin/env xsbtscript
!#

/***
scalaVersion := "2.9.1"

resolvers += "Typesafe Repository" at "http://repo.typesafe.com/typesafe/releases/"

libraryDependencies ++= Seq(
  "se.scalablesolutions.akka" % "akka-stm" % "1.2"
)
*/

import java.util.concurrent.CountDownLatch
import scala.collection.mutable.HashSet
import scala.testing.Benchmark
import akka.stm._
import akka.util.duration._
import compat.Platform

trait Sequence {
  def getValue():Int
  def getAndIncrement():Int
}

class SequenceByUnsafe extends Sequence {
  var value = 0
  def getValue() = value

  def getAndIncrement() = {
    value += 1
    value
  }
}

class SequenceBySynchronized extends SequenceByUnsafe {
  override def getAndIncrement() = synchronized{
    super.getAndIncrement()
  }
}

class SequenceBySTM extends Sequence {
  var value = Ref(0)
  def getValue() = value.get

  implicit val txFactory = TransactionFactory(maxRetries = 10000) 

//  var abort = 0
//  var commit = 0

  def getAndIncrement() = atomic {
    val result = value alter (_ + 1)
/*
    deferred {
      commit += 1
    }
    compensating {
      abort += 1
    }
*/
    result
  }
//  override def toString = "%d,%d,%f".format(abort, commit, if(commit>0) BigDecimal(abort) / BigDecimal(commit) * 100 else BigDecimal(0))
}


object Main {
  val THREAD_LOOP = 1000
  val THREAD_COUNT = 2

  def main(args: Array[String]) {
    args match{
      case Array("L") => {
        println(">>synchronized")
        threadTest(new SequenceBySynchronized)
      }
      case Array("S") => {
        println(">>STM")
        val sstm = new SequenceBySTM   
        threadTest(sstm)
      }
      case Array("U") => {
        println(">>unsafe")
        threadTest(new SequenceByUnsafe)
      }
    }
  }

  def threadTest(sequence:Sequence) {
    Platform.collectGarbage 
    val startLatch = new CountDownLatch(1);
    val threads = HashSet.empty[Thread]
    // スレッドの準備
    for (i <- 1 to THREAD_COUNT) {
      val thread = new Thread(new ThreadAccess(sequence, startLatch, THREAD_LOOP));
      threads += (thread)
      thread.start();
    }
    // 足並み揃えてゴー。
    startLatch.countDown();
    try {
      // みんなが終わるのを待つ
      for (thread <- threads) {
        thread.join();
      }
    } catch {
      case e: InterruptedException =>
        e.printStackTrace();
    }
  }
 

  // スレッドの処理
  class ThreadAccess(sequence: Sequence, startLatch: CountDownLatch, loopCount: Int) extends Runnable {
    import BenchUtil._
    override def run() {
      try {
        startLatch.await();
        bench(100){ 
          for(i <- 1 to (loopCount/100)){
            val counter = sequence.getAndIncrement();
          }
        } 
      } catch {
        case e: InterruptedException =>
          e.printStackTrace();
      }
    }
  }


object BenchUtil {

  private def avg(xs:List[BigDecimal]):BigDecimal =
    xs.sum / xs.size

  private def std(xs:List[BigDecimal]):BigDecimal = {
    val a = avg(xs) 
    Math.sqrt((xs.foldLeft(BigDecimal(0))((s,c) => s + (c-a) * (c+a)) / xs.size).toDouble)
  }

  private def median(xs:List[BigDecimal]) = xs.toSet.toList.sortWith(_ < _) match {
    case n :: Nil => n
    case xs if xs.size % 2 !=0 => xs(xs.size / 2)
    case xs if xs.size % 2 == 0 =>  {
      val a = xs(xs.size / 2 -1)
      val b = xs(xs.size / 2)
      (a + b)/2
    }
    case _ => throw new RuntimeException
  }

  private def mode(xs:List[BigDecimal]):BigDecimal =  
    xs.foldLeft(Map[BigDecimal,Int]().withDefaultValue(0)){(map,key) => map + (key -> (map(key) + 1))} maxBy(_._2) _1

  def bench(n:Int)(f: => Unit) {
    val truncate = n / 5
    val times = for(i <- List.range(1, n + 1, 1)) yield {
      val start = System.nanoTime
      f
      val stop = System.nanoTime
      Platform.collectGarbage      
      BigDecimal(stop - start) / 1000
    }
    val result = times.sortWith(_ < _).view(truncate, n - truncate).toList
    if (result.size > 0 ){
      println("threadId = %d, n = %d, avg = %11.2f, std = %11.2f, median = %11.2f, mode = %11.2f, min = %11.2f, max = %11.2f".
        format(Thread.currentThread.getId, result.size, avg(result), std(result), median(result), mode(result), result.min, result.max))
    }
  }
}


}

Main.main(args)

junichi-no-macbookpro:bin junichi$ ./thread-test.scala L
[info] Set current project to default-6bb5f7 (in build file:/Users/junichi/.conscript/boot/dc7aa3f025ce7a1a2e39/)
[warn] there were 1 deprecation warnings; re-run with -deprecation for details
[warn] one warning found

synchronized
threadId = 41, n = 60, avg = 22.32, std = 5.24, median = 25.00, mode = 19.00, min = 16.00, max = 36.00
threadId = 42, n = 60, avg = 25.88, std = 3.36, median = 22.00, mode = 27.00, min = 15.00, max = 28.00
junichi-no-macbookpro:bin junichi$ ./thread-test.scala S
[info] Set current project to default-6bb5f7 (in build file:/Users/junichi/.conscript/boot/dc7aa3f025ce7a1a2e39/)
STM
2011/12/05 9:00:41 org.multiverse.api.GlobalStmInstance

: Initializing GlobalStmInstance using factoryMethod 'org.multiverse.stms.alpha.AlphaStm.createFast'.

2011/12/05 9:00:41 org.multiverse.stms.alpha.AlphaStm

: Created a new AlphaStm instance

2011/12/05 9:00:41 org.multiverse.api.GlobalStmInstance

: Successfully initialized GlobalStmInstance using factoryMethod 'org.multiverse.stms.alpha.AlphaStm.createFast'.

threadId = 42, n = 60, avg = 39222.13, std = 38308.45, median = 75744.00, mode = 591.00, min = 578.00, max = 82639.00
threadId = 41, n = 60, avg = 62965.67, std = 59544.94, median = 76139.00, mode = 152672.00, min = 557.00, max = 153056.00
junichi-no-macbookpro:bin junichi$ ./thread-test.scala U
[info] Set current project to default-6bb5f7 (in build file:/Users/junichi/.conscript/boot/dc7aa3f025ce7a1a2e39/)
unsafe
threadId = 41, n = 60, avg = 24.15, std = 4.52, median = 22.50, mode = 28.00, min = 16.00, max = 29.00
threadId = 40, n = 60, avg = 21.45, std = 3.91, median = 21.00, mode = 20.00, min = 16.00, max = 28.00

-----STMの方を読み込みだけにしたテスト結果-----

synchronized
threadId = 29, n = 60, avg = 16.58, std = 2.25, median = 16.50, mode = 15.00, min = 13.00, max = 20.00
threadId = 30, n = 60, avg = 17.52, std = 2.32, median = 18.00, mode = 15.00, min = 14.00, max = 22.00
jkato:bin junichi_kato$ ./thread-test.scala S
[info] Set current project to default-5fe1fe (in build file:/private/var/lib/conscript/boot/0888c061d743d884c41b/)
STM
2011/12/05 22:34:00 org.multiverse.api.GlobalStmInstance

: Initializing GlobalStmInstance using factoryMethod 'org.multiverse.stms.alpha.AlphaStm.createFast'.

2011/12/05 22:34:00 org.multiverse.stms.alpha.AlphaStm

: Created a new AlphaStm instance

2011/12/05 22:34:00 org.multiverse.api.GlobalStmInstance

: Successfully initialized GlobalStmInstance using factoryMethod 'org.multiverse.stms.alpha.AlphaStm.createFast'.

threadId = 30, n = 60, avg = 119.85, std = 9.27, median = 124.00, mode = 126.00, min = 107.00, max = 137.00
threadId = 29, n = 60, avg = 120.80, std = 8.91, median = 123.50, mode = 127.00, min = 109.00, max = 137.00