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.Matchers import java.util.concurrent.ArrayBlockingQueue import org.scalatest.concurrent.Conductors class ArrayBlockingQueueSuite extends FunSuite with Matchers with Conductors { 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.
Although useful, bear in mind that a Conductor
's results are not guaranteed to be
accurate 100% of the time. The reason is that it uses java.lang.Thread
's getState
method to
decide when to advance the beat. This use goes against the advice given in the Javadoc documentation for
getState
, which says, "This method is designed for use in monitoring of the system state, not for
synchronization." In short, sometimes the return value of getState
occasionally may be inacurrate,
which in turn means that sometimes a Conductor
could decide to advance the beat too early. In practice,
Conductor
has proven to be very helpful when developing thread safe classes. It is also useful in
for regression tests, but you may have to tolerate occasional false negatives.
- Source
- Conductors.scala
- Alphabetic
- By Inheritance
- Conductor
- AnyRef
- Any
- Hide All
- Show All
- Public
- All
Instance Constructors
- new Conductor()
Value Members
-
final
def
!=(arg0: Any): Boolean
- Definition Classes
- AnyRef → Any
-
final
def
##(): Int
- Definition Classes
- AnyRef → Any
-
final
def
==(arg0: Any): Boolean
- Definition Classes
- AnyRef → Any
-
final
def
asInstanceOf[T0]: T0
- Definition Classes
- Any
-
def
beat: Int
The current value of the thread clock.
The current value of the thread clock.
- returns
the current beat value
-
def
clone(): AnyRef
- Attributes
- protected[java.lang]
- Definition Classes
- AnyRef
- Annotations
- @throws( ... )
-
def
conduct(interval: Interval)(implicit config: Conductors.PatienceConfig, pos: Position): Assertion
Conducts a multi-threaded test using the configured maximum allowed time between beats (the
timeout
) and the configured time to sleep between checks (theinterval
).Conducts a multi-threaded test using the configured maximum allowed time between beats (the
timeout
) and the configured time to sleep between checks (theinterval
).The maximum amount of time allowed between successive beats is configured by the
timeout
field of thePatienceConfig
passed implicitly as the last parameter. The interval to sleep between successive checks for progress is configured by the value contained in the passedinterval
parameter.- interval
the
Interval
configuration parameter- config
the
PatienceConfig
object containing the (used)timeout
and (unused)interval
parameters
-
def
conduct(timeout: Timeout)(implicit config: Conductors.PatienceConfig, pos: Position): Assertion
Conducts a multi-threaded test using the configured maximum allowed time between beats (the
timeout
) and the configured time to sleep between checks (theinterval
).Conducts a multi-threaded test using the configured maximum allowed time between beats (the
timeout
) and the configured time to sleep between checks (theinterval
).The maximum amount of time allowed between successive beats is configured by the value contained in the passed
timeout
parameter. The interval to sleep between successive checks for progress is configured by by theinterval
field of thePatienceConfig
passed implicitly as the last parameter.- timeout
the
Timeout
configuration parameter- config
the
PatienceConfig
object containing the (unused)timeout
and (used)interval
parameters
-
def
conduct(timeout: Timeout, interval: Interval)(implicit pos: Position): Assertion
Conducts a multi-threaded test using the configured maximum allowed time between beats (the
timeout
) and the configured time to sleep between checks (theinterval
).Conducts a multi-threaded test using the configured maximum allowed time between beats (the
timeout
) and the configured time to sleep between checks (theinterval
).The maximum amount of time allowed between successive beats is configured by the value contained in the passed
timeout
parameter. The interval to sleep between successive checks for progress is configured by the value contained in the passedinterval
parameter.- timeout
the
Timeout
configuration parameter- interval
the
Interval
configuration parameter
-
def
conduct()(implicit config: Conductors.PatienceConfig, pos: Position): Assertion
Conducts a multi-threaded test using the configured maximum allowed time between beats (the
timeout
) and the configured time to sleep between checks (theinterval
).Conducts a multi-threaded test using the configured maximum allowed time between beats (the
timeout
) and the configured time to sleep between checks (theinterval
).- config
the
PatienceConfig
object containing thetimeout
andinterval
parameters used to configure the multi-threaded test
-
def
conductingHasBegun: Boolean
Indicates whether either of the two overloaded
conduct
methods have been invoked.Indicates whether either of the two overloaded
conduct
methods have been invoked.This method returns true if either
conduct
method has been invoked. Theconduct
method may have returned or not. (In other words, atrue
result from this method does not mean theconduct
method has returned, just that it was already been invoked and,therefore, the multi-threaded scenario it conducts has definitely begun.) -
final
def
eq(arg0: AnyRef): Boolean
- Definition Classes
- AnyRef
-
def
equals(arg0: Any): Boolean
- Definition Classes
- AnyRef → Any
-
def
finalize(): Unit
- Attributes
- protected[java.lang]
- Definition Classes
- AnyRef
- Annotations
- @throws( classOf[java.lang.Throwable] )
-
final
def
getClass(): Class[_]
- Definition Classes
- AnyRef → Any
-
def
hashCode(): Int
- Definition Classes
- AnyRef → Any
-
def
isConductorFrozen: Boolean
Indicates whether the conductor has been frozen.
Indicates whether the conductor has been frozen.
Note: The only way a thread can freeze the conductor is by calling
withConductorFrozen
. -
final
def
isInstanceOf[T0]: Boolean
- Definition Classes
- Any
-
final
def
ne(arg0: AnyRef): Boolean
- Definition Classes
- AnyRef
-
final
def
notify(): Unit
- Definition Classes
- AnyRef
-
final
def
notifyAll(): Unit
- Definition Classes
- AnyRef
-
final
def
synchronized[T0](arg0: ⇒ T0): T0
- Definition Classes
- AnyRef
-
def
thread(fun: ⇒ Any): Thread
Creates a new thread that will execute the specified function.
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.
- fun
the function to be executed by the newly created thread
- returns
the newly created thread
-
def
threadNamed(name: String)(fun: ⇒ Any)(implicit pos: Position): Thread
Creates a new thread with the specified name that will execute the specified function.
Creates a new thread with the specified name that will execute the specified function.
This method may be safely called by any thread.
- 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
toString(): String
- Definition Classes
- AnyRef → Any
-
final
def
wait(): Unit
- Definition Classes
- AnyRef
- Annotations
- @throws( ... )
-
final
def
wait(arg0: Long, arg1: Int): Unit
- Definition Classes
- AnyRef
- Annotations
- @throws( ... )
-
final
def
wait(arg0: Long): Unit
- Definition Classes
- AnyRef
- Annotations
- @throws( ... )
-
def
waitForBeat(beat: Int)(implicit pos: Position): Succeeded.type
Blocks the current thread until the thread beat reaches the specified value, at which point the current thread will be unblocked.
Blocks the current thread until the thread beat reaches the specified value, at which point the current thread will be unblocked.
- beat
the tick value to wait for
- Exceptions thrown
NotAllowedException
if the abeat
less than or equal to zero is passed
-
def
whenFinished(fun: ⇒ Assertion)(implicit pos: Position): Assertion
Invokes
conduct
and afterconduct
method returns, ifconduct
returns normally (i.e., without throwing an exception), invokes the passed function.Invokes
conduct
and afterconduct
method returns, ifconduct
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 invokeconduct
and, if it returns noramlly, execute the passed function.Because
whenFinished
invokesconduct
, it can only be invoked once on aConductor
instance. As a result, if you need to pass a block of code towhenFinished
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 invokeconduct
and never invokewhenFinished
.- fun
the function to execute after
conduct
call returns
- Exceptions thrown
NotAllowedException
if the calling thread is not the thread that instantiated thisConductor
, or ifconduct
has already been invoked on this conductor.
-
def
withConductorFrozen[T](fun: ⇒ T): T
Executes the passed function with the
Conductor
frozen so that it won't advance the clock.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, theConductor
will be unfrozen so that the beat will advance when all threads are blocked, as normal.- fun
the function to execute while the
Conductor
is frozen.
Deprecated Value Members
-
def
thread(name: String)(fun: ⇒ Any)(implicit pos: Position): Thread
The overloaded thread method that takes a String name has been deprecated and will be removed in a future version of ScalaTest. Please use threadNamed instead.
The overloaded thread method that takes a String name has been deprecated and will be removed in a future version of ScalaTest. Please use threadNamed instead.
- Annotations
- @deprecated
- Deprecated
The overloaded thread method that takes a String name has been deprecated and will be removed in a future version of ScalaTest. Please use threadNamed instead.