diff --git a/TBF/Rig/TestMethods/GenesisCommunication/CommCompletedEventArgs.cs b/TBF/Rig/TestMethods/GenesisCommunication/CommCompletedEventArgs.cs
index 00e37280d..052fda168 100644
--- a/TBF/Rig/TestMethods/GenesisCommunication/CommCompletedEventArgs.cs
+++ b/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
{
+ /// Tracks one completion per worker, activity and group.
+ public sealed class GenesisWorkerGroupCompletion
+ {
+ private readonly int workerCount;
+ private readonly System.Collections.Generic.HashSet completed = new System.Collections.Generic.HashSet();
+ 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 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;
diff --git a/TBF/Rig/TestMethods/GenesisCommunication/GenesisCommunicationForm.cs b/TBF/Rig/TestMethods/GenesisCommunication/GenesisCommunicationForm.cs
index 1abcb57e4..b73f9f08f 100644
--- a/TBF/Rig/TestMethods/GenesisCommunication/GenesisCommunicationForm.cs
+++ b/TBF/Rig/TestMethods/GenesisCommunication/GenesisCommunicationForm.cs
@@ -204,14 +204,14 @@ namespace TBF.Rig.TestMethods.GenesisCommunication
static IList multiTestParams;
- static int currentActivityStep;
- static int currentGroup; /// form -> worker thread (0 = none)
+ static volatile int currentActivityStep;
+ 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 workerThreads;
static IList muxBrdOrGroup14Nrs;
- static bool stopWorkerThreads; /// form -> worker thread
+ static volatile bool stopWorkerThreads; /// form -> worker thread
/// Parameterless constructor (without watermeters, threads)
@@ -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
- ///
- /// Branch
- ///
- lock (this)
- {
- if (++completedCommCount < 4) return;
- completedCommCount = 0;
- }
-#endif
+ // Per-meter results update the UI only. Advance after every worker
+ // has finished all of its Group 1 assignments for this Group 2.
+ 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);
+
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;
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.");
+ }
+ }
+ }
+}
diff --git a/TBFTests/TBFTests.csproj b/TBFTests/TBFTests.csproj
index c65733cf5..65ccc1f2d 100644
--- a/TBFTests/TBFTests.csproj
+++ b/TBFTests/TBFTests.csproj
@@ -231,7 +231,7 @@
-->
-
+