//----------------------------------------------------------------------- // ----------------------------------------------------------------------- // file="MatrixParallelThread" // 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 StgSharp.Internal; using StgSharp.Internal.Intrinsic; using System; using System.Collections.Generic; using System.ComponentModel.DataAnnotations; using System.Diagnostics.CodeAnalysis; using System.Linq; using System.Runtime.CompilerServices; using System.Runtime.InteropServices; using System.Text; using System.Threading.Tasks; namespace StgSharp.Mathematics.Numeric { public unsafe partial class MatrixParallelThread : IDisposable { private IntPtr* _context; private bool disposedValue; private readonly ManualResetEventSlim _resetEvent; private readonly Thread _managedThread; internal MatrixParallelThread( int poolId ) { _resetEvent = new ManualResetEventSlim(false); _managedThread = new Thread(MatrixParallelWorkLoad); _managedThread.Start(); PoolId = poolId; } public int PoolId { get; init; } public ThreadPriority AfterWork { get; set; } internal ManualResetEventSlim? LeaderEvent { get; set; } internal IntrinsicContext* Intrinsic { get; set; } internal MatrixParallelWrap BindedWrap { get; set; } internal RoleType Role { get; set; } internal SIMDID Simd { get; set; } public void Dispose() { Dispose(disposing:true); GC.SuppressFinalize(this); } [MethodImpl(MethodImplOptions.AggressiveInlining)] public static unsafe void ExecuteTaskPackage( MatrixParallelQueue taskQueue ) { MatrixParallelTaskPackage* p = taskQueue.Package; switch (p->ComputeMode.OperationStyle) { case MatrixIndexStyle.BUFFER_INDEX: switch (p->ComputeMode.ParameterStyle) { case MatrixOperationParam.RIGHT_ANS_PARAM: BufferComputeUnary(p); break; case MatrixOperationParam.LEFT_RIGHT_ANS_PARAM: BufferComputeBinary(p); break; case MatrixOperationParam.RIGHT_ANS_SCALAR_PARAM: BufferComputeUnaryScalar(p); break; case MatrixOperationParam.LEFT_RIGHT_ANS_SCALAR_PARAM: BufferComputeBinaryScalar(p); break; case MatrixOperationParam.ANS_SCALAR_PARAM: BufferComputeNoOperatorScalar(p); break; default: break; } break; case MatrixIndexStyle.SPECIAL: SpecialBufferCompute(p); break; default: break; } } [MethodImpl(MethodImplOptions.AggressiveInlining)] public void Restart() { if (Role == RoleType.Leader) { LeaderEvent?.Set(); } else { _resetEvent.Set(); } } internal void MatrixParallelLeader() { Thread current = Thread.CurrentThread; current.Priority = ThreadPriority.Highest; LeaderEvent?.Wait(); LeaderEvent?.Reset(); // Console.WriteLine("Leader is computing"); BindedWrap.ReportThreadAccomplished(); // Console.WriteLine("leader's computing is done"); Role = RoleType.Worker; LeaderEvent = null; current.Priority = BindedWrap.AfterWork; } [DllImport( dllName:Native.LibName, EntryPoint = "sn_set_thread_affinity", CallingConvention = CallingConvention.Cdecl)] internal static extern void SetThreadAffinity( int coreId ); protected virtual void Dispose( bool disposing ) { if (!disposedValue) { if (disposing) { _resetEvent.Dispose(); _managedThread.Join(); } disposedValue = true; } } private unsafe void MatrixParallelWorkLoad() { // Bind OS-thread affinity once when the worker starts running SetThreadAffinity(PoolId); Thread.Sleep(0); while (true) { // Thread self = Thread.CurrentThread; _resetEvent.Wait(); _resetEvent.Reset(); AfterWork = BindedWrap.AfterWork; _managedThread.Priority = ThreadPriority.Highest; switch (Role) { case RoleType.Worker://worker // _ = Interlocked.Decrement(ref RemainThreadCount); // Console.WriteLine($"worker {Environment.CurrentManagedThreadId} start"); ExecuteTaskPackage(BindedWrap.TaskQueue); BindedWrap.ReportThreadAccomplished(); // Console.WriteLine($"worker {Environment.CurrentManagedThreadId} end"); break; case RoleType.SingleExternal: // TODO 将外部线程绑定到当前线程上,并执行任务包 break; default: break; } _managedThread.Priority = AfterWork; } } internal enum RoleType : int { Worker = 0, Leader = 1, SingleExternal = 2, } } }