From ec1b8dbd512ba19be4c563058bf71e088fda197b Mon Sep 17 00:00:00 2001 From: Youmu Liu <105042599+Nitload-NSI@users.noreply.github.com> Date: Thu, 6 Nov 2025 18:12:04 +0800 Subject: [PATCH] working on dynamic wrapping --- .../ConcurrentCapacityFixedStack.cs | 8 +- .../Memory/ConcurrentBufferStack.cs | 38 +- .../Threading/SamplingAdaptiveBackoff.cs | 457 ------------------ .../Numeric/MatrixCompute.Basic.cs | 8 +- .../Numeric/MatrixParallel.Publish.cs | 24 +- .../Numeric/MatrixParallel.cs | 57 +-- .../Numeric/MatrixParallelPool.cs | 148 +++--- .../Numeric/MatrixParallelWrap.cs | 7 + StgSharpDebug/Program.cs | 10 +- 9 files changed, 145 insertions(+), 612 deletions(-) delete mode 100644 StgSharp.Common/Threading/SamplingAdaptiveBackoff.cs diff --git a/StgSharp.Common/Collections/ConcurrentCapacityFixedStack.cs b/StgSharp.Common/Collections/ConcurrentCapacityFixedStack.cs index 69dadc9e..92884837 100644 --- a/StgSharp.Common/Collections/ConcurrentCapacityFixedStack.cs +++ b/StgSharp.Common/Collections/ConcurrentCapacityFixedStack.cs @@ -105,7 +105,7 @@ namespace StgSharp.Collections public T[] PopRange(int count) { if (count <= 0) { - return Array.Empty(); + return []; } T[] result = new T[count]; @@ -122,7 +122,7 @@ namespace StgSharp.Collections } if (i == 0) { - return Array.Empty(); + 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; } } diff --git a/StgSharp.Common/HighPerformance/Memory/ConcurrentBufferStack.cs b/StgSharp.Common/HighPerformance/Memory/ConcurrentBufferStack.cs index e045af13..1e2a2b51 100644 --- a/StgSharp.Common/HighPerformance/Memory/ConcurrentBufferStack.cs +++ b/StgSharp.Common/HighPerformance/Memory/ConcurrentBufferStack.cs @@ -47,7 +47,7 @@ namespace StgSharp.HighPerformance.Memory return new BufferStack(4); // Default small capacity } - BufferStack stack = new BufferStack(span.Length); + BufferStack 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 } diff --git a/StgSharp.Common/Threading/SamplingAdaptiveBackoff.cs b/StgSharp.Common/Threading/SamplingAdaptiveBackoff.cs deleted file mode 100644 index 71e7994c..00000000 --- a/StgSharp.Common/Threading/SamplingAdaptiveBackoff.cs +++ /dev/null @@ -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; - -/// -/// 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 -/// -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; - - /// - /// Gets the current method - /// - public readonly BackoffMethod CurrentMethod => _currentMethod; - - /// - /// Creates a new optimized sampling adaptive backoff strategy instance - /// - 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; - } - - /// - /// Gets the current sampling status (for debugging) - /// - public readonly (int attempts, int successes, int fallbacks, double successRate, double fallbackRate) GetCurrentSampleStats() - { - return (_currentSample.TotalAttempts, - _currentSample.SuccessfulAttempts, - _currentSample.FallbackCount, - _currentSample.SuccessRate, - _currentSample.FallbackRate); - } - - /// - /// Gets the historical performance of the specified method (for monitoring) - /// - public readonly (double successRate, double fallbackRate, double overallScore) GetMethodPerformance( - BackoffMethod method) - { - var (successRate, fallbackRate, score) = _packedPerformances.GetPerformance(method); - return (successRate, fallbackRate, score); - } - - /// - /// Called when attempting CAS (must be called regardless of success or failure) - /// - /// - /// Whether this CAS attempt was successful - /// - [MethodImpl(MethodImplOptions.AggressiveInlining)] - public void OnCASAttempt(bool success) - { - _currentSample.TotalAttempts++; - - if (success) - { - _currentSample.SuccessfulAttempts++; - _consecutiveFailures = 0; - _spinWait = default; // Reset SpinWait after success - } else - { - _consecutiveFailures++; - } - } - - /// - /// Called when CAS fails and fallback needs to be executed, performs backoff and may trigger - /// re-sampling - /// - [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(); - } - - /// - /// Determines initial method based on system characteristics - /// - [MethodImpl(MethodImplOptions.AggressiveInlining)] - private static BackoffMethod DetermineInitialMethod() - { - int logicalCores = Environment.ProcessorCount; - return logicalCores <= 16 ? BackoffMethod.SpinWait : BackoffMethod.Sleep0; - } - - /// - /// Determines the method to test in the next sampling window using optimized random selection - /// - [MethodImpl(MethodImplOptions.AggressiveInlining)] - private BackoffMethod DetermineNextTestMethod() - { - // Use bitwise operations for 80%/20% probability split - return (NextRandom() & 0xF) < 13 ? _candidateMethod : (BackoffMethod)(NextRandom() & 3); - } - - /// - /// Evaluates the current sampling window and may adjust the strategy - /// - [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; - } - } - - /// - /// Generates initial seed for RNG - /// - [MethodImpl(MethodImplOptions.AggressiveInlining)] - private static uint GenerateInitialSeed() - { - return (uint)(Environment.TickCount64 ^ (Thread.CurrentThread.ManagedThreadId * 0x9E3779B9U)); - } - - /// - /// Fast pseudo-random number generator using XorShift32 - /// - [MethodImpl(MethodImplOptions.AggressiveInlining)] - private uint NextRandom() - { - // XorShift32 - faster than LCG with better distribution - _rng ^= _rng << 13; - _rng ^= _rng >> 17; - _rng ^= _rng << 5; - return _rng; - } - - /// - /// Optimized backoff operation using jump table approach - /// - [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; - } - } - - /// - /// Specialized SpinWait backoff with optimized logic - /// - [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); - } - } - - /// - /// Switches to a new method with optimized state reset - /// - [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); - } - - } - /**/ - -} \ No newline at end of file diff --git a/StgSharp.Mathematics/Numeric/MatrixCompute.Basic.cs b/StgSharp.Mathematics/Numeric/MatrixCompute.Basic.cs index cfdc1a55..9ec89dec 100644 --- a/StgSharp.Mathematics/Numeric/MatrixCompute.Basic.cs +++ b/StgSharp.Mathematics/Numeric/MatrixCompute.Basic.cs @@ -49,8 +49,8 @@ namespace StgSharp.Mathematics.Numeric long count = (long)left.KernelColumnLength * left.KernelRowLength; if (count <= 1024) { - CapacityFixedStack pool = MatrixParallel.LeadParallel(); - if (!pool.TryPop(out MatrixParallelThread thread)) + ReadOnlySpan 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 pool = MatrixParallel.LeadParallel(); - if (!pool.TryPop(out MatrixParallelThread thread)) + ReadOnlySpan pool = MatrixParallel.LeadParallel(1); + if (pool.Length == 0) { /* * wait diff --git a/StgSharp.Mathematics/Numeric/MatrixParallel.Publish.cs b/StgSharp.Mathematics/Numeric/MatrixParallel.Publish.cs index 1e138806..47622543 100644 --- a/StgSharp.Mathematics/Numeric/MatrixParallel.Publish.cs +++ b/StgSharp.Mathematics/Numeric/MatrixParallel.Publish.cs @@ -820,9 +820,15 @@ namespace StgSharp.Mathematics.Numeric int tileWidth = (1024 / secCount) + 1; int tileCount = (primCount + tileWidth - 1) / tileWidth; - CapacityFixedStack pool = LeadParallel(); + ReadOnlySpan 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 taskBunch = new Stack(threadCount); MatrixParallelTaskPackage* 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; } diff --git a/StgSharp.Mathematics/Numeric/MatrixParallel.cs b/StgSharp.Mathematics/Numeric/MatrixParallel.cs index 4e1977a7..0ff8b4c9 100644 --- a/StgSharp.Mathematics/Numeric/MatrixParallel.cs +++ b/StgSharp.Mathematics/Numeric/MatrixParallel.cs @@ -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 _threads; + private static ConcurrentCapacityFixedStack _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 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 pool) + public static void ReleaseParallelLeadership(ref MatrixParallelThread[] pool) { #if NET9_0_OR_GREATER #else diff --git a/StgSharp.Mathematics/Numeric/MatrixParallelPool.cs b/StgSharp.Mathematics/Numeric/MatrixParallelPool.cs index 0370dfaa..c5efa236 100644 --- a/StgSharp.Mathematics/Numeric/MatrixParallelPool.cs +++ b/StgSharp.Mathematics/Numeric/MatrixParallelPool.cs @@ -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 wrapSpan) - { - SpareWrap = CapacityFixedStackBuilder.Create(wrapSpan); - SingleThreadAllocation = new CapacityFixedBag(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 wrapSpan) + private MatrixParallelThread[] _threads; + private bool _disposed = false; + private CapacityFixedStack _wraps; + + internal MatrixParallelHandle(MatrixParallelThread[] threadArray) { - SpareWrap = wrapSpan; - SingleThreadAllocation = new CapacityFixedBag(wrapSpan.Count); - SpareThreadCount = threadCount; + _threads = threadArray; + IsMultithread = threadArray.Length > 1; + SliceToWraps(); } - public int SpareThreadCount { get; set; } - - internal CapacityFixedBag SingleThreadAllocation { get; set; } - - internal CapacityFixedStack SpareWrap { get; set; } - - /* - public void ReturnIndividualHandle(ref MatrixParallelWrap.Handle handle) + public ReadOnlySpan 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 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); + } + } } diff --git a/StgSharp.Mathematics/Numeric/MatrixParallelWrap.cs b/StgSharp.Mathematics/Numeric/MatrixParallelWrap.cs index 21e4cbbb..16bb7bfb 100644 --- a/StgSharp.Mathematics/Numeric/MatrixParallelWrap.cs +++ b/StgSharp.Mathematics/Numeric/MatrixParallelWrap.cs @@ -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() { diff --git a/StgSharpDebug/Program.cs b/StgSharpDebug/Program.cs index b583fc05..38db3ab8 100644 --- a/StgSharpDebug/Program.cs +++ b/StgSharpDebug/Program.cs @@ -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 remaining = new System.Collections.Generic.List(); + List 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)