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

add scheduler method; fixing buffer parallel errors

Browse Source
This commit is contained in:
Youmu Liu
2025-11-19 18:09:25 +08:00
parent 45d1d08af8
commit 04fc55d31e
11 changed files with 272 additions and 339 deletions
Download Patch File Download Diff File Expand all files Collapse all files
StgSharp.Mathematics/Numeric/MatrixParallelWrap.cs
+119 -25
View File
@@ -44,10 +44,13 @@ namespace StgSharp.Mathematics.Numeric
internal MatrixParallelWrap(MatrixParallelThread[] threads, int begin, int count)
{
WrapTaskCapacity = count;
_threads = threads;
_begin = begin;
Count = threads.Length;
_threads[_begin].Role = 1;
for (int i = _begin; i < _begin + WrapTaskCapacity; i++) {
_threads[i].BindedWrap = this;
}
}
public MatrixParallelWrap()
@@ -63,15 +66,11 @@ namespace StgSharp.Mathematics.Numeric
internal int Magic { get; set; }
internal ManualResetEventSlim SchedulerReset { get; } = new(false);
private int Count { get; set; }
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void SetAllWorker()
{
foreach (MatrixParallelThread t in _threads) {
t.Restart();
for (int i = _begin; i < _begin + WrapTaskCapacity; i++) {
_threads[i].Restart();
}
}
@@ -79,25 +78,124 @@ namespace StgSharp.Mathematics.Numeric
{
while (tasks.TryDequeue(out nint handle))
{
MatrixParallelTaskPackageNonGeneric* task = (MatrixParallelTaskPackageNonGeneric*)handle;
PublishTaskToWorkers(task);
MatrixParallelTaskPackage* task = (MatrixParallelTaskPackage*)handle;
switch (task->ComputeMode.OperationStyle)
{
case MatrixOperationStyle.PANEL_OP:
break;
case MatrixOperationStyle.BUFFER_OP:
PublishBufferTaskToWorkers(task);
break;
case MatrixOperationStyle.KERNEL_OP:
PublishBufferTaskToWorkers(task);
break;
default:
break;
}
}
}
internal void PublishTaskToWorkers(MatrixParallelTaskPackageNonGeneric* taskPackage)
internal void SetScheduler()
{
int primCount = taskPackage->PrimCount;
int secCount = taskPackage->SecCount;
MatrixParallelThread thread = _threads[_begin];
thread.Restart();
}
int primAverage = primCount / Count;
int secAverage = secCount / Count;
bool shouldSliceSec = secAverage > 512;
for (int i = _begin; i < _begin + Count; i++)
private void PublishBufferTaskToWorkers(MatrixParallelTaskPackage* package)
{
long count = Unsafe.As<int, long>(ref package->PrimCount);
long recommendCount = count / 1024;
MatrixParallelThread[] pool = _threads;
if (pool.Length == 1)
{
Queue<IntPtr> tasks = _threads[i].Tasks;
MatrixParallelTaskPackageNonGeneric* subTask = MatrixParallelFactory.FromExistPackage(taskPackage);
subTask->PrimTileOffset = i * primAverage;
pool[0].Tasks.Enqueue((nint)package);
return;
}
long actual = count / pool.Length;
long extra = count % pool.Length;
long offset = 0;
MatrixParallelTaskPackage* current = package;
foreach (MatrixParallelThread thread in pool)
{
long wrapTaskLength = actual;
long ex = extra > 0 ? 1 : 0;
extra -= ex;
ref long offsetRef = ref Unsafe.As<int, long>(ref current->PrimTileOffset);
offsetRef += offset;
ref long lengthRef = ref Unsafe.As<int, long>(ref current->PrimCount);
lengthRef = wrapTaskLength + ex;
pool[0].Tasks.Enqueue((nint)current);
current = MatrixParallelFactory.FromExistPackage(current);
offset += wrapTaskLength;
}
}
private void PublishKernelTaskToWorkers(MatrixParallelTaskPackage* package)
{
int primCount = package->PrimCount;
int secCount = package->SecCount;
MatrixParallelThread[] pool = _threads;
if (pool.Length == 1)
{
pool[0].Tasks.Enqueue((nint)package);
return;
}
throw new NotImplementedException();
/*
int threadCount = pool.Length;
int tilePerThreadLess = tileCount / threadCount; // tiles per thread without extra slice
int columnAfterBaseTile = tileCount % threadCount * tileWidth; // tile sliced into columns
int extraColumnPerThread = columnAfterBaseTile / threadCount;
int extraColumnCount = columnAfterBaseTile % threadCount; // columns to be sliced
int total = primCount;
Stack<MatrixParallelWrap> taskBunch = new Stack<MatrixParallelWrap>(threadCount);
MatrixParallelTaskPackageNonGeneric* current = package;
foreach (MatrixParallelWrap wrap in pool.Wraps)
{
bool hasSlice = columnAfterBaseTile > 0;
bool isLargeSlice = extraColumnCount > 0;
// _count tiles for wrap
int wrapWidth = wrap!.WrapTaskCapacity;
int columnFromSlicingCount = (hasSlice ? extraColumnPerThread : 0) * wrapWidth;
int baseColumn = wrapWidth * tilePerThreadLess * tileWidth;
int columnInWrap = baseColumn + columnFromSlicingCount + (isLargeSlice ? 1 : 0);
// set tasks
if (columnInWrap == 0)
{
#if DEBUG
throw new InvalidOperationException("Zero tasks published.");
#else
break;
#endif
}
// update current task
current->PrimColumnCountInTile = columnInWrap;
current->SecColumnCountInTile = current->SecCount;
current->SecTileOffset = 0;
// remove tiles _count
extraColumnCount--;
columnAfterBaseTile -= columnFromSlicingCount;
total -= columnInWrap;
// set to wrap and push
wrap.WrapTask = current;
if (total <= 0)
{
break;
}
current = MatrixParallelFactory.FromExistPackage(current);
current->PrimTileOffset += columnInWrap;
}
/**/
return;
}
public class Handle(int inWrapIndex, MatrixParallelWrap wrap)
@@ -121,15 +219,11 @@ namespace StgSharp.Mathematics.Numeric
#region wrap task
internal MatrixParallelTaskPackageNonGeneric* WrapTask { get; set; }
internal unsafe Queue<nint> WrapTask { get; set; } = [];
internal int WrapTaskSliceCount { get; set; }
internal int WrapTaskCapacity
{
[MethodImpl(MethodImplOptions.AggressiveInlining)]
get => _threads.Length;
}
internal int WrapTaskCapacity { get ; init; }
#endregion
}