working on task publication

This commit is contained in:
Youmu Liu
2025-09-15 21:08:13 +08:00
parent ec25525838
commit fa54037a44
3 changed files with 24 additions and 15 deletions
@@ -39,7 +39,7 @@ using System.Threading.Tasks;
namespace StgSharp.Mathematics namespace StgSharp.Mathematics
{ {
public static partial class MatrixParallel internal static partial class MatrixParallel
{ {
[MethodImpl(MethodImplOptions.AggressiveInlining)] [MethodImpl(MethodImplOptions.AggressiveInlining)]
@@ -64,7 +64,7 @@ namespace StgSharp.Mathematics
* we allow each wrap to contain either 3 or 4 threads, providing flexibility while maintaining efficiency. * we allow each wrap to contain either 3 or 4 threads, providing flexibility while maintaining efficiency.
*/ */
public static partial class MatrixParallel internal static partial class MatrixParallel
{ {
private static Thread[] _threads; private static Thread[] _threads;
@@ -80,12 +80,14 @@ namespace StgSharp.Mathematics
public static bool SupportParallel { get; } = Environment.ProcessorCount > 6; public static bool SupportParallel { get; } = Environment.ProcessorCount > 6;
public static bool IsParallelMeaningful { get; } = Environment.ProcessorCount > 6; public static bool IsParallelMeaningful { get; } = Environment.ProcessorCount > 6 && IsInitialized;
public static MatrixParallelWrap.Handle LeaderHandle { get; /*private/**/ set; } public static MatrixParallelWrap.Handle LeaderHandle { get; private/**/ set; }
private static bool IsInitialized { get; set; } = false; private static bool IsInitialized { get; set; } = false;
private static Queue<IntPtr> LeaderTask { get; set; }
public static void EndParallel() public static void EndParallel()
{ {
Monitor.Exit(_lock); Monitor.Exit(_lock);
@@ -134,6 +136,7 @@ namespace StgSharp.Mathematics
} }
} }
LeaderHandle = new MatrixParallelWrap.Handle(0, _wraps[0]); LeaderHandle = new MatrixParallelWrap.Handle(0, _wraps[0]);
LeaderTask = LeaderHandle.Wrap.GetWorkerTaskQueue(0);
IsInitialized = true; IsInitialized = true;
} }
@@ -149,7 +152,20 @@ namespace StgSharp.Mathematics
Monitor.Enter(_lock); Monitor.Enter(_lock);
} }
public static void PublishTask() { } [MethodImpl(MethodImplOptions.AggressiveInlining)]
public static unsafe void MatrixParallelLeaderWork()
{
MatrixParallelWrap.ExecuteTaskPackage(LeaderTask);
_ = Interlocked.Decrement(ref RemainThreadCount);
}
public static unsafe void PublishTask(MatrixParallelTaskPackage* package)
{
int primCountPerWrap = package->PrimCount / _threads.Length, extraPrimCount = package->PrimCount % _threads.Length;
int secCountPerWrap = package->SecCount / _threads.Length, extraSecCount = package->SecCount % _threads.Length;
foreach (MatrixParallelWrap wrap in _wraps) { }
}
[MethodImpl(MethodImplOptions.AggressiveInlining)] [MethodImpl(MethodImplOptions.AggressiveInlining)]
public static void ResetThreadCount() public static void ResetThreadCount()
@@ -248,15 +264,12 @@ namespace StgSharp.Mathematics
int inWrapIndex = handle.InWrapIndex; int inWrapIndex = handle.InWrapIndex;
Thread self = Thread.CurrentThread; Thread self = Thread.CurrentThread;
AutoResetEvent reset = wrap.GetWorkerResetEvent(inWrapIndex); AutoResetEvent reset = wrap.GetWorkerResetEvent(inWrapIndex);
Console.WriteLine($"[Worker] 线程 {self.Name} 启动,Wrap ID: {wrap.ID}, 索引: {inWrapIndex}");
do do
{ {
_ = reset.WaitOne(); _ = reset.WaitOne();
Console.WriteLine($"[Worker] 线程 {self.Name} 收到任务信号,开始执行");
self.Priority = ThreadPriority.Highest; self.Priority = ThreadPriority.Highest;
ExecuteTaskPackage(wrap.GetWorkerTaskQueue(inWrapIndex)); ExecuteTaskPackage(wrap.GetWorkerTaskQueue(inWrapIndex));
int remainCount = Interlocked.Decrement(ref RemainThreadCount); _ = Interlocked.Decrement(ref RemainThreadCount);
Console.WriteLine($"[Worker] 线程 {self.Name} 完成任务,剩余线程数: {remainCount--}");
self.Priority = _AfterWork; self.Priority = _AfterWork;
} while (true); } while (true);
} }
@@ -269,25 +282,21 @@ namespace StgSharp.Mathematics
Queue<IntPtr> taskQueue = wrap.GetWorkerTaskQueue(inWrapIndex); Queue<IntPtr> taskQueue = wrap.GetWorkerTaskQueue(inWrapIndex);
AutoResetEvent resetEvent = wrap.SchedulerReset; AutoResetEvent resetEvent = wrap.SchedulerReset;
Thread self = Thread.CurrentThread; Thread self = Thread.CurrentThread;
Console.WriteLine($"[Scheduler] 调度器线程 {self.Name} 启动,Wrap ID: {wrap.ID}, 索引: {inWrapIndex}");
do do
{ {
_ = resetEvent.WaitOne(); _ = resetEvent.WaitOne();
Console.WriteLine($"[Scheduler] 调度器线程 {self.Name} 收到调度信号,开始分发任务");
self.Priority = ThreadPriority.Highest; self.Priority = ThreadPriority.Highest;
// TODO task publication to worker in wrap // TODO task publication to worker in wrap
wrap.SetAllWorker(); wrap.SetAllWorker();
Console.WriteLine($"[Scheduler] 调度器线程 {self.Name} 已通知所有工作线程");
ExecuteTaskPackage(taskQueue); ExecuteTaskPackage(taskQueue);
int remain = Interlocked.Decrement(ref RemainThreadCount); _ = Interlocked.Decrement(ref RemainThreadCount);
Console.WriteLine($"[Scheduler] 调度器线程 {self.Name} 完成调度任务,剩余线程数: {remain--}");
self.Priority = _AfterWork; self.Priority = _AfterWork;
} while (true); } while (true);
} }
[MethodImpl(MethodImplOptions.AggressiveInlining)] [MethodImpl(MethodImplOptions.AggressiveInlining)]
private static unsafe void ExecuteTaskPackage(Queue<IntPtr> taskQueue) public static unsafe void ExecuteTaskPackage(Queue<IntPtr> taskQueue)
{ {
while (taskQueue.TryDequeue(out nint handle)) while (taskQueue.TryDequeue(out nint handle))
{ {