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

Normalize all existing files to LF

Browse Source
This commit is contained in:
Youmu Liu
2026-03-06 18:31:11 +08:00
parent 0a0af53af1
commit 2149bbeede
323 changed files with 113191 additions and 113191 deletions
Download Patch File Download Diff File Expand all files Collapse all files
StgSharp.Common/PipeLine/CrossFramePipelineNode.cs
+91 -91
View File
@@ -1,91 +1,91 @@
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
// file="CrossFramePipelineNode"
// Project: StgSharp
// AuthorGroup: Nitload
// Copyright (c) Nitload. All rights reserved.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in all
// copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
//
// -----------------------------------------------------------------------
// -----------------------------------------------------------------------
using StgSharp.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);
}
}
}
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
// file="CrossFramePipelineNode"
// Project: StgSharp
// AuthorGroup: Nitload
// Copyright (c) Nitload. All rights reserved.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in all
// copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
//
// -----------------------------------------------------------------------
// -----------------------------------------------------------------------
using StgSharp.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/IConvertableToPipelineNode.cs
+82 -82
View File
@@ -1,82 +1,82 @@
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
// file="IConvertableToPipelineNode"
// 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.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);
}
}
}
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
// file="IConvertableToPipelineNode"
// 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.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/PipeLineRunner.cs
+242 -242
View File
@@ -1,242 +1,242 @@
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
// file="PipeLineRunner"
// Project: StgSharp
// AuthorGroup: Nitload
// Copyright (c) Nitload. All rights reserved.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in all
// copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
//
// -----------------------------------------------------------------------
// -----------------------------------------------------------------------
using StgSharp.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 ------------------------------------------
}
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
// file="PipeLineRunner"
// Project: StgSharp
// AuthorGroup: Nitload
// Copyright (c) Nitload. All rights reserved.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in all
// copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE.
//
// -----------------------------------------------------------------------
// -----------------------------------------------------------------------
using StgSharp.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
+114 -114
View File
@@ -1,114 +1,114 @@
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
// file="PipelineMarshal"
// 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;
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();
}
}
}
}
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
// file="PipelineMarshal"
// 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;
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 -266
View File
@@ -1,266 +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) { }
}
}
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
// 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
+80 -80
View File
@@ -1,80 +1,80 @@
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
// file="PipelineNodeLabel"
// 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.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;
}
}
}
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
// file="PipelineNodeLabel"
// 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.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
+187 -187
View File
@@ -1,187 +1,187 @@
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
// file="PipelineNodePort"
// 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.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();
}
}
}
}
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
// file="PipelineNodePort"
// 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.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.cs
+295 -295
View File
@@ -1,295 +1,295 @@
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
// file="PipelineScheduler"
// 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.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();
}
}
}
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
// file="PipelineScheduler"
// 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.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
+123 -123
View File
@@ -1,123 +1,123 @@
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
// file="PortArgs"
// 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.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 { }
}
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
// file="PortArgs"
// 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.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/SideNode.cs
+92 -92
View File
@@ -1,93 +1,93 @@
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
// file="SideNode"
// 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.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();
}
}
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
// file="SideNode"
// 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.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();
}
}
}