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

buffer op brief ok, but blocks on unkown reason

Browse Source
This commit is contained in:
Youmu Liu
2025-11-20 16:17:24 +08:00
parent a29a442723
commit c595076005
14 changed files with 434 additions and 325 deletions
Download Patch File Download Diff File Expand all files Collapse all files
StgSharp.Mathematics/Numeric/MatrixCompute.Basic.cs
+1 -1
View File
@@ -47,7 +47,7 @@ namespace StgSharp.Mathematics.Numeric
throw new IndexOutOfRangeException("Matrix dimensions must agree.");
}
long count = (long)left.KernelColumnLength * left.KernelRowLength;
Matrix<float> ans = Matrix<float>.FromDefault(left.ColumnLength, left.RowLength);
Matrix<float> ans = Matrix.FromDefault<float>(left.ColumnLength, left.RowLength);
if (count <= 256)
{
MatrixKernel<float>* leftPtr = left.Buffer;
StgSharp.Mathematics/Numeric/MatrixComputeModel.cs
+10
View File
@@ -47,6 +47,16 @@ namespace StgSharp.Mathematics.Numeric
MatrixParallelFactory.Release(p);
}
internal static unsafe void BufferComputeNoOperatorScalar(MatrixParallelTaskPackage* p)
{
int size = p->ElementSize * 16;
long count = Unsafe.As<int, long>(ref p->PrimCount);
long offset = Unsafe.As<int, long>(ref p->PrimTileOffset) * size;
((delegate* unmanaged[Cdecl]<void*, ScalarPacket*,long, void>)p->ComputeHandle)((void*)(p->Result + offset),
p->Scalar, count);
MatrixParallelFactory.Release(p);
}
internal static unsafe void BufferComputeUnaryScalar(MatrixParallelTaskPackage* p)
{
int size = p->ElementSize * 16;
StgSharp.Mathematics/Numeric/MatrixParallel.Publish.cs
+17 -7
View File
@@ -67,17 +67,21 @@ namespace StgSharp.Mathematics.Numeric
int primCount = package->PrimCount;
int secCount = package->SecCount;
long count = (long)primCount * secCount;
Console.WriteLine($"amount of kernels is {count}");
long recommendCount = count / 1024;
MatrixParallelThread[] pool = LeadParallel((int)recommendCount);
MatrixParallelHandle handle = new MatrixParallelHandle(pool);
if (handle.Wraps.Count() == 1)
if (handle.Wraps.Length == 1)
{
// Console.WriteLine($"Scheduling into {pool.Length} threads.");
ref long offsetRef = ref Unsafe.As<int, long>(ref package->PrimTileOffset);
offsetRef = 0;
ref long lengthRef = ref Unsafe.As<int, long>(ref package->PrimCount);
lengthRef = count;
handle.Wraps.First().WrapTask.Enqueue((nint)package);
handle.Wraps[0].WrapTask.Enqueue((nint)package);
// Console.WriteLine($"Scheduling {count} kernels into thread");
return handle;
}
long actual = count / pool.Length;
@@ -85,18 +89,25 @@ namespace StgSharp.Mathematics.Numeric
long offset = 0;
MatrixParallelTaskPackage* current = package;
// Console.WriteLine($"Scheduling into {pool.Length} threads.");
foreach (MatrixParallelWrap wrap in handle.Wraps)
{
long wrapTaskLength = actual * wrap.WrapTaskCapacity;
long ex = Math.Min(extra, wrap.WrapTaskCapacity);
long length = wrapTaskLength + ex;
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;
lengthRef = length;
wrap.WrapTask.Enqueue((nint) current);
Console.WriteLine($"Offset is {offset}");
Console.WriteLine($"Scheduling {length} kernels into {wrap.WrapTaskCapacity} threads");
Console.WriteLine();
current = MatrixParallelFactory.FromExistPackage(current);
offset += wrapTaskLength;
offset += length;
}
return handle;
}
@@ -117,9 +128,9 @@ namespace StgSharp.Mathematics.Numeric
}
/**/
MatrixParallelHandle handle = new MatrixParallelHandle(pool);
if (handle.Wraps.Count() == 1)
if (handle.Wraps.Length == 1)
{
handle.Wraps.First().WrapTask.Enqueue((nint)package);
handle.Wraps[0].WrapTask.Enqueue((nint)package);
return handle;
}
int threadCount = pool.Length;
@@ -129,7 +140,6 @@ namespace StgSharp.Mathematics.Numeric
int extraColumnCount = columnAfterBaseTile % threadCount; // columns to be sliced
int total = primCount;
Stack<MatrixParallelWrap> taskBunch = new Stack<MatrixParallelWrap>(threadCount);
MatrixParallelTaskPackage* current = package;
foreach (MatrixParallelWrap wrap in handle.Wraps)
{
StgSharp.Mathematics/Numeric/MatrixParallel.cs
+1 -2
View File
@@ -42,7 +42,6 @@ namespace StgSharp.Mathematics.Numeric
private static CapacityFixedStack<MatrixParallelThread> _threads;
private static volatile int RemainThreadCount = 0;
private static readonly
#if NET10_0_OR_GREATER
Lock
@@ -68,13 +67,13 @@ namespace StgSharp.Mathematics.Numeric
public static void LaunchParallel(MatrixParallelHandle handle, SleepMode afterWork)
{
RemainThreadCount = _threads.Count;
_AfterWork = (ThreadPriority)afterWork;
MatrixParallelThread leader = handle.GetLeader();
foreach (MatrixParallelWrap w in handle.Wraps) {
w.SetScheduler();
}
leader.MatrixParallelLeader();
handle.WaitAllWraps();
}
internal static void Init()
StgSharp.Mathematics/Numeric/MatrixParallelHandle.cs
+44 -33
View File
@@ -28,6 +28,7 @@
using StgSharp.Collections;
using System;
using System.Collections.Generic;
using System.ComponentModel.DataAnnotations;
using System.Diagnostics.CodeAnalysis;
using System.Linq;
using System.Runtime.CompilerServices;
@@ -36,7 +37,7 @@ using System.Threading.Tasks;
namespace StgSharp.Mathematics.Numeric
{
public struct MatrixParallelHandle : IDisposable
public class MatrixParallelHandle : IDisposable
{
private static readonly (int a, int b)[] LookupTable = [
@@ -55,10 +56,11 @@ namespace StgSharp.Mathematics.Numeric
(1, 4), // 19
(4, 2), // 20
];
private MatrixParallelThread[] _threads;
private MatrixParallelWrap[] _wraps;
private bool _disposed = false;
private CapacityFixedStack<MatrixParallelWrap> _wraps;
private int _count;
private ManualResetEventSlim _leaderEvent = new ManualResetEventSlim(false);
internal MatrixParallelHandle(MatrixParallelThread[] threadArray)
{
@@ -78,7 +80,7 @@ namespace StgSharp.Mathematics.Numeric
}
}
internal IEnumerable<MatrixParallelWrap> Wraps
internal ReadOnlySpan<MatrixParallelWrap> Wraps
{
get
{
@@ -89,21 +91,8 @@ namespace StgSharp.Mathematics.Numeric
}
}
internal ReadOnlySpan<MatrixParallelThread> ThreadsUnsafe
{
[MethodImpl(MethodImplOptions.AggressiveInlining)]
get => _threads;
}
private bool IsMultithread { get; init; }
public MatrixParallelThread DefineLeader()
{
MatrixParallelThread leader = _threads[1];
leader.Role = 2;
return leader;
}
public void Dispose()
{
if (_disposed) {
@@ -115,8 +104,27 @@ namespace StgSharp.Mathematics.Numeric
public MatrixParallelThread GetLeader()
{
Threads[0].Role = 2;
return Threads[0];
MatrixParallelThread leader = Threads[1];
leader.Role = MatrixParallelThread.RoleType.Leader;
leader.LeaderEvent = _leaderEvent;
return leader;
}
public void WaitAllWraps()
{
if (_disposed) {
throw new ObjectDisposedException(typeof(MatrixParallelHandle).Name);
}
SpinWait.SpinUntil(() => Volatile.Read(ref _count) == 0);
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
internal void ReportWrapAccomplished()
{
if (_disposed) {
throw new ObjectDisposedException(typeof(MatrixParallelHandle).Name);
}
Interlocked.Decrement(ref _count);
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
@@ -163,23 +171,26 @@ namespace StgSharp.Mathematics.Numeric
int count = _threads.Length;
if (count < 6)
{
MatrixParallelWrap wrap = new MatrixParallelWrap(_threads, 0, count);
MatrixParallelWrap wrap = new MatrixParallelWrap(this, 0, count);
_wraps = [ wrap ];
return;
}
(int c3,int c4) = FastDecompose(_threads.Length);
_wraps = new(c3 + c4);
int begin = 0;
for (int i = 0; i < c3; i++)
} else
{
_wraps.Push(new MatrixParallelWrap(_threads, begin, 3));
begin += 3;
}
for (int i = 0; i < c4; i++)
{
_wraps.Push(new MatrixParallelWrap(_threads, begin, 4));
begin += 4;
(int c3, int c4) = FastDecompose(_threads.Length);
_wraps = new MatrixParallelWrap[c3 + c4];
int begin = 0;
int i = 0;
for (; i < c3; i++)
{
_wraps[i] = new MatrixParallelWrap(this, begin, 3);
begin += 3;
}
for (; i < c3 + c4; i++)
{
_wraps[i] = new MatrixParallelWrap(this, begin, 4);
begin += 4;
}
}
_count = _wraps.Length;
}
}
StgSharp.Mathematics/Numeric/MatrixParallelTaskPackage.cs
+8 -8
View File
@@ -119,19 +119,19 @@ namespace StgSharp.Mathematics.Numeric
#region right enum
[FieldOffset(32)]public int RightPrimOffset; // 4
[FieldOffset(36)]public int RightPrimStride; // 4
[FieldOffset(40)]public int RightSecOffset; // 4
[FieldOffset(44)]public int RightSecStride; // 4
[FieldOffset(48)]public int RightPrimOffset; // 4
[FieldOffset(52)]public int RightPrimStride; // 4
[FieldOffset(56)]public int RightSecOffset; // 4
[FieldOffset(60)]public int RightSecStride; // 4
#endregion
#region ans enum
[FieldOffset(48)]public int ResultPrimOffset; // 4
[FieldOffset(52)]public int ResultPrimStride; // 4
[FieldOffset(56)]public int ResultSecOffset; // 4
[FieldOffset(60)] public int ResultSecStride; // 4
[FieldOffset(64)]public int ResultPrimOffset; // 4
[FieldOffset(68)]public int ResultPrimStride; // 4
[FieldOffset(72)]public int ResultSecOffset; // 4
[FieldOffset(76)] public int ResultSecStride; // 4
#endregion
StgSharp.Mathematics/Numeric/MatrixParallelThread.cs
+42 -21
View File
@@ -72,6 +72,8 @@ namespace StgSharp.Mathematics.Numeric
set => field = value;
}
internal ManualResetEventSlim? LeaderEvent { get; set; }
public void Dispose()
{
Dispose(disposing:true);
@@ -102,6 +104,7 @@ namespace StgSharp.Mathematics.Numeric
case MatrixOperationParam.LEFT_RIGHT_ANS_SCALAR_PARAM:
break;
case MatrixOperationParam.ANS_SCALAR_PARAM:
MatrixComputeModel.BufferComputeNoOperatorScalar(p);
break;
default:
break;
@@ -135,21 +138,30 @@ namespace StgSharp.Mathematics.Numeric
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void Restart()
{
_resetEvent.Set();
if (Role == RoleType.Leader)
{
LeaderEvent?.Set();
} else
{
_resetEvent.Set();
}
}
internal void MatrixParallelLeader()
{
Thread current = Thread.CurrentThread;
current.Priority = ThreadPriority.Highest;
Thread.Sleep(0);
LeaderEvent?.Wait();
LeaderEvent?.Reset();
// Console.WriteLine("Leader is computing");
#if WINDOWS
if (!UseUmaAffinityPath)
{
// NUMA/affinity path: bind to this thread's reserved mask, then restore
nuint previousMask = SetThreadAffinityMask(GetCurrentThread(), _affinity.Mask);
bool affinityChanged = previousMask != 0; // nonzero => success
Thread.Sleep(0);
ExecuteTaskPackage(Tasks);
if (affinityChanged) {
_ = SetThreadAffinityMask(GetCurrentThread(), previousMask);
@@ -164,7 +176,11 @@ namespace StgSharp.Mathematics.Numeric
ExecuteTaskPackage(Tasks);
#endif
this.Role = 1;
BindedWrap.ReportThreadAccomplished();
// Console.WriteLine("leader's computing is done");
Role = RoleType.Worker;
LeaderEvent = null;
current.Priority = ThreadPriority.Normal;
}
@@ -228,16 +244,27 @@ namespace StgSharp.Mathematics.Numeric
_resetEvent.Reset();
switch (Role)
{
case 0://worker
case RoleType.Worker://worker
// _ = Interlocked.Decrement(ref RemainThreadCount);
// Console.WriteLine("worker start");
ExecuteTaskPackage(Tasks);
BindedWrap.ReportThreadAccomplished();
// Console.WriteLine("worker end");
break;
case 1://scheduler
BindedWrap.PublishTaskToWorkers(BindedWrap.WrapTask);
case RoleType.Scheduler://scheduler
Console.WriteLine("scheduler start");
BindedWrap.PublishTaskToWorkers();
BindedWrap.SetAllWorker();
Console.WriteLine("scheduler done");
ExecuteTaskPackage(Tasks);
BindedWrap.ReportThreadAccomplished();
BindedWrap.WaitAllWorker();
BindedWrap.ReportWrapAccomplished();
Role = RoleType.Worker;
break;
case 2://leader, error because leader should be at outer thread but not here
case RoleType.Leader://leader, error because leader should be at outer thread but not here
break;
default:
break;
@@ -389,21 +416,15 @@ namespace StgSharp.Mathematics.Numeric
internal Queue<IntPtr> Tasks { get; } = [];
public int Role
public RoleType Role { get; internal set; }
public enum RoleType : int
{
get;
internal set
{
if ((Role & (~3u)) == 0)
{
field = value;
} else
{
#if DEBUG
throw new ArgumentOutOfRangeException(nameof(Role));
#endif
}
}
Worker = 0,
Scheduler = 1,
Leader = 2,
}
#endregion
StgSharp.Mathematics/Numeric/MatrixParallelWrap.cs
+56 -46
View File
@@ -30,6 +30,7 @@ using System;
using System.Collections.Generic;
using System.Diagnostics.CodeAnalysis;
using System.Linq;
using System.Net.WebSockets;
using System.Runtime.CompilerServices;
using System.Text;
using System.Threading.Tasks;
@@ -39,23 +40,22 @@ namespace StgSharp.Mathematics.Numeric
public unsafe class MatrixParallelWrap
{
private readonly MatrixParallelThread[] _threads;
private int _begin;
internal MatrixParallelWrap(MatrixParallelThread[] threads, int begin, int count)
private int _remainThreadCount;
private readonly MatrixParallelHandle _handle;
internal MatrixParallelWrap(MatrixParallelHandle handle, int begin, int count)
{
WrapTaskCapacity = count;
_threads = threads;
_handle = handle;
_begin = begin;
_threads[_begin].Role = 1;
for (int i = _begin; i < _begin + WrapTaskCapacity; i++) {
_threads[i].BindedWrap = this;
for (int i = 0; i < WrapTaskCapacity; i++)
{
Threads[i].Role = MatrixParallelThread.RoleType.Worker;
Threads[i].BindedWrap = this;
}
}
public MatrixParallelWrap()
{
_threads = new MatrixParallelThread[5];
Threads[0].Role = MatrixParallelThread.RoleType.Scheduler;
}
public static SleepMode AfterWork
@@ -66,19 +66,41 @@ namespace StgSharp.Mathematics.Numeric
internal int Magic { get; set; }
private ReadOnlySpan<MatrixParallelThread> Threads
{
[MethodImpl(MethodImplOptions.AggressiveInlining)]
get => _handle.Threads[_begin..(_begin + WrapTaskCapacity)];
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void ReportThreadAccomplished()
{
Interlocked.Decrement(ref _remainThreadCount);
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void ReportWrapAccomplished()
{
_handle.ReportWrapAccomplished();
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void SetAllWorker()
{
for (int i = _begin; i < _begin + WrapTaskCapacity; i++)
{
if (_threads[i].Role != 3) {
_threads[i].Restart();
}
for (int i = 1; i < WrapTaskCapacity; i++) {
Threads[i].Restart();
}
}
internal void PublishTaskToWorkers(Queue<IntPtr> tasks)
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void WaitAllWorker()
{
SpinWait.SpinUntil(() => Volatile.Read(ref _remainThreadCount) == 0);
}
internal void PublishTaskToWorkers()
{
Queue<nint> tasks = WrapTask;
while (tasks.TryDequeue(out nint handle))
{
MatrixParallelTaskPackage* task = (MatrixParallelTaskPackage*)handle;
@@ -100,45 +122,54 @@ namespace StgSharp.Mathematics.Numeric
internal void SetScheduler()
{
MatrixParallelThread thread = _threads[_begin];
MatrixParallelThread thread = Threads[0];
thread.Restart();
_ = Interlocked.Exchange(ref _remainThreadCount, WrapTaskCapacity);
}
private void PublishBufferTaskToWorkers(MatrixParallelTaskPackage* package)
{
long count = Unsafe.As<int, long>(ref package->PrimCount);
long recommendCount = count / 1024;
MatrixParallelThread[] pool = _threads;
ReadOnlySpan<MatrixParallelThread> pool = Threads;
if (pool.Length == 1)
{
pool[0].Tasks.Enqueue((nint)package);
return;
}
long count = Unsafe.As<int, long>(ref package->PrimCount);
long actual = count / pool.Length;
long extra = count % pool.Length;
long offset = 0;
MatrixParallelTaskPackage* current = package;
string msg = $"""
Total: count={count}, length={Threads.Length}{Environment.NewLine}
""";
foreach (MatrixParallelThread thread in pool)
{
long wrapTaskLength = actual;
long ex = extra > 0 ? 1 : 0;
long length = wrapTaskLength + ex;
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);
lengthRef = length;
thread.Tasks.Enqueue((nint)current);
msg += $"""
Slice to thread: total offset={offsetRef} local offset={offset}, length={lengthRef}{Environment.NewLine}
""";
current = MatrixParallelFactory.FromExistPackage(current);
offset += wrapTaskLength;
offset = length;
}
Console.WriteLine(msg);
}
private void PublishKernelTaskToWorkers(MatrixParallelTaskPackage* package)
{
int primCount = package->PrimCount;
int secCount = package->SecCount;
MatrixParallelThread[] pool = _threads;
ReadOnlySpan<MatrixParallelThread> pool = Threads;
if (pool.Length == 1)
{
@@ -201,31 +232,10 @@ namespace StgSharp.Mathematics.Numeric
return;
}
public class Handle(int inWrapIndex, MatrixParallelWrap wrap)
{
public int InWrapIndex { get; } = inWrapIndex;
/// <summary>
/// Role of the thread in the wrap. 1 is Scheduler, 0 are Workers. 2 is leader, beyond
/// wrap's control.
/// </summary>
public int Role { get; internal set; }
public Queue<IntPtr> TaskCache { get; } = [];
internal ManualResetEventSlim ResetEvent { get; } = new(false);
internal MatrixParallelWrap Wrap { get; } = wrap;
}
#region wrap task
internal unsafe Queue<nint> WrapTask { get; set; } = [];
internal int WrapTaskSliceCount { get; set; }
internal int WrapTaskCapacity { get ; init; }
#endregion