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(); 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."); } } } }