youmuliu/stgsharp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Daily update on 20250618-051714 by Nitload Youmu. Additional Message : Brief support to collection operation in EXPRESS. A collection type added in core lib and compilestack.

Browse Source 
Signed-off-by: Nitload <YoumuLiu_NSI@whv0t.onmicrosoft.com>
This commit is contained in:
Nitload
2025-06-18 05:17:14 +08:00
parent 74c01effe4
commit c624a8ca38
300 changed files with 12227 additions and 4303 deletions
Download Patch File Download Diff File Expand all files Collapse all files
StgSharp.Common/PipeLine/CrossFramePipelineNode.cs
+5 -5
View File
@@ -63,9 +63,9 @@ namespace StgSharp.PipeLine
private protected set => _mainExecution = value;
}
public IEnumerable<string> InputInterfacesName => _inputName;
public IEnumerable<string> InputPortName => _inputName;
public IEnumerable<string> OutputInterfacesName => _outputName;
public IEnumerable<string> OutputPortName => _outputName;
public PipelineNodeOperation Operation
{
@@ -73,8 +73,8 @@ namespace StgSharp.PipeLine
}
public void NodeMain(
in Dictionary<string, PipelineNodeImport> input,
in Dictionary<string, PipelineNodeExport> output )
in Dictionary<string, PipelineNodeInPort> input,
in Dictionary<string, PipelineNodeOutPort> output )
{
if( currentCount < cycleCount )
{
@@ -86,7 +86,7 @@ namespace StgSharp.PipeLine
}
//Console.WriteLine(currentCount);
PipelineNodeExport.SkipAll( output );
PipelineNodeOutPort.SkipAll( output );
}
}
StgSharp.Common/PipeLine/IConvertableToPipelineNode.cs
+10 -10
View File
@@ -36,21 +36,21 @@ using System.Text;
namespace StgSharp.PipeLine
{
public delegate void PipelineNodeOperation(
in Dictionary<string, PipelineNodeImport> input,
in Dictionary<string, PipelineNodeExport> output );
in Dictionary<string, PipelineNodeInPort> input,
in Dictionary<string, PipelineNodeOutPort> output );
public interface IConvertableToPipelineNode
{
public IEnumerable<string> InputInterfacesName { get; }
public IEnumerable<string> InputPortName { get; }
public IEnumerable<string> OutputInterfacesName { get; }
public IEnumerable<string> OutputPortName { get; }
PipelineNodeOperation Operation { get; }
void NodeMain(
in Dictionary<string, PipelineNodeImport> input,
in Dictionary<string, PipelineNodeExport> output );
in Dictionary<string, PipelineNodeInPort> input,
in Dictionary<string, PipelineNodeOutPort> output );
}
@@ -71,15 +71,15 @@ namespace StgSharp.PipeLine
_execution = execution;
}
public IEnumerable<string> InputInterfacesName => _input;
public IEnumerable<string> InputPortName => _input;
public IEnumerable<string> OutputInterfacesName => _output;
public IEnumerable<string> OutputPortName => _output;
public PipelineNodeOperation Operation => _execution;
public void NodeMain(
in Dictionary<string, PipelineNodeImport> input,
in Dictionary<string, PipelineNodeExport> output )
in Dictionary<string, PipelineNodeInPort> input,
in Dictionary<string, PipelineNodeOutPort> output )
{
_execution.Invoke( input, output );
}
StgSharp.Common/PipeLine/PipeLineRunner.cs
+84 -78
View File
@@ -51,69 +51,26 @@ namespace StgSharp.PipeLine
private static ConcurrentStack<PipelineScheduler> pipelineInWaiting
= new ConcurrentStack<PipelineScheduler>();
private static int _callingThread;
private static SemaphoreSlim bpLock = new SemaphoreSlim( 0, 1 );
private static SemaphoreSlim pipelineLock = new SemaphoreSlim( 0, 1 );
/// <summary>
/// <para> Only the following occasions of calling are valid: </para> <para> 1. no bp is
/// running or the only bp is quitting. </para><para> 2. calling from the main thread or a
/// thread the current bp is running on.</para>
/// </summary>
internal static bool IsValidCalling
{
get
{
if( ( NodeRunner.CurrentBlueprint == null ) || //no bp running at all
( ( pipelineInWaiting.IsEmpty ) && ( NodeRunner.CurrentBlueprint.IsRunning ==
false ) ) )
{
// a bp is quitting
//in case no bp is running
return true;
}
int? id = Task.CurrentId;
if( id is null )
{
// in case calling from main thread and current running bp
return _callingThread == Environment.CurrentManagedThreadId;
}
foreach( NodeRunner item in allAvailableTask )
{
if( item.Id == id )
{
//if calling from any bp running task
return true;
}
}
return false;
}
}
public static void Run( PipelineScheduler blueprint )
public static void Run( PipelineScheduler pipeline )
{
//Console.WriteLine();
if( blueprint == null )
if( pipeline == null )
{
return;
}
if( blueprint.IsRunning ) {
if( pipeline.IsRunning ) {
throw new InvalidOperationException( "Attempt to run a blueprint instance twice." );
}
if( IsValidCalling )
{
//Console.WriteLine("valid and run");
if( InterruptAndRunNewBp( blueprint ) )
{
while( InterruptAndRunNewBp( NodeRunner.CurrentBlueprint ) ) { }
}
} else
{
bpLock.Wait();
if( InterruptAndRunNewBp( blueprint ) )
{
while( InterruptAndRunNewBp( NodeRunner.CurrentBlueprint ) ) { }
}
bpLock.Release();
if( !IsValidCalling() ) {
pipelineLock.Wait();
}
if( InterruptAndRunNewPipeline( pipeline ) )
{
while( InterruptAndRunNewPipeline( NodeRunner.CurrentPipeline! ) ) { }
}
pipelineLock.Release();
Interlocked.Exchange( ref _callingThread, -1 );
}
@@ -137,7 +94,10 @@ namespace StgSharp.PipeLine
public static void Terminate()
{
cts.Cancel();
NodeRunner.CurrentBlueprint.Terminate();
if( NodeRunner.CurrentPipeline is null ) {
return;
}
NodeRunner.CurrentPipeline.Terminate();
foreach( NodeRunner task in allAvailableTask ) {
task.Cts.Cancel();
}
@@ -147,26 +107,28 @@ namespace StgSharp.PipeLine
pipelineInWaiting.Clear();
}
internal static bool InterruptAndRunNewBp( PipelineScheduler newBp )
internal static bool InterruptAndRunNewPipeline( PipelineScheduler newPipeline )
{
ArgumentNullException.ThrowIfNull( newBp );
if( ( newBp != NodeRunner.CurrentBlueprint ) && ( NodeRunner.CurrentBlueprint != null ) )
ArgumentNullException.ThrowIfNull( newPipeline );
if( ( newPipeline != NodeRunner.CurrentPipeline ) &&
( NodeRunner.CurrentPipeline != null ) )
{
//if a running bp exists
pipelineInWaiting.Push( NodeRunner.CurrentBlueprint );
pipelineInWaiting.Push( NodeRunner.CurrentPipeline );
}
//load new bp to current bp
Interlocked.Exchange( ref NodeRunner.CurrentBlueprint!, newBp );
NodeRunner.CurrentPipeline = newPipeline;
Interlocked.Exchange( ref _callingThread, Environment.CurrentManagedThreadId );
//run beginning node to init it first.
if( !NodeRunner.CurrentBlueprint.IsRunning )
if( !NodeRunner.CurrentPipeline.IsRunning )
{
( NodeRunner.CurrentBlueprint.BeginLayer as BeginningNode )!.Run();
( NodeRunner.CurrentPipeline.BeginLayer as BeginningNode )!.Run();
}
//refresh all tasks
/*
for( int i = 0; i < allAvailableTask.Length; i++ )
{
NodeRunner item = allAvailableTask[ i ];
@@ -174,19 +136,64 @@ namespace StgSharp.PipeLine
allAvailableTask[ i ] = NodeRunner.Run();
}
}
/**/
PipelineNode node;
while( !cts.Token.IsCancellationRequested && NodeRunner.CurrentBlueprint
.RequestNextNativeNode(
out node ) )
while( !cts.Token.IsCancellationRequested &&
NodeRunner.CurrentPipeline.RequestNextMainThreadNode( out node ) )
{
node.Run();
if( !NodeRunner.CurrentBlueprint.IsRunning )
if( !NodeRunner.CurrentPipeline.IsRunning )
{
break;
}
}
( NodeRunner.CurrentBlueprint.EndLayer as EndingNode )!.Run();
return pipelineInWaiting.TryPeek( out NodeRunner.CurrentBlueprint );
while( ( NodeRunner.CurrentPipeline != null ) &&
NodeRunner.CurrentPipeline.IsRunning &&
!cts.Token.IsCancellationRequested &&
NodeRunner.CurrentPipeline!.RequestNexConcurrentNode( out node ) )
{
node.Run();
if( !NodeRunner.CurrentPipeline!.IsRunning )
{
break;
}
}
( NodeRunner.CurrentPipeline!.EndLayer as EndingNode )!.Run();
bool isEmpty = pipelineInWaiting.TryPop( out PipelineScheduler? pipeline );
NodeRunner.CurrentPipeline = pipeline!;
return isEmpty;
}
/// <summary>
/// <para> Only the following occasions of calling are valid: </para> <para>1. no bp is
/// running or the only bp is quitting.</para><para>2. calling from the main thread or a
/// thread the current bp is running on. </para>
/// </summary>
internal static bool IsValidCalling()
{
if( ( NodeRunner.CurrentPipeline == null ) || //no bp running at all
( ( pipelineInWaiting.IsEmpty ) && ( NodeRunner.CurrentPipeline.IsRunning == false ) ) )
{
// a bp is quitting
//in case no bp is running
return true;
}
int? id = Task.CurrentId;
if( id is null )
{
// in case calling from main thread and current running bp
return _callingThread == Environment.CurrentManagedThreadId;
}
foreach( NodeRunner item in allAvailableTask )
{
if( item.Id == id )
{
//if calling from any bp running task
return true;
}
}
return false;
}
private static NodeRunner[] CreateBlueprintRunnerArray()
@@ -200,7 +207,7 @@ namespace StgSharp.PipeLine
internal class NodeRunner : CancellableTask
{
private static PipelineScheduler _current;
private static volatile PipelineScheduler? _current;
internal NodeRunner()
: base()
@@ -208,9 +215,10 @@ namespace StgSharp.PipeLine
_startup = InternalRun;
}
internal static ref PipelineScheduler CurrentBlueprint
internal static PipelineScheduler? CurrentPipeline
{
get => ref _current;
get => _current;
set => _current = value;
}
internal static NodeRunner Run()
@@ -225,16 +233,14 @@ namespace StgSharp.PipeLine
private static void InternalRun( CancellationToken token )
{
//Console.WriteLine("Task Begin");
PipelineNode item;
while( ( CurrentBlueprint != null ) && CurrentBlueprint.IsRunning && !token.IsCancellationRequested && CurrentBlueprint.RequestNexGlobalNode(
out item ) )
{
//Console.WriteLine($"Get node of {item.Name}, going to run.");
Console.WriteLine( $"Global thread {Environment.CurrentManagedThreadId} starts." );
while( ( CurrentPipeline != null ) &&
CurrentPipeline.IsRunning &&
!token.IsCancellationRequested &&
CurrentPipeline.RequestNexConcurrentNode( out item ) ) {
item.Run();
}
//Console.WriteLine("Task Completed");
}
}//------------------------------------- End of Class ------------------------------------------
StgSharp.Common/PipeLine/PipelineMarshal.cs
+118
View File
@@ -0,0 +1,118 @@
//-----------------------------------------------------------------------
//-----------------------------------------------------------------------
// file="PipelineMarshal.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 System;
using System.Collections;
using System.Collections.Generic;
using System.Linq;
using System.Net.Http;
using System.Text;
using System.Threading.Tasks;
namespace StgSharp.PipeLine
{
public static class PipelineMarshal
{
public static IEnumerable<PipelineNode> GetNodeSortedEnumeration(
PipelineScheduler scheduler )
{
ArgumentNullException.ThrowIfNull( scheduler );
return new PipelineNodeEnumerator( scheduler )
{
_count = scheduler._allNode.Count,
};
}
private class PipelineNodeEnumerator : ICollection<PipelineNode>
{
private readonly List<List<PipelineNode>> _mt, _c;
public int _count;
public PipelineNodeEnumerator( PipelineScheduler scheduler )
{
if( !scheduler.TryGetAllNode( out _mt, out _c ) ) {
throw new InvalidOperationException( "Binded scheduler is not sorted." );
}
}
public bool IsReadOnly => true;
public int Count => _count;
public IEnumerator<PipelineNode> GetEnumerator()
{
for( int i = 1; i < _mt.Count; i++ )
{
foreach( PipelineNode node in _mt[ i ] ) {
yield return node;
}
foreach( PipelineNode node in _c[ i ] ) {
yield return node;
}
}
}
void ICollection<PipelineNode>.Add( PipelineNode item )
{
throw new NotSupportedException();
}
void ICollection<PipelineNode>.Clear()
{
throw new NotSupportedException();
}
bool ICollection<PipelineNode>.Contains( PipelineNode item )
{
throw new NotSupportedException();
}
void ICollection<PipelineNode>.CopyTo( PipelineNode[] array, int arrayIndex )
{
throw new NotSupportedException();
}
bool ICollection<PipelineNode>.Remove( PipelineNode item )
{
throw new NotSupportedException();
}
IEnumerator IEnumerable.GetEnumerator()
{
return GetEnumerator();
}
}
}
}
StgSharp.Common/PipeLine/PipelineNodeLabel.cs
+77
View File
@@ -0,0 +1,77 @@
//-----------------------------------------------------------------------
//-----------------------------------------------------------------------
// file="PipelineNodeLabel.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 System;
using System.Collections.Generic;
using System.Diagnostics.CodeAnalysis;
using System.Linq;
using System.Runtime.CompilerServices;
using System.Text;
using System.Threading.Tasks;
namespace StgSharp.PipeLine
{
public abstract record class PipelineNodeLabel
{
public static PipelineNodeLabel Empty => PipelineNodeLabelEmpty.Only;
public abstract int ComputeHashCode();
public override int GetHashCode()
{
return ComputeHashCode();
}
}
public record class PipelineNodeStringLabel : PipelineNodeLabel
{
public PipelineNodeStringLabel( string name )
{
Name = name;
}
public string Name { get; }
public override int ComputeHashCode()
{
return string.GetHashCode( Name );
}
}
internal record class PipelineNodeLabelEmpty : PipelineNodeLabel
{
public static PipelineNodeLabelEmpty Only { get; } = new PipelineNodeLabelEmpty();
public override int ComputeHashCode()
{
return 0;
}
}
}
StgSharp.Common/PipeLine/PipelineNodePort.cs
+104 -71
View File
@@ -38,120 +38,153 @@ using System.Threading.Tasks;
namespace StgSharp.PipeLine
{
public class PipelineNodeExport
public class PipelineNodeInPort
{
private HashSet<PipelineNodeImport> _linkerSet = new HashSet<PipelineNodeImport>();
private SemaphoreSlim formerCompleteSemaphore;
private HashSet<PipelineNodeOutPort> _linkerSet = new HashSet<PipelineNodeOutPort>();
public PipelineNodeExport( PipelineNode container, string name )
public PipelineNodeInPort( PipelineNode container, string label )
{
Container = container;
Name = name;
formerCompleteSemaphore = new SemaphoreSlim( initialCount: 0, maxCount: 1 );
Label = label;
}
public IEnumerable<PipelineNodeOutPort> LinkedPorts => _linkerSet;
public PipelineNode Container { get; init; }
public SemaphoreSlim Waiter { get; init; }
public string Name { get; init; }
public string Label { get; init; }
public void Connect( PipelineNodeImport port )
internal HashSet<PipelineNodeOutPort> Linkers => _linkerSet;
public void Connect( PipelineNodeOutPort port )
{
ArgumentNullException.ThrowIfNull( port );
if( _linkerSet.Add( port ) ) {
Container.Next.Add( port.Container );
}
}
public static void SkipAll( Dictionary<string, PipelineNodeExport> ports )
{
if( ports is null ) {
return;
}
foreach( PipelineNodeExport port in ports.Values ) {
port.Waiter.Release();
}
}
public void TransmitValue( IPipeLineConnectionPayload value )
{
foreach( PipelineNodeImport item in _linkerSet ) { }
}
public void WaitAll( IEnumerable<PipelineNodeExport> ports )
{
if( ports is null ) {
return;
}
foreach( PipelineNodeExport port in ports ) {
port.Waiter.Wait();
}
}
}
public class PipelineNodeImport
{
private HashSet<PipelineNodeExport> _linkerSet = new HashSet<PipelineNodeExport>();
public PipelineNodeImport( PipelineNode container, string name )
{
Container = container;
Name = name;
}
public IEnumerable<PipelineNodeExport> LinkedPorts => _linkerSet;
public PipelineNode Container { get; init; }
public SemaphoreSlim Waiter { get; init; }
public string Name { get; init; }
public void Connect( PipelineNodeExport port )
{
ArgumentNullException.ThrowIfNull( port );
if( _linkerSet.Add( port ) ) {
if( _linkerSet.Add( port ) )
{
port.Linkers.Add( this );
Container.Previous.Add( port.Container );
port.Container.Next.Add( Container );
}
}
public IEnumerator<PipelineNode> GetFormerNodes()
{
foreach( PipelineNodeExport port in _linkerSet ) {
foreach( PipelineNodeOutPort port in _linkerSet ) {
yield return port.Container;
}
}
public void Wait()
{
foreach( PipelineNodeExport connector in _linkerSet ) {
connector.Waiter.Wait();
}
PipelineNodeOutPort.WaitAll( _linkerSet );
}
public static void WaitAll( IDictionary<string, PipelineNodeImport> ports )
public static void WaitAll( IDictionary<string, PipelineNodeInPort> ports )
{
if( ports is null ) {
return;
}
foreach( PipelineNodeImport port in ports.Values ) {
foreach( PipelineNodeInPort port in ports.Values ) {
port.Wait();
}
}
public static void WaitAll( IEnumerable<PipelineNodeImport> ports )
public static void WaitAll( IEnumerable<PipelineNodeInPort> ports )
{
if( ports is null ) {
return;
}
foreach( PipelineNodeImport port in ports ) {
foreach( PipelineNodeInPort port in ports ) {
port.Wait();
}
}
}
public class PipelineNodeOutPort
{
private HashSet<PipelineNodeInPort> _linkerSet = new HashSet<PipelineNodeInPort>();
private volatile int _passCount;
private SemaphoreSlim formerCompleteSemaphore;
public PipelineNodeOutPort( PipelineNode container, string label )
{
Container = container;
Label = label;
formerCompleteSemaphore = new SemaphoreSlim( initialCount: 0, maxCount: 1 );
}
public PipelineNode Container { get; init; }
public string Label { get; init; }
internal HashSet<PipelineNodeInPort> Linkers => _linkerSet;
private SemaphoreSlim Waiter => formerCompleteSemaphore;
public void Connect( PipelineNodeInPort port )
{
ArgumentNullException.ThrowIfNull( port );
if( _linkerSet.Add( port ) )
{
port.Linkers.Add( this );
Container.Next.Add( port.Container );
port.Container.Previous.Add( Container );
}
}
public void Release()
{
if( _passCount == 0 )
{
Waiter.Release();
} else
{
return;
}
}
public static void SkipAll( Dictionary<string, PipelineNodeOutPort> ports )
{
if( ports is null ) {
return;
}
foreach( PipelineNodeOutPort port in ports.Values ) {
port.Release();
}
}
public void TransmitValue( IPipeLineConnectionPayload value )
{
foreach( PipelineNodeInPort item in _linkerSet ) { }
}
public static void WaitAll( IEnumerable<PipelineNodeOutPort> ports )
{
if( ports is null ) {
return;
}
foreach( PipelineNodeOutPort port in ports ) {
port.Wait();
}
}
internal void Wait()
{
Waiter.Wait();
if( _passCount >= _linkerSet.Count )
{
Interlocked.Exchange( ref _passCount, 0 );
} else
{
Interlocked.Increment( ref _passCount );
Waiter.Release();
}
}
}
}
StgSharp.Common/PipeLine/PipelineScheduler.cs
+161 -80
View File
@@ -49,44 +49,41 @@ namespace StgSharp.PipeLine
{
private BeginningNode _beginning;
private Dictionary<string, PipelineNode> allNode;
private bool _isAvailable = false;
private EndingNode _ending;
private List<PipelineNode>
_concurrentNodes = new List<PipelineNode>(),
_mainThreadNodes = new List<PipelineNode>();
private List<PipelineNode> globalNodeList;
private List<PipelineNode> mainThreadNodeList;
private SemaphoreSlim runningStat;
internal int currentGlobalNodeIndex;
internal int currentNativeNodeIndex;
private int mtLevel, cLevel,mtIndex, cIndex;
private List<List<PipelineNode>> _concurrentNodes, _mainThreadNodes ;
private SemaphoreSlim runningStat,_concurrentLock = new SemaphoreSlim( 1, 1 );
internal Dictionary<PipelineNodeLabel, PipelineNode> _allNode;
public PipelineScheduler()
{
allNode = new Dictionary<string, PipelineNode>();
globalNodeList = new List<PipelineNode>();
mainThreadNodeList = new List<PipelineNode>();
_allNode = new Dictionary<PipelineNodeLabel, PipelineNode>();
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<string> InputInterfacesName
public IEnumerable<string> InputPortName
{
get { return _beginning.OutputPorts.Keys; }
get => _beginning.InputPorts.Keys;
}
public IEnumerable<string> OutputInterfacesName
public IEnumerable<string> OutputPortName
{
get { return _beginning.OutputPorts.Keys; }
get => _beginning.OutputPorts.Keys;
}
public PipelineNode BeginLayer => _beginning;
@@ -104,112 +101,196 @@ namespace StgSharp.PipeLine
{
get => runningStat;
}
/**/
public void AddNode( PipelineNode node, bool isNative )
{
ArgumentNullException.ThrowIfNull( node );
node.IsNative = isNative;
allNode.Add( node.Name, node );
_allNode.TryAdd( node.Label, node );
}
public void ArrangeNodeOrder()
public void ArrangeNodes()
{
LinkedList<PipelineNode> _cache = new LinkedList<PipelineNode>();
foreach( PipelineNode node in _beginning.Next )
if( _isAvailable ) {
return;
}
int maxLevel = 1;
Queue<PipelineNode> completedNodes = new Queue<PipelineNode>();
foreach( PipelineNode item in _beginning.Next )
{
if( node.IsNative )
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 )
{
_mainThreadNodes.Add( node );
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<List<PipelineNode>>( maxLevel + 1 );
_mainThreadNodes = new List<List<PipelineNode>>( 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.Add( node );
}
foreach( PipelineNode item in node.Next )
{
item.Previous.Remove( node );
_cache.AddLast( item );
_concurrentNodes[ item.Level ].Add( item );
}
}
while( _cache.Count > 0 )
{
LinkedListNode<PipelineNode> begin = _cache.First!,current = begin,zero;
do
{
PipelineNode node = current.Value;
if( current.Value.Previous.Count == 0 )
{
zero = current;
current = current.Next!;
if( node.IsNative )
{
_mainThreadNodes.Add( node );
} else
{
_concurrentNodes.Add( node );
}
foreach( PipelineNode item in node.Next )
{
item.Previous.Remove( node );
_cache.AddLast( item );
}
_cache.Remove( zero );
} else
{
current = current.Next!;
}
} while (current.Next != begin);
foreach( List<PipelineNode> item in _concurrentNodes ) {
item.TrimExcess();
}
foreach( List<PipelineNode> item in _mainThreadNodes ) {
item.TrimExcess();
}
_isAvailable = true;
}
public PipelineNode Create( [NotNull]IConvertableToPipelineNode body, string name )
public PipelineNode Create(
PipelineNodeLabel label,
[NotNull] IConvertableToPipelineNode body )
{
PipelineNode node = new PipelineNode( name, body, false );
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 name, IPipeLineConnectionPayload arg)[] parameters )
public void SetInput( params (string label, IPipeLineConnectionPayload arg)[] parameters )
{
_beginning.SetInputData( parameters );
}
public void Terminate()
{
Interlocked.Exchange( ref currentGlobalNodeIndex, int.MaxValue );
Interlocked.Exchange( ref currentNativeNodeIndex, int.MaxValue );
_concurrentLock.Wait();
(mtLevel, cLevel, mtIndex, cIndex) = (int.MaxValue, int.MaxValue, int.MaxValue, int.MaxValue);
_concurrentLock.Release();
}
internal bool RequestNexGlobalNode( out PipelineNode node )
public bool TryGetAllNode(
out List<List<PipelineNode>> mainThreadNodes,
out List<List<PipelineNode>> concurrentNodes )
{
if( currentGlobalNodeIndex < globalNodeList.Count )
if( _isAvailable )
{
node = globalNodeList[ currentGlobalNodeIndex ];
Interlocked.Increment( ref currentGlobalNodeIndex );
mainThreadNodes = _mainThreadNodes;
concurrentNodes = _concurrentNodes;
return true;
}
node = default!;
mainThreadNodes = null;
concurrentNodes = null;
return false;
}
internal bool RequestNextNativeNode( out PipelineNode node )
internal bool RequestNexConcurrentNode( out PipelineNode node )
{
if( currentNativeNodeIndex < mainThreadNodeList.Count )
int index, level;
if( cLevel >= _concurrentNodes.Count )
{
node = mainThreadNodeList[ currentNativeNodeIndex ];
Interlocked.Increment( ref currentNativeNodeIndex );
return true;
node = null!;
return false;
}
node = default!;
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<string, PipelineNodeImport> input,
in Dictionary<string, PipelineNodeExport> output )
in Dictionary<string, PipelineNodeInPort> input,
in Dictionary<string, PipelineNodeOutPort> output )
{
PipeLineRunner.Run( this );
PipelineNodeExport.SkipAll( output );
PipelineNodeOutPort.SkipAll( output );
}
~PipelineScheduler()
StgSharp.Common/PipeLine/PortArgs.cs
+5 -5
View File
@@ -44,11 +44,11 @@ namespace StgSharp.PipeLine
{
private protected IPipeLineConnectionPayload args;
private protected PipelineNodeExport input;
private protected PipelineNodeImport output;
private protected PipelineNodeInPort output;
private protected PipelineNodeOutPort input;
private protected SemaphoreSlim formerCompleteSemaphore;
public PipelineConnector( PipelineNodeExport former, PipelineNodeImport after )
public PipelineConnector( PipelineNodeOutPort former, PipelineNodeInPort after )
{
this.input = former;
this.output = after;
@@ -61,9 +61,9 @@ namespace StgSharp.PipeLine
set => args = value;
}
public PipelineNodeExport Input => input;
public PipelineNodeInPort Output => output;
public PipelineNodeImport Output => output;
public PipelineNodeOutPort Input => input;
public static void SkipAll( Dictionary<string, PipelineConnector> ports )
{
StgSharp.Common/PipeLine/SideNode.cs
+13 -7
View File
@@ -28,7 +28,6 @@
//
//-----------------------------------------------------------------------
//-----------------------------------------------------------------------
using System;
using System.Collections.Generic;
using System.Globalization;
@@ -43,18 +42,24 @@ namespace StgSharp.PipeLine
internal sealed class BeginningNode : PipelineNode
{
public const string OutName = "default";
private PipelineScheduler _bp;
public BeginningNode( PipelineScheduler bp )
: base( DefaultOperation, name: "BeginningNode", true, ["default"], ["default"] )
: base(
DefaultOperation, new PipelineNodeStringLabel( "BeginningNode" ), true, ["default"],
[OutName] )
{
_bp = bp;
}
public PipelineNodeOutPort DefaultOut => OutputPorts[ OutName ];
public override void Run()
{
_bp.RunningStat.Wait();
PipelineNodeExport.SkipAll( _output );
PipelineNodeOutPort.SkipAll( _output );
}
internal void SetInputData( IEnumerable<(string, IPipeLineConnectionPayload)> data )
@@ -75,16 +80,17 @@ namespace StgSharp.PipeLine
private PipelineScheduler _bp;
public EndingNode( PipelineScheduler bp )
: base( DefaultOperation, name: "EndingNode", true, ["default"], ["default"] )
: base(
DefaultOperation, name: new PipelineNodeStringLabel( "EndingNode" ), true, ["default"],
["default"] )
{
_bp = bp;
}
public override void Run()
{
PipelineNodeImport.WaitAll( _input );
Interlocked.Exchange( ref _bp.currentGlobalNodeIndex, 0 );
Interlocked.Exchange( ref _bp.currentNativeNodeIndex, 0 );
PipelineNodeInPort.WaitAll( _input );
_bp.ResetIndex();
_bp.RunningStat.Release();
}