youmuliu/stgsharp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Arrange code structure. Introduce dynamic scheduler and multi leader.

Browse Source
This commit is contained in:
Youmu Liu
2025-10-21 16:39:34 +08:00
parent 838ce73133
commit b26ec45433
13 changed files with 388 additions and 240 deletions
Download Patch File Download Diff File Expand all files Collapse all files
StgSharp.Mathematics/Numeric/MatrixParallel.cs
+2 -170
View File
@@ -25,7 +25,7 @@
//
// -----------------------------------------------------------------------
// -----------------------------------------------------------------------
using StgSharp.Commom.Collections;
using StgSharp.Collections;
using System;
using System.Collections.Generic;
using System.Diagnostics.CodeAnalysis;
@@ -189,175 +189,6 @@ namespace StgSharp.Mathematics.Numeric
}
}
public unsafe class MatrixParallelWrap
{
private readonly Queue<IntPtr>[] _cache;
private readonly Thread[] _threads;
private readonly ManualResetEventSlim[] _workerReset;
public MatrixParallelWrap(int id, int wrapSize, bool isFirst)
{
ID = id;
int begin = isFirst ? 1 : 0, end = wrapSize - 1;
_cache = new Queue<IntPtr>[wrapSize];
_workerReset = new ManualResetEventSlim[end];
_threads = new Thread[wrapSize];
for (int i = 0; i < end; i++) {
_workerReset[i] = new ManualResetEventSlim(false);
}
for (int i = 0; i < _cache.Length; i++) {
_cache[i] = new Queue<IntPtr>(8);
}
for (int i = begin; i < end; i++)
{
_threads[i] = new Thread(MatrixParallelWorker)
{
IsBackground = true,
Priority = ThreadPriority.BelowNormal,
Name = $"MPT#{ID}_{i}"
};
_threads[i].Start(new Handle(i, this));
}
_threads[end] = new Thread(MatrixParallelScheduler)
{
IsBackground = true,
Priority = ThreadPriority.BelowNormal,
Name = $"MPT#{ID}_{end}*"
};
_threads[end].Start(new Handle(end, this));
}
public int ID { get; init; }
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 ManualResetEventSlim GetWorkerResetEvent(int id)
{
return _workerReset[id];
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public Queue<IntPtr> GetWorkerTaskQueue(int idInWrap)
{
return _cache[idInWrap];
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void SetAllWorker()
{
foreach (ManualResetEventSlim item in _workerReset) {
item.Set();
}
}
public record class Handle(int InWrapIndex, MatrixParallelWrap Wrap);
#region wrap task
internal MatrixParallelTaskPackageNonGeneric* WrapTask { get; set; }
internal int WrapTaskSliceCount { get; set; }
internal int WrapTaskCapacity
{
[MethodImpl(MethodImplOptions.AggressiveInlining)]
get => _threads.Length;
}
#endregion
#region ThreadOperation
private static unsafe void MatrixParallelWorker([AllowNull] object obj)
{
Handle handle = (obj as Handle)!;
MatrixParallelWrap wrap = handle.Wrap;
int inWrapIndex = handle.InWrapIndex;
Thread self = Thread.CurrentThread;
ManualResetEventSlim reset = wrap.GetWorkerResetEvent(inWrapIndex);
do
{
reset.Wait();
reset.Reset();
self.Priority = ThreadPriority.Highest;
ExecuteTaskPackage(wrap.GetWorkerTaskQueue(inWrapIndex));
_ = Interlocked.Decrement(ref RemainThreadCount);
self.Priority = _AfterWork;
} while (true);
}
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 = _AfterWork;
} while (true);
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public static unsafe void ExecuteTaskPackage(Queue<IntPtr> taskQueue)
{
while (taskQueue.TryDequeue(out nint handle))
{
MatrixParallelTaskPackageNonGeneric* p = (MatrixParallelTaskPackageNonGeneric*)handle;
switch (p->ComputeMode)
{
case 1:
Compute_Unary(p);
break;
case 2:
Compute_UnaryScalar(p);
break;
case 3:
Compute_Binary(p);
break;
case 4:
Compute_BinaryScalar(p);
break;
default:
break;
}
}
}
#endregion
}
public enum SleepMode
{
QuickWakeup = ThreadPriority.AboveNormal,
NormalSleep = ThreadPriority.BelowNormal,
DeepSleep = ThreadPriority.Lowest,
}
#region wrap field
private static CapacityFixedStack<MatrixParallelWrap> Wraps;
@@ -370,6 +201,7 @@ namespace StgSharp.Mathematics.Numeric
#else
Monitor.Enter(_lock);
#endif
return Wraps;
}