Fix Genesis scheduling and synchronization for up to 10 workers

TBF/Rig/TestMethods/GenesisCommunication/CommCompletedEventArgs.cs

@@ -1,4 +1,4 @@
-///
+///
 /// Copyright (c) 2015-2019 Sensus Metering Systems
 /// Author: Milan Hanajík
 ///
@@ -7,8 +7,44 @@ using TBF.Rig.RegisterReaders.GenesisRegReader.implementations;
 
 namespace TBF.Rig.TestMethods.GenesisCommunication
 {
+    /// <summary>Tracks one completion per worker, activity and group.</summary>
+    public sealed class GenesisWorkerGroupCompletion
+    {
+        private readonly int workerCount;
+        private readonly System.Collections.Generic.HashSet<int> completed = new System.Collections.Generic.HashSet<int>();
+        private int activity, group;
+        private bool released;
+        public GenesisWorkerGroupCompletion(int workerCount)
+        {
+            if (workerCount < 1 || workerCount > 10) throw new ArgumentOutOfRangeException(nameof(workerCount));
+            this.workerCount = workerCount;
+        }
+        public void Begin(int activityStep, int groupNumber)
+        {
+            lock (completed) { activity = activityStep; group = groupNumber; released = false; completed.Clear(); }
+        }
+        public bool Complete(int activityStep, int groupNumber, int worker)
+        {
+            lock (completed)
+            {
+                if (released || activityStep != activity || groupNumber != group || worker < 0 || worker >= workerCount) return false;
+                if (!completed.Add(worker) || completed.Count != workerCount) return false;
+                released = true;
+                return true;
+            }
+        }
+        public static System.Collections.Generic.IEnumerable<int> BoardIndexes(int worker, int workers, int boards)
+        {
+            if (workers < 1 || workers > 10 || worker < 0 || worker >= workers || boards < 0) throw new ArgumentOutOfRangeException();
+            for (int index = worker; index < boards; index += workers) yield return index;
+        }
+    }
+
     public class CommCompletedEventArgs : EventArgs
     {
+        public bool WorkerGroupCompleted;
+        public int ActivityStep;
+        public int Group;
         public int ThreadId;
         public int WMNr0;                   /// 0-based water meter position
         public GenesisSmartReader Ihead;

TBF/Rig/TestMethods/GenesisCommunication/GenesisCommunicationForm.cs

@@ -204,14 +204,14 @@ namespace TBF.Rig.TestMethods.GenesisCommunication
 		static IList<iPerlCommunicationParams> multiTestParams;
 
 
-		static int currentActivityStep;
+		static volatile int currentActivityStep;
-		static int currentGroup;                    /// form -> worker thread (0 = none)
+		static volatile int currentGroup;                    /// form -> worker thread (0 = none)
         static int lastGroup;
-        static int completedCommCount;              /// Number of completed communication steps
+        private GenesisWorkerGroupCompletion groupCompletion;
 
         static IList<Thread> workerThreads;
         static IList<int> muxBrdOrGroup14Nrs;
-        static bool stopWorkerThreads;				/// form -> worker thread
+        static volatile bool stopWorkerThreads;				/// form -> worker thread
 
 
 		/// <summary> Parameterless constructor (without watermeters, threads) </summary>
@@ -384,7 +384,6 @@ namespace TBF.Rig.TestMethods.GenesisCommunication
 			///
 			currentActivityStep = 0;
 			currentGroup = 0;
-			completedCommCount = 0;
 			stopWorkerThreads = false;
 
             /// group numbers are >=1, lastGroup == 0 means there is no group
@@ -409,6 +408,9 @@ namespace TBF.Rig.TestMethods.GenesisCommunication
                 if (!muxBrdOrGroup14Nrs.Contains(iPerl.MuxBoardNrOrGroup14)) muxBrdOrGroup14Nrs.Add(iPerl.MuxBoardNrOrGroup14);
             }
 
+            groupCompletion = new GenesisWorkerGroupCompletion(workerThreads.Count);
+            groupCompletion.Begin(0, 1);
+            log.InfoFormat("Genesis workers configured: Threads={0}, Group1Count={1}, LastGroup2={2}", workerThreads.Count, muxBrdOrGroup14Nrs.Count, lastGroup);
             log.WarnFormat("nrThreads = {0}", workerThreads.Count);
 		}
 
@@ -666,6 +668,8 @@ namespace TBF.Rig.TestMethods.GenesisCommunication
 
 			for (int i = 0; i < multiTestParams.Count; i++ )
 			{
+                while (activityStep != currentActivityStep && !stopWorkerThreads) Thread.Sleep(50);
+                if (stopWorkerThreads) break;
                 Test currentTest = tests[i];
 				iPerlCommunicationParams currentTestParams = multiTestParams[i];
                 string currentActivity = currentTestParams.Activity;	/// Current activity
@@ -707,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}");
@@ -716,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++)
@@ -729,10 +733,12 @@ namespace TBF.Rig.TestMethods.GenesisCommunication
 
 					if (stopWorkerThreads) break;
 
+                    if (!isHoldActivity)
+                    {
 #if TURA_SPECIAL
                     int threadIx = threadID;        /// Just one thread for TURA_SPECIAL
 #else
-					for (int threadIx = threadID; threadIx < threadID + 10; threadIx += cfg.NrThreads)// - max threads = 4 => 10
+					foreach (int threadIx in GenesisWorkerGroupCompletion.BoardIndexes(threadID, workerThreads.Count, muxBrdOrGroup14Nrs.Count))
 #endif
                     {
 						bool wmFound = false;
@@ -884,8 +890,13 @@ namespace TBF.Rig.TestMethods.GenesisCommunication
 						if (stopWorkerThreads) break;
 					}
 
-					if (stopWorkerThreads) break;
+                    }
-				} /// for (int group
+
+                    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)
+                    { WorkerGroupCompleted = true, ActivityStep = activityStep, Group = group });
+                } /// for (int group
 
 				TBF.UiBridge.Bridge.OnTestProgress(null, new TBF.UiBridge.TestProgressEventArgs(tests[i], Progress.Completed));
 				activityStep++;
@@ -1608,21 +1619,17 @@ namespace TBF.Rig.TestMethods.GenesisCommunication
                     }
                 }
 
-#if !TURA_SPECIAL
+                // Per-meter results update the UI only. Advance after every worker
-                ///
+                // has finished all of its Group 1 assignments for this Group 2.
-                /// Branch
+                if (!data.WorkerGroupCompleted || !groupCompletion.Complete(data.ActivityStep, data.Group, data.ThreadId)) return;
-                ///
+                log.InfoFormat("Genesis group completed: Activity={0}, Group2={1}, Workers={2}", data.ActivityStep, data.Group, workerThreads.Count);
-                lock (this)
+
-                {
-                    if (++completedCommCount < 4) return;
-                    completedCommCount = 0;
-                }
-#endif
 
                 if (currentGroup < lastGroup)
                 {
                     /// Go to the next step / next group
                     currentGroup++;
+                    groupCompletion.Begin(currentActivityStep, currentGroup);
                 }
                 else if (currentActivityStep + 1 < multiTestParams.Count)
                 {
@@ -1630,16 +1637,17 @@ namespace TBF.Rig.TestMethods.GenesisCommunication
                     currentActivityStep++;
                     activityLabel.Text = multiTestParams[currentActivityStep].Activity;
                     currentGroup++;
+                    groupCompletion.Begin(currentActivityStep, currentGroup);
                 }
                 else
                 {
-	                log.Debug("Wait for finish workerThreads 4 seconds");
+	                log.Debug("Waiting for all Genesis workers; results remain visible for 4 seconds afterwards.");
 
-	                Thread lastThread = workerThreads[data.ThreadId];
+	                Thread[] finishingThreads = workerThreads.ToArray();
 
 	                Task.Run(() =>
 	                {
-		                lastThread.Join(2000);
+		                foreach (var thread in finishingThreads) thread.Join();
 
 		                if (!CanUseUi()) return;
 

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

TBFTests/TBFTests.csproj

@@ -231,7 +231,7 @@
   <Target Name="AfterBuild">
   </Target>
   -->
-  <ItemGroup><Compile Include="GenesisRecoveryTests.cs" /></ItemGroup>
+  <ItemGroup><Compile Include="GenesisRecoveryTests.cs" /><Compile Include="GenesisParallelSchedulingTests.cs" /></ItemGroup>
   <!-- Project dependencies may copy an older SQLite interop DLL with a newer timestamp. -->
   <Target Name="EnsureMatchingSQLiteInterop" AfterTargets="Build">
     <Copy SourceFiles="@(SQLiteInteropFiles)"