optimize para performance of buffered stack

This commit is contained in:
Youmu Liu
2025-11-05 20:33:58 +08:00
parent 9c4c71dc5b
commit ea0e6f3ff6
10 changed files with 516 additions and 91 deletions
@@ -0,0 +1,287 @@
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
// file="ConcurrentCapacityFixedStack"
// 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.Collections;
using System.Collections.Generic;
using System.Runtime.CompilerServices;
using System.Threading;
namespace StgSharp.Collections
{
/// <summary>
/// Lock-free, fixed-capacity stack implemented on top of an array using atomic index
/// operations. - No linked list; only array slots - Uses CAS on a shared top index to
/// reserve/release slots - Per-slot state flag to safely publish/consume values under races
/// </summary>
public class ConcurrentCapacityFixedStack<T> : IEnumerable<T>
{
private readonly int[] _states; // 0 = empty, 1 = full
private readonly T[] _values;
private readonly int _capacity;
// Number of full slots (observable count)
private int _count;
// Next push position (0.._capacity). Only modified via CAS.
private int _top;
public ConcurrentCapacityFixedStack(int capacity)
{
if (capacity <= 0) {
throw new ArgumentOutOfRangeException(nameof(capacity));
}
_capacity = capacity;
_values = new T[capacity];
_states = new int[capacity];
_top = 0;
_count = 0;
}
public bool IsEmpty => Count == 0;
public int Capacity => _capacity;
public int Count => Volatile.Read(ref _count);
public void Clear()
{
// Pop all elements to return slots to empty state
while (TryPop(out _)) { }
}
public IEnumerator<T> GetEnumerator()
{
T[] snap = ToArraySnapshot();
for (int i = 0; i < snap.Length; i++) {
yield return snap[i];
}
}
public T Peek()
{
if (TryPeek(out T? v)) {
return v;
}
throw new InvalidOperationException("Stack is empty.");
}
public T Pop()
{
if (TryPop(out T value)) {
return value;
}
throw new InvalidOperationException("Stack is empty.");
}
public T[] PopRange(int count)
{
if (count <= 0) {
return Array.Empty<T>();
}
T[] result = new T[count];
int i = 0;
for (; i < count; i++)
{
if (!TryPop(out result[i]))
{
break;
}
}
if (i == count) {
return result;
}
if (i == 0) {
return Array.Empty<T>();
}
T[] trimmed = new T[i];
Array.Copy(result, trimmed, i);
return trimmed;
}
public void Push(T item)
{
SpinWait sw = default;
while (true)
{
int oldTop = Volatile.Read(ref _top);
if (oldTop >= _capacity) {
throw new InvalidOperationException("Stack is full.");
}
if (Interlocked.CompareExchange(ref _top, oldTop + 1, oldTop) != oldTop)
{
sw.SpinOnce();
continue; // failed to reserve slot
}
// We exclusively own slot oldTop. Publish value then state.
_values[oldTop] = item;
Volatile.Write(ref _states[oldTop], 1); // publish
_ = Interlocked.Increment(ref _count);
return;
}
}
public void PushRange(ReadOnlySpan<T> span)
{
for (int i = 0; i < span.Length; i++) {
Push(span[i]);
}
}
/// <summary>
/// Best-effort snapshot for enumeration; not strongly consistent under concurrent
/// mutations.
/// </summary>
public T[] ToArraySnapshot()
{
int expected = Math.Min(Volatile.Read(ref _count), _capacity);
if (expected == 0) {
return Array.Empty<T>();
}
T[] snap = new T[expected];
int copied = 0;
int idx = Volatile.Read(ref _top) - 1;
while (idx >= 0 && copied < expected)
{
if (Volatile.Read(ref _states[idx]) == 1) {
snap[copied++] = _values[idx];
}
idx--;
}
if (copied == snap.Length) {
return snap;
}
T[] trimmed = new T[copied];
Array.Copy(snap, trimmed, copied);
return trimmed;
}
public bool TryPeek(out T value)
{
value = default!;
SpinWait sw = default;
while (true)
{
int top = Volatile.Read(ref _top);
if (top <= 0) {
return false;
}
int idx = top - 1;
// If not yet published or racing, retry/spin
if (Volatile.Read(ref _states[idx]) == 0)
{
if (Volatile.Read(ref _top) != top)
{
continue; // head moved, retry
}
sw.SpinOnce();
continue;
}
T v = _values[idx];
// Validate consistency
if (Volatile.Read(ref _top) == top && Volatile.Read(ref _states[idx]) == 1)
{
value = v;
return true;
}
}
}
public bool TryPop(out T value)
{
value = default!;
SpinWait sw = default;
while (true)
{
int oldTop = Volatile.Read(ref _top);
if (oldTop <= 0)
{
return false; // empty
}
int newTop = oldTop - 1; // index to consume
if (Interlocked.CompareExchange(ref _top, newTop, oldTop) != oldTop)
{
sw.SpinOnce();
continue; // failed to reserve element
}
// If producer hasn't published yet, roll back our reservation and retry.
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);
sw.SpinOnce();
continue;
}
value = _values[newTop];
_values[newTop] = default!; // avoid ref retention
Volatile.Write(ref _states[newTop], 0); // mark empty
Interlocked.Decrement(ref _count);
return true;
}
}
public int TryPopRange(Span<T> destination)
{
if (destination.IsEmpty) {
return 0;
}
int i = 0;
for (; i < destination.Length; i++)
{
if (!TryPop(out destination[i]))
{
break;
}
}
return i;
}
IEnumerator IEnumerable.GetEnumerator() => GetEnumerator();
}
}
+16 -16
View File
@@ -171,22 +171,22 @@ namespace StgSharp
#pragma warning restore CA1027
{
Joystick_1 = glConst.GLFW_JOYSTICK_1,
Joystick_2 = glConst.GLFW_JOYSTICK_2,
Joystick_3 = glConst.GLFW_JOYSTICK_3,
Joystick_4 = glConst.GLFW_JOYSTICK_4,
Joystick_5 = glConst.GLFW_JOYSTICK_5,
Joystick_6 = glConst.GLFW_JOYSTICK_6,
Joystick_7 = glConst.GLFW_JOYSTICK_7,
Joystick_8 = glConst.GLFW_JOYSTICK_8,
Joystick_9 = glConst.GLFW_JOYSTICK_9,
Joystick_10 = glConst.GLFW_JOYSTICK_10,
Joystick_11 = glConst.GLFW_JOYSTICK_11,
Joystick_12 = glConst.GLFW_JOYSTICK_12,
Joystick_13 = glConst.GLFW_JOYSTICK_13,
Joystick_14 = glConst.GLFW_JOYSTICK_14,
Joystick_15 = glConst.GLFW_JOYSTICK_15,
Joystick_16 = glConst.GLFW_JOYSTICK_16,
Joystick_1 = Constant.GLFW_JOYSTICK_1,
Joystick_2 = Constant.GLFW_JOYSTICK_2,
Joystick_3 = Constant.GLFW_JOYSTICK_3,
Joystick_4 = Constant.GLFW_JOYSTICK_4,
Joystick_5 = Constant.GLFW_JOYSTICK_5,
Joystick_6 = Constant.GLFW_JOYSTICK_6,
Joystick_7 = Constant.GLFW_JOYSTICK_7,
Joystick_8 = Constant.GLFW_JOYSTICK_8,
Joystick_9 = Constant.GLFW_JOYSTICK_9,
Joystick_10 = Constant.GLFW_JOYSTICK_10,
Joystick_11 = Constant.GLFW_JOYSTICK_11,
Joystick_12 = Constant.GLFW_JOYSTICK_12,
Joystick_13 = Constant.GLFW_JOYSTICK_13,
Joystick_14 = Constant.GLFW_JOYSTICK_14,
Joystick_15 = Constant.GLFW_JOYSTICK_15,
Joystick_16 = Constant.GLFW_JOYSTICK_16,
First = Joystick_1,
Last = Joystick_16
@@ -65,7 +65,7 @@ namespace StgSharp.HighPerformance.Memory
return new ConcurrentBufferStack<T>(4); // Default small capacity
}
ConcurrentBufferStack<T> stack = new ConcurrentBufferStack<T>(span.Length);
ConcurrentBufferStack<T> stack = new(span.Length);
// Use internal method for efficient bulk initialization
@@ -106,12 +106,20 @@ namespace StgSharp.HighPerformance.Memory
public void Clear()
{
Header* header = (Header*)_buffer;
_lock.EnterExpansionProcess();
try
{
header->Top = 0;
Volatile.Write(ref _count, 0);
_lock.EnterBufferCopy(); // exclusive while mutating Top
try
{
Header* header = (Header*)_buffer;
header->Top = 0;
Volatile.Write(ref _count, 0);
}
finally
{
_lock.ExitBufferCopy();
}
}
finally
{
@@ -127,29 +135,29 @@ namespace StgSharp.HighPerformance.Memory
public T Pop()
{
Header* header = (Header*)_buffer;
while (true)
{
nuint oldTop = Volatile.Read(ref header->Top);
if (oldTop <= 0) {
throw new InvalidOperationException("Stack underflow");
}
if (Interlocked.CompareExchange(ref header->Top, oldTop - 1, oldTop) == oldTop)
_lock.EnterBufferRead(); // prevent buffer being reallocated while reading/modifying Top
try
{
_lock.EnterBufferRead();
try
Header* header = (Header*)_buffer;
nuint oldTop = Volatile.Read(ref header->Top);
if (oldTop <= 0) {
throw new InvalidOperationException("Stack underflow");
}
if (Interlocked.CompareExchange(ref header->Top, oldTop - 1, oldTop) == oldTop)
{
T* dataStart = (T*)(_buffer + sizeof(Header));
_ = Interlocked.Decrement(ref _count);
return dataStart[oldTop - 1];
}
finally
{
_lock.ExitBufferRead();
}
}
Thread.Sleep(0); // Yield to avoid busy waiting
finally
{
_lock.ExitBufferRead();
}
Thread.Sleep(0);
}
}
@@ -157,14 +165,16 @@ namespace StgSharp.HighPerformance.Memory
{
while (true)
{
// Ensure capacity, expanding if needed
_lock.EnterMetaDataRead();
try
{
Header* header = (Header*)_buffer;
nuint oldTop = Volatile.Read(ref header->Top);
nuint topSnapshot = Volatile.Read(ref header->Top);
nuint cap = Volatile.Read(ref header->Capacity);
if (oldTop >= cap)
if (topSnapshot >= cap)
{
// Need expansion
_lock.ExitMetaDataRead();
_lock.EnterExpansionProcess();
try
@@ -193,22 +203,6 @@ namespace StgSharp.HighPerformance.Memory
}
continue;
}
if (Interlocked.CompareExchange(ref header->Top, oldTop + 1, oldTop) == oldTop)
{
_lock.EnterBufferRead();
try
{
T* dataStart = (T*)(_buffer + sizeof(Header));
dataStart[oldTop] = item;
_ = Interlocked.Increment(ref _count);
return;
}
finally
{
_lock.ExitBufferRead();
}
}
}
finally
{
@@ -216,39 +210,63 @@ namespace StgSharp.HighPerformance.Memory
_lock.ExitMetaDataRead();
}
}
Thread.Sleep(0); // Yield to avoid busy waiting
// Reserve and write under exclusive buffer lock to avoid consumers reading unwritten data
_lock.EnterBufferCopy();
try
{
Header* header = (Header*)_buffer;
nuint idx = header->Top; // reserve next slot
if (idx >= header->Capacity)
{
// lost race against expander or other producer; retry
continue;
}
T* dataStart = (T*)(_buffer + sizeof(Header));
dataStart[idx] = item; // write first
header->Top = idx + 1; // then publish by increasing top
_ = Interlocked.Increment(ref _count);
return;
}
finally
{
_lock.ExitBufferCopy();
}
}
}
public bool TryPop(out T value)
{
Header* header = (Header*)_buffer;
while (true)
{
nuint oldTop = Volatile.Read(ref header->Top);
if (oldTop <= 0)
_lock.EnterBufferRead();
try
{
value = default;
return false;
}
Header* header = (Header*)_buffer;
nuint oldTop = Volatile.Read(ref header->Top);
if (oldTop <= 0)
{
value = default;
return false;
}
if (Interlocked.CompareExchange(ref header->Top, oldTop - 1, oldTop) == oldTop)
{
_lock.EnterBufferRead();
try
if (Interlocked.CompareExchange(ref header->Top, oldTop - 1, oldTop) == oldTop)
{
T* dataStart = (T*)(_buffer + sizeof(Header));
value = dataStart[oldTop - 1];
_ = Interlocked.Decrement(ref _count);
return true;
}
finally
{
_lock.ExitBufferRead();
}
return true;
}
Thread.Sleep(0); // Yield to avoid busy waiting
finally
{
_lock.ExitBufferRead();
}
value = default;
return true;
}
Thread.Sleep(0); // Yield to avoid busy waiting
}
// Internal method for efficient bulk initialization
+1 -1
View File
@@ -38,7 +38,7 @@ namespace StgSharp.Internal
internal static partial class GraphicFramework
{
internal static uint[] GLtype = [];
internal static uint[] GLtype = new uint[16];
static GraphicFramework()
{
@@ -42,7 +42,7 @@ namespace StgSharp.Threading
public class BufferExpansionLock : IDisposable
{
private static readonly ThreadLocal<int> Count = new();
private static readonly ThreadLocal<int> Count = new(() => 0);
private readonly ReaderWriterLockSlim _copyLock;
private readonly ReaderWriterLockSlim _expansionLock;
@@ -74,13 +74,11 @@ namespace StgSharp.Threading
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void EnterBufferRead()
{
if (++Count.Value == 1)
{
int c = Count.Value;
if (c == 0) {
_copyLock.EnterReadLock();
} else
{
Count.Value++;
}
Count.Value = c + 1;
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
@@ -104,12 +102,10 @@ namespace StgSharp.Threading
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void ExitBufferRead()
{
if (--Count.Value == 0)
{
int c = Count.Value - 1;
Count.Value = c;
if (c == 0) {
_copyLock.ExitReadLock();
} else
{
Count.Value--;
}
}
@@ -91,6 +91,7 @@ namespace StgSharp.Mathematics.Numeric
* wait
*/
}
MatrixParallel.ReleaseParallelLeadership(ref pool);
MatrixKernel<float>* leftPtr = left.Buffer;
MatrixKernel<float>* rightPtr = right.Buffer;
MatrixKernel<float>* ansPtr = ans.Buffer;
@@ -98,6 +99,8 @@ namespace StgSharp.Mathematics.Numeric
NativeIntrinsic.Intrinsic.f32_sub(leftPtr + i, rightPtr + i, ansPtr + i);
}
// thread.
return ans;
}
Stack<MatrixParallelWrap> taskGroup = MatrixParallel.PublicTask(left.GetColumnEnumeration(), right.GetColumnEnumeration(), ans.GetColumnEnumeration(),
@@ -820,9 +820,9 @@ namespace StgSharp.Mathematics.Numeric
int tileWidth = (1024 / secCount) + 1;
int tileCount = (primCount + tileWidth - 1) / tileWidth;
MatrixParallelPool pool = LeadParallel();
CapacityFixedStack<MatrixParallelThread> pool = LeadParallel();
int threadCount = pool.SpareThreadCount;
int threadCount = pool.Count;
int tilePerThreadLess = tileCount / threadCount; // tiles per thread without extra slice
int columnAfterBaseTile = tileCount % threadCount * tileWidth; // tile sliced into columns
int extraColumnPerThread = columnAfterBaseTile / threadCount;
@@ -833,7 +833,7 @@ namespace StgSharp.Mathematics.Numeric
MatrixParallelTaskPackage<T>* current = package;
while (true)
{
if (!pool.TryRequestWrap(out MatrixParallelWrap? wrap) || wrap is null)
if (!pool.TryPop(out MatrixParallelThread? wrap) || wrap is null)
{
#if DEBUG
throw new InvalidOperationException("Someone steal parallel wraps from the pool.");
@@ -846,7 +846,7 @@ namespace StgSharp.Mathematics.Numeric
bool isLargeSlice = extraColumnCount > 0;
// count tiles for wrap
int wrapWidth = wrap!.WrapTaskCapacity;
int wrapWidth =/**/ 3/* wrap!.WrapTaskCapacity/**/;
int columnFromSlicingCount = (hasSlice ? extraColumnPerThread : 0) * wrapWidth;
int baseColumn = wrapWidth * tilePerThreadLess * tileWidth;
int columnInWrap = baseColumn + columnFromSlicingCount + (isLargeSlice ? 1 : 0);
@@ -872,8 +872,9 @@ namespace StgSharp.Mathematics.Numeric
total -= columnInWrap;
// set to wrap and push
wrap.WrapTask = (MatrixParallelTaskPackageNonGeneric*)current;
taskBunch.Push(wrap);
wrap.Tasks.Enqueue((IntPtr)current);
// taskBunch.Push(wrap);
if (total <= 0)
@@ -65,6 +65,7 @@ namespace StgSharp.Mathematics.Numeric
internal CapacityFixedStack<MatrixParallelWrap> SpareWrap { get; set; }
/*
public void ReturnIndividualHandle(ref MatrixParallelWrap.Handle handle)
{
MatrixParallelWrap wrap = handle.Wrap;
@@ -108,6 +109,6 @@ namespace StgSharp.Mathematics.Numeric
wrap = null!;
return false;
}
/**/
}
}
@@ -96,6 +96,11 @@ namespace StgSharp.Mathematics.Numeric
_resetEvent.Set();
}
public void ReturnToPool()
{
// MatrixParallel.(this);
}
protected virtual void Dispose(bool disposing)
{
if (!disposedValue)
+117 -3
View File
@@ -2,8 +2,8 @@
// -----------------------------------------------------------------------
// file="Program"
// Project: StgSharp
// AuthorGroup: Nitload Space
// Copyright (c) Nitload Space. All rights reserved.
// 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
@@ -40,10 +40,12 @@ using StgSharp.Script;
using StgSharp.Script.Express;
using System;
using System.Collections.Concurrent;
using System.Diagnostics;
using System.Runtime.CompilerServices;
using System.Runtime.Intrinsics;
using System.Runtime.Intrinsics.X86;
using System.Threading;
using System.Xml.XPath;
namespace StgSharp
@@ -51,7 +53,7 @@ namespace StgSharp
internal static partial class Program
{
private static string testExpPath = @"D:\Programmer\stepcode\data\ap203\ap203.exp";
private static string testExpPath = @"D:\\Programmer\\stepcode\\data\\ap203\\ap203.exp";
internal static unsafe void Main(string[] args)
{
@@ -59,6 +61,118 @@ namespace StgSharp
{
World.Init();
// New test with synchronized start:
// - 8 producer threads, each pushing 1000 unique integers
// - 8 consumer threads, each attempting to pop 1000 times
// - Consumers print one line per successful pop
// - Finally, print popped/remaining counts and values
const int producerThreads = 8;
const int consumerThreads = 8;
const int perProducer = 1000;
const int perConsumerAttempts = 1000;
ConcurrentBufferStack<int> stack = new ConcurrentBufferStack<int>();
Thread[] producers = new Thread[producerThreads];
Thread[] consumers = new Thread[consumerThreads];
ConcurrentBag<int> popped = new ConcurrentBag<int>();
int totalThreads = producerThreads + consumerThreads;
using CountdownEvent ready = new CountdownEvent(totalThreads);
using ManualResetEventSlim startGate = new ManualResetEventSlim(false);
for (int c = 0; c < consumerThreads; c++)
{
consumers[c] = new Thread(() => {
ready.Signal();
startGate.Wait();
for (int k = 0; k < perConsumerAttempts; k++)
{
if (stack.TryPop(out int v))
{
popped.Add(v);
Console.WriteLine($"pop: {v}");
} else
{
Thread.Yield();
}
}
})
{
IsBackground = true
};
}
for (int p = 0; p < producerThreads; p++)
{
int captured = p;
producers[p] = new Thread(() => {
ready.Signal();
startGate.Wait();
int begin = captured * perProducer;
for (int j = 0; j < perProducer; j++)
{
stack.Push(begin + j);
if ((j & 0xFF) == 0) {
Thread.Yield();
}
}
})
{
IsBackground = true
};
}
foreach (Thread t in consumers) {
t.Start();
}
foreach (Thread t in producers) {
t.Start();
}
// Ensure all threads are ready, then start simultaneously
ready.Wait();
startGate.Set();
foreach (Thread t in producers) {
t.Join();
}
foreach (Thread t in consumers) {
t.Join();
}
// Gather remaining data
List<int> remaining = new System.Collections.Generic.List<int>();
while (stack.TryPop(out int value)) {
remaining.Add(value);
}
Console.WriteLine($"Popped count: {popped.Count}");
Console.WriteLine("Popped values:");
foreach (int v in popped)
{
Console.Write(v);
Console.Write('\t');
}
Console.WriteLine();
Console.WriteLine($"Remaining count: {remaining.Count}");
Console.WriteLine("Remaining values:");
foreach (int v in remaining)
{
Console.Write(v);
Console.Write('\t');
}
Console.WriteLine();
// Sanity check
Console.WriteLine($"Produced: {producerThreads * perProducer}, Popped+Remaining: {popped.Count + remaining.Count}");
/*
// TestMatrixParallel();
// 原有的矩阵测试代码