Put thread pool to individual type define as thread resource manager.

This commit is contained in:
Youmu Liu
2025-10-22 20:35:05 +08:00
parent 3788db4cd0
commit 3707e5bd74
4 changed files with 93 additions and 20 deletions
@@ -28,6 +28,7 @@
using System; using System;
using System.Collections; using System.Collections;
using System.Collections.Generic; using System.Collections.Generic;
using System.Reflection;
namespace StgSharp.Collections namespace StgSharp.Collections
{ {
@@ -72,6 +73,16 @@ namespace StgSharp.Collections
} }
} }
public void RemoveAt(int index)
{
if (index >= 0)
{
_buffer[index] = _buffer[Count - 1];
_buffer[Count - 1] = default!;
Count--;
}
}
IEnumerator IEnumerable.GetEnumerator() IEnumerator IEnumerable.GetEnumerator()
{ {
return GetEnumerator(); return GetEnumerator();
@@ -820,27 +820,27 @@ namespace StgSharp.Mathematics.Numeric
int tileWidth = (1024 / secCount) + 1; int tileWidth = (1024 / secCount) + 1;
int tileCount = (primCount / tileWidth) + 1; int tileCount = (primCount / tileWidth) + 1;
CapacityFixedStack<MatrixParallelWrap> wraps = LeadParallel(); MatrixParallelPool pool = LeadParallel();
int tilePerThreadLess = tileCount / wraps.Count; int tilePerThreadLess = tileCount / pool.SpareThreadCount;
int tileToSlice = tileCount % wraps.Count; int tileToSlice = tileCount % pool.SpareThreadCount;
int slicingPerWrap = tileToSlice / wraps.Count; int slicingPerWrap = tileToSlice / pool.SpareThreadCount;
int extraTiles = tileToSlice % wraps.Count; int extraTiles = tileToSlice % pool.SpareThreadCount;
int total = tileCount; int total = tileCount;
int begin = 0; int begin = 0;
Stack<MatrixParallelWrap> taskBunch = new Stack<MatrixParallelWrap>(wraps.Count); Stack<MatrixParallelWrap> taskBunch = new Stack<MatrixParallelWrap>(pool.SpareThreadCount);
while (tileCount > 0) while (tileCount > 0)
{ {
MatrixParallelWrap wrap = wraps.Pop(); _ = pool.TryRequestWrap(out MatrixParallelWrap? wrap);
bool hasSlice = tileToSlice > 0; bool hasSlice = tileToSlice > 0;
bool isLargeSlice = extraTiles > 0; bool isLargeSlice = extraTiles > 0;
// count tiles for wrap // count tiles for wrap
// TODO 此处应该是线程数量而不是线程束数量 // TODO 此处应该是线程数量而不是线程束数量
// TODO 需要想办法正确同步线程数量信息 // TODO 需要想办法正确同步线程数量信息
int toSlice = hasSlice ? wrap.WrapTaskCapacity * slicingPerWrap : 0; int toSlice = hasSlice ? wrap!.WrapTaskCapacity * slicingPerWrap : 0;
int baseTile = wrap.WrapTaskCapacity * tilePerThreadLess; int baseTile = wrap!.WrapTaskCapacity * tilePerThreadLess;
int tileInWrap = baseTile + toSlice + (isLargeSlice ? 1 : 0); int tileInWrap = baseTile + toSlice + (isLargeSlice ? 1 : 0);
int columnInWrap = tileInWrap * tileWidth; int columnInWrap = tileInWrap * tileWidth;
@@ -857,7 +857,7 @@ namespace StgSharp.Mathematics.Numeric
begin += tileInWrap; begin += tileInWrap;
taskBunch.Push(wrap); taskBunch.Push(wrap);
} }
ReleaseParallelLeadership(ref wraps); ReleaseParallelLeadership(ref pool);
return taskBunch; return taskBunch;
} }
@@ -28,6 +28,7 @@
using StgSharp.Collections; using StgSharp.Collections;
using System; using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Diagnostics.CodeAnalysis;
using System.Linq; using System.Linq;
using System.Text; using System.Text;
using System.Threading.Tasks; using System.Threading.Tasks;
@@ -46,34 +47,63 @@ namespace StgSharp.Mathematics.Numeric
{ {
SpareWrap = CapacityFixedStackBuilder.Create(wrapSpan); SpareWrap = CapacityFixedStackBuilder.Create(wrapSpan);
SingleThreadAllocation = new CapacityFixedBag<MatrixParallelWrap>(wrapSpan.Length); SingleThreadAllocation = new CapacityFixedBag<MatrixParallelWrap>(wrapSpan.Length);
SpareThreads = threadCount; SpareThreadCount = threadCount;
} }
internal MatrixParallelPool(int threadCount, CapacityFixedStack<MatrixParallelWrap> wrapSpan) internal MatrixParallelPool(int threadCount, CapacityFixedStack<MatrixParallelWrap> wrapSpan)
{ {
SpareWrap = wrapSpan; SpareWrap = wrapSpan;
SingleThreadAllocation = new CapacityFixedBag<MatrixParallelWrap>(wrapSpan.Count); SingleThreadAllocation = new CapacityFixedBag<MatrixParallelWrap>(wrapSpan.Count);
SpareThreads = threadCount; SpareThreadCount = threadCount;
} }
public int SpareThreads { get; set; } public int SpareThreadCount { get; set; }
internal CapacityFixedBag<MatrixParallelWrap> SingleThreadAllocation { get; set; } internal CapacityFixedBag<MatrixParallelWrap> SingleThreadAllocation { get; set; }
internal CapacityFixedStack<MatrixParallelWrap> SpareWrap { get; set; } internal CapacityFixedStack<MatrixParallelWrap> SpareWrap { get; set; }
public bool TryRequestLeaderHandle(out MatrixParallelWrap.Handle handle) { } public void ReturnIndividualHandle(ref MatrixParallelWrap.Handle handle)
{
MatrixParallelWrap wrap = handle.Wrap;
handle.Role = 0;
if (wrap.IsAllAvailable)
{
SingleThreadAllocation.RemoveAt(wrap.Magic);
SpareWrap.Push(wrap);
SpareThreadCount += wrap.WrapTaskCapacity;
}
}
public bool TryRequestWrap(out MatrixParallelWrap? wrap) public bool TryRentIndividualHandle(out MatrixParallelWrap.Handle handle)
{
foreach (MatrixParallelWrap item in SingleThreadAllocation)
{
if (item.TryGetLeaderHandle(out handle!)) {
return true;
}
}
if (TryRequestWrap(out MatrixParallelWrap wrap))
{
wrap!.Magic = SingleThreadAllocation.Count;
SingleThreadAllocation.Add(wrap);
if (wrap.TryGetLeaderHandle(out handle!)) {
return true;
}
}
handle = null!;
return false;
}
public bool TryRequestWrap([NotNull]out MatrixParallelWrap? wrap)
{ {
if (SpareWrap.TryPop(out wrap)) if (SpareWrap.TryPop(out wrap))
{ {
SpareThreads -= wrap.WrapTaskCapacity; SpareThreadCount -= wrap.WrapTaskCapacity;
return true; return true;
} else
{
return false;
} }
wrap = null!;
return false;
} }
} }
@@ -45,6 +45,7 @@ namespace StgSharp.Mathematics.Numeric
{ {
ID = id; ID = id;
_threads = new Thread[wrapSize]; _threads = new Thread[wrapSize];
_resetHandle = new Handle[wrapSize];
for (int i = 0; i < wrapSize; i++) for (int i = 0; i < wrapSize; i++)
{ {
_threads[i] = new Thread(MatrixParallelWorker) _threads[i] = new Thread(MatrixParallelWorker)
@@ -53,7 +54,23 @@ namespace StgSharp.Mathematics.Numeric
Priority = ThreadPriority.BelowNormal, Priority = ThreadPriority.BelowNormal,
Name = $"MPT#{ID}_{i}" Name = $"MPT#{ID}_{i}"
}; };
_threads[i].Start(new Handle(i, this)); Handle handle = new Handle(i, this);
_resetHandle[i] = handle;
_threads[i].Start(handle);
}
}
public bool IsAllAvailable
{
get
{
foreach (Handle item in _resetHandle)
{
if (item.Role != 0) {
return false;
}
}
return true;
} }
} }
@@ -75,6 +92,21 @@ namespace StgSharp.Mathematics.Numeric
} }
} }
internal bool TryGetLeaderHandle(out Handle? handle)
{
foreach (Handle item in _resetHandle)
{
if (item.Role != 2)
{
handle = item;
item.Role = 2;
return true;
}
}
handle = null;
return false;
}
public class Handle(int inWrapIndex, MatrixParallelWrap wrap) public class Handle(int inWrapIndex, MatrixParallelWrap wrap)
{ {