Arrange code structure. Introduce dynamic scheduler and multi leader.

This commit is contained in:
Youmu Liu
2025-10-21 18:06:20 +08:00
parent b26ec45433
commit 91a0add1d1
3 changed files with 40 additions and 64 deletions
+13 -19
View File
@@ -81,8 +81,6 @@ namespace StgSharp.Mathematics.Numeric
public static bool IsParallelMeaningful => Environment.ProcessorCount > 6 && IsInitialized; public static bool IsParallelMeaningful => Environment.ProcessorCount > 6 && IsInitialized;
public static MatrixParallelWrap.Handle LeaderHandle { get; private/**/ set; }
private static bool IsInitialized { get; set; } = false; private static bool IsInitialized { get; set; } = false;
private static Queue<IntPtr> LeaderTask { get; set; } private static Queue<IntPtr> LeaderTask { get; set; }
@@ -126,38 +124,34 @@ namespace StgSharp.Mathematics.Numeric
{ {
if (count == 3) if (count == 3)
{ {
Wraps = [ new MatrixParallelWrap(0, count, true) ]; Pool = new(3, [ new MatrixParallelWrap(0, count) ]);
(LargeWrapCount, SmallWrapCount) = (0, 1); (LargeWrapCount, SmallWrapCount) = (0, 1);
LeaderHandle = new MatrixParallelWrap.Handle(0, Wraps.Peek());
} else } else
{ {
count = 4; count = 4;
Wraps = [ new MatrixParallelWrap(0, count, true) ]; Pool = new(4, [ new MatrixParallelWrap(0, count) ]);
(LargeWrapCount, SmallWrapCount) = (1, 0); (LargeWrapCount, SmallWrapCount) = (1, 0);
LeaderHandle = new MatrixParallelWrap.Handle(0, Wraps.Peek());
} }
} else if (count == 6) } else if (count == 6)
{ {
Wraps = [ Pool = new(6, [
new MatrixParallelWrap(0, 3, true), new MatrixParallelWrap(0, 3),
new MatrixParallelWrap(3, 3, false) new MatrixParallelWrap(3, 3)
]; ]);
(LargeWrapCount, SmallWrapCount) = (0, 2); (LargeWrapCount, SmallWrapCount) = (0, 2);
LeaderHandle = new MatrixParallelWrap.Handle(0, Wraps.Peek());
} else } else
{ {
(int _3wrap, int _4wrap) = FastDecompose(count); (int _3wrap, int _4wrap) = FastDecompose(count);
Wraps = new(_4wrap + _3wrap); CapacityFixedStack<MatrixParallelWrap> stack = new(_4wrap + _3wrap);
int id = 0, idx = 0; int id = 0, idx = 0;
for (; idx < _3wrap; idx++, id += 3) { for (; idx < _3wrap; idx++, id += 3) {
Wraps.Push(new MatrixParallelWrap(idx, 3, id == 0)); stack.Push(new MatrixParallelWrap(idx, 3));
} }
for (; idx < _3wrap + _4wrap; idx++, id += 4) { for (; idx < _3wrap + _4wrap; idx++, id += 4) {
Wraps.Push(new MatrixParallelWrap(id, 4, id == 0)); stack.Push(new MatrixParallelWrap(id, 4));
} }
LeaderHandle = new MatrixParallelWrap.Handle(0, Wraps.Peek()); Pool = new MatrixParallelPool(count, stack);
} }
LeaderTask = LeaderHandle.Wrap.GetWorkerTaskQueue(0);
IsInitialized = true; IsInitialized = true;
} }
@@ -191,18 +185,18 @@ namespace StgSharp.Mathematics.Numeric
#region wrap field #region wrap field
private static CapacityFixedStack<MatrixParallelWrap> Wraps; private static MatrixParallelPool Pool;
private static int LargeWrapCount; private static int LargeWrapCount;
private static int SmallWrapCount; private static int SmallWrapCount;
public static CapacityFixedStack<MatrixParallelWrap> LeadParallel() public static MatrixParallelPool LeadParallel()
{ {
#if NET9_0_OR_GREATER #if NET9_0_OR_GREATER
#else #else
Monitor.Enter(_lock); Monitor.Enter(_lock);
#endif #endif
return Wraps; return Pool;
} }
public static void ReleaseParallelLeadership(ref CapacityFixedStack<MatrixParallelWrap> wraps) public static void ReleaseParallelLeadership(ref CapacityFixedStack<MatrixParallelWrap> wraps)
@@ -49,14 +49,20 @@ namespace StgSharp.Mathematics.Numeric
SpareThreads = threadCount; SpareThreads = threadCount;
} }
internal MatrixParallelPool(int threadCount, CapacityFixedStack<MatrixParallelWrap> wrapSpan)
{
SpareWrap = wrapSpan;
SingleThreadAllocation = new CapacityFixedBag<MatrixParallelWrap>(wrapSpan.Count);
SpareThreads = threadCount;
}
public int SpareThreads { get; set; } public int SpareThreads { get; set; }
internal CapacityFixedBag<MatrixParallelWrap> SingleThreadAllocation { get; set; } internal CapacityFixedBag<MatrixParallelWrap> SingleThreadAllocation { get; set; }
internal CapacityFixedStack<MatrixParallelWrap> SpareWrap { get; set; } internal CapacityFixedStack<MatrixParallelWrap> SpareWrap { get; set; }
public bool TryRequest(out ) public bool TryRequestLeaderHandle(out MatrixParallelWrap.Handle handle) { }
public bool TryRequestWrap(out MatrixParallelWrap? wrap) public bool TryRequestWrap(out MatrixParallelWrap? wrap)
{ {
@@ -38,18 +38,15 @@ namespace StgSharp.Mathematics.Numeric
public unsafe class MatrixParallelWrap public unsafe class MatrixParallelWrap
{ {
private readonly Queue<IntPtr>[] _cache;
private readonly Handle[] _resetHandle; private readonly Handle[] _resetHandle;
private readonly Thread[] _threads; private readonly Thread[] _threads;
public MatrixParallelWrap(int id, int wrapSize) public MatrixParallelWrap(int id, int wrapSize)
{ {
ID = id; ID = id;
_cache = new Queue<IntPtr>[wrapSize];
_threads = new Thread[wrapSize]; _threads = new Thread[wrapSize];
for (int i = 0; i < wrapSize; i++) for (int i = 0; i < wrapSize; i++)
{ {
_cache[i] = new Queue<IntPtr>(8);
_threads[i] = new Thread(MatrixParallelWorker) _threads[i] = new Thread(MatrixParallelWorker)
{ {
IsBackground = true, IsBackground = true,
@@ -70,21 +67,6 @@ namespace StgSharp.Mathematics.Numeric
internal ManualResetEventSlim SchedulerReset { get; } = new(false); internal ManualResetEventSlim SchedulerReset { get; } = new(false);
public Span<IntPtr> EnqueuePackage(Span<IntPtr> packageSpan)
{
int idx = 0;
for (; idx < _cache.Length; idx++) {
_cache[idx].Enqueue(packageSpan[idx]);
}
return packageSpan[idx..];
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public Queue<IntPtr> GetWorkerTaskQueue(int idInWrap)
{
return _cache[idInWrap];
}
[MethodImpl(MethodImplOptions.AggressiveInlining)] [MethodImpl(MethodImplOptions.AggressiveInlining)]
public void SetAllWorker() public void SetAllWorker()
{ {
@@ -104,6 +86,8 @@ namespace StgSharp.Mathematics.Numeric
/// </summary> /// </summary>
public int Role { get; internal set; } public int Role { get; internal set; }
public Queue<IntPtr> TaskCache { get; } = [];
internal ManualResetEventSlim ResetEvent { get; } = new(false); internal ManualResetEventSlim ResetEvent { get; } = new(false);
internal MatrixParallelWrap Wrap { get; } = wrap; internal MatrixParallelWrap Wrap { get; } = wrap;
@@ -138,33 +122,25 @@ namespace StgSharp.Mathematics.Numeric
{ {
reset.Wait(); reset.Wait();
reset.Reset(); reset.Reset();
Queue<IntPtr> taskQueue = handle.TaskCache;
self.Priority = ThreadPriority.Highest; self.Priority = ThreadPriority.Highest;
ExecuteTaskPackage(wrap.GetWorkerTaskQueue(inWrapIndex)); switch (handle.Role)
{
case 0://worker
// _ = Interlocked.Decrement(ref RemainThreadCount);
ExecuteTaskPackage(handle.TaskCache);
break;
case 1://scheduler
// TODO task publication to worker in wrap
wrap.SetAllWorker();
ExecuteTaskPackage(taskQueue);
// _ = Interlocked.Decrement(ref RemainThreadCount); break;
self.Priority = (ThreadPriority)AfterWork; case 2://leader, error because leader should be at outer thread but not here
} while (true); break;
} default:
break;
private static unsafe void MatrixParallelScheduler([AllowNull] object obj) }
{
Handle handle = (obj as Handle)!;
MatrixParallelWrap wrap = handle.Wrap;
int inWrapIndex = handle.InWrapIndex;
Queue<IntPtr> taskQueue = wrap.GetWorkerTaskQueue(inWrapIndex);
ManualResetEventSlim resetEvent = wrap.SchedulerReset;
Thread self = Thread.CurrentThread;
do
{
resetEvent.Wait();
resetEvent.Reset();
self.Priority = ThreadPriority.Highest;
// TODO task publication to worker in wrap
wrap.SetAllWorker();
ExecuteTaskPackage(taskQueue);
// _ = Interlocked.Decrement(ref RemainThreadCount);
self.Priority = (ThreadPriority)AfterWork; self.Priority = (ThreadPriority)AfterWork;
} while (true); } while (true);
} }