[rebuid]use dynamic wrapping instead of static wrapping

This commit is contained in:
Youmu Liu
2025-11-03 20:03:50 +08:00
parent ee97dfddfb
commit 9c4c71dc5b
15 changed files with 331 additions and 595 deletions
@@ -39,44 +39,13 @@ namespace StgSharp.Mathematics.Numeric
public unsafe class MatrixParallelWrap
{
private readonly Handle[] _resetHandle;
private readonly Thread[] _threads;
private readonly MatrixParallelThread[] _threads;
public MatrixParallelWrap(int id, int wrapSize)
public MatrixParallelWrap()
{
ID = id;
_threads = new Thread[wrapSize];
_resetHandle = new Handle[wrapSize];
for (int i = 0; i < wrapSize; i++)
{
_threads[i] = new Thread(MatrixParallelWorker)
{
IsBackground = true,
Priority = ThreadPriority.BelowNormal,
Name = $"MPT#{ID}_{i}"
};
Handle handle = new Handle(i, this);
_resetHandle[i] = handle;
_threads[i].Start(handle);
}
_threads = new MatrixParallelThread[5];
}
public bool IsAllAvailable
{
get
{
foreach (Handle item in _resetHandle)
{
if (item.Role != 0) {
return false;
}
}
return true;
}
}
public int ID { get; init; }
public static SleepMode AfterWork
{
get;
@@ -87,27 +56,39 @@ namespace StgSharp.Mathematics.Numeric
internal ManualResetEventSlim SchedulerReset { get; } = new(false);
private int Count { get; set; }
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void SetAllWorker()
{
foreach (Handle item in _resetHandle) {
item.ResetEvent.Set();
foreach (MatrixParallelThread t in _threads) {
t.Restart();
}
}
internal bool TryGetLeaderHandle(out Handle? handle)
internal void PublishTaskToWorkers(Queue<IntPtr> tasks)
{
foreach (Handle item in _resetHandle)
while (tasks.TryDequeue(out nint handle))
{
if (item.Role != 2)
{
handle = item;
item.Role = 2;
return true;
}
MatrixParallelTaskPackageNonGeneric* task = (MatrixParallelTaskPackageNonGeneric*)handle;
PublishTaskToWorkers(task);
}
}
internal void PublishTaskToWorkers(MatrixParallelTaskPackageNonGeneric* taskPackage)
{
int primCount = taskPackage->PrimCount;
int secCount = taskPackage->SecCount;
int primAverage = primCount / Count;
int secAverage = secCount / Count;
bool shouldSliceSec = secAverage > 512;
for (int i = 0; i < Count; i++)
{
Queue<IntPtr> tasks = _threads[i].Tasks;
MatrixParallelTaskPackageNonGeneric* subTask = MatrixParallelFactory.FromExistPackage(taskPackage);
subTask->PrimTileOffset = i * primAverage;
}
handle = null;
return false;
}
public class Handle(int inWrapIndex, MatrixParallelWrap wrap)
@@ -141,71 +122,6 @@ namespace StgSharp.Mathematics.Numeric
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 = handle.ResetEvent;
do
{
reset.Wait();
reset.Reset();
Queue<IntPtr> taskQueue = handle.TaskCache;
self.Priority = ThreadPriority.Highest;
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);
break;
case 2://leader, error because leader should be at outer thread but not here
break;
default:
break;
}
self.Priority = (ThreadPriority)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:
MatrixComputeModel.Compute_Unary(p);
break;
case 2:
MatrixComputeModel.Compute_UnaryScalar(p);
break;
case 3:
MatrixComputeModel.Compute_Binary(p);
break;
case 4:
MatrixComputeModel.Compute_BinaryScalar(p);
break;
default:
break;
}
}
}
#endregion
}
@@ -217,18 +133,4 @@ namespace StgSharp.Mathematics.Numeric
DeepSleep = ThreadPriority.Lowest,
}
public class MatrixComputeResource
{
public IEnumerable<MatrixParallelWrap> Wraps { get; set; }
public int WorkingThreadCount { get; set; }
/// <summary>
/// Closure task leader should do to collect results.
/// </summary>
public unsafe IntPtr LeaderClosurePackage { get; set; }
}
}