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

small size buffer op ok

Browse Source
This commit is contained in:
Youmu Liu
2025-11-14 18:28:15 +08:00
parent 1417785726
commit a8c522ab23
19 changed files with 578 additions and 1048 deletions
Download Patch File Download Diff File Expand all files Collapse all files
StgSharp.Mathematics/Numeric/MatrixParallelThread.cs
+185 -10
View File
@@ -30,6 +30,7 @@ using System.Collections.Generic;
using System.Diagnostics.CodeAnalysis;
using System.Linq;
using System.Runtime.CompilerServices;
using System.Runtime.InteropServices;
using System.Text;
using System.Threading.Tasks;
@@ -38,12 +39,19 @@ namespace StgSharp.Mathematics.Numeric
public class MatrixParallelThread : IDisposable
{
// Affinity binding info for this thread
private AffinityBinding _affinity;
private bool _affinityApplied;
private bool disposedValue;
private ManualResetEventSlim _resetEvent;
private Thread _managedThread;
public MatrixParallelThread()
{
// Allocate a unique logical processor for this worker (mutually exclusive)
_affinity = AffinityAllocator.Allocate();
_managedThread = new Thread(MatrixParallelWorker);
_resetEvent = new ManualResetEventSlim(false);
}
@@ -71,19 +79,47 @@ namespace StgSharp.Mathematics.Numeric
while (taskQueue.TryDequeue(out nint handle))
{
MatrixParallelTaskPackageNonGeneric* p = (MatrixParallelTaskPackageNonGeneric*)handle;
switch (p->ComputeMode)
switch (p->ComputeMode.OperationStyle)
{
case 1:
MatrixComputeModel.Compute_Unary(p);
case MatrixOperationStyle.PANEL_OP:
break;
case 2:
MatrixComputeModel.Compute_UnaryScalar(p);
case MatrixOperationStyle.BUFFER_OP:
switch (p->ComputeMode.ParameterStyle)
{
case MatrixOperationParam.RIGHT_ANS_PARAM:
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:
break;
case MatrixOperationParam.ANS_SCALAR_PARAM:
break;
default:
break;
}
break;
case 3:
MatrixComputeModel.Compute_Binary(p);
break;
case 4:
MatrixComputeModel.Compute_BinaryScalar(p);
case MatrixOperationStyle.KERNEL_OP:
switch (p->ComputeMode.ParameterStyle)
{
case MatrixOperationParam.RIGHT_ANS_PARAM:
MatrixComputeModel.Compute_Unary(p);
break;
case MatrixOperationParam.RIGHT_ANS_SCALAR_PARAM:
MatrixComputeModel.Compute_UnaryScalar(p);
break;
case MatrixOperationParam.LEFT_RIGHT_ANS_PARAM:
MatrixComputeModel.Compute_Binary(p);
break;
case MatrixOperationParam.LEFT_RIGHT_ANS_SCALAR_PARAM:
MatrixComputeModel.Compute_BinaryScalar(p);
break;
default:
break;
}
break;
default:
break;
@@ -111,12 +147,18 @@ namespace StgSharp.Mathematics.Numeric
_managedThread.Join();
}
// Release the reserved logical processor back to the pool
AffinityAllocator.Release(_affinity.LogicalProcessor);
disposedValue = true;
}
}
private unsafe void MatrixParallelWorker()
{
// Bind OS-thread affinity once when the worker starts running
ApplyAffinityIfNeeded();
// Thread self = Thread.CurrentThread;
_resetEvent.Wait();
_resetEvent.Reset();
@@ -141,6 +183,139 @@ namespace StgSharp.Mathematics.Numeric
_managedThread.Priority = AfterWork;
}
#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);
#elif LINUX
#endif
#region wrap belong
internal MatrixParallelWrap BindedWrap { get; set; } = null!;