//----------------------------------------------------------------------- //----------------------------------------------------------------------- // file="PipelineScheduler.cs" // Project: StgSharp // AuthorGroup: Nitload Space // Copyright (c) Nitload Space. 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.Stg; using System; using System.Collections.Generic; using System.Diagnostics.CodeAnalysis; using System.Linq; using System.Runtime.CompilerServices; using System.Threading; using System.Xml; using System.Xml.Linq; namespace StgSharp.PipeLine { /// /// A sets of complex logic designed for running in multi threads environment. Warning: do not /// run a instance in multi threads more than once. /// public partial class PipelineScheduler : IConvertableToPipelineNode { private BeginningNode _beginning; private bool _isAvailable = false; private EndingNode _ending; private int mtLevel, cLevel,mtIndex, cIndex; private List> _concurrentNodes, _mainThreadNodes ; private SemaphoreSlim runningStat,_concurrentLock = new SemaphoreSlim( 1, 1 ); internal Dictionary _allNode; public PipelineScheduler() { _allNode = new Dictionary(); runningStat = new SemaphoreSlim( 1, 1 ); _beginning = new BeginningNode( this ); _ending = new EndingNode( this ); } /**/ public PipelineNode this[ PipelineNodeLabel label ] { get { return _allNode[ label ]; } } public bool IsRunning { get => runningStat.CurrentCount == 0; } public IEnumerable InputPortName { get => _beginning.InputPorts.Keys; } public IEnumerable OutputPortName { get => _beginning.OutputPorts.Keys; } public PipelineNode BeginLayer => _beginning; public PipelineNode EndLayer => _ending; public PipelineNodeOperation Operation { get => ( this as IConvertableToPipelineNode ).NodeMain; } /**/ internal SemaphoreSlim RunningStat { get => runningStat; } public void AddNode( PipelineNode node, bool isNative ) { ArgumentNullException.ThrowIfNull( node ); node.IsNative = isNative; _allNode.TryAdd( node.Label, node ); } public void ArrangeNodes() { if( _isAvailable ) { return; } int maxLevel = 1; Queue completedNodes = new Queue(); foreach( PipelineNode item in _beginning.Next ) { item.TryIncreaseLevelTo( 1 ); completedNodes.Enqueue( item ); } //calc level of nodes O(V) while( completedNodes.Count > 0 ) { PipelineNode u = completedNodes.Dequeue(); foreach( PipelineNode v in u.Next ) { if( !v.TryIncreaseLevelTo( u.Level + 1 ) ) { throw new InvalidOperationException( "Ring in DAG detected." ); } if( v.RemainingIncreaseCount == 0 ) { completedNodes.Enqueue( v ); maxLevel = int.Max( maxLevel, v.Level ); } } } //bucket sort of nodes, O(E) _concurrentNodes = new List>( maxLevel + 1 ); _mainThreadNodes = new List>( maxLevel + 1 ); for( int i = 0; i <= maxLevel; i++ ) { _concurrentNodes.Add( [] ); _mainThreadNodes.Add( [] ); } foreach( PipelineNode item in _allNode.Values ) { if( item.IsNative ) { _mainThreadNodes[ item.Level ].Add( item ); } else { _concurrentNodes[ item.Level ].Add( item ); } } foreach( List item in _concurrentNodes ) { item.TrimExcess(); } foreach( List item in _mainThreadNodes ) { item.TrimExcess(); } _isAvailable = true; } public PipelineNode Create( PipelineNodeLabel label, [NotNull] IConvertableToPipelineNode body ) { if( _allNode.TryGetValue( label, out PipelineNode? node ) ) { throw new InvalidOperationException( "Label has exist in scheduler." ); } node = new PipelineNode( label, body, false ); _allNode.Add( label, node ); return node; } public PipelineNode GetNode( PipelineNodeLabel label ) { if( _allNode.TryGetValue( label, out PipelineNode? node ) ) { return node; } else { throw new KeyNotFoundException(); } } public void LinkToInput( PipelineNodeInPort port ) { ArgumentNullException.ThrowIfNull( port ); PipelineNodeOutPort p = _beginning.DefaultOut; p.Connect( port ); } public void LinkToInput( PipelineNodeInPort port, string inputPortName ) { ArgumentNullException.ThrowIfNull( port ); PipelineNodeOutPort p = _beginning.DefineCertainOutputPort( inputPortName ); p.Connect( port ); } [MethodImpl( MethodImplOptions.AggressiveInlining )] public void SetInput( params (string label, IPipeLineConnectionPayload arg)[] parameters ) { _beginning.SetInputData( parameters ); } public void Terminate() { _concurrentLock.Wait(); (mtLevel, cLevel, mtIndex, cIndex) = (int.MaxValue, int.MaxValue, int.MaxValue, int.MaxValue); _concurrentLock.Release(); } public bool TryGetAllNode( out List> mainThreadNodes, out List> concurrentNodes ) { if( _isAvailable ) { mainThreadNodes = _mainThreadNodes; concurrentNodes = _concurrentNodes; return true; } mainThreadNodes = null; concurrentNodes = null; return false; } internal bool RequestNexConcurrentNode( out PipelineNode node ) { int index, level; if( cLevel >= _concurrentNodes.Count ) { node = null!; return false; } _concurrentLock.Wait(); while( cIndex >= _concurrentNodes[ cLevel ].Count ) { Interlocked.Increment( ref cLevel ); Interlocked.Exchange( ref cIndex, 0 ); if( cLevel >= _concurrentNodes.Count ) { node = null!; return false; } } index = cIndex; level = cLevel; Interlocked.Increment( ref cIndex ); _concurrentLock.Release(); node = _concurrentNodes[ level ][ index ]; return true; } internal bool RequestNextMainThreadNode( out PipelineNode node ) { if( mtLevel >= _mainThreadNodes.Count ) { node = null!; return false; } while( mtIndex >= _mainThreadNodes[ cLevel ].Count ) { Interlocked.Increment( ref mtLevel ); Interlocked.Exchange( ref mtIndex, 0 ); if( mtLevel >= _mainThreadNodes.Count ) { node = null!; return false; } } node = _mainThreadNodes[ mtLevel ][ mtIndex ]; return true; } internal void ResetIndex() { (mtLevel, cLevel, mtIndex, cIndex) = (0, 0, 0, 0); _concurrentLock.Release(); } void IConvertableToPipelineNode.NodeMain( in Dictionary input, in Dictionary output ) { PipeLineRunner.Run( this ); PipelineNodeOutPort.SkipAll( output ); } ~PipelineScheduler() { runningStat.Dispose(); } } }