tbf/TBFTests/GenesisParallelSchedulingTests.cs

using System;
using System.Linq;
using System.Threading;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using TBF.Rig.TestMethods.GenesisCommunication;

namespace TBFTests
{
    [TestClass]
    [TestCategory("GenesisParallelScheduling")]
    public class GenesisParallelSchedulingTests
    {
        [DataTestMethod]
        [DataRow(1)] [DataRow(2)] [DataRow(3)] [DataRow(4)] [DataRow(5)]
        [DataRow(6)] [DataRow(7)] [DataRow(8)] [DataRow(9)] [DataRow(10)]
        public void EveryBoardIsAssignedExactlyOnceForAllSupportedWorkerCounts(int workers)
        {
            foreach (int boards in new[] { 0, 1, 2, 7, 10 })
            {
                var actual = Enumerable.Range(0, workers)
                    .SelectMany(worker => GenesisWorkerGroupCompletion.BoardIndexes(worker, workers, boards)).OrderBy(x => x).ToArray();
                CollectionAssert.AreEqual(Enumerable.Range(0, boards).ToArray(), actual);
            }
        }

        [DataTestMethod]
        [DataRow(1)] [DataRow(2)] [DataRow(3)] [DataRow(4)] [DataRow(5)]
        [DataRow(6)] [DataRow(7)] [DataRow(8)] [DataRow(9)] [DataRow(10)]
        public void GroupAdvancesOnlyAfterEveryDistinctWorkerFinishes(int count)
        {
            var completion = new GenesisWorkerGroupCompletion(count);
            completion.Begin(0, 1);
            Assert.IsFalse(completion.Complete(0, 2, 0));
            Assert.IsFalse(completion.Complete(1, 1, 0));
            Assert.IsFalse(completion.Complete(0, 1, -1));
            Assert.IsFalse(completion.Complete(0, 1, count));
            for (int worker = 0; worker < count; worker++)
            {
                Assert.AreEqual(worker == count - 1, completion.Complete(0, 1, worker));
                Assert.IsFalse(completion.Complete(0, 1, worker), "Duplicate completion must not advance the group.");
            }
            completion.Begin(0, 2);
            Assert.IsFalse(completion.Complete(0, 1, 0), "Late event from the previous group.");
            for (int worker = count - 1; worker >= 0; worker--)
                Assert.AreEqual(worker == 0, completion.Complete(0, 2, worker));
            completion.Begin(1, 1);
            Assert.IsFalse(completion.Complete(0, 2, 0));
        }

        [TestMethod]
        public void MultipleActivitiesAndGroupsRequireAllWorkersIncludingIdleWorkers()
        {
            var completion = new GenesisWorkerGroupCompletion(10);
            for (int activity = 0; activity < 3; activity++)
            for (int group = 1; group <= 10; group++)
            {
                completion.Begin(activity, group);
                Assert.IsFalse(completion.Complete(activity - 1, group, 0));
                Assert.IsFalse(completion.Complete(activity, group - 1, 0));
                // Also models HOLD ON: no board requests, but each worker must finish.
                for (int worker = 0; worker < 10; worker++)
                    Assert.AreEqual(worker == 9, completion.Complete(activity, group, worker));
            }
        }

        [TestMethod]
        public void TenWorkersCanRunConcurrentlyAndSlowTenthWorkerHoldsGroup()
        {
            var completion = new GenesisWorkerGroupCompletion(10);
            completion.Begin(0, 1);
            using (var started = new CountdownEvent(10))
            using (var firstNine = new CountdownEvent(9))
            using (var run = new ManualResetEventSlim(false))
            using (var slow = new ManualResetEventSlim(false))
            {
                int advances = 0;
                var threads = Enumerable.Range(0, 10).Select(worker => new Thread(() =>
                {
                    started.Signal();
                    run.Wait();
                    if (worker == 9) slow.Wait();
                    if (completion.Complete(0, 1, worker)) Interlocked.Increment(ref advances);
                    if (worker != 9) firstNine.Signal();
                }) { IsBackground = true }).ToArray();
                foreach (var thread in threads) thread.Start();
                try
                {
                    Assert.IsTrue(started.Wait(5000), "All ten workers must start before any finishes.");
                    run.Set();
                    Assert.IsTrue(firstNine.Wait(5000));
                    Assert.AreEqual(0, Volatile.Read(ref advances), "Nine completions must not release a ten-worker group.");
                    slow.Set();
                }
                finally
                {
                    run.Set(); slow.Set();
                    foreach (var thread in threads) thread.Join(5000);
                }
                Assert.AreEqual(1, advances);
            }
        }

        [TestMethod]
        public void Parallel_Group1OneToTen_Group2One_AllTenCallsOverlap()
        {
            VerifyProcessingScenario(10, 10, 1);
        }

        [TestMethod]
        public void Serial_Group1One_Group2OneToTen_NoCallsOverlap()
        {
            VerifyProcessingScenario(10, 1, 10);
        }

        [TestMethod]
        public void Combined_FiveGroup1Boards_TwoGroup2Groups_ParallelWithinSerialBetween()
        {
            VerifyProcessingScenario(10, 5, 2);
        }

        [TestMethod]
        public void Combined_FourWorkers_TenBoards_ThreeGroups_EveryMeterRunsOnce()
        {
            VerifyProcessingScenario(4, 10, 3);
        }

        [TestMethod]
        public void Serial_OneWorker_TenGroup1Boards_AllCallsRunOnce()
        {
            VerifyProcessingScenario(1, 10, 1);
        }

        // Hardware-free harness using the production assignment and completion helpers.
        // Calls are held at a rendezvous, so overlap is proven without timing guesses.
        private static void VerifyProcessingScenario(int workers, int boards, int groups)
        {
            var completion = new GenesisWorkerGroupCompletion(workers);
            completion.Begin(0, 1);
            var sync = new object();
            int currentGroup = 1, active = 0, advances = 0;
            bool abort = false;
            var calls = new int[groups, boards];
            var finished = new int[groups];
            var peaks = new int[groups];
            var failures = new System.Collections.Generic.List<Exception>();
            int parallelism = Math.Min(workers, boards);
            var rendezvous = Enumerable.Range(0, groups).Select(_ => new CountdownEvent(parallelism)).ToArray();
            var threads = Enumerable.Range(0, workers).Select(worker => new Thread(() =>
            {
                try
                {
                    for (int group = 1; group <= groups; group++)
                    {
                        lock (sync)
                        {
                            while (currentGroup != group && !abort)
                                if (!Monitor.Wait(sync, 10000)) throw new TimeoutException("Group did not advance.");
                            if (abort) return;
                        }
                        bool firstCall = true;
                        foreach (int board in GenesisWorkerGroupCompletion.BoardIndexes(worker, workers, boards))
                        {
                            lock (sync)
                            {
                                for (int previous = 0; previous < group - 1; previous++)
                                    Assert.AreEqual(boards, finished[previous], "Next Group 2 started before previous group completed.");
                                calls[group - 1, board]++;
                                active++;
                                peaks[group - 1] = Math.Max(peaks[group - 1], active);
                            }
                            if (firstCall)
                            {
                                rendezvous[group - 1].Signal();
                                Assert.IsTrue(rendezvous[group - 1].Wait(10000), "Assigned workers did not enter calls concurrently.");
                                firstCall = false;
                            }
                            lock (sync) { active--; finished[group - 1]++; }
                        }
                        if (completion.Complete(0, group, worker))
                        {
                            lock (sync)
                            {
                                Assert.AreEqual(boards, finished[group - 1]);
                                Assert.AreEqual(0, active);
                                advances++;
                                completion.Begin(0, group + 1);
                                currentGroup++;
                                Monitor.PulseAll(sync);
                            }
                        }
                    }
                }
                catch (Exception error)
                {
                    lock (sync) { failures.Add(error); abort = true; Monitor.PulseAll(sync); }
                }
            }) { IsBackground = true }).ToArray();
            foreach (var thread in threads) thread.Start();
            bool allJoined = true;
            foreach (var thread in threads) allJoined &= thread.Join(15000);
            Assert.IsTrue(allJoined, "Workers did not terminate.");
            foreach (var item in rendezvous) item.Dispose();
            Assert.AreEqual(0, failures.Count, string.Join("\n", failures.Select(x => x.ToString())));
            Assert.AreEqual(groups, advances);
            for (int group = 0; group < groups; group++)
            {
                Assert.AreEqual(parallelism, peaks[group], "Unexpected maximum concurrent calls.");
                for (int board = 0; board < boards; board++)
                    Assert.AreEqual(1, calls[group, board], "Meter was skipped or called more than once.");
            }
        }
    }
}