From 04fc55d31ee6a14b4a628a58568ce31fb2da007d Mon Sep 17 00:00:00 2001 From: Youmu Liu <105042599+Nitload-NSI@users.noreply.github.com> Date: Wed, 19 Nov 2025 18:09:25 +0800 Subject: [PATCH] add scheduler method; fixing buffer parallel errors --- .../Collections/CapacityFixedStack.cs | 6 +- .../Numeric/MatrixCompute.Basic.cs | 44 +++--- .../Numeric/MatrixComputeModel.cs | 38 +++-- .../Numeric/MatrixParallel.Publish.cs | 62 ++++++-- .../Numeric/MatrixParallel.cs | 2 +- .../Numeric/MatrixParallelHandle.cs | 10 +- .../Numeric/MatrixParallelTaskPackage.cs | 105 +------------ .../Numeric/MatrixParallelThread.cs | 51 ++++--- .../Numeric/MatrixParallelWrap.cs | 144 +++++++++++++++--- .../Numeric/MatrixTaskFactory.cs | 29 ++-- StgSharpDebug/Program.cs | 120 +-------------- 11 files changed, 272 insertions(+), 339 deletions(-) diff --git a/StgSharp.Common/Collections/CapacityFixedStack.cs b/StgSharp.Common/Collections/CapacityFixedStack.cs index bdb29dd7..32874bd6 100644 --- a/StgSharp.Common/Collections/CapacityFixedStack.cs +++ b/StgSharp.Common/Collections/CapacityFixedStack.cs @@ -93,7 +93,8 @@ namespace StgSharp.Collections [MethodImpl(MethodImplOptions.AggressiveInlining)] public T[] PopRange(int count) { - int i = int.Min(count, _index) - 1; + int i = int.Min(count, _index); + _index -= i; return _values[0..i]; } @@ -147,8 +148,7 @@ namespace StgSharp.Collections public static CapacityFixedStack Create(ReadOnlySpan values) { CapacityFixedStack stack = new(values.Length); - Span sp = stack.GetValueLayout(); - values.CopyTo(sp); + stack.PushRange(values); return stack; } diff --git a/StgSharp.Mathematics/Numeric/MatrixCompute.Basic.cs b/StgSharp.Mathematics/Numeric/MatrixCompute.Basic.cs index 05d9bf53..c8e6eed8 100644 --- a/StgSharp.Mathematics/Numeric/MatrixCompute.Basic.cs +++ b/StgSharp.Mathematics/Numeric/MatrixCompute.Basic.cs @@ -68,10 +68,10 @@ namespace StgSharp.Mathematics.Numeric MatrixParallel.ReturnParallelResources(ref pool); return ans; } - MatrixParallelTaskPackage* package = MatrixParallelFactory.CreateBaseTask(); - package->Left = left.Buffer; - package->Right = right.Buffer; - package->Result = ans.Buffer; + MatrixParallelTaskPackage* package = MatrixParallelFactory.CreateBaseTask(); + package->Left = (nint)left.Buffer; + package->Right = (nint)right.Buffer; + package->Result = (nint)ans.Buffer; package->ElementSize = sizeof(float); package->PrimCount = left.KernelColumnLength; package->SecCount = left.KernelRowLength; @@ -80,41 +80,37 @@ namespace StgSharp.Mathematics.Numeric MatrixParallelHandle taskGroup = MatrixParallel.ScheduleTask(package); MatrixParallel.LaunchParallel(taskGroup, SleepMode.DeepSleep); - /* - * run parallel - */ - MatrixParallelFactory.Release(package); taskGroup.Dispose(); return ans; } - public static unsafe void Fill(this Matrix right, T value) where T : unmanaged, INumber + public static unsafe void Fill(this Matrix ans, T value) where T : unmanaged, INumber { - long count = (long)right.KernelColumnLength * right.KernelRowLength; - ScalarPacket* scaler = MatrixParallelFactory.CreateScalarPacket(); - scaler->Data(0) = value; + long count = (long)ans.KernelColumnLength * ans.KernelRowLength; + ScalarPacket* scalar = MatrixParallelFactory.CreateScalarPacket(); + scalar->Data(0) = value; if (count <= 256) { - NativeIntrinsic.Intrinsic.f32_fill(right.Buffer, scaler, count); + NativeIntrinsic.Intrinsic.f32_fill(ans.Buffer, scalar, count); return; } else if (count <= 1024) { MatrixParallelThread[] pool = MatrixParallel.LeadParallel(1); - - NativeIntrinsic.Intrinsic.f32_fill(right.Buffer, scaler, count); - + NativeIntrinsic.Intrinsic.f32_fill(ans.Buffer, scalar, count); MatrixParallel.ReturnParallelResources(ref pool); return; } - MatrixParallelHandle taskGroup =/**/ default - /*MatrixParallel.PublicNoAnswerTask(left.GetColumnEnumeration(), scaler, - (delegate* unmanaged[Cdecl]*, ScalarPacket*, void>)NativeIntrinsic.Intrinsic.f32_add) - /**/; + MatrixParallelTaskPackage* package = MatrixParallelFactory.CreateBaseTask(); + package->Result = (nint)ans.Buffer; + package->ElementSize = sizeof(float); + package->PrimCount = ans.KernelColumnLength; + package->SecCount = ans.KernelRowLength; + package->ComputeMode = new MatrixOpMode(MatrixOperationStyle.BUFFER_OP, MatrixOperationParam.ANS_SCALAR_PARAM); + package->ComputeHandle = (IntPtr)NativeIntrinsic.Intrinsic.f32_fill; + package->Scalar = scalar; + MatrixParallelHandle taskGroup = MatrixParallel.ScheduleTask(package); MatrixParallel.LaunchParallel(taskGroup, SleepMode.DeepSleep); - /* - * run parallel - */ - MatrixParallelFactory.Release(scaler); + taskGroup.Dispose(); } diff --git a/StgSharp.Mathematics/Numeric/MatrixComputeModel.cs b/StgSharp.Mathematics/Numeric/MatrixComputeModel.cs index 619f71ae..bea0bb03 100644 --- a/StgSharp.Mathematics/Numeric/MatrixComputeModel.cs +++ b/StgSharp.Mathematics/Numeric/MatrixComputeModel.cs @@ -34,31 +34,33 @@ namespace StgSharp.Mathematics.Numeric internal static class MatrixComputeModel { - internal static unsafe void BufferComputeBinary(MatrixParallelTaskPackageNonGeneric* p) + internal static unsafe void BufferComputeBinary(MatrixParallelTaskPackage* p) { int size = p->ElementSize * 16; - int count = p->PrimCount * p->SecCount; - int offset = p->LeftPrimOffset * p->SecCount; + long count = Unsafe.As(ref p->PrimCount); + long offset = Unsafe.As(ref p->PrimTileOffset) * size; - ((delegate* unmanaged[Cdecl])p->ComputeHandle)((void*)(p->Left + offset * size), - (void*)(p->Right + offset * size), - (void*)(p->Result + offset * size), - count); + ((delegate* unmanaged[Cdecl])p->ComputeHandle)((void*)(p->Left + offset), + (void*)(p->Right + offset), + (void*)(p->Result + offset), + count); + MatrixParallelFactory.Release(p); } - internal static unsafe void BufferComputeUnaryScalar(MatrixParallelTaskPackageNonGeneric* p) + internal static unsafe void BufferComputeUnaryScalar(MatrixParallelTaskPackage* p) { int size = p->ElementSize * 16; - int count = p->PrimCount * p->SecCount; - int offset = p->LeftPrimOffset * p->SecCount; + long count = Unsafe.As(ref p->PrimCount); + long offset = Unsafe.As(ref p->PrimTileOffset) * size; - ((delegate* unmanaged[Cdecl])p->ComputeHandle)((void*)(p->Right + offset * size), - (void*)(p->Result + offset * size), + ((delegate* unmanaged[Cdecl])p->ComputeHandle)((void*)(p->Right + offset), + (void*)(p->Result + offset), count); + MatrixParallelFactory.Release(p); } [MethodImpl(MethodImplOptions.AggressiveInlining)] - internal static unsafe void Compute_Binary(MatrixParallelTaskPackageNonGeneric* p) + internal static unsafe void Compute_Binary(MatrixParallelTaskPackage* p) { int eSize = p->ElementSize; @@ -106,10 +108,11 @@ namespace StgSharp.Mathematics.Numeric op(left + (nint)lc, right + (nint)rc, ans + (nint)ac); } } + MatrixParallelFactory.Release(p); } [MethodImpl(MethodImplOptions.AggressiveInlining)] - internal static unsafe void Compute_BinaryScalar(MatrixParallelTaskPackageNonGeneric* p) + internal static unsafe void Compute_BinaryScalar(MatrixParallelTaskPackage* p) { int eSize = p->ElementSize; @@ -157,10 +160,11 @@ namespace StgSharp.Mathematics.Numeric op(left + (nint)lc, right + (nint)rc, ans + (nint)ac, p->Scalar); } } + MatrixParallelFactory.Release(p); } [MethodImpl(MethodImplOptions.AggressiveInlining)] - internal static unsafe void Compute_Unary(MatrixParallelTaskPackageNonGeneric* p) + internal static unsafe void Compute_Unary(MatrixParallelTaskPackage* p) { int eSize = p->ElementSize; @@ -201,10 +205,11 @@ namespace StgSharp.Mathematics.Numeric op(right + (nint)rc, ans + (nint)ac); } } + MatrixParallelFactory.Release(p); } [MethodImpl(MethodImplOptions.AggressiveInlining)] - internal static unsafe void Compute_UnaryScalar(MatrixParallelTaskPackageNonGeneric* p) + internal static unsafe void Compute_UnaryScalar(MatrixParallelTaskPackage* p) { int eSize = p->ElementSize; @@ -245,6 +250,7 @@ namespace StgSharp.Mathematics.Numeric op(right + (nint)rc, ans + (nint)ac, p->Scalar); } } + MatrixParallelFactory.Release(p); } } diff --git a/StgSharp.Mathematics/Numeric/MatrixParallel.Publish.cs b/StgSharp.Mathematics/Numeric/MatrixParallel.Publish.cs index 37acd586..29f35f4e 100644 --- a/StgSharp.Mathematics/Numeric/MatrixParallel.Publish.cs +++ b/StgSharp.Mathematics/Numeric/MatrixParallel.Publish.cs @@ -41,12 +41,14 @@ namespace StgSharp.Mathematics.Numeric public static partial class MatrixParallel { + private const int MIN_TASK_PER_THREAD = 256; + private const int REC_TASK_PER_THREAD = 1024; + [MethodImpl(MethodImplOptions.AggressiveInlining)] - internal static unsafe MatrixParallelHandle ScheduleTask(MatrixParallelTaskPackage* package) - where T : unmanaged,INumber + internal static unsafe MatrixParallelHandle ScheduleTask(MatrixParallelTaskPackage* package) { - MatrixParallelTaskPackageNonGeneric* nonGeneric = (MatrixParallelTaskPackageNonGeneric*)package; + MatrixParallelTaskPackage* nonGeneric = package; switch (nonGeneric->ComputeMode.OperationStyle) { case MatrixOperationStyle.PANEL_OP: @@ -60,12 +62,46 @@ namespace StgSharp.Mathematics.Numeric } } - private static unsafe MatrixParallelHandle ScheduleBufferTask(MatrixParallelTaskPackageNonGeneric* package) + private static unsafe MatrixParallelHandle ScheduleBufferTask(MatrixParallelTaskPackage* package) { - throw new NotImplementedException(); + int primCount = package->PrimCount; + int secCount = package->SecCount; + long count = (long)primCount * secCount; + + long recommendCount = count / 1024; + MatrixParallelThread[] pool = LeadParallel((int)recommendCount); + MatrixParallelHandle handle = new MatrixParallelHandle(pool); + if (handle.Wraps.Count() == 1) + { + ref long offsetRef = ref Unsafe.As(ref package->PrimTileOffset); + offsetRef = 0; + ref long lengthRef = ref Unsafe.As(ref package->PrimCount); + lengthRef = count; + handle.Wraps.First().WrapTask.Enqueue((nint)package); + return handle; + } + long actual = count / pool.Length; + long extra = count % pool.Length; + + long offset = 0; + MatrixParallelTaskPackage* current = package; + foreach (MatrixParallelWrap wrap in handle.Wraps) + { + long wrapTaskLength = actual * wrap.WrapTaskCapacity; + long ex = Math.Min(extra, wrap.WrapTaskCapacity); + extra -= ex; + ref long offsetRef = ref Unsafe.As(ref current->PrimTileOffset); + offsetRef = offset; + ref long lengthRef = ref Unsafe.As(ref current->PrimCount); + lengthRef = wrapTaskLength + ex; + wrap.WrapTask.Enqueue((nint) current); + current = MatrixParallelFactory.FromExistPackage(current); + offset += wrapTaskLength; + } + return handle; } - private static unsafe MatrixParallelHandle ScheduleKernelTask(MatrixParallelTaskPackageNonGeneric* package) + private static unsafe MatrixParallelHandle ScheduleKernelTask(MatrixParallelTaskPackage* package) { int primCount = package->PrimCount; int secCount = package->SecCount; @@ -73,17 +109,17 @@ namespace StgSharp.Mathematics.Numeric int tileCount = (primCount + tileWidth - 1) / tileWidth; MatrixParallelThread[] pool = LeadParallel(tileCount); - + /* if (pool.Length == 0) { - /* - * wait until has enough threads - */ + // no need + // LeadParallel has contained capacity check } + /**/ MatrixParallelHandle handle = new MatrixParallelHandle(pool); if (handle.Wraps.Count() == 1) { - handle.Wraps.First().WrapTask = package; + handle.Wraps.First().WrapTask.Enqueue((nint)package); return handle; } int threadCount = pool.Length; @@ -94,7 +130,7 @@ namespace StgSharp.Mathematics.Numeric int total = primCount; Stack taskBunch = new Stack(threadCount); - MatrixParallelTaskPackageNonGeneric* current = package; + MatrixParallelTaskPackage* current = package; foreach (MatrixParallelWrap wrap in handle.Wraps) { bool hasSlice = columnAfterBaseTile > 0; @@ -127,7 +163,7 @@ namespace StgSharp.Mathematics.Numeric total -= columnInWrap; // set to wrap and push - wrap.WrapTask = current; + wrap.WrapTask.Enqueue((nint)current); if (total <= 0) { diff --git a/StgSharp.Mathematics/Numeric/MatrixParallel.cs b/StgSharp.Mathematics/Numeric/MatrixParallel.cs index fcb5ce05..0ca60967 100644 --- a/StgSharp.Mathematics/Numeric/MatrixParallel.cs +++ b/StgSharp.Mathematics/Numeric/MatrixParallel.cs @@ -71,7 +71,7 @@ namespace StgSharp.Mathematics.Numeric RemainThreadCount = _threads.Count; _AfterWork = (ThreadPriority)afterWork; foreach (MatrixParallelWrap w in handle.Wraps) { - w.SchedulerReset.Set(); + w.SetScheduler(); } } diff --git a/StgSharp.Mathematics/Numeric/MatrixParallelHandle.cs b/StgSharp.Mathematics/Numeric/MatrixParallelHandle.cs index 0f7c3ce6..17fb976a 100644 --- a/StgSharp.Mathematics/Numeric/MatrixParallelHandle.cs +++ b/StgSharp.Mathematics/Numeric/MatrixParallelHandle.cs @@ -1,9 +1,9 @@ //----------------------------------------------------------------------- // ----------------------------------------------------------------------- -// file="MatrixParallelPool" +// file="MatrixParallelHandle" // 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 @@ -39,7 +39,8 @@ namespace StgSharp.Mathematics.Numeric public struct MatrixParallelHandle : IDisposable { - private static readonly (int a, int b)[] LookupTable = [ (1, 1), // 7 + 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 @@ -157,6 +158,7 @@ namespace StgSharp.Mathematics.Numeric { MatrixParallelWrap wrap = new MatrixParallelWrap(_threads, 0, count); _wraps = [ wrap ]; + return; } (int c3,int c4) = FastDecompose(_threads.Length); _wraps = new(c3 + c4); diff --git a/StgSharp.Mathematics/Numeric/MatrixParallelTaskPackage.cs b/StgSharp.Mathematics/Numeric/MatrixParallelTaskPackage.cs index cca6d2d6..538352f4 100644 --- a/StgSharp.Mathematics/Numeric/MatrixParallelTaskPackage.cs +++ b/StgSharp.Mathematics/Numeric/MatrixParallelTaskPackage.cs @@ -82,16 +82,13 @@ namespace StgSharp.Mathematics.Numeric } [StructLayout(LayoutKind.Explicit, Size = 128)] - internal unsafe struct MatrixParallelTaskPackageNonGeneric + public unsafe struct MatrixParallelTaskPackage { [MethodImpl(MethodImplOptions.AggressiveInlining)] - public static unsafe void Copy( - MatrixParallelTaskPackageNonGeneric* source, - MatrixParallelTaskPackageNonGeneric* target) + public static unsafe void Copy(MatrixParallelTaskPackage* source, MatrixParallelTaskPackage* target) { - Buffer.MemoryCopy(source, target, sizeof(MatrixParallelTaskPackageNonGeneric), - sizeof(MatrixParallelTaskPackageNonGeneric)); + Buffer.MemoryCopy(source, target, sizeof(MatrixParallelTaskPackage), sizeof(MatrixParallelTaskPackage)); } #region source @@ -113,7 +110,7 @@ namespace StgSharp.Mathematics.Numeric #region left enum - [FieldOffset(32)] public int LeftPrimOffset; // 4 + [FieldOffset(32)]public int LeftPrimOffset; // 4 [FieldOffset(36)]public int LeftPrimStride; // 4 [FieldOffset(40)]public int LeftSecOffset; // 4 [FieldOffset(44)]public int LeftSecStride; // 4 @@ -122,7 +119,7 @@ namespace StgSharp.Mathematics.Numeric #region right enum - [FieldOffset(32)] public int RightPrimOffset; // 4 + [FieldOffset(32)]public int RightPrimOffset; // 4 [FieldOffset(36)]public int RightPrimStride; // 4 [FieldOffset(40)]public int RightSecOffset; // 4 [FieldOffset(44)]public int RightSecStride; // 4 @@ -147,9 +144,9 @@ namespace StgSharp.Mathematics.Numeric #region global profile - [FieldOffset(96)]public int PrimCount; // 4 - [FieldOffset(100)]public MatrixOpMode ComputeMode; // 4 - [FieldOffset(104)]public int SecCount; // 4 + [FieldOffset(96)]public int PrimCount; // 4 + [FieldOffset(100)]public int SecCount; // 4 + [FieldOffset(104)]public MatrixOpMode ComputeMode; // 4 [FieldOffset(108)]private int ReservedInt0; // 4 #endregion @@ -163,90 +160,4 @@ namespace StgSharp.Mathematics.Numeric #endregion } - - [StructLayout(LayoutKind.Explicit, Size =128)] - internal unsafe struct MatrixParallelTaskPackage where T : unmanaged, INumber - { - - public MatrixParallelTaskPackage() - { - Unsafe.SkipInit(out this); - } - - [MethodImpl(MethodImplOptions.AggressiveInlining)] - public static unsafe void Copy(MatrixParallelTaskPackage* source, MatrixParallelTaskPackage* target) - { - Buffer.MemoryCopy(source, target, sizeof(MatrixParallelTaskPackage), - sizeof(MatrixParallelTaskPackage)); - } - - #region source - - // Group 1: Left/Right matrix kernel base address - [FieldOffset(0)]public MatrixKernel* Left; // 8 - [FieldOffset(8)] public MatrixKernel* Right; // 8 - - #endregion - - #region ans - - // Group 2: Result kernel base address + reserved - [FieldOffset(16)] public MatrixKernel* Result; // 8 - [FieldOffset(24)] public int ElementSize = sizeof(T); - [FieldOffset(28)] public int ReservedPtr0; // 4 (extension/reserved) - - #endregion - - #region left enum - - [FieldOffset(32)]public int LeftPrimOffset; // 4 - [FieldOffset(36)]public int LeftPrimStride; // 4 - [FieldOffset(40)]public int LeftSecOffset; // 4 - [FieldOffset(44)] public int LeftSecStride; // 4 - - #endregion - - #region right enum - - [FieldOffset(48)]public int RightPrimOffset; // 4 - [FieldOffset(52)]public int RightPrimStride; // 4 - [FieldOffset(56)]public int RightSecOffset; // 4 - [FieldOffset(60)]public int RightSecStride; // 4 - - #endregion - - #region ans enum - - [FieldOffset(64)]public int ResultPrimOffset; // 4 - [FieldOffset(68)]public int ResultPrimStride; // 4 - [FieldOffset(72)]public int ResultSecOffset; // 4 - [FieldOffset(76)] public int ResultSecStride; // 4 - - #endregion - - #region scalar/compute - - [FieldOffset(80)] public ScalarPacket* Scalar; // 8 - [FieldOffset(88)]public IntPtr ComputeHandle; // 8 - - #endregion - - #region global profile - - [FieldOffset(96)] public int PrimCount; // 4 - [FieldOffset(100)]public MatrixOpMode ComputeMode; // 4 - [FieldOffset(104)]public int SecCount; // 4 - [FieldOffset(108)] private readonly int ReservedInt0; // 4 - - #endregion - - #region count offset - - [FieldOffset(112)]public int PrimTileOffset; // 4 - [FieldOffset(116)]public int SecTileOffset; // 4 - [FieldOffset(120)]public int PrimColumnCountInTile; // 4 - [FieldOffset(124)]public int SecColumnCountInTile; // 4 - - #endregion - } } \ No newline at end of file diff --git a/StgSharp.Mathematics/Numeric/MatrixParallelThread.cs b/StgSharp.Mathematics/Numeric/MatrixParallelThread.cs index 84cdf2b3..8a55f434 100644 --- a/StgSharp.Mathematics/Numeric/MatrixParallelThread.cs +++ b/StgSharp.Mathematics/Numeric/MatrixParallelThread.cs @@ -27,6 +27,7 @@ // ----------------------------------------------------------------------- using System; using System.Collections.Generic; +using System.ComponentModel.DataAnnotations; using System.Diagnostics.CodeAnalysis; using System.Linq; using System.Runtime.CompilerServices; @@ -53,6 +54,7 @@ namespace StgSharp.Mathematics.Numeric _affinity = AffinityAllocator.Allocate(); _managedThread = new Thread(MatrixParallelWorker); + _managedThread.Start(); _resetEvent = new ManualResetEventSlim(false); } @@ -78,7 +80,7 @@ namespace StgSharp.Mathematics.Numeric { while (taskQueue.TryDequeue(out nint handle)) { - MatrixParallelTaskPackageNonGeneric* p = (MatrixParallelTaskPackageNonGeneric*)handle; + MatrixParallelTaskPackage* p = (MatrixParallelTaskPackage*)handle; switch (p->ComputeMode.OperationStyle) { case MatrixOperationStyle.PANEL_OP: @@ -127,16 +129,12 @@ namespace StgSharp.Mathematics.Numeric } } + [MethodImpl(MethodImplOptions.AggressiveInlining)] public void Restart() { _resetEvent.Set(); } - public void ReturnToPool() - { - // MatrixParallel.(this); - } - protected virtual void Dispose(bool disposing) { if (!disposedValue) @@ -159,27 +157,30 @@ namespace StgSharp.Mathematics.Numeric // Bind OS-thread affinity once when the worker starts running ApplyAffinityIfNeeded(); - // Thread self = Thread.CurrentThread; - _resetEvent.Wait(); - _resetEvent.Reset(); - _managedThread.Priority = ThreadPriority.Highest; - switch (Role) + while (true) { - case 0://worker - // _ = Interlocked.Decrement(ref RemainThreadCount); - ExecuteTaskPackage(Tasks); - break; - case 1://scheduler - BindedWrap.PublishTaskToWorkers(Tasks); - BindedWrap.SetAllWorker(); - ExecuteTaskPackage(Tasks); - break; - case 2://leader, error because leader should be at outer thread but not here - break; - default: - break; + // Thread self = Thread.CurrentThread; + _managedThread.Priority = ThreadPriority.Highest; + _resetEvent.Wait(); + _resetEvent.Reset(); + switch (Role) + { + case 0://worker + // _ = Interlocked.Decrement(ref RemainThreadCount); + ExecuteTaskPackage(Tasks); + break; + case 1://scheduler + BindedWrap.PublishTaskToWorkers(BindedWrap.WrapTask); + BindedWrap.SetAllWorker(); + ExecuteTaskPackage(Tasks); + break; + case 2://leader, error because leader should be at outer thread but not here + break; + default: + break; + } + _managedThread.Priority = AfterWork; } - _managedThread.Priority = AfterWork; } #region core affinity diff --git a/StgSharp.Mathematics/Numeric/MatrixParallelWrap.cs b/StgSharp.Mathematics/Numeric/MatrixParallelWrap.cs index e1d80fbd..fa6a42f4 100644 --- a/StgSharp.Mathematics/Numeric/MatrixParallelWrap.cs +++ b/StgSharp.Mathematics/Numeric/MatrixParallelWrap.cs @@ -44,10 +44,13 @@ namespace StgSharp.Mathematics.Numeric internal MatrixParallelWrap(MatrixParallelThread[] threads, int begin, int count) { + WrapTaskCapacity = count; _threads = threads; _begin = begin; - Count = threads.Length; _threads[_begin].Role = 1; + for (int i = _begin; i < _begin + WrapTaskCapacity; i++) { + _threads[i].BindedWrap = this; + } } public MatrixParallelWrap() @@ -63,15 +66,11 @@ namespace StgSharp.Mathematics.Numeric internal int Magic { get; set; } - internal ManualResetEventSlim SchedulerReset { get; } = new(false); - - private int Count { get; set; } - [MethodImpl(MethodImplOptions.AggressiveInlining)] public void SetAllWorker() { - foreach (MatrixParallelThread t in _threads) { - t.Restart(); + for (int i = _begin; i < _begin + WrapTaskCapacity; i++) { + _threads[i].Restart(); } } @@ -79,25 +78,124 @@ namespace StgSharp.Mathematics.Numeric { while (tasks.TryDequeue(out nint handle)) { - MatrixParallelTaskPackageNonGeneric* task = (MatrixParallelTaskPackageNonGeneric*)handle; - PublishTaskToWorkers(task); + MatrixParallelTaskPackage* task = (MatrixParallelTaskPackage*)handle; + switch (task->ComputeMode.OperationStyle) + { + case MatrixOperationStyle.PANEL_OP: + break; + case MatrixOperationStyle.BUFFER_OP: + PublishBufferTaskToWorkers(task); + break; + case MatrixOperationStyle.KERNEL_OP: + PublishBufferTaskToWorkers(task); + break; + default: + break; + } } } - internal void PublishTaskToWorkers(MatrixParallelTaskPackageNonGeneric* taskPackage) + internal void SetScheduler() { - int primCount = taskPackage->PrimCount; - int secCount = taskPackage->SecCount; + MatrixParallelThread thread = _threads[_begin]; + thread.Restart(); + } - int primAverage = primCount / Count; - int secAverage = secCount / Count; - bool shouldSliceSec = secAverage > 512; - for (int i = _begin; i < _begin + Count; i++) + private void PublishBufferTaskToWorkers(MatrixParallelTaskPackage* package) + { + long count = Unsafe.As(ref package->PrimCount); + long recommendCount = count / 1024; + MatrixParallelThread[] pool = _threads; + if (pool.Length == 1) { - Queue tasks = _threads[i].Tasks; - MatrixParallelTaskPackageNonGeneric* subTask = MatrixParallelFactory.FromExistPackage(taskPackage); - subTask->PrimTileOffset = i * primAverage; + pool[0].Tasks.Enqueue((nint)package); + return; } + long actual = count / pool.Length; + long extra = count % pool.Length; + + long offset = 0; + MatrixParallelTaskPackage* current = package; + foreach (MatrixParallelThread thread in pool) + { + long wrapTaskLength = actual; + long ex = extra > 0 ? 1 : 0; + extra -= ex; + ref long offsetRef = ref Unsafe.As(ref current->PrimTileOffset); + offsetRef += offset; + ref long lengthRef = ref Unsafe.As(ref current->PrimCount); + lengthRef = wrapTaskLength + ex; + pool[0].Tasks.Enqueue((nint)current); + current = MatrixParallelFactory.FromExistPackage(current); + offset += wrapTaskLength; + } + } + + private void PublishKernelTaskToWorkers(MatrixParallelTaskPackage* package) + { + int primCount = package->PrimCount; + int secCount = package->SecCount; + MatrixParallelThread[] pool = _threads; + + if (pool.Length == 1) + { + pool[0].Tasks.Enqueue((nint)package); + return; + } + throw new NotImplementedException(); + /* + int threadCount = pool.Length; + int tilePerThreadLess = tileCount / threadCount; // tiles per thread without extra slice + int columnAfterBaseTile = tileCount % threadCount * tileWidth; // tile sliced into columns + int extraColumnPerThread = columnAfterBaseTile / threadCount; + int extraColumnCount = columnAfterBaseTile % threadCount; // columns to be sliced + + int total = primCount; + Stack taskBunch = new Stack(threadCount); + MatrixParallelTaskPackageNonGeneric* current = package; + foreach (MatrixParallelWrap wrap in pool.Wraps) + { + bool hasSlice = columnAfterBaseTile > 0; + bool isLargeSlice = extraColumnCount > 0; + + // _count tiles for wrap + int wrapWidth = wrap!.WrapTaskCapacity; + int columnFromSlicingCount = (hasSlice ? extraColumnPerThread : 0) * wrapWidth; + int baseColumn = wrapWidth * tilePerThreadLess * tileWidth; + int columnInWrap = baseColumn + columnFromSlicingCount + (isLargeSlice ? 1 : 0); + + // set tasks + if (columnInWrap == 0) + { +#if DEBUG + throw new InvalidOperationException("Zero tasks published."); +#else + break; +#endif + } + + // update current task + current->PrimColumnCountInTile = columnInWrap; + current->SecColumnCountInTile = current->SecCount; + current->SecTileOffset = 0; + + // remove tiles _count + extraColumnCount--; + columnAfterBaseTile -= columnFromSlicingCount; + total -= columnInWrap; + + // set to wrap and push + wrap.WrapTask = current; + + if (total <= 0) + { + break; + } + current = MatrixParallelFactory.FromExistPackage(current); + current->PrimTileOffset += columnInWrap; + } + /**/ + return; } public class Handle(int inWrapIndex, MatrixParallelWrap wrap) @@ -121,15 +219,11 @@ namespace StgSharp.Mathematics.Numeric #region wrap task - internal MatrixParallelTaskPackageNonGeneric* WrapTask { get; set; } + internal unsafe Queue WrapTask { get; set; } = []; internal int WrapTaskSliceCount { get; set; } - internal int WrapTaskCapacity - { - [MethodImpl(MethodImplOptions.AggressiveInlining)] - get => _threads.Length; - } + internal int WrapTaskCapacity { get ; init; } #endregion } diff --git a/StgSharp.Mathematics/Numeric/MatrixTaskFactory.cs b/StgSharp.Mathematics/Numeric/MatrixTaskFactory.cs index 98374399..ea7bace0 100644 --- a/StgSharp.Mathematics/Numeric/MatrixTaskFactory.cs +++ b/StgSharp.Mathematics/Numeric/MatrixTaskFactory.cs @@ -55,13 +55,13 @@ namespace StgSharp.Mathematics.Numeric private static SlabAllocator _scalarAllocator ; - private static SlabAllocator _wrapAllocator; + private static SlabAllocator _wrapAllocator; [MethodImpl(MethodImplOptions.AggressiveInlining)] - public static MatrixParallelTaskPackage* CreateBaseTask() where T: unmanaged, INumber + public static MatrixParallelTaskPackage* CreateBaseTask() where T : unmanaged, INumber { - MatrixParallelTaskPackage* p = (MatrixParallelTaskPackage*)_wrapAllocator.Allocate(); - Span s = new Span((byte*)p, sizeof(MatrixParallelTaskPackage)); + MatrixParallelTaskPackage* p = (MatrixParallelTaskPackage*)_wrapAllocator.Allocate(); + Span s = new Span((byte*)p, sizeof(MatrixParallelTaskPackage)); s.Fill(0); return p; } @@ -74,20 +74,19 @@ namespace StgSharp.Mathematics.Numeric } [MethodImpl(MethodImplOptions.AggressiveInlining)] - public static unsafe MatrixParallelTaskPackage* FromExistPackage(MatrixParallelTaskPackage* exist) - where T: unmanaged, INumber + public static unsafe MatrixParallelTaskPackage* FromExistPackage(MatrixParallelTaskPackage* exist) + where T : unmanaged, INumber { - MatrixParallelTaskPackage* ptr = (MatrixParallelTaskPackage*)_wrapAllocator.Allocate(); - MatrixParallelTaskPackage.Copy(exist, ptr); + MatrixParallelTaskPackage* ptr = (MatrixParallelTaskPackage*)_wrapAllocator.Allocate(); + MatrixParallelTaskPackage.Copy(exist, ptr); return ptr; } [MethodImpl(MethodImplOptions.AggressiveInlining)] - public static unsafe MatrixParallelTaskPackageNonGeneric* FromExistPackage( - MatrixParallelTaskPackageNonGeneric* exist) + public static unsafe MatrixParallelTaskPackage* FromExistPackage(MatrixParallelTaskPackage* exist) { - MatrixParallelTaskPackageNonGeneric* ptr = (MatrixParallelTaskPackageNonGeneric*)_wrapAllocator.Allocate(); - MatrixParallelTaskPackageNonGeneric.Copy(exist, ptr); + MatrixParallelTaskPackage* ptr = (MatrixParallelTaskPackage*)_wrapAllocator.Allocate(); + MatrixParallelTaskPackage.Copy(exist, ptr); return ptr; } @@ -99,7 +98,7 @@ namespace StgSharp.Mathematics.Numeric } [MethodImpl(MethodImplOptions.AggressiveInlining)] - public static void Release(MatrixParallelTaskPackageNonGeneric* source) + public static void Release(MatrixParallelTaskPackage* source) { if (source->Scalar != null) { _scalarAllocator.Free((nuint)source->Scalar); @@ -108,7 +107,7 @@ namespace StgSharp.Mathematics.Numeric } [MethodImpl(MethodImplOptions.AggressiveInlining)] - public static void Release(MatrixParallelTaskPackage* source) where T: unmanaged, INumber + public static void Release(MatrixParallelTaskPackage* source) where T : unmanaged, INumber { if (source->Scalar != null) { _scalarAllocator.Free((nuint)source->Scalar); @@ -123,7 +122,7 @@ namespace StgSharp.Mathematics.Numeric { _intermediateResult = new hlsfAllocator(64 * 16 * 16); _scalarAllocator = SlabAllocator.Create(64 * 1024, SlabBufferLayout.Chunked); - _wrapAllocator = SlabAllocator. + _wrapAllocator = SlabAllocator. Create(128 * 1024, SlabBufferLayout.Chunked); } } diff --git a/StgSharpDebug/Program.cs b/StgSharpDebug/Program.cs index b21578b7..85016753 100644 --- a/StgSharpDebug/Program.cs +++ b/StgSharpDebug/Program.cs @@ -62,133 +62,20 @@ namespace StgSharp try { World.Initialize().UseModule().UseModule().InitializeModule(); - TestRender(); - /* - // 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; + // TestRender(); - 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 + 10 * 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 = []; - 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}"); - - /* // 原有的矩阵测试代码 HybridLayerSegregatedFitAllocator hlsf = new HybridLayerSegregatedFitAllocator(64 * 400 * 400); - Matrix m = Matrix.FromHlsf(20, 20, hlsf); + Matrix m = Matrix.FromHlsf(200, 200, hlsf); m.Fill(1.0f); Matrix n = m.Add(m); for (int i = 0; i < 20; i++) { for (int j = 0; j < 20; j++) { - Console.Write($"{n[i, j].ToString("0.000")}\t"); + Console.Write($"{n[i, j]:0.0E+00} \t"); } Console.WriteLine(); } @@ -198,6 +85,7 @@ namespace StgSharp catch (Exception ex) { Console.WriteLine(ex); + Console.ReadLine(); /* World.WriteLog($"{ex}", LogType.Error); if( true ) { }