From ea0e6f3ff66ac2b8f052d20def70a56ff2ccffd7 Mon Sep 17 00:00:00 2001 From: Youmu Liu <105042599+Nitload-NSI@users.noreply.github.com> Date: Wed, 5 Nov 2025 20:33:58 +0800 Subject: [PATCH] optimize para performance of buffered stack --- .../ConcurrentCapacityFixedStack.cs | 287 ++++++++++++++++++ StgSharp.Common/Controlling/KeyEnum.cs | 32 +- .../Memory/ConcurrentBufferStack.cs | 124 ++++---- StgSharp.Common/Internal/Graphic/gl.cs | 2 +- .../Threading/BufferExpansionLock.cs | 18 +- .../Numeric/MatrixCompute.Basic.cs | 3 + .../Numeric/MatrixParallel.Publish.cs | 13 +- .../Numeric/MatrixParallelPool.cs | 3 +- .../Numeric/MatrixParallelThread.cs | 5 + StgSharpDebug/Program.cs | 120 +++++++- 10 files changed, 516 insertions(+), 91 deletions(-) create mode 100644 StgSharp.Common/Collections/ConcurrentCapacityFixedStack.cs diff --git a/StgSharp.Common/Collections/ConcurrentCapacityFixedStack.cs b/StgSharp.Common/Collections/ConcurrentCapacityFixedStack.cs new file mode 100644 index 00000000..69dadc9e --- /dev/null +++ b/StgSharp.Common/Collections/ConcurrentCapacityFixedStack.cs @@ -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 +{ + /// + /// 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 + /// + public class ConcurrentCapacityFixedStack : IEnumerable + { + + 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 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[] 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[] 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 span) + { + for (int i = 0; i < span.Length; i++) { + Push(span[i]); + } + } + + /// + /// Best-effort snapshot for enumeration; not strongly consistent under concurrent + /// mutations. + /// + public T[] ToArraySnapshot() + { + int expected = Math.Min(Volatile.Read(ref _count), _capacity); + if (expected == 0) { + return Array.Empty(); + } + + 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 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(); + + } +} diff --git a/StgSharp.Common/Controlling/KeyEnum.cs b/StgSharp.Common/Controlling/KeyEnum.cs index 5505e64a..3e94f4ed 100644 --- a/StgSharp.Common/Controlling/KeyEnum.cs +++ b/StgSharp.Common/Controlling/KeyEnum.cs @@ -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 diff --git a/StgSharp.Common/HighPerformance/Memory/ConcurrentBufferStack.cs b/StgSharp.Common/HighPerformance/Memory/ConcurrentBufferStack.cs index 99171b6b..e045af13 100644 --- a/StgSharp.Common/HighPerformance/Memory/ConcurrentBufferStack.cs +++ b/StgSharp.Common/HighPerformance/Memory/ConcurrentBufferStack.cs @@ -65,7 +65,7 @@ namespace StgSharp.HighPerformance.Memory return new ConcurrentBufferStack(4); // Default small capacity } - ConcurrentBufferStack stack = new ConcurrentBufferStack(span.Length); + ConcurrentBufferStack 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 diff --git a/StgSharp.Common/Internal/Graphic/gl.cs b/StgSharp.Common/Internal/Graphic/gl.cs index 07ac1aaa..02a7dc6d 100644 --- a/StgSharp.Common/Internal/Graphic/gl.cs +++ b/StgSharp.Common/Internal/Graphic/gl.cs @@ -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() { diff --git a/StgSharp.Common/Threading/BufferExpansionLock.cs b/StgSharp.Common/Threading/BufferExpansionLock.cs index 6ecbc3f9..df5fbedb 100644 --- a/StgSharp.Common/Threading/BufferExpansionLock.cs +++ b/StgSharp.Common/Threading/BufferExpansionLock.cs @@ -42,7 +42,7 @@ namespace StgSharp.Threading public class BufferExpansionLock : IDisposable { - private static readonly ThreadLocal Count = new(); + private static readonly ThreadLocal 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--; } } diff --git a/StgSharp.Mathematics/Numeric/MatrixCompute.Basic.cs b/StgSharp.Mathematics/Numeric/MatrixCompute.Basic.cs index 2af8c137..cfdc1a55 100644 --- a/StgSharp.Mathematics/Numeric/MatrixCompute.Basic.cs +++ b/StgSharp.Mathematics/Numeric/MatrixCompute.Basic.cs @@ -91,6 +91,7 @@ namespace StgSharp.Mathematics.Numeric * wait */ } + MatrixParallel.ReleaseParallelLeadership(ref pool); MatrixKernel* leftPtr = left.Buffer; MatrixKernel* rightPtr = right.Buffer; MatrixKernel* 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 taskGroup = MatrixParallel.PublicTask(left.GetColumnEnumeration(), right.GetColumnEnumeration(), ans.GetColumnEnumeration(), diff --git a/StgSharp.Mathematics/Numeric/MatrixParallel.Publish.cs b/StgSharp.Mathematics/Numeric/MatrixParallel.Publish.cs index 894aa176..1e138806 100644 --- a/StgSharp.Mathematics/Numeric/MatrixParallel.Publish.cs +++ b/StgSharp.Mathematics/Numeric/MatrixParallel.Publish.cs @@ -820,9 +820,9 @@ namespace StgSharp.Mathematics.Numeric int tileWidth = (1024 / secCount) + 1; int tileCount = (primCount + tileWidth - 1) / tileWidth; - MatrixParallelPool pool = LeadParallel(); + CapacityFixedStack 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* 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) diff --git a/StgSharp.Mathematics/Numeric/MatrixParallelPool.cs b/StgSharp.Mathematics/Numeric/MatrixParallelPool.cs index bcaa6b75..0370dfaa 100644 --- a/StgSharp.Mathematics/Numeric/MatrixParallelPool.cs +++ b/StgSharp.Mathematics/Numeric/MatrixParallelPool.cs @@ -65,6 +65,7 @@ namespace StgSharp.Mathematics.Numeric internal CapacityFixedStack 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; } - + /**/ } } diff --git a/StgSharp.Mathematics/Numeric/MatrixParallelThread.cs b/StgSharp.Mathematics/Numeric/MatrixParallelThread.cs index c4fa8070..62e6b10e 100644 --- a/StgSharp.Mathematics/Numeric/MatrixParallelThread.cs +++ b/StgSharp.Mathematics/Numeric/MatrixParallelThread.cs @@ -96,6 +96,11 @@ namespace StgSharp.Mathematics.Numeric _resetEvent.Set(); } + public void ReturnToPool() + { + // MatrixParallel.(this); + } + protected virtual void Dispose(bool disposing) { if (!disposedValue) diff --git a/StgSharpDebug/Program.cs b/StgSharpDebug/Program.cs index f4835a5b..b583fc05 100644 --- a/StgSharpDebug/Program.cs +++ b/StgSharpDebug/Program.cs @@ -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 stack = new ConcurrentBufferStack(); + Thread[] producers = new Thread[producerThreads]; + Thread[] consumers = new Thread[consumerThreads]; + ConcurrentBag popped = new ConcurrentBag(); + + 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 remaining = new System.Collections.Generic.List(); + 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(); // 原有的矩阵测试代码