MemoryChannel integration test

In a previous post, I discussed the concurrency issues with the initial MemoryChannel implementation and how unit tests were insufficient to uncover them.

I came up with these basic requirements/invariants to guide my integration test design:

• Data from separately sent buffers must not be mixed or interleaved. That is, if sender 1 writes "FFFF" and sender 2 writes "CCCC", the receiver may only see "CCCCFFFF" or "FFFFCCCC" (depending on how the writes were serialized).
• Assuming the channel is fully drained, all sent buffers must eventually be delivered; data must not be lost.
• The requirements above must hold given a single receiver and one or more senders operating concurrently.

The basic test flow would thus be as follows:

1. Create MemoryChannel.
2. Start senders on background threads; run until canceled.
3. Start async receiver; validate data after each receive operation.
4. Run for some predetermined duration.
5. Cancel senders; wait for tasks to complete.
6. Dispose MemoryChannel; this will unblock the last receive operation.
7. Validate sent and received data size.

In the final refactoring, the main body of the test code ended up as follows:

MemoryChannel channel = new MemoryChannel();
int[] sentDataSizes = new int[] { 11, 19, 29, 41, 53, 71, 89, 101 };

using (CancellationTokenSource cts = new CancellationTokenSource())
{
    DataOracle oracle = new DataOracle();
    Sender[] senders = this.CreateSenders(channel, sentDataSizes, oracle);

    ValidatingReceiver receiver = new ValidatingReceiver(channel, this.logger, this.receiveBufferSize, oracle);

    Task[] senderTasks = new Task[senders.Length];
    Task receiverTask = this.StartSendersAndReceiver(cts.Token, senders, receiver, senderTasks);
    Thread.Sleep(this.duration);

    cts.Cancel();
    Task.WaitAll(senderTasks);

    channel.Dispose();
    receiverTask.Wait();

    ValidateTransferredByteCount(senderTasks, receiverTask);
}

My data validation strategy uses sent buffers of prime number sizes filled with a specific byte value per sender and a larger power of two for receive buffer size; for example, sender 1 uses a buffer of size 11 filled with 0x1, sender 2 uses a buffer of size 19 filled with 0x2 and so on, with a receiver asking for 256 bytes at a time. My thinking was that it would be easier to detect state corruption if I used numbers like these (I admittedly have not done any rigorous mathematical proofs of this…). I implemented a ValidatingReceiver which scans all received buffers and looks for runs of the same byte value. On each byte value change, the results are fed to a DataOracle which knows the mapping of expected byte values to buffer length multiples:

public void VerifyLastSeen(byte lastSeen, int lastCount)
{
    int expectedCountMultiple;
    if (!this.patterns.TryGetValue(lastSeen, out expectedCountMultiple))
    {
        string message = string.Format(
            CultureInfo.InvariantCulture,
            "State corruption detected; byte 0x{0:X} was unexpected.",
            lastSeen);
        throw new InvalidOperationException(message);
    }

    if (lastCount % expectedCountMultiple != 0)
    {
        string message = string.Format(
            CultureInfo.InvariantCulture,
            "State corruption detected; count of {0} for byte 0x{1:X} is not a multiple of {2}.",
            lastCount,
            lastSeen,
            expectedCountMultiple);
        throw new InvalidOperationException(message);
    }
}

By the final commit, I had a fully automated integration test app which gave me reasonable confidence that my logic was correct:

[ . . . ]
[0035.048/T01] Receive loop with 8 senders, 5.0 sec, send before receive=True, receive buffer=256...
[0035.048/T01] Sender B=11/F=0x1 starting...
[0035.048/T01] Sender B=19/F=0x2 starting...
[0035.048/T01] Sender B=29/F=0x3 starting...
[0035.049/T01] Sender B=41/F=0x4 starting...
[0035.049/T01] Sender B=53/F=0x5 starting...
[0035.049/T01] Sender B=71/F=0x6 starting...
[0035.049/T01] Sender B=89/F=0x7 starting...
[0035.050/T01] Sender B=101/F=0x8 starting...
[0035.050/T01] Receiver starting...
[0040.050/T35] Sender B=41/F=0x4 completed. Sent 409631 bytes.
[0040.050/T39] Sender B=101/F=0x8 completed. Sent 1009091 bytes.
[0040.051/T36] Sender B=53/F=0x5 completed. Sent 529470 bytes.
[0040.051/T38] Sender B=89/F=0x7 completed. Sent 889288 bytes.
[0040.051/T37] Sender B=71/F=0x6 completed. Sent 709290 bytes.
[0040.051/T33] Sender B=19/F=0x2 completed. Sent 189886 bytes.
[0040.051/T32] Sender B=11/F=0x1 completed. Sent 109956 bytes.
[0040.051/T34] Sender B=29/F=0x3 completed. Sent 289826 bytes.
[0040.053/T01] Receiver completed. Received 4136438 bytes.
[0040.053/T01] Done.