From c8b5732046e2b1b246c5a04c58b8669208090c93 Mon Sep 17 00:00:00 2001 From: Michal Buzik Date: Thu, 10 Sep 2026 22:09:30 +0200 Subject: [PATCH] Fix Genesis scheduling and synchronization for up to 10 workers (cherry picked from commit 684513ee593a47561310f69b5f405b2ff4de2504) --- .../GenesisCommunicationForm.cs | 10 +- TBFTests/GenesisParallelSchedulingTests.cs | 213 ++++++++++++++++++ 2 files changed, 220 insertions(+), 3 deletions(-) create mode 100644 TBFTests/GenesisParallelSchedulingTests.cs diff --git a/TBF/Rig/TestMethods/GenesisCommunication/GenesisCommunicationForm.cs b/TBF/Rig/TestMethods/GenesisCommunication/GenesisCommunicationForm.cs index f4d84edf7..4205662da 100644 --- a/TBF/Rig/TestMethods/GenesisCommunication/GenesisCommunicationForm.cs +++ b/TBF/Rig/TestMethods/GenesisCommunication/GenesisCommunicationForm.cs @@ -711,7 +711,8 @@ namespace TBF.Rig.TestMethods.GenesisCommunication } //HOLD ON - if activity is like HoldSlotStr, ignore loop and do activity like HoldSlotStr - if (currentActivity.ToLower().Equals(HoldSlotStr.ToLower())) + bool isHoldActivity = currentActivity.ToLower().Equals(HoldSlotStr.ToLower()); + if (isHoldActivity) { log.Debug($"Activity '{currentActivity}' is HoldSlotStr, skipping loop and performing HoldOn operation. Thread: {threadID}"); @@ -720,7 +721,6 @@ namespace TBF.Rig.TestMethods.GenesisCommunication string resultStr = string.Empty; error = HoldOn(threadID, ref resultStr); // wait for the hold on to complete - one time for each thread ProcessWaterMetersDialogConfirmation(threadID, resultStr,error); - return; } for (int group = 1; group <= lastGroup; group++) @@ -733,6 +733,8 @@ namespace TBF.Rig.TestMethods.GenesisCommunication if (stopWorkerThreads) break; + if (!isHoldActivity) + { #if TURA_SPECIAL int threadIx = threadID; /// Just one thread for TURA_SPECIAL #else @@ -890,6 +892,8 @@ namespace TBF.Rig.TestMethods.GenesisCommunication if (stopWorkerThreads) break; } + } + if (stopWorkerThreads) break; log.DebugFormat("Genesis worker group completed: Activity={0}, Group2={1}, Worker={2}", activityStep, group, threadID); OnCommCompleted(null, new CommCompletedEventArgs(threadID, -1, null, null, string.Empty, CommErr.None) @@ -1639,7 +1643,7 @@ namespace TBF.Rig.TestMethods.GenesisCommunication } else { - log.Debug("Wait for finish workerThreads 4 seconds"); + log.Debug("Waiting for all Genesis workers; results remain visible for 4 seconds afterwards."); Thread[] finishingThreads = workerThreads.ToArray(); diff --git a/TBFTests/GenesisParallelSchedulingTests.cs b/TBFTests/GenesisParallelSchedulingTests.cs new file mode 100644 index 000000000..096ab752c --- /dev/null +++ b/TBFTests/GenesisParallelSchedulingTests.cs @@ -0,0 +1,213 @@ +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."); + } + } + } +}