org.scalatest.concurrent

class Conductor

[source: org/scalatest/concurrent/Conductor.scala]

final class Conductor
extends AnyRef
Class that facilitates the testing of classes, traits, and libraries designed to be used by multiple threads concurrently.

A Conductor conducts a multi-threaded scenario by maintaining a clock of "beats." Beats are numbered starting with 0. You can ask a Conductor to run threads that interact with the class, trait, or library (the subject) you want to test. A thread can call the Conductor's waitForBeat method, which will cause the thread to block until that beat has been reached. The Conductor will advance the beat only when all threads participating in the test are blocked. By tying the timing of thread activities to specific beats, you can write tests for concurrent systems that have deterministic interleavings of threads.

A Conductor object has a three-phase lifecycle. It begins its life in the setup phase. During this phase, you can start threads by invoking the thread method on the Conductor. When conduct is invoked on a Conductor, it enters the conducting phase. During this phase it conducts the one multi-threaded scenario it was designed to conduct. After all participating threads have exited, either by returning normally or throwing an exception, the conduct method will complete, either by returning normally or throwing an exception. As soon as the conduct method completes, the Conductor enters its defunct phase. Once the Conductor has conducted a multi-threaded scenario, it is defunct and can't be reused. To run the same test again, you'll need to create a new instance of Conductor.

Here's an example of the use of Conductor to test the ArrayBlockingQueue class from java.util.concurrent:

 import org.scalatest.fixture.FunSuite
 import org.scalatest.matchers.ShouldMatchers
 import java.util.concurrent.ArrayBlockingQueue

 class ArrayBlockingQueueSuite extends FunSuite with ShouldMatchers {
 
   test("calling put on a full queue blocks the producer thread") {

     val conductor = new Conductor
     import conductor._

     val buf = new ArrayBlockingQueue[Int](1)
 
     thread("producer") {
       buf put 42
       buf put 17
       beat should be (1)
     }
 
     thread("consumer") {
       waitForBeat(1)
       buf.take should be (42)
       buf.take should be (17)
     }
 
     whenFinished {
       buf should be ('empty)
     }
   }
 }
 

When the test shown is run, it will create one thread named producer and another named consumer. The producer thread will eventually execute the code passed as a by-name parameter to thread("producer"):

 buf put 42
 buf put 17
 beat should be (1)
 
Similarly, the consumer thread will eventually execute the code passed as a by-name parameter to thread("consumer"):

 waitForBeat(1)
 buf.take should be (42)
 buf.take should be (17)
 

The thread calls create the threads and starts them, but they will not immediately execute the by-name parameter passed to them. They will first block, waiting for the Conductor to give them a green light to proceed.

The next call in the test is whenFinished. This method will first call conduct on the Conductor, which will wait until all threads that were created (in this case, producer and consumer) are at the "starting line", i.e., they have all started and are blocked, waiting on the green light. The conduct method will then give these threads the green light and they will all start executing their blocks concurrently.

When the threads are given the green light, the beat is 0. The first thing the producer thread does is put 42 in into the queue. As the queue is empty at this point, this succeeds. The producer thread next attempts to put a 17 into the queue, but because the queue has size 1, this can't succeed until the consumer thread has read the 42 from the queue. This hasn't happened yet, so producer blocks. Meanwhile, the consumer thread's first act is to call waitForBeat(1). Because the beat starts out at 0, this call will block the consumer thread. As a result, once the producer thread has executed buf put 17 and the consumer thread has executed waitForBeat(1), both threads will be blocked.

The Conductor maintains a clock that wakes up periodically and checks to see if all threads participating in the multi-threaded scenario (in this case, producer and consumer) are blocked. If so, it increments the beat. Thus sometime later the beat will be incremented, from 0 to 1. Because consumer was waiting for beat 1, it will wake up (i.e., the waitForBeat(1) call will return) and execute the next line of code in its block, buf.take should be (42). This will succeed, because the producer thread had previously (during beat 0) put 42 into the queue. This act will also make producer runnable again, because it was blocked on the second put, which was waiting for another thread to read that 42.

Now both threads are unblocked and able to execute their next statement. The order is non-deterministic, and can even be simultaneous if running on multiple cores. If the consumer thread happens to execute buf.take should be (17) first, it will block (buf.take will not return), because the queue is at that point empty. At some point later, the producer thread will execute buf put 17, which will unblock the consumer thread. Again both threads will be runnable and the order non-deterministic and possibly simulataneous. The producer thread may charge ahead and run its next statement, beat should be (1). This will succeed because the beat is indeed 1 at this point. As this is the last statement in the producer's block, the producer thread will exit normally (it won't throw an exception). At some point later the consumer thread will be allowed to complete its last statement, the buf.take call will return 17. The consumer thread will execute 17 should be (17). This will succeed and as this was the last statement in its block, the consumer will return normally.

If either the producer or consumer thread had completed abruptbly with an exception, the conduct method (which was called by whenFinished) would have completed abruptly with an exception to indicate the test failed. However, since both threads returned normally, conduct will return. Because conduct doesn't throw an exception, whenFinished will execute the block of code passed as a by-name parameter to it: buf should be ('empty). This will succeed, because the queue is indeed empty at this point. The whenFinished method will then return, and because the whenFinished call was the last statement in the test and it didn't throw an exception, the test completes successfully.

This test tests ArrayBlockingQueue, to make sure it works as expected. If there were a bug in ArrayBlockingQueue such as a put called on a full queue didn't block, but instead overwrote the previous value, this test would detect it. However, if there were a bug in ArrayBlockingQueue such that a call to take called on an empty queue never blocked and always returned 0, this test might not detect it. The reason is that whether the consumer thread will ever call take on an empty queue during this test is non-deterministic. It depends on how the threads get scheduled during beat 1. What is deterministic in this test, because the consumer thread blocks during beat 0, is that the producer thread will definitely attempt to write to a full queue. To make sure the other scenario is tested, you'd need a different test:

 test("calling take on an empty queue blocks the consumer thread") {

   val conductor = new Conductor
   import conductor._

   val buf = new ArrayBlockingQueue[Int](1)

   thread("producer") {
     waitForBeat(1)
     buf put 42
     buf put 17
   }

   thread("consumer") {
     buf.take should be (42)
     buf.take should be (17)
     beat should be (1)
   }

   whenFinished {
     buf should be ('empty)
   }
 }
 

In this test, the producer thread will block, waiting for beat 1. The consumer thread will invoke buf.take as its first act. This will block, because the queue is empty. Because both threads are blocked, the Conductor will at some point later increment the beat to 1. This will awaken the producer thread. It will return from its waitForBeat(1) call and execute buf put 42. This will unblock the consumer thread, which will take the 42, and so on.

The problem that Conductor is designed to address is the difficulty, caused by the non-deterministic nature of thread scheduling, of testing classes, traits, and libraries that are intended to be used by multiple threads. If you just create a test in which one thread reads from an ArrayBlockingQueue and another writes to it, you can't be sure that you have tested all possible interleavings of threads, no matter how many times you run the test. The purpose of Conductor is to enable you to write tests with deterministic interleavings of threads. If you write one test for each possible interleaving of threads, then you can be sure you have all the scenarios tested. The two tests shown here, for example, ensure that both the scenario in which a producer thread tries to write to a full queue and the scenario in which a consumer thread tries to take from an empty queue are tested.

Class Conductor was inspired by the MultithreadedTC project, created by Bill Pugh and Nat Ayewah of the University of Maryland.

Author
Josh Cough
Bill Venners
Method Summary
def beat : Int
The current value of the thread clock.
def conduct : Unit
Conducts a multithreaded test with a default clock period of 10 milliseconds and default run limit of 5 seconds.
def conduct (clockPeriod : Int, timeout : Int) : Unit
Conducts a multithreaded test with the specified clock period (in milliseconds) and timeout (in seconds).
def conductingHasBegun : Boolean
Indicates whether either of the two overloaded conduct methods have been invoked.
def isConductorFrozen : Boolean
Indicates whether the conductor has been frozen.
def thread (fun : => Unit) : java.lang.Thread
Creates a new thread that will execute the specified function.
def thread (name : java.lang.String)(fun : => Unit) : java.lang.Thread
Creates a new thread with the specified name that will execute the specified function.
def waitForBeat (beat : Int) : Unit
Blocks the current thread until the thread beat reaches the specified value, at which point the current thread will be unblocked.
def whenFinished (fun : => Unit) : Unit
Invokes conduct and after conduct method returns, if conduct returns normally (i.e., without throwing an exception), invokes the passed function.
def withConductorFrozen [T](fun : => T) : Unit
Executes the passed function with the Conductor frozen so that it won't advance the clock.
Methods inherited from AnyRef
getClass, hashCode, equals, clone, toString, notify, notifyAll, wait, wait, wait, finalize, ==, !=, eq, ne, synchronized
Methods inherited from Any
==, !=, isInstanceOf, asInstanceOf
Method Details
def thread(fun : => Unit) : java.lang.Thread
Creates a new thread that will execute the specified function.

The name of the thread will be of the form Conductor-Thread-N, where N is some integer.

This method may be safely called by any thread.

Parameters
fun - the function to be executed by the newly created thread
Returns
the newly created thread

def thread(name : java.lang.String)(fun : => Unit) : java.lang.Thread
Creates a new thread with the specified name that will execute the specified function.

This method may be safely called by any thread.

Parameters
name - the name of the newly created thread
fun - the function to be executed by the newly created thread
Returns
the newly created thread

def whenFinished(fun : => Unit) : Unit
Invokes conduct and after conduct method returns, if conduct returns normally (i.e., without throwing an exception), invokes the passed function.

If conduct completes abruptly with an exception, this method will complete abruptly with the same exception and not execute the passed function.

This method must be called by the thread that instantiated this Conductor, and that same thread will invoke conduct and, if it returns noramlly, execute the passed function.

Because whenFinished invokes conduct, it can only be invoked once on a Conductor instance. As a result, if you need to pass a block of code to whenFinished it should be the last statement of your test. If you don't have a block of code that needs to be run once all the threads have finished successfully, then you can simply invoke conduct and never invoke whenFinished.

Parameters
fun - the function to execute after conduct call returns
Throws
NotAllowedException - if the calling thread is not the thread that instantiated this Conductor, or if conduct has already been invoked on this conductor.

def waitForBeat(beat : Int) : Unit
Blocks the current thread until the thread beat reaches the specified value, at which point the current thread will be unblocked.
Parameters
beat - the tick value to wait for
Throws
NotAllowedException - if the a beat less than or equal to zero is passed

def beat : Int
The current value of the thread clock.
Returns
the current beat value

def withConductorFrozen[T](fun : => T) : Unit
Executes the passed function with the Conductor frozen so that it won't advance the clock.

While the Conductor is frozen, the beat will not advance. Once the passed function has completed executing, the Conductor will be unfrozen so that the beat will advance when all threads are blocked, as normal.

Parameters
fun - the function to execute while the Conductor is frozen.

def isConductorFrozen : Boolean
Indicates whether the conductor has been frozen.

Note: The only way a thread can freeze the conductor is by calling withConductorFrozen.


def conduct : Unit
Conducts a multithreaded test with a default clock period of 10 milliseconds and default run limit of 5 seconds.

def conductingHasBegun : Boolean
Indicates whether either of the two overloaded conduct methods have been invoked.

This method returns true if either conduct method has been invoked. The conduct method may have returned or not. (In other words, a true result from this method does not mean the conduct method has returned, just that it was already been invoked and,therefore, the multi-threaded scenario it conducts has definitely begun.)


def conduct(clockPeriod : Int, timeout : Int) : Unit
Conducts a multithreaded test with the specified clock period (in milliseconds) and timeout (in seconds).

A Conductor instance maintains an internal clock, which will wake up periodically and check to see if it should advance the beat, abort the test, or go back to sleep. It sleeps clockPeriod milliseconds each time. It will abort the test if either deadlock is suspected or the beat has not advanced for the number of seconds specified as timeout. Suspected deadlock will be declared if for some number of consecutive clock cycles, all test threads are in the BLOCKED or WAITING states and none of them are waiting for a beat.

Parameters
clockPeriod - The period (in ms) the clock will sleep each time it sleeps
timeout - The maximum allowed time between successive advances of the beat. If this time is exceeded, the Conductor will abort the test.
Throws
Throwable - The first error or exception that is thrown by one of the test threads, or a TestFailedException if the test was aborted due to a timeout or suspected deadlock.


Copyright (C) 2001-2010 Artima, Inc. All rights reserved.