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

working on dynamic wrapping

Browse Source
This commit is contained in:
Youmu Liu
2025-11-06 18:12:04 +08:00
parent ea0e6f3ff6
commit ec1b8dbd51
9 changed files with 145 additions and 612 deletions
Download Patch File Download Diff File Expand all files Collapse all files
StgSharp.Common/Collections/ConcurrentCapacityFixedStack.cs
+4 -4
View File
@@ -105,7 +105,7 @@ namespace StgSharp.Collections
public T[] PopRange(int count)
{
if (count <= 0) {
return Array.Empty<T>();
return [];
}
T[] result = new T[count];
@@ -122,7 +122,7 @@ namespace StgSharp.Collections
}
if (i == 0) {
return Array.Empty<T>();
return [];
}
T[] trimmed = new T[i];
@@ -251,7 +251,7 @@ namespace StgSharp.Collections
if (Volatile.Read(ref _states[newTop]) == 0)
{
// attempt to restore top; ok if it fails due to concurrent operations
Interlocked.CompareExchange(ref _top, newTop + 1, newTop);
_ = Interlocked.CompareExchange(ref _top, newTop + 1, newTop);
sw.SpinOnce();
continue;
}
@@ -259,7 +259,7 @@ namespace StgSharp.Collections
value = _values[newTop];
_values[newTop] = default!; // avoid ref retention
Volatile.Write(ref _states[newTop], 0); // mark empty
Interlocked.Decrement(ref _count);
_ = Interlocked.Decrement(ref _count);
return true;
}
}
StgSharp.Common/HighPerformance/Memory/ConcurrentBufferStack.cs
+18 -20
View File
@@ -47,7 +47,7 @@ namespace StgSharp.HighPerformance.Memory
return new BufferStack<T>(4); // Default small capacity
}
BufferStack<T> stack = new BufferStack<T>(span.Length);
BufferStack<T> stack = new(span.Length);
// Use internal method for efficient bulk initialization
@@ -82,7 +82,7 @@ namespace StgSharp.HighPerformance.Memory
private byte* _buffer;
private readonly BufferExpansionLock _lock = new();
private ulong _count;
private long _count;
public ConcurrentBufferStack()
{
@@ -101,7 +101,7 @@ namespace StgSharp.HighPerformance.Memory
public ulong Count
{
[MethodImpl(MethodImplOptions.AggressiveInlining)]
get => Volatile.Read(ref _count);
get => (ulong)Volatile.Read(ref _count);
}
public void Clear()
@@ -141,7 +141,7 @@ namespace StgSharp.HighPerformance.Memory
try
{
Header* header = (Header*)_buffer;
nuint oldTop = Volatile.Read(ref header->Top);
int oldTop = Volatile.Read(ref header->Top);
if (oldTop <= 0) {
throw new InvalidOperationException("Stack underflow");
}
@@ -170,8 +170,8 @@ namespace StgSharp.HighPerformance.Memory
try
{
Header* header = (Header*)_buffer;
nuint topSnapshot = Volatile.Read(ref header->Top);
nuint cap = Volatile.Read(ref header->Capacity);
int topSnapshot = Volatile.Read(ref header->Top);
int cap = Volatile.Read(ref header->Capacity);
if (topSnapshot >= cap)
{
// Need expansion
@@ -182,8 +182,8 @@ namespace StgSharp.HighPerformance.Memory
header = (Header*)_buffer;
if (header->Top >= header->Capacity)
{
nuint newCap = header->Capacity * 2;
nuint newSize = (nuint)sizeof(Header) + (newCap * (nuint)sizeof(T));
int newCap = header->Capacity * 2;
nuint newSize = (nuint)sizeof(Header) + ((nuint)newCap * (nuint)sizeof(T));
_lock.EnterBufferCopy();
try
@@ -216,7 +216,7 @@ namespace StgSharp.HighPerformance.Memory
try
{
Header* header = (Header*)_buffer;
nuint idx = header->Top; // reserve next slot
int idx = header->Top; // reserve next slot
if (idx >= header->Capacity)
{
// lost race against expander or other producer; retry
@@ -244,7 +244,7 @@ namespace StgSharp.HighPerformance.Memory
try
{
Header* header = (Header*)_buffer;
nuint oldTop = Volatile.Read(ref header->Top);
int oldTop = Volatile.Read(ref header->Top);
if (oldTop <= 0)
{
value = default;
@@ -263,10 +263,8 @@ namespace StgSharp.HighPerformance.Memory
{
_lock.ExitBufferRead();
}
value = default;
return true;
Thread.Sleep(0);
}
Thread.Sleep(0); // Yield to avoid busy waiting
}
// Internal method for efficient bulk initialization
@@ -282,28 +280,28 @@ namespace StgSharp.HighPerformance.Memory
fixed (T* sourcePtr = span) {
Buffer.MemoryCopy(sourcePtr, dataStart,
header->Capacity * (nuint)sizeof(T),
(nuint)header->Capacity * (nuint)sizeof(T),
(nuint)span.Length * (nuint)sizeof(T));
}
header->Top = (nuint)span.Length;
Volatile.Write(ref _count, (ulong)span.Length);
header->Top = span.Length;
Volatile.Write(ref _count, span.Length);
}
private void AllocateBuffer(int capacity)
{
nuint size = (nuint)sizeof(Header) + ((nuint)capacity * (nuint)sizeof(T));
nuint size = (nuint)sizeof(Header) + (nuint)capacity * (nuint)sizeof(T);
_buffer = (byte*)NativeMemory.Alloc(size);
Header* header = (Header*)_buffer;
header->Top = 0;
header->Capacity = (nuint)capacity;
header->Capacity = capacity;
}
private struct Header
{
public nuint Capacity; // max capacity of the stack (number of elements)
public nuint Top; // next available index in the stack
public int Capacity; // max capacity of the stack (number of elements)
public int Top; // next available index in the stack
}
StgSharp.Common/Threading/SamplingAdaptiveBackoff.cs
-457
View File
@@ -1,457 +0,0 @@
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
// file="SamplingAdaptiveBackoff"
// Project: StgSharp
// AuthorGroup: Nitload
// Copyright (c) Nitload. All rights reserved.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in all
// copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
//
// -----------------------------------------------------------------------
// -----------------------------------------------------------------------
using System;
using System.Diagnostics;
using System.Runtime.CompilerServices;
using System.Threading;
/// <summary>
/// Optimized sampling-based adaptive backoff strategy that re-samples and evaluates the current
/// contention situation and adjusts the strategy after a certain number of fallbacks
/// </summary>
public struct SamplingAdaptiveBackoff
{
/*
private const int CONFIDENCE_THRESHOLD = 3; // Requires N consecutive sampling confirmations to switch strategies
// Configuration constants
private const int FALLBACK_SAMPLING_THRESHOLD = 64; // Triggers re-sampling every 64 fallbacks
private const int INITIAL_SPINWAIT_THRESHOLD = 8; // Cheap spinning for the first few SpinWait attempts
private const int MAX_SAMPLING_WINDOW_MS = 1000; // Maximum sampling window time
private const int MIN_SAMPLING_WINDOW_MS = 100; // Minimum sampling window time
private const int PERFORMANCE_HISTORY_MASK = 7; // Use 3-bit mask for 8 entries (power of 2)
private const double SIGNIFICANT_IMPROVEMENT = 0.15; // 15% improvement is considered significant
private BackoffMethod _candidateMethod;
private BackoffMethod _currentMethod;
private byte _confidenceCounter;
private byte _consecutiveFailures;
// Packed performance data to reduce memory footprint
private PackedMethodPerformance _packedPerformances; // Bit-packed performance data for all 3 methods
private SamplingWindow _currentSample;
// Instance state
private SpinWait _spinWait;
private uint _rng;
/// <summary>
/// Gets the current method
/// </summary>
public readonly BackoffMethod CurrentMethod => _currentMethod;
/// <summary>
/// Creates a new optimized sampling adaptive backoff strategy instance
/// </summary>
public static SamplingAdaptiveBackoff Create()
{
// Choose initial strategy based on system characteristics
BackoffMethod initialMethod = DetermineInitialMethod();
SamplingAdaptiveBackoff instance = new()
{
_spinWait = default,
_currentMethod = initialMethod,
_consecutiveFailures = 0,
_confidenceCounter = 0,
_candidateMethod = initialMethod,
_rng = GenerateInitialSeed(),
_packedPerformances = PackedMethodPerformance.Initialize()
};
// Start the first sampling window
instance._currentSample.Reset(initialMethod);
return instance;
}
/// <summary>
/// Gets the current sampling status (for debugging)
/// </summary>
public readonly (int attempts, int successes, int fallbacks, double successRate, double fallbackRate) GetCurrentSampleStats()
{
return (_currentSample.TotalAttempts,
_currentSample.SuccessfulAttempts,
_currentSample.FallbackCount,
_currentSample.SuccessRate,
_currentSample.FallbackRate);
}
/// <summary>
/// Gets the historical performance of the specified method (for monitoring)
/// </summary>
public readonly (double successRate, double fallbackRate, double overallScore) GetMethodPerformance(
BackoffMethod method)
{
var (successRate, fallbackRate, score) = _packedPerformances.GetPerformance(method);
return (successRate, fallbackRate, score);
}
/// <summary>
/// Called when attempting CAS (must be called regardless of success or failure)
/// </summary>
/// <param name="success">
/// Whether this CAS attempt was successful
/// </param>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void OnCASAttempt(bool success)
{
_currentSample.TotalAttempts++;
if (success)
{
_currentSample.SuccessfulAttempts++;
_consecutiveFailures = 0;
_spinWait = default; // Reset SpinWait after success
} else
{
_consecutiveFailures++;
}
}
/// <summary>
/// Called when CAS fails and fallback needs to be executed, performs backoff and may trigger
/// re-sampling
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void OnFallbackAndBackoff()
{
_currentSample.FallbackCount++;
// Check if re-sampling threshold is reached using bitwise operations for faster comparison
if ((_currentSample.FallbackCount & ~(FALLBACK_SAMPLING_THRESHOLD - 1)) != 0 ||
_currentSample.ElapsedTicks > MAX_SAMPLING_WINDOW_MS)
{
if (_currentSample.ElapsedTicks >= MIN_SAMPLING_WINDOW_MS) {
EvaluateCurrentSample();
}
}
// Perform backoff using jump table approach for better branch prediction
PerformBackoffOptimized();
}
/// <summary>
/// Determines initial method based on system characteristics
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
private static BackoffMethod DetermineInitialMethod()
{
int logicalCores = Environment.ProcessorCount;
return logicalCores <= 16 ? BackoffMethod.SpinWait : BackoffMethod.Sleep0;
}
/// <summary>
/// Determines the method to test in the next sampling window using optimized random selection
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
private BackoffMethod DetermineNextTestMethod()
{
// Use bitwise operations for 80%/20% probability split
return (NextRandom() & 0xF) < 13 ? _candidateMethod : (BackoffMethod)(NextRandom() & 3);
}
/// <summary>
/// Evaluates the current sampling window and may adjust the strategy
/// </summary>
[MethodImpl(MethodImplOptions.NoInlining)]
private void EvaluateCurrentSample()
{
// Update performance record for the current method using packed data
_packedPerformances.UpdatePerformance(
_currentSample.TestedMethod, _currentSample.SuccessRate, _currentSample.FallbackRate,
_currentSample.ElapsedTicks / Math.Max(1, _currentSample.TotalAttempts));
// Find the currently best performing method
BackoffMethod bestMethod = _packedPerformances.FindBestMethod();
// Use branchless logic for confidence tracking
bool methodChanged = bestMethod != _currentMethod;
bool candidateMatches = bestMethod == _candidateMethod;
_confidenceCounter = methodChanged ? (candidateMatches ? (byte)(_confidenceCounter + 1) : (byte)1) : (byte)0;
if (methodChanged && candidateMatches && _confidenceCounter >= CONFIDENCE_THRESHOLD)
{
SwitchToMethod(bestMethod);
} else if (methodChanged && !candidateMatches)
{
_candidateMethod = bestMethod;
} else if (!methodChanged) {
_candidateMethod = _currentMethod;
}
// Start new sampling window
BackoffMethod nextTestMethod = DetermineNextTestMethod();
_currentSample.Reset(nextTestMethod);
if (nextTestMethod != _currentMethod)
{
_currentMethod = nextTestMethod;
_spinWait = default;
}
}
/// <summary>
/// Generates initial seed for RNG
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
private static uint GenerateInitialSeed()
{
return (uint)(Environment.TickCount64 ^ (Thread.CurrentThread.ManagedThreadId * 0x9E3779B9U));
}
/// <summary>
/// Fast pseudo-random number generator using XorShift32
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
private uint NextRandom()
{
// XorShift32 - faster than LCG with better distribution
_rng ^= _rng << 13;
_rng ^= _rng >> 17;
_rng ^= _rng << 5;
return _rng;
}
/// <summary>
/// Optimized backoff operation using jump table approach
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
private void PerformBackoffOptimized()
{
// Use computed goto pattern for better branch prediction
switch (_currentMethod)
{
case BackoffMethod.SpinWait:
PerformSpinWaitBackoff();
break;
case BackoffMethod.Yield:
_ = Thread.Yield();
break;
case BackoffMethod.Sleep0:
Thread.Sleep(0);
break;
default:
break;
}
}
/// <summary>
/// Specialized SpinWait backoff with optimized logic
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
private void PerformSpinWaitBackoff()
{
if (_consecutiveFailures > INITIAL_SPINWAIT_THRESHOLD && _spinWait.NextSpinWillYield)
{
// Optimized random yield count using bitwise operations
int yields = (int)(NextRandom() & 7) + 1;
for (int i = 0; i < yields; i++) {
_ = Thread.Yield();
}
} else
{
_spinWait.SpinOnce(sleep1Threshold:-1);
}
}
/// <summary>
/// Switches to a new method with optimized state reset
/// </summary>
[MethodImpl(MethodImplOptions.AggressiveInlining)]
private void SwitchToMethod(BackoffMethod method)
{
_currentMethod = method;
_candidateMethod = method;
_confidenceCounter = 0;
_spinWait = default;
}
public enum BackoffMethod : byte
{
SpinWait = 0,
Yield = 1,
Sleep0 = 2
}
// Optimized sampling window state with better packing
private struct SamplingWindow
{
public BackoffMethod TestedMethod; // Current method being tested
public int FallbackCount; // Number of fallbacks
public int SuccessfulAttempts; // Number of successful attempts
public int TotalAttempts; // Total number of attempts
public long StartTicks; // Sampling window start time
public readonly double SuccessRate
{
[MethodImpl(MethodImplOptions.AggressiveInlining)]
get => TotalAttempts == 0 ? 0.0 : (double)SuccessfulAttempts / TotalAttempts;
}
public readonly double FallbackRate
{
[MethodImpl(MethodImplOptions.AggressiveInlining)]
get => TotalAttempts == 0 ? 0.0 : (double)FallbackCount / TotalAttempts;
}
public readonly long ElapsedTicks
{
[MethodImpl(MethodImplOptions.AggressiveInlining)]
get => Environment.TickCount64 - StartTicks;
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void Reset(BackoffMethod method)
{
StartTicks = Environment.TickCount64;
TotalAttempts = 0;
SuccessfulAttempts = 0;
FallbackCount = 0;
TestedMethod = method;
}
}
// Bit-packed performance data structure for memory efficiency
private struct PackedMethodPerformance
{
// Pack all 3 methods' performance into compact format
private ulong _packed0; // Method 0 (SpinWait) data
private ulong _packed1; // Method 1 (Yield) data
private ulong _packed2; // Method 2 (Sleep0) data
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public readonly BackoffMethod FindBestMethod()
{
(double _, double _, double score0) = UnpackPerformanceData(_packed0);
(double _, double _, double score1) = UnpackPerformanceData(_packed1);
(double _, double _, double score2) = UnpackPerformanceData(_packed2);
// Use branchless comparison for best performance
BackoffMethod best = score1 > score0 + SIGNIFICANT_IMPROVEMENT ?
BackoffMethod.Yield :
BackoffMethod.SpinWait;
best = score2 > GetScore(best) + SIGNIFICANT_IMPROVEMENT ? BackoffMethod.Sleep0 : best;
return best;
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public readonly (double successRate, double fallbackRate, double overallScore) GetPerformance(
BackoffMethod method)
{
ulong packed = method switch
{
BackoffMethod.SpinWait => _packed0,
BackoffMethod.Yield => _packed1,
BackoffMethod.Sleep0 => _packed2,
_ => _packed0
};
return UnpackPerformanceData(packed);
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public static PackedMethodPerformance Initialize()
{
return new PackedMethodPerformance
{
_packed0 = PackPerformanceData(0.5, 0.5, 0),
_packed1 = PackPerformanceData(0.5, 0.5, 0),
_packed2 = PackPerformanceData(0.5, 0.5, 0)
};
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void UpdatePerformance(BackoffMethod method, double successRate, double fallbackRate, long avgLatency)
{
ulong packed = PackPerformanceData(successRate, fallbackRate, avgLatency);
switch (method)
{
case BackoffMethod.SpinWait:
_packed0 = packed;
break;
case BackoffMethod.Yield:
_packed1 = packed;
break;
case BackoffMethod.Sleep0:
_packed2 = packed;
break;
}
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
private readonly double GetScore(BackoffMethod method)
{
return method switch
{
BackoffMethod.SpinWait => UnpackPerformanceData(_packed0).overallScore,
BackoffMethod.Yield => UnpackPerformanceData(_packed1).overallScore,
BackoffMethod.Sleep0 => UnpackPerformanceData(_packed2).overallScore,
_ => 0.0
};
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
private static ulong PackPerformanceData(double successRate, double fallbackRate, long avgLatency)
{
// Pack data into 64-bit value: 20 bits success rate, 20 bits fallback rate, 24 bits latency
uint success = (uint)(successRate * 1048575); // 20 bits (2^20 - 1)
uint fallback = (uint)(fallbackRate * 1048575); // 20 bits
uint latency = (uint)Math.Min(avgLatency, 16777215); // 24 bits (2^24 - 1)
return ((ulong)success << 44) | ((ulong)fallback << 24) | latency;
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
private static (double successRate, double fallbackRate, double overallScore) UnpackPerformanceData(
ulong packed)
{
double successRate = ((packed >> 44) & 0xFFFFF) / 1048575.0;
double fallbackRate = ((packed >> 24) & 0xFFFFF) / 1048575.0;
long avgLatency = (long)(packed & 0xFFFFFF);
// Calculate overall score with optimized formula
double baseScore = successRate * 0.6 + (1.0 - fallbackRate) * 0.4;
double latencyPenalty = avgLatency > 0 ? Math.Min(0.1, avgLatency / 10000.0) : 0;
double overallScore = baseScore - latencyPenalty;
return (successRate, fallbackRate, overallScore);
}
}
/**/
}
StgSharp.Mathematics/Numeric/MatrixCompute.Basic.cs
+4 -4
View File
@@ -49,8 +49,8 @@ namespace StgSharp.Mathematics.Numeric
long count = (long)left.KernelColumnLength * left.KernelRowLength;
if (count <= 1024)
{
CapacityFixedStack<MatrixParallelThread> pool = MatrixParallel.LeadParallel();
if (!pool.TryPop(out MatrixParallelThread thread))
ReadOnlySpan<MatrixParallelThread> pool = MatrixParallel.LeadParallel(1);
if (pool.Length == 0)
{
/*
* wait
@@ -84,8 +84,8 @@ namespace StgSharp.Mathematics.Numeric
if (count <= 1024)
{
// TODO: this is very awkward, blocking thread renting, need a better way to handle small tasks
CapacityFixedStack<MatrixParallelThread> pool = MatrixParallel.LeadParallel();
if (!pool.TryPop(out MatrixParallelThread thread))
ReadOnlySpan<MatrixParallelThread> pool = MatrixParallel.LeadParallel(1);
if (pool.Length == 0)
{
/*
* wait
StgSharp.Mathematics/Numeric/MatrixParallel.Publish.cs
+10 -14
View File
@@ -820,9 +820,15 @@ namespace StgSharp.Mathematics.Numeric
int tileWidth = (1024 / secCount) + 1;
int tileCount = (primCount + tileWidth - 1) / tileWidth;
CapacityFixedStack<MatrixParallelThread> pool = LeadParallel();
ReadOnlySpan<MatrixParallelThread> pool = LeadParallel(tileCount);
int threadCount = pool.Count;
int threadCount = pool.Length;
if (pool.Length == 0)
{
/*
* wait until has enough threads
*/
}
int tilePerThreadLess = tileCount / threadCount; // tiles per thread without extra slice
int columnAfterBaseTile = tileCount % threadCount * tileWidth; // tile sliced into columns
int extraColumnPerThread = columnAfterBaseTile / threadCount;
@@ -831,17 +837,8 @@ namespace StgSharp.Mathematics.Numeric
int total = primCount;
Stack<MatrixParallelWrap> taskBunch = new Stack<MatrixParallelWrap>(threadCount);
MatrixParallelTaskPackage<T>* current = package;
while (true)
foreach (MatrixParallelThread thread in pool)
{
if (!pool.TryPop(out MatrixParallelThread? wrap) || wrap is null)
{
#if DEBUG
throw new InvalidOperationException("Someone steal parallel wraps from the pool.");
#else
break;
#endif
}
bool hasSlice = columnAfterBaseTile > 0;
bool isLargeSlice = extraColumnCount > 0;
@@ -872,7 +869,7 @@ namespace StgSharp.Mathematics.Numeric
total -= columnInWrap;
// set to wrap and push
wrap.Tasks.Enqueue((IntPtr)current);
// wrap.Tasks.Enqueue((IntPtr)current);
// taskBunch.Push(wrap);
@@ -884,7 +881,6 @@ namespace StgSharp.Mathematics.Numeric
current = MatrixParallelFactory.FromExistPackage(current);
current->PrimTileOffset += columnInWrap;
}
ReleaseParallelLeadership(ref pool);
return taskBunch;
}
StgSharp.Mathematics/Numeric/MatrixParallel.cs
+4 -53
View File
@@ -34,45 +34,12 @@ using System.Threading;
namespace StgSharp.Mathematics.Numeric
{
/*
* Thread wrapping to accelerate matrix operations by dividing threads into groups (pool).
* For M pool, each containing an average of N threads, we satisfy the condition M * N = T - 4,
* where T is the total thread count of the CPU. Typically, for modern consumer CPUs, T is around 16.
* For simplicity, let M * N = 16.
*
* The total time complexity for publishing tasks is O(M + N):
* - O(M) accounts for distributing tasks across M pool.
* - O(N) accounts for threads within each wrap fetching tasks from the wrap queue.
*
* Using the inequality Sqrt(M * N) <= (M + N) / 2 (derived from the arithmetic-geometric mean inequality),
* the optimal configuration occurs when M = N. For M * N = 16, this gives M = N = 4.
*
* To ensure compatibility across a wide range of CPU architectures,
* we allow each wrap to contain either 3 or 4 threads, providing flexibility while maintaining efficiency.
*/
public static partial class MatrixParallel
{
private static readonly (int a, int b)[] LookupTable = [
(1, 1), // 7
(0, 2), // 8 (no non-zero solution)
(3, 0), // 9 (no non-zero solution)
(2, 1), // 10
(1, 2), // 11
(0, 3), // 12 (no non-zero solution)
(3, 1), // 13
(2, 2), // 14
(1, 3), // 15
(4, 1), // 16
(3, 2), // 17
(2, 3), // 18
(1, 4), // 19
(4, 2), // 20
];
private static bool _meaningful;
private static CapacityFixedStack<MatrixParallelThread> _threads;
private static ConcurrentCapacityFixedStack<MatrixParallelThread> _threads;
private static volatile int RemainThreadCount = 0;
private static readonly object _lock = new();
@@ -91,22 +58,6 @@ namespace StgSharp.Mathematics.Numeric
Monitor.Exit(_lock);
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public static (int count_3, int count_4) FastDecompose(int c)
{
if (c < 7) {
return (0, 0);
}
int t = c - 7;
int k = t / 14;
int d = t % 14;
(int a0, int b0) = LookupTable[d];
return (a0 + (2 * k), b0 + (2 * k));
}
public static void Init()
{
if (IsInitialized) {
@@ -144,17 +95,17 @@ namespace StgSharp.Mathematics.Numeric
#region wrap field
public static CapacityFixedStack<MatrixParallelThread> LeadParallel()
public static MatrixParallelThread[] LeadParallel(int recommendCount)
{
#if NET9_0_OR_GREATER
#else
Monitor.Enter(_lock);
#endif
return _threads;
return _threads.PopRange(recommendCount);
}
public static void ReleaseParallelLeadership(ref CapacityFixedStack<MatrixParallelThread> pool)
public static void ReleaseParallelLeadership(ref MatrixParallelThread[] pool)
{
#if NET9_0_OR_GREATER
#else
StgSharp.Mathematics/Numeric/MatrixParallelPool.cs
+92 -56
View File
@@ -36,79 +36,115 @@ using System.Threading.Tasks;
namespace StgSharp.Mathematics.Numeric
{
public class MatrixParallelPool
public struct MatrixParallelHandle : IDisposable
{
private MatrixParallelPool()
{
throw new NotSupportedException();
}
/*
internal MatrixParallelPool(int threadCount, ReadOnlySpan<MatrixParallelWrap> wrapSpan)
{
SpareWrap = CapacityFixedStackBuilder.Create(wrapSpan);
SingleThreadAllocation = new CapacityFixedBag<MatrixParallelWrap>(wrapSpan.Length);
SpareThreadCount = threadCount;
}
/**/
private static readonly (int a, int b)[] LookupTable = [
(1, 1), // 7
(0, 2), // 8 (no non-zero solution)
(3, 0), // 9 (no non-zero solution)
(2, 1), // 10
(1, 2), // 11
(0, 3), // 12 (no non-zero solution)
(3, 1), // 13
(2, 2), // 14
(1, 3), // 15
(4, 1), // 16
(3, 2), // 17
(2, 3), // 18
(1, 4), // 19
(4, 2), // 20
];
internal MatrixParallelPool(int threadCount, CapacityFixedStack<MatrixParallelWrap> wrapSpan)
private MatrixParallelThread[] _threads;
private bool _disposed = false;
private CapacityFixedStack<MatrixParallelWrap> _wraps;
internal MatrixParallelHandle(MatrixParallelThread[] threadArray)
{
SpareWrap = wrapSpan;
SingleThreadAllocation = new CapacityFixedBag<MatrixParallelWrap>(wrapSpan.Count);
SpareThreadCount = threadCount;
_threads = threadArray;
IsMultithread = threadArray.Length > 1;
SliceToWraps();
}
public int SpareThreadCount { get; set; }
internal CapacityFixedBag<MatrixParallelWrap> SingleThreadAllocation { get; set; }
internal CapacityFixedStack<MatrixParallelWrap> SpareWrap { get; set; }
/*
public void ReturnIndividualHandle(ref MatrixParallelWrap.Handle handle)
public ReadOnlySpan<MatrixParallelThread> Threads
{
MatrixParallelWrap wrap = handle.Wrap;
handle.Role = 0;
if (wrap.IsAllAvailable)
get
{
SingleThreadAllocation.RemoveAt(wrap.Magic);
SpareWrap.Push(wrap);
SpareThreadCount += wrap.WrapTaskCapacity;
}
}
public bool TryRentIndividualHandle(out MatrixParallelWrap.Handle handle)
{
foreach (MatrixParallelWrap item in SingleThreadAllocation)
{
if (item.TryGetLeaderHandle(out handle!)) {
return true;
if (_disposed) {
throw new ObjectDisposedException(typeof(MatrixParallelHandle).Name);
}
return _threads;
}
if (TryRequestWrap(out MatrixParallelWrap wrap))
{
wrap!.Magic = SingleThreadAllocation.Count;
SingleThreadAllocation.Add(wrap);
if (wrap.TryGetLeaderHandle(out handle!)) {
return true;
}
}
internal ReadOnlySpan<MatrixParallelThread> ThreadsUnsafe
{
[MethodImpl(MethodImplOptions.AggressiveInlining)]
get => _threads;
}
private bool IsMultithread { get; init; }
public void Dispose()
{
if (_disposed) {
return;
}
handle = null!;
return false;
MatrixParallel.ReleaseParallelLeadership(ref _threads);
_disposed = true;
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public bool TryRequestWrap([NotNull] out MatrixParallelWrap? wrap)
private static (int count_3, int count_4) FastDecompose(int c)
{
if (SpareWrap.TryPop(out wrap))
/*
* Thread wrapping to accelerate matrix operations by dividing threads into groups (pool).
* For M pool, each containing an average of N threads, we satisfy the condition M * N = T - 4,
* where T is the total thread count of the CPU. Typically, for modern consumer CPUs, T is around 16.
* For simplicity, let M * N = 16.
*
* The total time complexity for publishing tasks is O(M + N):
* - O(M) accounts for distributing tasks across M pool.
* - O(N) accounts for threads within each wrap fetching tasks from the wrap queue.
*
* Using the inequality Sqrt(M * N) <= (M + N) / 2 (derived from the arithmetic-geometric mean inequality),
* the optimal configuration occurs when M = N. For M * N = 16, this gives M = N = 4.
*
* To ensure compatibility across a wide range of CPU architectures,
* we allow each wrap to contain either 3 or 4 threads, providing flexibility while maintaining efficiency.
*/
if (c < 7)
{
SpareThreadCount -= wrap.WrapTaskCapacity;
return true;
return (0, 0);
}
wrap = null!;
return false;
int t = c - 7;
int k = t / 14;
int d = t % 14;
(int a0, int b0) = LookupTable[d];
return (a0 + (2 * k), b0 + (2 * k));
}
/**/
private void SliceToWraps()
{
if (_disposed) {
throw new ObjectDisposedException(typeof(MatrixParallelHandle).Name);
}
if (_wraps is not null) {
return;
}
int count = _threads.Length;
if (count < 6)
{
MatrixParallelWrap wrap = new MatrixParallelWrap(_threads);
_wraps = [ wrap ];
}
(int c3,int c4) = FastDecompose(_threads.Length);
_wraps = new(c3 + c4);
}
}
}
StgSharp.Mathematics/Numeric/MatrixParallelWrap.cs
+7
View File
@@ -40,6 +40,13 @@ namespace StgSharp.Mathematics.Numeric
{
private readonly MatrixParallelThread[] _threads;
private int _begin;
internal MatrixParallelWrap(MatrixParallelThread[] threads)
{
_threads = threads;
Count = threads.Length;
}
public MatrixParallelWrap()
{
StgSharpDebug/Program.cs
+6 -4
View File
@@ -92,7 +92,8 @@ namespace StgSharp
if (stack.TryPop(out int v))
{
popped.Add(v);
Console.WriteLine($"pop: {v}");
// Console.WriteLine($"pop: {v}");
} else
{
Thread.Yield();
@@ -111,7 +112,7 @@ namespace StgSharp
ready.Signal();
startGate.Wait();
int begin = captured * perProducer;
int begin = captured * perProducer + 10 * perProducer;
for (int j = 0; j < perProducer; j++)
{
stack.Push(begin + j);
@@ -146,12 +147,13 @@ namespace StgSharp
}
// Gather remaining data
List<int> remaining = new System.Collections.Generic.List<int>();
List<int> remaining = [];
while (stack.TryPop(out int value)) {
remaining.Add(value);
}
Console.WriteLine($"Popped count: {popped.Count}");
/**/
Console.WriteLine("Popped values:");
foreach (int v in popped)
{
@@ -159,7 +161,7 @@ namespace StgSharp
Console.Write('\t');
}
Console.WriteLine();
/**/
Console.WriteLine($"Remaining count: {remaining.Count}");
Console.WriteLine("Remaining values:");
foreach (int v in remaining)