レースコンディションテスト(Scala)

RaceCondTest.scala

package actor
import java.util.concurrent.CountDownLatch
import scala.collection.mutable.HashSet
import akka.stm._

class Sequence {
  var value = 0
  def getValue() = value
  def getAndIncrement() = synchronized {
    value += 1
    value
  }
  // STMでやる場合は以下
  //  var value = Ref(0)
  //  def getValue() = value.get
  //  def getAndIncrement() = atomic {
  //    value alter (_ + 1)
  //  }
}

object Main {
  val THREAD_LOOP = 2;
  val THREAD_COUNT = 1000;

  def main(args: Array[String]) {
    for (l <- 0 until 10000) {
      val sequence = new Sequence();
      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();
      }
      // 答え合わせ
      if (THREAD_LOOP * THREAD_COUNT != sequence.getValue()) {
        println(
          "%d = NG !!!, total counter = %d".format(l,
            sequence.getValue()));
      }
    }
  }

  // スレッドの処理
  class ThreadAccess(sequence: Sequence, startLatch: CountDownLatch, loopCount: Int) extends Runnable {

    override def run() {
      try {
        val threadId = Thread.currentThread().getId();
        startLatch.await();
        for (i <- 1 to loopCount) {
          val counter = sequence.getAndIncrement();
        }
      } catch {
        case e: InterruptedException =>
          e.printStackTrace();
      }
    }
  }

}