Writing Multithreaded Unit Tests - 5/23/2011

Test driven development is the linchpin of writing quality code. Writing unit tests first insures each class, method, and line of code behaves as expected. What happens when the code is multithreaded, such as starting and responding to asynchronous requests? Suddenly unit tests can seem difficult, if not impossible to write. In this article we are going to explore multithreaded unit testing in C# using Moq and NUnit.

The full source code for the examples in this article can be found at Google code. We will be working in Example.ThreadedUnitTesting.sln. Our first project will have the same name. I am assuming the reader has a basic understanding of multithreaded programming.

For this example we are going to work with a system that has to solve a network algorithm. It will start with gathering data or facts, and then send them to the class responsible for determining production routes for the facts to run through for their transformations. Finally those routes with the initial fact get posted. The three processes run at different speeds. So we will build a class that can manage the synchronization of the three processes. The responsibilities for reading, calculating, and posting are all in other classes that will also run in threads. How those classes work is not a concern, just their operation contract as defined by their interfaces. We will start with what we are going to interact with. The interfaces we are interacting with are the following:

01 public interface IThread
02 {
03     void Stop();
04     void Signal();
05 }
06 
07 public interface IFactReader : IThread
08 {
09     void GetAllFactsAsync(Queue destinationQueue, SignalWork callBack);
10 }
11 
12 public interface IFactMapper : IThread
13 {
14     void CalcProductionStopsAsync(Queue factQueue, Queue productionQueue, SignalWork callBack);
15 }
16 
17 public interface IMapWriter : IThread
18 {
19     void PostProductionStopsAsync(Queue sourceQueue, SignalWork callBack);
20 }
21 
22 public interface ILogger
23 {
24     void Log(LogLevel level, string message);
25     void Log(LogLevel level, Exception exception);
26 }
27 

Following the practice of building our interfaces first, we will and interface that will declare the operations contract of the class that we will be writing and testing.

01 public interface IFactHandler : IThread
02 {
03     void ProcessFacts(object data);
04 }
05 

We created the IThread interface to give us a common framework for talking with the classes. To start we will write a test for ProcessFacts().

First we will implement the IThreaded interface methods in FactHandler. If you have ReSharper then you can use it to implement the interface with exceptions being thrown for each methods. We will also add the interfaces we are going to work with. Otherwise code the class as follows:

01 public class FactHandler : IFactHandler
02 {
03     public IFactReader FactReader { get; set; }
04     public IMapWriter MapWriter { get; set; }
05     public IFactMapper FactMapper { get; set; }
06     public ILogger Log { get; set; }
07 
08     private Queue _factQueue = Queue.Synchronized(new Queue());
09     private Queue _mapQueue = Queue.Synchronized(new Queue());
10 
11     public void Stop()
12     {
13         throw new NotImplementedException();
14     }
15 
16     public void Signal()
17     {
18         throw new NotImplementedException();
19     }
20 
21     public void ProcessFacts(object data)
22     {
23         throw new NotImplementedException();
24     }
25 }
26 

Our goal is to create a handler class that will facilitate the communication between the three worker classes that are creating our production facility maps. This handler must be able to start, stop, and signal the worker threads, and log errors. The unit test for the logger will not be covered in this article.

The test fixture will set up the Moq mocks and our target. We will start with creating a target and a mock for the IFactReader.

01 private FactHandler target;
02 private Mock<IFactReader> _readerMock;
03 

The target is what we are testing, the mock will implement the interface we are coding to. We will initialize our mock in the setup portion of the test fixture.

01 [SetUp]
02 public void SetUp()
03 {
04     target = new FactHandler();
05     _readerMock = new Mock<IFactReader>();
06     target.FactReader = _readerMock.Object;
07 }
08 

All Moq objects have an Object property that holds the instantiation of the interface or class being mocked. Assign that to the interface on the target. We will do this for all interfaces the target needs to run. When we are done the test fixture will look something like this:

01 [TestFixture]
02 public class FactHandler_Tests
03 {
04     private FactHandler target;
05     private Mock<IFactReader> _readerMock;
06     private Mock<IMapWriter> _writerMock;
07     private Mock<IFactMapper> _mapperMock;
08     private Mock<ILogger> _loggerMock;
09 
10     [SetUp]
11     public void SetUp()
12     {
13         target = new FactHandler();
14         _mapperMock = new Mock<IFactMapper>();
15         _loggerMock = new Mock<ILogger>();
16         _writerMock = new Mock<IMapWriter>();
17         _readerMock = new Mock<IFactReader>();
18         target.FactReader = _readerMock.Object;
19         target.MapWriter = _writerMock.Object;
20         target.Log = _loggerMock.Object;
21         target.FactMapper = _mapperMock.Object;
22     }
23 
24 }
25 

With our mocks we can inject whatever threading primitives we need to insure the target is behaving. For most our tests, we will use the EventWaitHandle. Remember that unit tests are whitebox tests. We ought to know what the target is doing to make our tests effective. In this case, we know the target will start the fact reader as a thread with GetAllFactsAsync() as the start up method. Using Moq we will set up a thread safe call to GetAllFactsAsync that will assign the running thread to a variable, signal the main test thread, and exit. From there we can compare the ids of the target thread and the mock thread to confirm the target is properly starting a new thread. Finally, since the target is expected to run in a thread, our unit tests will start the target in a thread. A word of warning about writing multithreaded tests: it is possible that some tests cannot pass until the entire target has been coded. If we have not implemented Stop(), the target may not shut down gracefully causing intermittent errors. If this appears to be happening, ignore the failure and move to resolving the other failed test before coming back.

Start with setting up the GetAllFactsAsync method in Moq.

01 Thread readerThread = null;
02 
03 _readerMock.Setup(x => x.GetAllFactsAsync(It.IsAny<Queue>(), It.IsAny<SignalWork>()))
04     .Callback(() =>
05     {
06         readerThread = Thread.CurrentThread;
07         waitHandle.Set();
08     });
09 

C# will take care of the scoping of the readerThread local variable that is being accessed in the anonymous callback method. Next we will create the thread to run the target.

01 Thread targetThread = new Thread(target.ProcessFacts);
02 targetThread.Start();
03 

We expect the target to start the thread in the mock, so at this point we wait for the signal. We will cap the wait time at four seconds, which is very generous considering how little code is is running. This prevents the test from hanging the test runner. Once we get the event or time out, we tell the target to stop running, we then join the threads and let them finish.

01 waitHandle.WaitOne(4000);
02 target.Stop();
03 targetThread.Join();
04 readerThread.Join();
05 

Finally we check the ManagedThreadIds of the two threads to verify they are different. Even though the threads have terminated, a thread handle is valid so long as it has not been closed, or in C# so long as there is a reference to it.

01 Assert.AreNotEqual(targetThread.ManagedThreadId, readerThread.ManagedThreadId);
02 

And there we have our first multithreaded test. At this point it looks like this:

01 [Test]
02 public void ProcessFacts_StartsThreadedFactReader_Test()
03 {
04     EventWaitHandle waitHandle = new EventWaitHandle(false, EventResetMode.AutoReset);
05     Thread readerThread = null;
06 
07     _readerMock.Setup(x => x.GetAllFactsAsync(It.IsAny<Queue>(), It.IsAny<SignalWork>()))
08         .Callback(() =>
09         {
10             readerThread = Thread.CurrentThread;
11             waitHandle.Set();
12         });
13 
14     Thread targetThread = new Thread(target.ProcessFacts);
15     targetThread.Start();
16     waitHandle.WaitOne(4000);
17     target.Stop();
18     targetThread.Join();
19     readerThread.Join();
20     Assert.AreNotEqual(targetThread.ManagedThreadId, readerThread.ManagedThreadId);
21 }
22 
23 

Of course, the tests fail since all we do is throw exceptions. I'll leave getting the tests passing to the reader. Or you can check the code out of the repository.

The next test will be similar in structure. In this test we need to confirm that the stop method is called when the target stops. This one is a bit more tricky because we need to mock the execution loop of the reader thread but is guaranteed to exit even if the test fails. To do this we will add a second wait handle. We will also create a call counter for the stop method.

01 EventWaitHandle startedHandle = new EventWaitHandle(false, EventResetMode.AutoReset);
02 EventWaitHandle waitHandle = new EventWaitHandle(false, EventResetMode.AutoReset);
03 EventWaitHandle stopHandle = new EventWaitHandle(false, EventResetMode.AutoReset);
04 int calledCount = 0;
05 
06 

You'll notice that I also added a startedHandle. This will be used for flow control of the test. We ought not to call Stop() before we even spun everything up. We will create the GetAllFactsAsync to wait up to five seconds for Stop() to get called. Stop will increment the called count then set the stop event.

01 _readerMock.Setup(x => x.GetAllFactsAsync(It.IsAny<Queue>(), It.IsAny<SignalWork>()))
02     .Callback(() =>
03                   {
04                       startedHandle.Set();
05                       stopHandle.WaitOne(5000);
06                       waitHandle.Set();
07                   });
08 
09 _readerMock.Setup(x => x.Stop())
10     .Callback(() =>
11                   {
12                       ++calledCount;
13                       stopHandle.Set();
14                   });
15 
16 

Now we are ready to create the target thread, run it and call its stop method.

01 Thread targetThread = new Thread(target.ProcessFacts);
02 targetThread.Start();
03 startedHandle.WaitOne(1000);
04 target.Stop();
05 waitHandle.WaitOne(4000);
06 targetThread.Join();
07 Assert.AreEqual(1, calledCount);
08 
09 

In this case we are no longer concerned with joining the internal threads created by the target, since the Stop method should be implemented with this test. In the prior tests it would be okay to rely on the stop method, since it gets confirmed in this test. I added the joining to the internal threads to stabilize the tests prior to implementing Stop().

01 [Test]
02 public void Stop_StopsReaderThread_Test()
03 {
04     EventWaitHandle startedHandle = new EventWaitHandle(false, EventResetMode.AutoReset);
05     EventWaitHandle waitHandle = new EventWaitHandle(false, EventResetMode.AutoReset);
06     EventWaitHandle stopHandle = new EventWaitHandle(false, EventResetMode.AutoReset);
07     int calledCount = 0;
08 
09     _readerMock.Setup(x => x.GetAllFactsAsync(It.IsAny<Queue>(), It.IsAny<SignalWork>()))
10         .Callback(() =>
11                       {
12                           startedHandle.Set();
13                           stopHandle.WaitOne(5000);
14                           waitHandle.Set();
15                       });
16 
17     _readerMock.Setup(x => x.Stop())
18         .Callback(() =>
19                       {
20                           ++calledCount;
21                           stopHandle.Set();
22                       });
23     Thread targetThread = new Thread(target.ProcessFacts);
24     targetThread.Start();
25     startedHandle.WaitOne(1000);
26     target.Stop();
27     waitHandle.WaitOne(4000);
28     targetThread.Join();
29     Assert.AreEqual(1, calledCount);
30 }
31 

Stopping the thread is good, but how do I know that the target is in fact waiting for the Stop() call before exiting? This test will be similar to the Stop() test except that we will inspect the state of the thread after we know it has started and we wait an arbitrary amount of time.

01 [Test]
02 public void ProcessFacts_RunsUntilStopIsCalled_Test()
03 {
04     EventWaitHandle startedHandle = new EventWaitHandle(false, EventResetMode.AutoReset);
05     EventWaitHandle stopHandle = new EventWaitHandle(false, EventResetMode.AutoReset);
06 
07     _readerMock.Setup(x => x.GetAllFactsAsync(It.IsAny<Queue>(), It.IsAny<SignalWork>()))
08         .Callback(() =>
09         {
10             startedHandle.Set();
11             stopHandle.WaitOne(15000);
12         });
13 
14     _readerMock.Setup(x => x.Stop())
15         .Callback(() => stopHandle.Set());
16 
17     Thread targetThread = new Thread(target.ProcessFacts);
18     targetThread.Start();
19     startedHandle.WaitOne(1000);
20     Thread.Sleep(1000);
21     Assert.IsTrue(targetThread.IsAlive);
22     target.Stop();
23     targetThread.Join();
24 }
25 
26 

The final tests we will cover in this article will be for the signal. Each thread takes as a start up parameter either the thread safe fact queue, the thread safe product queue, or both. And they all take a reference to the Signal() method. When a worker thread enqueues or dequeues an item, it will signal the target. The target must then signal all other threads.

In this test, we will set up a counter for all three working threads to ensure they all got called. To do this, we will use the Monitor primitive to synchronize the start up of the threads.

01 object locker = new object();
02 int counter = 0;
03 
04 EventWaitHandle writeStop = new EventWaitHandle(false, EventResetMode.AutoReset);
05 _writerMock.Setup(x => x.PostProductionStopsAsync(It.IsAny<Queue>(), It.IsAny<SignalWork>()))
06     .Callback(() =>
07                   {
08                       lock (locker)
09                       {
10                           ++counter;
11                           Monitor.Pulse(locker);
12                       }
13                       writeStop.WaitOne(30000);
14                   });
15 _writerMock.Setup(x => x.Stop())
16     .Callback(() => writeStop.Set());
17 

This establishes the start up and shutdown code for the mock. This will be repeated for all three threads the target uses. We will wrap the increment in the lock, since ++ is not thread safe. All three threads will need their mock setup.

01 _mapperMock.Setup(x => x.Signal())
02     .Callback(() =>
03                   {
04                       lock (locker)
05                       {
06                           calledCount++;
07                       }
08                   });
09 

Now we are ready to start up the threads and signal them. To make sure all threads are running we will lock on our object and wait for the start up counter to reach three. We will do this with a Pulse.

01 Thread targetThread = new Thread(target.ProcessFacts);
02 targetThread.Start();
03 lock (locker)
04 {
05     while ( counter != 3)
06     {
07         Monitor.Wait(locker);
08     }
09 }
10 
11 

Then we write the actual test.

01 target.Signal();
02 target.Stop();
03 targetThread.Join();
04 Assert.AreEqual(3, calledCount);
05 

Our test looks like this:

01     [Test]
02     public void Signal_SignalsAllThreads_Test()
03     {
04         object locker = new object();
05         int counter = 0;
06 
07         EventWaitHandle writeStop = new EventWaitHandle(false, EventResetMode.AutoReset);
08         _writerMock.Setup(x => x.PostProductionStopsAsync(It.IsAny<Queue>(), It.IsAny<SignalWork>()))
09             .Callback(() =>
10                           {
11                               lock (locker)
12                               {
13                                   ++counter;
14                                   Monitor.Pulse(locker);
15                               }
16                               writeStop.WaitOne(30000);
17                           });
18         _writerMock.Setup(x => x.Stop())
19             .Callback(() => writeStop.Set());
20         EventWaitHandle mapperStop = new EventWaitHandle(false, EventResetMode.AutoReset);
21         _mapperMock.Setup(x => x.CalcProductionStopsAsync(It.IsAny<Queue>(), It.IsAny<Queue>(), It.IsAny<SignalWork>()))
22             .Callback(() =>
23             {
24                 lock (locker)
25                 {
26                     ++counter;
27                     Monitor.Pulse(locker);
28                 }
29                 mapperStop.WaitOne(30000);
30             });
31         _mapperMock.Setup(x => x.Stop())
32             .Callback(() => mapperStop.Set());
33         EventWaitHandle readStop = new EventWaitHandle(false, EventResetMode.AutoReset);
34         _readerMock.Setup(x => x.GetAllFactsAsync(It.IsAny<Queue>(), It.IsAny<SignalWork>()))
35             .Callback(() =>
36             {
37                 lock (locker)
38                 {
39                     ++counter;
40                     Monitor.Pulse(locker);
41                 }
42                 readStop.WaitOne(30000);
43             });
44         _readerMock.Setup(x => x.Stop())
45             .Callback(() => readStop.Set());
46 
47         int calledCount = 0;
48 
49         _readerMock.Setup(x => x.Signal())
50             .Callback(() =>
51                           {
52                               lock (locker)
53                               {
54                                   calledCount++;
55                                   Monitor.Pulse(locker);
56                               }
57                           });
58         _mapperMock.Setup(x => x.Signal())
59             .Callback(() =>
60             {
61                 lock (locker)
62                 {
63                     calledCount++;
64                     Monitor.Pulse(locker);
65                 }
66             });
67         _writerMock.Setup(x => x.Signal())
68             .Callback(() =>
69             {
70                 lock (locker)
71                 {
72                     calledCount++;
73                     Monitor.Pulse(locker);
74                 }
75             });
76 
77         Thread targetThread = new Thread(target.ProcessFacts);
78         targetThread.Start();
79         lock (locker)
80         {
81             while ( counter != 3)
82             {
83                 Monitor.Wait(locker);
84             }
85         }
86 
87         target.Signal();
88         target.Stop();
89         targetThread.Join();
90         Assert.AreEqual(3, calledCount);
91     }
92 }
93 
94 

Why did I go through all the work to create such elaborate threading logic to write my tests? Could I have just used .Sleep() to give my tests time to set up? Yes, I could have. But I want to know the test are going to work, not guess they will work. Putting the threading logic into my tests helps me write code that is more likely to work in a production environment.

For further reading on multithreaded programing in C#, I recommend Threading in C#.

Comments