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

MulAvailable

Browse Source
This commit is contained in:
Nitload
2025-12-16 23:13:17 +08:00
parent 28ea5d1d1e
commit 3486375470
2637 changed files with 23787 additions and 838773 deletions
Download Patch File Download Diff File Expand all files Collapse all files
StgSharp.Common/PipeLine/Blueprint.NodeCraeater.cs
-58
View File
@@ -1,58 +0,0 @@
//-----------------------------------------------------------------------
//-----------------------------------------------------------------------
// file="PipeLine.NodeCraeater.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.Text;
namespace StgSharp.PipeLine
{
public partial class BlueprintScheduler
{
public PipeLineNode CreateBlueprintNode(
string name,
BlueprintNodeOperation operation,
string[] inputPort,
string[] outputPort,
bool isNative )
{
PipeLineNode node = new PipeLineNode(
operation, name, isNative, inputPort, outputPort );
if( isNative ) {
mainThreadNodeList.Add( node );
} else {
globalNodeList.Add( node );
}
return node;
}
}
}
StgSharp.Common/PipeLine/Blueprint.cs
-177
View File
@@ -1,177 +0,0 @@
//-----------------------------------------------------------------------
//-----------------------------------------------------------------------
// file="Blueprint.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.Linq;
using System.Runtime.CompilerServices;
using System.Threading;
namespace StgSharp.PipeLine
{
/// <summary>
/// A sets of complex logic designed for running in multi threads environment. Warning: do not
/// run a <see cref="BlueprintScheduler" /> instance in multi threads more than once.
/// </summary>
public partial class BlueprintScheduler : IConvertableToBlueprintNode
{
private List<PipeLineNode>[] _concurrentNodes, _mainThreadNodes;
private BeginningNode beginNode;
private Dictionary<string, PipeLineNode> allNode;
private EndingNode endNode;
private Func<bool> terminate;
private List<PipeLineNode> globalNodeList;
private List<PipeLineNode> mainThreadNodeList;
private SemaphoreSlim runningStat;
internal int currentGlobalNodeIndex;
internal int currentNativeNodeIndex;
public BlueprintScheduler()
{
allNode = new Dictionary<string, PipeLineNode>();
globalNodeList = new List<PipeLineNode>();
mainThreadNodeList = new List<PipeLineNode>();
runningStat = new SemaphoreSlim( 1, 1 );
beginNode = new BeginningNode( this );
endNode = new EndingNode( this );
}
public BlueprintNodeOperation Operation
{
get => ExecuteMain;
}
public bool IsRunning
{
get => runningStat.CurrentCount == 0;
}
public IEnumerable<string> InputInterfacesName
{
get { return beginNode.OutputInterfaces.Keys; }
}
public IEnumerable<string> OutputInterfacesName
{
get { return beginNode.OutputInterfaces.Keys; }
}
public PipeLineNode BeginLayer => beginNode;
public PipeLineNode EndLayer => endNode;
/**/
internal SemaphoreSlim RunningStat
{
get => runningStat;
}
/**/
public void AddNode( PipeLineNode node, bool isNative )
{
_ = node ?? throw new ArgumentNullException( nameof( node ) );
node.IsNative = isNative;
allNode.Add( node.Name, node );
}
public void ExecuteMain(
in Dictionary<string, PipelineConnector> input,
in Dictionary<string, PipelineConnector> output )
{
PipeLineRunner.Run( this );
}
[MethodImpl( MethodImplOptions.AggressiveInlining )]
public (string name, PipeLineConnectorArgs arg)[] GetOutput()
{
return endNode.OutputData;
}
public void Init()
{
foreach( KeyValuePair<string, PipeLineNode> item in allNode ) {
if( item.Value.IsNative ) {
mainThreadNodeList.Add( item.Value );
} else {
globalNodeList.Add( item.Value );
}
}
mainThreadNodeList.Sort();
globalNodeList.Sort();
}
[MethodImpl( MethodImplOptions.AggressiveInlining )]
public void SetInput(
params (string name, PipeLineConnectorArgs arg)[] parameters )
{
beginNode.InputData = parameters;
}
public void Terminate()
{
Interlocked.Exchange( ref currentGlobalNodeIndex, int.MaxValue );
Interlocked.Exchange( ref currentNativeNodeIndex, int.MaxValue );
}
internal bool RequestNexGlobalNode( out PipeLineNode node )
{
if( currentGlobalNodeIndex < globalNodeList.Count ) {
node = globalNodeList[ currentGlobalNodeIndex ];
Interlocked.Increment( ref currentGlobalNodeIndex );
return true;
}
node = default!;
return false;
}
internal bool RequestNextNativeNode( out PipeLineNode node )
{
if( currentNativeNodeIndex < mainThreadNodeList.Count ) {
node = mainThreadNodeList[ currentNativeNodeIndex ];
Interlocked.Increment( ref currentNativeNodeIndex );
return true;
}
node = default!;
return false;
}
~BlueprintScheduler()
{
runningStat.Dispose();
}
}
}
StgSharp.Common/PipeLine/Connector.cs
-168
View File
@@ -1,168 +0,0 @@
//-----------------------------------------------------------------------
//-----------------------------------------------------------------------
// file="Connector.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.Linq;
using System.Numerics;
using System.Runtime.CompilerServices;
using System.Security.Authentication.ExtendedProtection;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
namespace StgSharp.PipeLine
{
public class PipelineConnector
{
private protected PipeLineNode after;
private protected PipeLineNode former;
private protected PipeLineConnectorArgs args;
private protected SemaphoreSlim formerCompleteSemaphore;
public PipelineConnector( PipeLineNode former, PipeLineNode after )
{
this.former = former;
this.after = after;
formerCompleteSemaphore = new SemaphoreSlim(
initialCount: 0, maxCount: 1 );
}
public PipeLineNode After => after;
public PipeLineNode Former => former;
public PipeLineConnectorArgs Args
{
get => args;
set => args = value;
}
public int Level => former.Level;
public static void SkipAll(
Dictionary<string, PipelineConnector> ports )
{
if( ports == null ) {
return;
}
foreach( KeyValuePair<string, PipelineConnector> item in ports ) {
if( item.Value != null ) {
item.Value.CompleteAndSkip();
}
}
}
public static void WaitAll(
Dictionary<string, PipelineConnector> ports )
{
if( ports == null ) {
return;
}
foreach( KeyValuePair<string, PipelineConnector> item in ports ) {
if( item.Value != null ) {
item.Value.WaitForStart();
}
}
}
public virtual void WaitForStart()
{
//Console.WriteLine($"Waiting \t{former.ContextName}\t->\t{after.ContextName}");
formerCompleteSemaphore.Wait();
}
~PipelineConnector()
{
formerCompleteSemaphore.Dispose();
}
#region pipline op
[MethodImpl( MethodImplOptions.AggressiveInlining )]
public void CompleteAndSkip()
{
//Console.WriteLine($"Complete\t{former.Name}\t->\t{after.Name}");
formerCompleteSemaphore.Release();
}
[MethodImpl( MethodImplOptions.AggressiveInlining )]
public void CompleteAndWriteData( PipeLineConnectorArgs args )
{
this.args = args;
CompleteAndSkip();
}
#endregion
}
public abstract class PipeLineConnectorArgs
{
public object? Value
{
get;
set;
}
public object? ValueDefault
{
get;
}
}
public sealed class SealedBlueprintPiipeline : PipelineConnector
{
internal SealedBlueprintPiipeline(
PipeLineNode former,
PipeLineNode after )
: base( former, after ) { }
public static SealedBlueprintPiipeline SealInput(
PipeLineNode node,
string portName )
{
return new SealedBlueprintPiipeline( PipeLineNode.Empty, node );
}
public static SealedBlueprintPiipeline SealOutput(
PipeLineNode node,
string portName )
{
return new SealedBlueprintPiipeline( node, PipeLineNode.Empty );
}
public override sealed void WaitForStart() { }
}
}
StgSharp.Common/PipeLine/CrossFramePipelineNode.cs
-93
View File
@@ -1,93 +0,0 @@
//-----------------------------------------------------------------------
//-----------------------------------------------------------------------
// file="CrossFramePipelineNode.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.PipeLine;
using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
namespace StgSharp.PipeLine
{
public class CrossFrameOperation : IConvertableToPipelineNode
{
private static string[] _inputName = ["TaskIn"];
private static string[] _outputName = ["TaskOut"];
private Action _mainExecution;
private int currentCount;
private readonly int cycleCount;
public CrossFrameOperation( Action startup, int cycleCount )
{
MainExecution = startup;
this.cycleCount = ( cycleCount > 1 ) ?
cycleCount : throw new ArgumentException(
$"{$"The {typeof(CrossFrameOperation).Name} crosses only one cycle, "}{$"please use a {typeof(PipelineNode).Name} instead."}" );
this.cycleCount = cycleCount;
currentCount = cycleCount;
}
public Action MainExecution
{
get => _mainExecution;
private protected set => _mainExecution = value;
}
public IEnumerable<string> InputPortName => _inputName;
public IEnumerable<string> OutputPortName => _outputName;
public PipelineNodeOperation Operation
{
get => NodeMain;
}
public void NodeMain(
in Dictionary<string, PipelineNodeInPort> input,
in Dictionary<string, PipelineNodeOutPort> output )
{
if( currentCount < cycleCount )
{
Interlocked.Increment( ref currentCount );
} else
{
Interlocked.Exchange( ref currentCount, 0 );
Task.Run( MainExecution );
}
//Console.WriteLine(currentCount);
PipelineNodeOutPort.SkipAll( output );
}
}
}
StgSharp.Common/PipeLine/IConvertableToBlueprintNode.cs
-88
View File
@@ -1,88 +0,0 @@
//-----------------------------------------------------------------------
//-----------------------------------------------------------------------
// file="IConvertableToBlueprintNode.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.Runtime.CompilerServices;
using System.Text;
namespace StgSharp.PipeLine
{
public delegate void PipelineNodeOperation(
in Dictionary<string, PipelineNodeImport> input,
in Dictionary<string, PipelineNodeExport> output );
public interface IConvertableToBlueprintNode
{
PipelineNodeOperation Operation { get; }
public IEnumerable<string> InputInterfacesName { get; }
public IEnumerable<string> OutputInterfacesName { get; }
void NodeMain(
in Dictionary<string, PipelineNodeImport> input,
in Dictionary<string, PipelineNodeExport> output );
}
internal class DefaultConvertableToBlueprintNode : IConvertableToBlueprintNode
{
private string[] _input;
private string[] _output;
private PipelineNodeOperation _execution;
public DefaultConvertableToBlueprintNode(
PipelineNodeOperation execution,
string[] input,
string[] output )
{
_input = input;
_output = output;
_execution = execution;
}
public PipelineNodeOperation Operation => _execution;
public IEnumerable<string> InputInterfacesName => _input;
public IEnumerable<string> OutputInterfacesName => _output;
public void NodeMain(
in Dictionary<string, PipelineNodeImport> input,
in Dictionary<string, PipelineNodeExport> output )
{
_execution.Invoke( input, output );
}
}
}
StgSharp.Common/PipeLine/IConvertableToPipelineNode.cs
-88
View File
@@ -1,88 +0,0 @@
//-----------------------------------------------------------------------
//-----------------------------------------------------------------------
// file="IConvertableToPipelineNode.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.Runtime.CompilerServices;
using System.Text;
namespace StgSharp.PipeLine
{
public delegate void PipelineNodeOperation(
in Dictionary<string, PipelineNodeInPort> input,
in Dictionary<string, PipelineNodeOutPort> output );
public interface IConvertableToPipelineNode
{
public IEnumerable<string> InputPortName { get; }
public IEnumerable<string> OutputPortName { get; }
PipelineNodeOperation Operation { get; }
void NodeMain(
in Dictionary<string, PipelineNodeInPort> input,
in Dictionary<string, PipelineNodeOutPort> output );
}
internal class DefaultConvertableToBlueprintNode : IConvertableToPipelineNode
{
private string[] _input;
private string[] _output;
private PipelineNodeOperation _execution;
public DefaultConvertableToBlueprintNode(
PipelineNodeOperation execution,
string[] input,
string[] output )
{
_input = input;
_output = output;
_execution = execution;
}
public IEnumerable<string> InputPortName => _input;
public IEnumerable<string> OutputPortName => _output;
public PipelineNodeOperation Operation => _execution;
public void NodeMain(
in Dictionary<string, PipelineNodeInPort> input,
in Dictionary<string, PipelineNodeOutPort> output )
{
_execution.Invoke( input, output );
}
}
}
StgSharp.Common/PipeLine/PipeLineNode.cs
-300
View File
@@ -1,300 +0,0 @@
//-----------------------------------------------------------------------
//-----------------------------------------------------------------------
// file="PipeLineNode.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.Entities;
using System;
using System.Collections.Generic;
using System.Diagnostics.CodeAnalysis;
using System.Linq;
using System.Reflection;
using System.Runtime.CompilerServices;
using System.Runtime.Serialization;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
namespace StgSharp.PipeLine
{
/// <summary>
///
/// </summary>
/// <risk>
/// Reset pipeline has risk, this operation is a direct operation to port in current node, the
/// port on the other node cannot get the change.
/// </risk>
#pragma warning disable CA1036
public class PipeLineNode : IComparable<PipeLineNode>
#pragma warning restore CA1036
{
private static Dictionary<string, PipelineConnector> _emptyLayer = new();
private static PipeLineNode _emptyNode = new PipeLineNode(
string.Empty, null!, false );
private IConvertableToBlueprintNode _mainOperation;
private string _name;
private protected bool _levelAvailable, _nativeRequested;
private protected Dictionary<string, PipelineConnector> _inputInterfaces;
private protected Dictionary<string, PipelineConnector> _outputInterfaces;
private protected int _level;
internal PipeLineNode(
string name,
IConvertableToBlueprintNode node,
bool isNative )
{
_levelAvailable = false;
_name = name;
_nativeRequested = isNative;
_mainOperation = node;
_inputInterfaces = new Dictionary<string, PipelineConnector>();
_outputInterfaces = new Dictionary<string, PipelineConnector>();
foreach( string portName in node.InputInterfacesName ) {
_inputInterfaces.Add( portName, null! );
}
foreach( string portName in node.OutputInterfacesName ) {
_outputInterfaces.Add( portName, null! );
}
}
#pragma warning disable CS8618
internal PipeLineNode(
BlueprintNodeOperation operation,
string name,
bool isNative,
string[] inputPorts,
string[] outputPorts )
#pragma warning restore CS8618
{
if( ( inputPorts == null ) || ( inputPorts.Length == 0 ) ) {
throw new ArgumentNullException( nameof( inputPorts ) );
}
if( ( outputPorts == null ) || ( outputPorts.Length == 0 ) ) {
throw new ArgumentNullException( nameof( outputPorts ) );
}
_levelAvailable = false;
_name = name;
_nativeRequested = isNative;
operation = operation ?? DefaultOperation;
_mainOperation = new DefaultConvertableToBlueprintNode(
operation, inputPorts, outputPorts );
_inputInterfaces = new Dictionary<string, PipelineConnector>();
_outputInterfaces = new Dictionary<string, PipelineConnector>();
foreach( string portName in inputPorts ) {
_inputInterfaces.Add(
portName,
SealedBlueprintPiipeline.SealOutput( this, name ) );
}
foreach( string portName in outputPorts ) {
_outputInterfaces.Add( portName, null! );
}
}
public bool IsNative
{
get => _nativeRequested;
internal set => _nativeRequested = value;
}
public static Dictionary<string, PipelineConnector> EmptyPortLayer
{
get => _emptyLayer;
}
public int Level
{
get
{
if( _levelAvailable ) {
return _level;
}
_level = 0;
foreach( KeyValuePair<string, PipelineConnector> item in _inputInterfaces ) {
if( item.Value == null ) {
continue;
}
if( _level < item.Value.Level ) {
_level = item.Value.Level;
}
}
_level += 1;
_levelAvailable = true;
return _level;
}
}
public static PipeLineNode Empty
{
get => _emptyNode;
}
public string Name => _name;
internal Dictionary<string, PipelineConnector> InputInterfaces => _inputInterfaces;
internal Dictionary<string, PipelineConnector> OutputInterfaces => _outputInterfaces;
internal IConvertableToBlueprintNode Value
{
get => _mainOperation;
}
public void Append(
[NotNull] PipeLineNode next,
string thisOutputPortName,
string afterInputPortName )
{
this._levelAvailable = false;
PipelineConnector pipeline = new PipelineConnector( this, next );
if( next is EndingNode enode ) {
enode.SetCertainInputPort( afterInputPortName, pipeline );
} else {
next.SetCertainInputPort( afterInputPortName, pipeline );
}
SetCertainOutputPort( thisOutputPortName, pipeline );
}
public int CompareTo( PipeLineNode other )
{
if( null == other ) {
throw new ArgumentNullException( nameof( other ) );
}
return this.Level.CompareTo( other.Level );
}
public void Prepend(
[NotNull] PipeLineNode former,
string formerOutputPortName,
string thisInputPortName )
{
this._levelAvailable = false;
PipelineConnector pipeline = new PipelineConnector( former, this );
if( former is BeginningNode bnode ) {
bnode.SetCertainOutputPort( formerOutputPortName, pipeline );
} else {
former.SetCertainOutputPort( formerOutputPortName, pipeline );
}
SetCertainInputPort( thisInputPortName, pipeline );
}
public virtual void Run()
{
try {
//Console.WriteLine(_name);
PipelineConnector.WaitAll( _inputInterfaces );
if( _mainOperation != null ) {
_mainOperation.Operation(
in _inputInterfaces, in _outputInterfaces );
}
}
catch( Exception ) {
throw;
}
}
protected static void DefaultOperation(
in Dictionary<string, PipelineConnector> input,
in Dictionary<string, PipelineConnector> output )
{
PipelineConnector.SkipAll( output );
}
#region port operation
public PipelineConnector GetInputPort( string name )
{
PipelineConnector ret;
if( !_inputInterfaces.TryGetValue( name, out ret ) ) {
throw new ArgumentOutOfRangeException(
$"Cannot find the interface named {name} in node {Name}." );
}
return ret;
}
public PipelineConnector GetOutputPort( string name )
{
PipelineConnector ret;
if( !_outputInterfaces.TryGetValue( name, out ret ) ) {
throw new ArgumentOutOfRangeException(
$"Cannot find the interface named {name} in node {Name}." );
}
return ret;
}
public void SetCertainInputPort(
string name,
PipelineConnector pipeline )
{
if( !_inputInterfaces.ContainsKey( name ) ) {
throw new ArgumentException(
$"Cannot find the interface named {name} in node {Name}." );
}
_inputInterfaces[ name ] = pipeline;
}
public void SetCertainOutputPort(
string name,
PipelineConnector pipeline )
{
if( !_outputInterfaces.ContainsKey( name ) ) {
throw new ArgumentOutOfRangeException(
$"Cannot find the interface named {name} in node {Name}." );
}
_outputInterfaces[ name ] = pipeline;
}
#endregion
}//--------------------------------- end of class -----------------------------------
public class BlueprintNodeRunException : Exception
{
protected BlueprintNodeRunException(
SerializationInfo info,
StreamingContext context )
: base( info, context ) { }
public BlueprintNodeRunException() { }
public BlueprintNodeRunException( Exception ex )
: base( string.Empty, ex ) { }
public BlueprintNodeRunException( string message ) : base( message ) { }
public BlueprintNodeRunException(
string message,
Exception innerException )
: base( message, innerException ) { }
}
}
StgSharp.Common/PipeLine/PipeLineRunner.cs
-247
View File
@@ -1,247 +0,0 @@
//-----------------------------------------------------------------------
//-----------------------------------------------------------------------
// file="PipeLineRunner.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.Threading;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Data;
using System.Diagnostics;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
namespace StgSharp.PipeLine
{
public static class PipeLineRunner
{
private static NodeRunner[] allAvailableTask = CreateBlueprintRunnerArray();
private static CancellationTokenSource cts = new CancellationTokenSource();
private static ConcurrentStack<PipelineScheduler> pipelineInWaiting
= new ConcurrentStack<PipelineScheduler>();
private static int _callingThread;
private static SemaphoreSlim pipelineLock = new SemaphoreSlim( 0, 1 );
public static void Run( PipelineScheduler pipeline )
{
//Console.WriteLine();
if( pipeline == null )
{
return;
}
if( pipeline.IsRunning ) {
throw new InvalidOperationException( "Attempt to run a blueprint instance twice." );
}
if( !IsValidCalling() ) {
pipelineLock.Wait();
}
if( InterruptAndRunNewPipeline( pipeline ) )
{
while( InterruptAndRunNewPipeline( NodeRunner.CurrentPipeline! ) ) { }
}
pipelineLock.Release();
Interlocked.Exchange( ref _callingThread, -1 );
}
public static void SetMaxTaskCount( int size )
{
if( size < 1 ) {
size = 1;
}
if( allAvailableTask.Length >= size )
{
for( int i = size; i < allAvailableTask.Length; i++ )
{
if( allAvailableTask[ i ] != null ) {
allAvailableTask[ i ].Cancel();
}
}
}
Array.Resize( ref allAvailableTask, size );
}
public static void Terminate()
{
cts.Cancel();
if( NodeRunner.CurrentPipeline is null ) {
return;
}
NodeRunner.CurrentPipeline.Terminate();
foreach( NodeRunner task in allAvailableTask ) {
task.Cts.Cancel();
}
foreach( PipelineScheduler bp in pipelineInWaiting ) {
bp.Terminate();
}
pipelineInWaiting.Clear();
}
internal static bool InterruptAndRunNewPipeline( PipelineScheduler newPipeline )
{
ArgumentNullException.ThrowIfNull( newPipeline );
if( ( newPipeline != NodeRunner.CurrentPipeline ) &&
( NodeRunner.CurrentPipeline != null ) )
{
//if a running bp exists
pipelineInWaiting.Push( NodeRunner.CurrentPipeline );
}
//load new bp to current bp
NodeRunner.CurrentPipeline = newPipeline;
Interlocked.Exchange( ref _callingThread, Environment.CurrentManagedThreadId );
//run beginning node to init it first.
if( !NodeRunner.CurrentPipeline.IsRunning )
{
( NodeRunner.CurrentPipeline.BeginLayer as BeginningNode )!.Run();
}
//refresh all tasks
/*
for( int i = 0; i < allAvailableTask.Length; i++ )
{
NodeRunner item = allAvailableTask[ i ];
if( ( item is null ) || item.IsCompleted ) {
allAvailableTask[ i ] = NodeRunner.Run();
}
}
/**/
PipelineNode node;
while( !cts.Token.IsCancellationRequested &&
NodeRunner.CurrentPipeline.RequestNextMainThreadNode( out node ) )
{
node.Run();
if( !NodeRunner.CurrentPipeline.IsRunning )
{
break;
}
}
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()
{
int count = Environment.ProcessorCount;
return new NodeRunner[ count < 2 ? count : count - 2];
}
}
internal class NodeRunner : CancellableTask
{
private static volatile PipelineScheduler? _current;
internal NodeRunner()
: base()
{
_startup = InternalRun;
}
internal static PipelineScheduler? CurrentPipeline
{
get => _current;
set => _current = value;
}
internal static NodeRunner Run()
{
NodeRunner ret = new NodeRunner
{
_startup = InternalRun
};
ret._internalTask = Task.Run( ret.InternalAction );
return ret;
}
private static void InternalRun( CancellationToken token )
{
PipelineNode item;
Console.WriteLine( $"Global thread {Environment.CurrentManagedThreadId} starts." );
while( ( CurrentPipeline != null ) &&
CurrentPipeline.IsRunning &&
!token.IsCancellationRequested &&
CurrentPipeline.RequestNexConcurrentNode( out item ) ) {
item.Run();
}
}
}//------------------------------------- End of Class ------------------------------------------
}
StgSharp.Common/PipeLine/PipelineMarshal.cs
-118
View File
@@ -1,118 +0,0 @@
//-----------------------------------------------------------------------
//-----------------------------------------------------------------------
// 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/PipelineNode.cs
+266
View File
@@ -0,0 +1,266 @@
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
// file="PipelineNode"
// 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 System;
using System.Collections.Generic;
using System.Diagnostics.CodeAnalysis;
using System.Linq;
using System.Reflection;
using System.Runtime.CompilerServices;
using System.Runtime.Serialization;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
namespace StgSharp.PipeLine
{
/// <summary>
///
/// </summary>
/// <risk>
/// Reset pipeline has risk, this operation is a direct operation to port in current node, the
/// port on the other node cannot get the change.
/// </risk>
#pragma warning disable CA1036
public class PipelineNode
#pragma warning restore CA1036
{
private static readonly Dictionary<string, PipelineConnector> _emptyLayer = [];
private static PipelineNode _emptyNode = new PipelineNode(
PipelineNodeLabel.Empty, null!, false);
private IConvertableToPipelineNode _mainOperation;
private PipelineNodeLabel _label;
private protected bool _nativeRequested;
private protected Dictionary<string, PipelineNodeInPort> _input;
private protected Dictionary<string, PipelineNodeOutPort> _output;
private protected int _level, _count;
internal PipelineNode(PipelineNodeLabel label, IConvertableToPipelineNode node, bool isNative)
{
_label = label;
_nativeRequested = isNative;
_mainOperation = node;
InputPorts = new();
OutputPorts = new();
foreach (string port in node.InputPortName) {
InputPorts.Add(port, new PipelineNodeInPort(this, port));
}
foreach (string port in node.OutputPortName) {
OutputPorts.Add(port, new PipelineNodeOutPort(this, port));
}
}
#pragma warning disable CS8618
internal PipelineNode(
PipelineNodeOperation operation,
PipelineNodeLabel name,
bool isNative,
string[] inputNames,
string[] outputNames)
#pragma warning restore CS8618
{
if ((inputNames == null) || (inputNames.Length == 0)) {
throw new ArgumentNullException(nameof(inputNames));
}
if ((outputNames == null) || (outputNames.Length == 0)) {
throw new ArgumentNullException(nameof(outputNames));
}
_label = name;
_nativeRequested = isNative;
if (operation == null) {
operation = DefaultOperation;
}
_mainOperation = new DefaultConvertableToBlueprintNode(
operation, inputNames, outputNames);
InputPorts = new Dictionary<string, PipelineNodeInPort>();
OutputPorts = new Dictionary<string, PipelineNodeOutPort>();
foreach (string port in inputNames) {
InputPorts.Add(port, new PipelineNodeInPort(this, port));
}
foreach (string port in outputNames) {
OutputPorts.Add(port, new PipelineNodeOutPort(this, port));
}
}
public bool IsNative
{
get => _nativeRequested;
internal set => _nativeRequested = value;
}
public static Dictionary<string, PipelineConnector> EmptyPortLayer => _emptyLayer;
public IConvertableToPipelineNode NodeOperation => _mainOperation;
public int Level
{
get => _level;
private set => _level = value;
}
public static PipelineNode Empty => _emptyNode;
public PipelineNodeLabel Label => _label;
internal Dictionary<string, PipelineNodeInPort> InputPorts
{
get => _input;
private set => _input = value;
}
internal Dictionary<string, PipelineNodeOutPort> OutputPorts
{
get => _output;
private set => _output = value;
}
internal HashSet<PipelineNode> Previous { get; } = [];
internal HashSet<PipelineNode> Next { get; } = [];
internal IConvertableToPipelineNode Value => _mainOperation;
internal int RemainingIncreaseCount => Previous.Count - _count;
public static void Connect(PipelineNode front, string frontNodePort, string backNodePort, PipelineNode back)
{
ArgumentNullException.ThrowIfNull(front);
ArgumentNullException.ThrowIfNull(back);
if (!front.OutputPorts
.TryGetValue(frontNodePort,
out PipelineNodeOutPort? frontPort)) {
throw new ArgumentOutOfRangeException($"Cannot find port named {frontNodePort}");
}
if (!back.InputPorts.TryGetValue(backNodePort, out PipelineNodeInPort? backPort)) {
throw new ArgumentOutOfRangeException($"Cannot find port named {backNodePort}");
}
backPort.Connect(frontPort);
frontPort.Connect(backPort);
}
public IEnumerator<PipelineNode> GetFormerNodes()
{
foreach (PipelineNodeInPort port in InputPorts.Values)
{
IEnumerator<PipelineNode> enumerator = port.GetFormerNodes();
while (enumerator.MoveNext()) {
yield return enumerator.Current;
}
}
}
public virtual void Run()
{
try
{
// Console.WriteLine(_label);
PipelineNodeInPort.WaitAll(InputPorts);
if (_mainOperation != null) {
_mainOperation.NodeMain(in _input, in _output);
}
}
catch (Exception)
{
throw;
}
}
internal bool TryIncreaseLevelTo(int level)
{
_count++;
if (_count > Previous.Count) {
return false;
}
Level = int.Max(level, this.Level);
return true;
}
protected static void DefaultOperation(
in Dictionary<string, PipelineNodeInPort> input,
in Dictionary<string, PipelineNodeOutPort> output)
{
PipelineNodeOutPort.SkipAll(output);
}
#region port operation
public PipelineNodeInPort GetInputPort(string lable)
{
if (!InputPorts.TryGetValue(lable, out PipelineNodeInPort? ret)) {
throw new ArgumentOutOfRangeException(
$"Cannot find the interface named {lable} in node {Label}.");
}
return ret;
}
public PipelineNodeOutPort GetOutputPort(string lable)
{
if (!OutputPorts.TryGetValue(lable, out PipelineNodeOutPort? ret)) {
throw new ArgumentOutOfRangeException(
$"Cannot find the interface named {lable} in node {Label}.");
}
return ret;
}
public PipelineNodeInPort? DefineCertainInputPort(string label)
{
if (InputPorts.ContainsKey(label)) {
return null;
}
PipelineNodeInPort port = new(this, label);
InputPorts[label] = port;
return port;
}
public PipelineNodeOutPort? DefineCertainOutputPort(string label)
{
if (OutputPorts.ContainsKey(label)) {
return null;
}
PipelineNodeOutPort port = new(this, label);
OutputPorts[label] = port;
return port;
}
#endregion
}//--------------------------------- end of class -----------------------------------
public class PipelineNodeRunException : Exception
{
public PipelineNodeRunException() { }
public PipelineNodeRunException(Exception ex) : base(string.Empty, ex) { }
public PipelineNodeRunException(string message) : base(message) { }
public PipelineNodeRunException(string message, Exception innerException) : base(message, innerException) { }
}
}
StgSharp.Common/PipeLine/PipelineNodeLabel.cs
-77
View File
@@ -1,77 +0,0 @@
//-----------------------------------------------------------------------
//-----------------------------------------------------------------------
// 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
-190
View File
@@ -1,190 +0,0 @@
//-----------------------------------------------------------------------
//-----------------------------------------------------------------------
// file="PipelineNodePort.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.Linq;
using System.Runtime.ExceptionServices;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
namespace StgSharp.PipeLine
{
public class PipelineNodeInPort
{
private HashSet<PipelineNodeOutPort> _linkerSet = new HashSet<PipelineNodeOutPort>();
public PipelineNodeInPort( PipelineNode container, string label )
{
Container = container;
Label = label;
}
public IEnumerable<PipelineNodeOutPort> LinkedPorts => _linkerSet;
public PipelineNode Container { get; init; }
public SemaphoreSlim Waiter { get; init; }
public string Label { get; init; }
internal HashSet<PipelineNodeOutPort> Linkers => _linkerSet;
public void Connect( PipelineNodeOutPort port )
{
ArgumentNullException.ThrowIfNull( port );
if( _linkerSet.Add( port ) )
{
port.Linkers.Add( this );
Container.Previous.Add( port.Container );
port.Container.Next.Add( Container );
}
}
public IEnumerator<PipelineNode> GetFormerNodes()
{
foreach( PipelineNodeOutPort port in _linkerSet ) {
yield return port.Container;
}
}
public void Wait()
{
PipelineNodeOutPort.WaitAll( _linkerSet );
}
public static void WaitAll( IDictionary<string, PipelineNodeInPort> ports )
{
if( ports is null ) {
return;
}
foreach( PipelineNodeInPort port in ports.Values ) {
port.Wait();
}
}
public static void WaitAll( IEnumerable<PipelineNodeInPort> ports )
{
if( ports is null ) {
return;
}
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.NodeCraeater.cs
-60
View File
@@ -1,60 +0,0 @@
//-----------------------------------------------------------------------
//-----------------------------------------------------------------------
// file="PipelineScheduler.NodeCraeater.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.Text;
namespace StgSharp.PipeLine
{
public partial class PipelineScheduler
{
public PipelineNode CreateBlueprintNode(
string name,
PipelineNodeOperation operation,
string[] inputPort,
string[] outputPort,
bool isNative )
{
PipelineNode node = new PipelineNode( operation, name, isNative, inputPort,
outputPort );
if( isNative )
{
mainThreadNodeList.Add( node );
} else
{
globalNodeList.Add( node );
}
return node;
}
}
}
StgSharp.Common/PipeLine/PipelineScheduler.cs
-302
View File
@@ -1,302 +0,0 @@
//-----------------------------------------------------------------------
//-----------------------------------------------------------------------
// 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
{
/// <summary>
/// A sets of complex logic designed for running in multi threads environment. Warning: do not
/// run a <see cref="PipelineScheduler" /> instance in multi threads more than once.
/// </summary>
public partial class PipelineScheduler : IConvertableToPipelineNode
{
private BeginningNode _beginning;
private bool _isAvailable = false;
private EndingNode _ending;
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<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> InputPortName
{
get => _beginning.InputPorts.Keys;
}
public IEnumerable<string> 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<PipelineNode> completedNodes = new Queue<PipelineNode>();
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<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[ item.Level ].Add( item );
}
}
foreach( List<PipelineNode> item in _concurrentNodes ) {
item.TrimExcess();
}
foreach( List<PipelineNode> 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<List<PipelineNode>> mainThreadNodes,
out List<List<PipelineNode>> 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<string, PipelineNodeInPort> input,
in Dictionary<string, PipelineNodeOutPort> output )
{
PipeLineRunner.Run( this );
PipelineNodeOutPort.SkipAll( output );
}
~PipelineScheduler()
{
runningStat.Dispose();
}
}
}
StgSharp.Common/PipeLine/PortArgs.cs
-126
View File
@@ -1,126 +0,0 @@
//-----------------------------------------------------------------------
//-----------------------------------------------------------------------
// file="PortArgs.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.Linq;
using System.Numerics;
using System.Runtime.CompilerServices;
using System.Security.Authentication.ExtendedProtection;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
namespace StgSharp.PipeLine
{
public class PipelineConnector
{
private protected IPipeLineConnectionPayload args;
private protected PipelineNodeInPort output;
private protected PipelineNodeOutPort input;
private protected SemaphoreSlim formerCompleteSemaphore;
public PipelineConnector( PipelineNodeOutPort former, PipelineNodeInPort after )
{
this.input = former;
this.output = after;
formerCompleteSemaphore = new SemaphoreSlim( initialCount: 0, maxCount: 1 );
}
public IPipeLineConnectionPayload Args
{
get => args;
set => args = value;
}
public PipelineNodeInPort Output => output;
public PipelineNodeOutPort Input => input;
public static void SkipAll( Dictionary<string, PipelineConnector> ports )
{
if( ports == null ) {
return;
}
foreach( KeyValuePair<string, PipelineConnector> item in ports )
{
if( item.Value != null ) {
item.Value.CompleteAndSkip();
}
}
}
public static void WaitAll( Dictionary<string, PipelineConnector> ports )
{
if( ports == null ) {
return;
}
foreach( KeyValuePair<string, PipelineConnector> item in ports )
{
if( item.Value != null ) {
item.Value.WaitForStart();
}
}
}
[MethodImpl( MethodImplOptions.AggressiveInlining )]
public virtual void WaitForStart()
{
//Console.WriteLine($"Waiting \t{input.ContextName}\t->\t{output.ContextName}");
formerCompleteSemaphore.Wait();
}
~PipelineConnector()
{
formerCompleteSemaphore.Dispose();
}
#region pipline op
[MethodImpl( MethodImplOptions.AggressiveInlining )]
public void CompleteAndSkip()
{
//Console.WriteLine($"Complete\t{input.Name}\t->\t{output.Name}");
formerCompleteSemaphore.Release();
}
[MethodImpl( MethodImplOptions.AggressiveInlining )]
public void CompleteAndWriteData( IPipeLineConnectionPayload args )
{
this.args = args;
CompleteAndSkip();
}
#endregion
}
public abstract class IPipeLineConnectionPayload { }
}
StgSharp.Common/PipeLine/RouteSwitch.cs
-112
View File
@@ -1,112 +0,0 @@
//-----------------------------------------------------------------------
//-----------------------------------------------------------------------
// file="RouteSwitch.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.Linq;
using System.Reflection.Emit;
using System.Security.Cryptography.X509Certificates;
using System.Text;
namespace StgSharp.PipeLine
{
public class RouteSwitch : PipeLineNode
{
private protected Dictionary<string, (PipelineConnector pipeline, Func<bool> openingRule)> outputDictionary;
public RouteSwitch( string name, bool isNative, string inputPorts )
: base(
DefaultOperation, name, isNative, [inputPorts], [string.Empty] )
{
outputDictionary = new Dictionary<string, (PipelineConnector pipeline, Func<bool> openingRule)>(
);
}
public new int Level
{
get
{
if( _levelAvailable ) {
return _level;
}
_level = 0;
KeyValuePair<string, PipelineConnector> item = _inputInterfaces.First(
);
if( item.Value == null ) {
throw new InvalidOperationException();
}
if( _level <= item.Value.Level ) {
_level = item.Value.Level;
}
return _level;
}
}
public new void AddBefore(
PipeLineNode after,
string thisOutputPortName,
string afterInputPortName )
{
this.AppendNode( after, afterInputPortName, () => true );
}
public void AppendNode(
PipeLineNode node,
string portName,
Func<bool> openningRule )
{
if( node == null ) {
throw new ArgumentNullException( nameof( node ) );
}
string indexName = $"{node.Name}.{portName}";
PipelineConnector p = new PipelineConnector( this, node );
node.InputInterfaces[ portName ] = p;
outputDictionary.Add( indexName, (p, openningRule) );
}
public new void Run()
{
try {
PipelineConnector input = InputInterfaces.First().Value;
input.WaitForStart();
foreach( KeyValuePair<string, (PipelineConnector pipeline, Func<bool> openingRule)> item in outputDictionary ) {
if( item.Value.openingRule() ) {
item.Value.pipeline.CompleteAndWriteData( input.Args );
}
}
}
catch( Exception ) {
throw;
}
}
}
}
StgSharp.Common/PipeLine/SideNode.cs
-98
View File
@@ -1,98 +0,0 @@
//-----------------------------------------------------------------------
//-----------------------------------------------------------------------
// file="SideNode.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.Globalization;
using System.Linq;
using System.Security.Principal;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
namespace StgSharp.PipeLine
{
internal sealed class BeginningNode : PipelineNode
{
public const string OutName = "default";
private PipelineScheduler _bp;
public BeginningNode( PipelineScheduler bp )
: base(
DefaultOperation, new PipelineNodeStringLabel( "BeginningNode" ), true, ["default"],
[OutName] )
{
_bp = bp;
}
public PipelineNodeOutPort DefaultOut => OutputPorts[ OutName ];
public override void Run()
{
_bp.RunningStat.Wait();
PipelineNodeOutPort.SkipAll( _output );
}
internal void SetInputData( IEnumerable<(string, IPipeLineConnectionPayload)> data )
{
if( data is null ) {
return;
}
foreach( (string, IPipeLineConnectionPayload) item in data ) {
OutputPorts[ item.Item1 ].TransmitValue( item.Item2 );
}
}
}
internal sealed class EndingNode : PipelineNode
{
private PipelineScheduler _bp;
public EndingNode( PipelineScheduler bp )
: base(
DefaultOperation, name: new PipelineNodeStringLabel( "EndingNode" ), true, ["default"],
["default"] )
{
_bp = bp;
}
public override void Run()
{
PipelineNodeInPort.WaitAll( _input );
_bp.ResetIndex();
_bp.RunningStat.Release();
}
}
}