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

Switch matrix parallel to task stealing, type and wrap are refacored, thread execution core has not yet

Browse Source
This commit is contained in:
Youmu Liu
2026-02-24 15:12:23 +08:00
parent d99c1e3aa6
commit d62e59cb93
21 changed files with 1000 additions and 965 deletions
Download Patch File Download Diff File Expand all files Collapse all files
StgSharp.Mathematics/Numeric/MatrixParallelThread.cs
+45 -278
View File
@@ -25,6 +25,7 @@
//
// -----------------------------------------------------------------------
// -----------------------------------------------------------------------
using StgSharp.Internal;
using System;
using System.Collections.Generic;
using System.ComponentModel.DataAnnotations;
@@ -37,42 +38,42 @@ using System.Threading.Tasks;
namespace StgSharp.Mathematics.Numeric
{
public partial class MatrixParallelThread : IDisposable
public unsafe partial class MatrixParallelThread : IDisposable
{
// Affinity binding info for this thread
private AffinityBinding _affinity;
private bool _affinityApplied;
private IntPtr* _context;
private bool disposedValue;
private readonly ManualResetEventSlim _resetEvent;
private readonly Thread _managedThread;
public MatrixParallelThread()
{
// Allocate a unique logical processor for this worker (mutually exclusive)
_affinity = AffinityAllocator.Allocate();
_resetEvent = new ManualResetEventSlim(false);
_managedThread = new Thread(MatrixParallelWorker);
_managedThread.Start();
}
// Global runtime switch: true => UMA/no-affinity; false => NUMA/affinity (auto-detected)
public static bool UseUmaAffinityPath { get; } = DetectUmaAffinityPath();
// ~MatrixComputeThread()
// {
// Dispose(disposing: false);
// }
public ThreadPriority AfterWork
{
get => field;
set => field = value;
}
public ThreadPriority AfterWork { get; set; }
internal ManualResetEventSlim? LeaderEvent { get; set; }
[DllImport(
dllName:Native.LibName,
EntryPoint = "sn_set_thread_affinity",
CallingConvention = CallingConvention.Cdecl)]
internal static extern void SetThreadAffinity(
ulong affinityMask
);
#region wrap belong
internal MatrixParallelWrap BindedWrap { get; set; }
public void Dispose()
{
Dispose(disposing:true);
@@ -81,45 +82,39 @@ namespace StgSharp.Mathematics.Numeric
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public static unsafe void ExecuteTaskPackage(
Queue<IntPtr> taskQueue
MatrixParallelQueue taskQueue
)
{
while (taskQueue.TryDequeue(out nint handle))
MatrixParallelTaskPackage* p = taskQueue.Package;
switch (p->ComputeMode.OperationStyle)
{
MatrixParallelTaskPackage* p = (MatrixParallelTaskPackage*)handle;
if ((short)p->ComputeMode.ParameterStyle < 0)
{
MatrixComputeModel.SpecialBufferCompute(p);
} else
{
switch (p->ComputeMode.OperationStyle)
case MatrixIndexStyle.BUFFER_INDEX:
switch (p->ComputeMode.ParameterStyle)
{
case MatrixIndexStyle.BUFFER_INDEX:
switch (p->ComputeMode.ParameterStyle)
{
case MatrixOperationParam.RIGHT_ANS_PARAM:
MatrixComputeModel.BufferComputeUnary(p);
break;
case MatrixOperationParam.LEFT_RIGHT_ANS_PARAM:
MatrixComputeModel.BufferComputeBinary(p);
break;
case MatrixOperationParam.RIGHT_ANS_SCALAR_PARAM:
MatrixComputeModel.BufferComputeUnaryScalar(p);
break;
case MatrixOperationParam.LEFT_RIGHT_ANS_SCALAR_PARAM:
MatrixComputeModel.BufferComputeBinaryScalar(p);
break;
case MatrixOperationParam.ANS_SCALAR_PARAM:
MatrixComputeModel.BufferComputeNoOperatorScalar(p);
break;
default:
break;
}
case MatrixOperationParam.RIGHT_ANS_PARAM:
BufferComputeUnary(p);
break;
case MatrixOperationParam.LEFT_RIGHT_ANS_PARAM:
BufferComputeBinary(p);
break;
case MatrixOperationParam.RIGHT_ANS_SCALAR_PARAM:
BufferComputeUnaryScalar(p);
break;
case MatrixOperationParam.LEFT_RIGHT_ANS_SCALAR_PARAM:
BufferComputeBinaryScalar(p);
break;
case MatrixOperationParam.ANS_SCALAR_PARAM:
BufferComputeNoOperatorScalar(p);
break;
default:
break;
}
}
break;
case MatrixIndexStyle.SPECIAL:
SpecialBufferCompute(p);
break;
default:
break;
}
}
@@ -143,33 +138,12 @@ namespace StgSharp.Mathematics.Numeric
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);
}
} else
{
// UMA path: no affinity binding
ExecuteTaskPackage(Tasks);
}
#else
// Non-Windows: no affinity control
ExecuteTaskPackage(Tasks);
#endif
BindedWrap.ReportThreadAccomplished();
// Console.WriteLine("leader's computing is done");
Role = RoleType.Worker;
LeaderEvent = null;
current.Priority = ThreadPriority.Normal;
current.Priority = BindedWrap.AfterWork;
}
protected virtual void Dispose(
@@ -184,51 +158,18 @@ namespace StgSharp.Mathematics.Numeric
_managedThread.Join();
}
// Release the reserved logical processor back to the pool
AffinityAllocator.Release(_affinity.LogicalProcessor);
disposedValue = true;
}
}
private static bool DetectUmaAffinityPath()
{
#if WINDOWS
try
{
// Quick heuristic: single CPU looks like UMA
if (Environment.ProcessorCount <= 1)
{
return true;
}
// Query NUMA topology: highest node number 0 => single node => UMA
uint highest = 0;
if (GetNumaHighestNodeNumber(out highest)) {
return highest == 0;
}
// If query fails, prefer UMA (safer: no affinity changes)
return true;
}
catch
{
return true;
}
#else
// Non-Windows: treat as UMA (no affinity API available)
return true;
#endif
}
private unsafe void MatrixParallelWorker()
{
// Bind OS-thread affinity once when the worker starts running
ApplyAffinityIfNeeded();
while (true)
{
// Thread self = Thread.CurrentThread;
AfterWork = BindedWrap.AfterWork;
_managedThread.Priority = ThreadPriority.Highest;
_resetEvent.Wait();
_resetEvent.Reset();
@@ -237,23 +178,11 @@ namespace StgSharp.Mathematics.Numeric
case RoleType.Worker://worker
// _ = Interlocked.Decrement(ref RemainThreadCount);
// Console.WriteLine($"worker {Environment.CurrentManagedThreadId} start");
ExecuteTaskPackage(Tasks);
ExecuteTaskPackage(BindedWrap.TaskQueue);
BindedWrap.ReportThreadAccomplished();
// Console.WriteLine($"worker {Environment.CurrentManagedThreadId} end");
break;
case RoleType.Scheduler://scheduler
// Console.WriteLine($"scheduler {Environment.CurrentManagedThreadId} start");
BindedWrap.PublishTaskToWorkers();
BindedWrap.SetAllWorker();
// Console.WriteLine($"scheduler {Environment.CurrentManagedThreadId} done");
ExecuteTaskPackage(Tasks);
BindedWrap.ReportThreadAccomplished();
BindedWrap.WaitAllWorker();
BindedWrap.ReportWrapAccomplished();
Role = RoleType.Worker;
break;
case RoleType.Leader://leader, error because leader should be at outer thread but not here
break;
default:
@@ -263,174 +192,12 @@ namespace StgSharp.Mathematics.Numeric
}
}
#region core affinity
/*
* data to record core affinity
* preparation for future NUMA affinity setting
*/
// Represents the binding information for a worker thread
internal readonly struct AffinityBinding
{
public readonly int LogicalProcessor;
public readonly nuint Mask;
public AffinityBinding(
int logicalProcessor,
nuint mask
)
{
LogicalProcessor = logicalProcessor;
Mask = mask;
}
}
// Thread-safe allocator that hands out unique logical processors (mutually exclusive)
internal static class AffinityAllocator
{
// Bitset of reserved logical processors (0..63). One bit per CPU.
private static long s_reservedMask; // treat as unsigned bits
public static AffinityBinding Allocate()
{
int cpuCount = Environment.ProcessorCount;
if (cpuCount > 64)
{
cpuCount = 64; // SetThreadAffinityMask supports up to 64-bit mask
}
while (true)
{
long snapshot = Volatile.Read(ref s_reservedMask);
// Find first zero bit
int candidate = -1;
long bit = 0;
for (int i = 0; i < cpuCount; i++)
{
bit = 1L << i;
if ((snapshot & bit) == 0)
{
candidate = i;
break;
}
}
if (candidate < 0)
{
// No free CPU found: fall back to CPU 0 (shared)
return new AffinityBinding(0, (nuint)1);
}
long desired = snapshot | bit;
if (Interlocked.CompareExchange(ref s_reservedMask, desired, snapshot) == snapshot)
{
// Successfully reserved candidate
nuint mask = (nuint)(1UL << candidate);
return new AffinityBinding(candidate, mask);
}
// else: another thread raced us; retry
}
}
public static void Release(
int logicalProcessor
)
{
if ((uint)logicalProcessor >= 64u) {
return;
}
long bit = 1L << logicalProcessor;
while (true)
{
long snapshot = Volatile.Read(ref s_reservedMask);
if ((snapshot & bit) == 0)
{
return; // already free
}
long desired = snapshot & ~bit;
if (Interlocked.CompareExchange(ref s_reservedMask, desired, snapshot) == snapshot) {
return;
}
}
}
}
private void ApplyAffinityIfNeeded()
{
if (_affinityApplied) {
return;
}
#if WINDOWS
try
{
// On Windows pin the current native thread to the reserved logical processor
SetThreadAffinityMask(GetCurrentThread(), _affinity.Mask);
}
catch
{
// Ignore binding errors; continue without affinity
}
#elif LINUX
#endif
_affinityApplied = true;
}
#endregion
#if WINDOWS
[DllImport("kernel32.dll")]
private static extern IntPtr GetCurrentThread();
[DllImport("kernel32.dll")]
private static extern nuint SetThreadAffinityMask(
IntPtr hThread,
nuint dwThreadAffinityMask
);
[LibraryImport("kernel32.dll")]
[return: MarshalAs(UnmanagedType.Bool)]
private static partial bool GetNumaHighestNodeNumber(
out uint HighestNodeNumber
);
#elif LINUX
#endif
#region wrap belong
internal MatrixParallelWrap BindedWrap { get; set; } = null!;
internal Queue<IntPtr> Tasks { get; } = [];
public RoleType Role
{
get;
internal set
{
// Console.WriteLine($"Thread {_managedThread.ManagedThreadId} set to {value}, from thread {Environment.CurrentManagedThreadId}");
field = value;
}
}
public RoleType Role { get; internal set; }
public enum RoleType : int
{
Worker = 0,
Scheduler = 1,
Leader = 2,
}