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

daily update on 20240620 by nitload youmu

Browse Source
This commit is contained in:
Nitload
2024-06-20 07:47:00 +08:00
parent 7fb220c523
commit b2cb0cbddf
39 changed files with 1252 additions and 550 deletions
Download Patch File Download Diff File Expand all files Collapse all files
源代码/StgSharp/Logic/Blueprint/Blueprint.cs
+72 -145
View File
@@ -45,33 +45,60 @@ using System.Xml.Serialization;
namespace StgSharp.Logic
{
/// <summary>
/// A sets of complex logic designed for running in multi threads environment.
/// Warning: do not run a <see cref="Blueprint"/> instance in multi threads more than once.
///
/// </summary>
public partial class Blueprint : IConvertableToBlueprintNode
{
private int runningFlag;
private BeginningNode beginNode;
private Dictionary<string, BlueprintNode> allNode;
private EndingNode endNode;
private Func<bool> terminate;
private int currentNodeIndex;
private List<Task> allThread;
internal int currentGlobalNodeIndex;
internal int currentNativeNodeIndex;
private List<BlueprintNode> nativeNodeList;
private List<BlueprintNode> nodeList;
private List<BlueprintNode> globalNodeList;
public Blueprint()
{
allNode = new Dictionary<string, BlueprintNode>();
allThread = new List<Task>();
nodeList = new List<BlueprintNode>();
globalNodeList = new List<BlueprintNode>();
nativeNodeList = new List<BlueprintNode>();
terminate = () => true;
beginNode = new BeginningNode();
endNode = new EndingNode();
//terminate = () => true;
beginNode = new BeginningNode(this);
endNode = new EndingNode(this);
}
public BlueprintNode BeginLayer => beginNode;
public BlueprintNode EndLayer => endNode;
public bool IsRunning => allThread.Any(t => t.Status == TaskStatus.Running);
public bool IsRunning
{
get => runningFlag == 1;
internal set => Interlocked.Exchange(ref runningFlag ,(value ? 1 : 0));
}
/**/
internal Func<bool> ShouldClose
{
get => terminate;
set => terminate = value;
}
/**/
public void AddNode(BlueprintNode node, bool isNative)
{
_ = node ?? throw new ArgumentNullException(nameof(node));
node.IsNative = isNative;
allNode.Add(node.Name, node);
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public (string name, BlueprintPipelineArgs arg)[] GetOutput()
@@ -79,115 +106,27 @@ namespace StgSharp.Logic
return endNode.OutputData;
}
public void Init(int threadCount)
public void Init()
{
if (threadCount > nodeList.Count)
foreach (KeyValuePair<string, BlueprintNode> item in allNode)
{
InternalIO.InternalWriteLog(
"Thread setting error: amount of threads of this blueprint is larger than amount of nodes.", LogType.Warning);
threadCount = nodeList.Count;
}
if (threadCount!=0)
{
foreach (var item in allNode)
if (item.Value.IsNative)
{
if (item.Value.IsNative)
{
nativeNodeList.Add(item.Value);
}
else
{
nodeList.Add(item.Value);
}
nativeNodeList.Add(item.Value);
}
nativeNodeList.Sort();
PrepareTasks(threadCount);
}
else
{
foreach (var item in allNode)
else
{
nodeList.Add(item.Value);
globalNodeList.Add(item.Value);
}
}
nodeList.Sort();
nativeNodeList.Sort();
globalNodeList.Sort();
}
public void PrepareTasks(int count)
public void ExecuteMain(in Dictionary<string, BlueprintPipeline> input,
in Dictionary<string, BlueprintPipeline> output)
{
for (int i = 0; i < count; i++)
{
allThread.Add(new Task(IndividualTask));
}
}
private void IndividualTask()
{
do
{
BlueprintNode node;
while (RequestNextNode(out node))
{
node.Run();
}
} while (!terminate());
}
/// <summary>
/// Run all nodes repeatedly until it is instructed to terminate.
/// </summary>
/// <param name="terminateInstructor">Instructor to terminate all tasks. Returns <see langword="true"/> if tasks should terminate</param>
public void RunRepeatedly(Func<bool> terminateInstructor)
{
if (allThread.Count != 0)
{
MultiThreadRun(terminateInstructor);
return;
}
SingleThreadRun(terminateInstructor);
}
private void MultiThreadRun(Func<bool> terminateInstructor)
{
if (terminate != terminateInstructor)
{
terminate = terminateInstructor;
if (allThread.Count != 0)
{
foreach (var t in allThread)
{
t.Start();
}
}
}
beginNode.Run();
foreach (var node in nativeNodeList)
{
node.Run();
}
EndLayer.Run();
}
private void SingleThreadRun(Func<bool> terminateInstructor)
{
beginNode.Run();
foreach (var item in nodeList)
{
item.Run();
}
EndLayer.Run();
}
public void AddNativeNode(BlueprintNode node)
{
node.IsNative = true;
allNode.Add(node.Name, node);
}
public void AddAsyncNode(BlueprintNode node)
{
node.IsNative = false;
allNode.Add(node.Name, node);
BlueprintRunner.Run(this, () => true);
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
@@ -196,59 +135,47 @@ namespace StgSharp.Logic
beginNode.InputData = parameters;
}
private bool RequestNextNode(out BlueprintNode node)
internal bool RequestNexGlobalNode(out BlueprintNode node)
{
if (currentNodeIndex < nodeList.Count)
if (currentGlobalNodeIndex < globalNodeList.Count)
{
BlueprintNode ret = nodeList[currentNodeIndex];
Interlocked.Increment(ref currentNodeIndex);
node = ret;
node = globalNodeList[currentGlobalNodeIndex];
Interlocked.Increment(ref currentGlobalNodeIndex);
return true;
}
node = default!;
return false;
}
internal bool RequestNextNativeNode(out BlueprintNode node)
{
if (currentNativeNodeIndex < nativeNodeList.Count)
{
node = nativeNodeList[currentNativeNodeIndex];
Interlocked.Increment(ref currentNativeNodeIndex);
return true;
}
Interlocked.Exchange(ref currentNodeIndex, 0);
node = default!;
return false;
}
string[] IConvertableToBlueprintNode.InputInterfacesName
{
get
{
return beginNode.OutputInterfaces.Select(
p => p.Key).ToArray();
}
}
BlueprintNodeAction IConvertableToBlueprintNode.MainExecution
{
get
{
return (in Dictionary<string, BlueprintPipeline> input,
in Dictionary<string, BlueprintPipeline> output)
=>
{
beginNode.Run();
foreach (var node in nativeNodeList)
{
node.Run();
}
endNode.Run();
};
}
get { return beginNode.OutputInterfaces.Select(p => p.Key).ToArray(); }
}
string[] IConvertableToBlueprintNode.OutputInterfacesName
{
get
{
return beginNode.OutputInterfaces.Select(
p => p.Key).ToArray();
}
get { return beginNode.OutputInterfaces.Select(p => p.Key).ToArray(); }
}
}
internal class NodeClearException : Exception
{
public void Terminate()
{
Interlocked.Exchange(ref currentGlobalNodeIndex, int.MaxValue);
Interlocked.Exchange(ref currentNativeNodeIndex, int.MaxValue);
}
}
}
源代码/StgSharp/Logic/Blueprint/BlueprintNode.cs
+69 -76
View File
@@ -35,6 +35,7 @@ using StgSharp.Entities;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Runtime.CompilerServices;
using System.Runtime.Serialization;
using System.Text;
using System.Threading;
@@ -42,11 +43,6 @@ using System.Threading.Tasks;
namespace StgSharp.Logic
{
public delegate void BlueprintNodeAction(
in Dictionary<string, BlueprintPipeline> inpput,
in Dictionary<string, BlueprintPipeline> output);
/// <summary>
///
@@ -59,86 +55,69 @@ namespace StgSharp.Logic
public class BlueprintNode : IComparable<BlueprintNode>
#pragma warning restore CA1036
{
private BlueprintNodeAction mainOperation;
private protected bool levelAvailable, nativeRequested;
private protected int _level;
private string name;
private protected Dictionary<string, BlueprintPipeline> _inputInterfaces;
private protected Dictionary<string, BlueprintPipeline> _outputInterfaces;
private protected int _level;
private IConvertableToBlueprintNode mainOperation;
private string name;
internal BlueprintNode(string name,
BlueprintNodeAction operation, bool isNative,
string[] inputPorts,
string[] outputPorts)
IConvertableToBlueprintNode node, bool isNative)
{
this.levelAvailable = false;
this.name = name;
nativeRequested = isNative;
mainOperation = operation;
mainOperation = node;
_inputInterfaces = new Dictionary<string, BlueprintPipeline>();
_outputInterfaces = new Dictionary<string, BlueprintPipeline>();
if (inputPorts.Length == 0 )
foreach (string portName in node.InputInterfacesName)
{
throw new ArgumentNullException(nameof(inputPorts));
_inputInterfaces.Add(portName, null!);
}
if (outputPorts.Length == 0)
foreach (string portName in node.OutputInterfacesName)
{
throw new ArgumentNullException(nameof(outputPorts));
}
foreach (var portName in inputPorts)
{
_inputInterfaces.Add(portName, null);
}
foreach (var portName in outputPorts)
{
_outputInterfaces.Add(portName, null);
_outputInterfaces.Add(portName, null!);
}
}
public BlueprintNode(string name, IConvertableToBlueprintNode convertable)
{
this.levelAvailable = false;
this.name = name;
_inputInterfaces = new Dictionary<string, BlueprintPipeline>();
_outputInterfaces = new Dictionary<string, BlueprintPipeline>();
mainOperation = convertable.MainExecution;
foreach (var soket in convertable.InputInterfacesName)
{
_inputInterfaces.Add(soket, null);
}
foreach (var soket in convertable.OutputInterfacesName)
{
_outputInterfaces.Add(soket, null);
}
}
public BlueprintNode(string name,
public BlueprintNode(string name, bool isNative,
string[] inputPorts,
string[] outputPorts
)
{
this.levelAvailable = false;
if (inputPorts == null || inputPorts.Length == 0)
{
throw new ArgumentNullException(nameof(inputPorts));
}
if (outputPorts == null || outputPorts.Length == 0)
{
throw new ArgumentNullException(nameof(outputPorts));
}
levelAvailable = false;
this.name = name;
IsNative = isNative;
_inputInterfaces = new Dictionary<string, BlueprintPipeline>();
_outputInterfaces = new Dictionary<string, BlueprintPipeline>();
if (inputPorts.Length * outputPorts.Length == 0)
foreach (string portName in inputPorts)
{
throw new Exception();
_inputInterfaces.Add(portName, null!);
}
foreach (var portName in inputPorts)
foreach (string portName in outputPorts)
{
_inputInterfaces.Add(portName, null);
}
foreach (var portName in outputPorts)
{
_outputInterfaces.Add(portName, null);
_outputInterfaces.Add(portName, null!);
}
}
internal IConvertableToBlueprintNode Value
{
get => mainOperation;
}
public bool IsNative
{
get => nativeRequested;
@@ -154,7 +133,7 @@ namespace StgSharp.Logic
return _level;
}
_level = 0;
foreach (var item in _inputInterfaces)
foreach (KeyValuePair<string, BlueprintPipeline> item in _inputInterfaces)
{
if (item.Value == null)
{
@@ -213,21 +192,6 @@ namespace StgSharp.Logic
SetCertainOutputPort(thisOutputPortName, pipeline);
}
public void Run()
{
try
{
//Console.WriteLine(name);
BlueprintPipeline.WaitAll(_inputInterfaces);
mainOperation(in _inputInterfaces, in _outputInterfaces);
}
catch (Exception)
{
throw;
}
}
public int CompareTo(BlueprintNode other)
{
#pragma warning disable CA1062
@@ -235,12 +199,30 @@ namespace StgSharp.Logic
#pragma warning restore CA1062
}
public virtual void Run()
{
try
{
//Console.WriteLine(name);
BlueprintPipeline.WaitAll(_inputInterfaces);
if (mainOperation != null)
{
mainOperation.ExecuteMain(in _inputInterfaces, in _outputInterfaces);
}
}
catch (Exception)
{
throw;
}
}
#region port operation
public BlueprintPipeline GetCertainInputPort(string name)
{
BlueprintPipeline ret;
if (!_inputInterfaces.TryGetValue(name,out ret))
if (!_inputInterfaces.TryGetValue(name, out ret))
{
throw new Exception($"Cannot find the interface named {name} in node {Name}.");
}
@@ -283,15 +265,29 @@ namespace StgSharp.Logic
public interface IConvertableToBlueprintNode
{
#pragma warning disable CA1819
public string[] InputInterfacesName { get; }
public BlueprintNodeAction MainExecution { get; }
public string[] OutputInterfacesName { get; }
#pragma warning restore CA1819
#pragma warning disable CA1819
public string[] OutputInterfacesName { get; }
#pragma warning restore CA1819
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void ExecuteMain(
in Dictionary<string, BlueprintPipeline> input,
in Dictionary<string, BlueprintPipeline> output);
}
public class BlueprintNodeRunException : Exception
{
protected BlueprintNodeRunException(SerializationInfo info, StreamingContext context) : base(info, context)
{
}
public BlueprintNodeRunException()
{
}
@@ -308,9 +304,6 @@ namespace StgSharp.Logic
{
}
protected BlueprintNodeRunException(SerializationInfo info, StreamingContext context) : base(info, context)
{
}
}
}
源代码/StgSharp/Logic/Blueprint/BlueprintRunner.cs
+172
View File
@@ -0,0 +1,172 @@
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.Logic
{
public static class BlueprintRunner
{
private static ConcurrentStack<Blueprint> blueprintInWaiting = new ConcurrentStack<Blueprint>();
private static NodeRunner[] allAvailableTask = new NodeRunner[1];
private static ReaderWriterLockSlim bpLock = new ReaderWriterLockSlim();
private static CancellationTokenSource cts = new CancellationTokenSource();
public static void Run(Blueprint blueprint, Func<bool> terminateInstructor)
{
//Console.WriteLine();
if (blueprint == null) return;
if (terminateInstructor == null)
{
throw new ArgumentNullException(nameof(terminateInstructor));
}
if (blueprint.IsRunning)
{
throw new InvalidOperationException( $"Attempt to run a blueprint twice." );
}
blueprint.ShouldClose = terminateInstructor;
if (IsInternalCalling)
{
if(InterruptAndRunNewBp(blueprint))
do { } while(InterruptAndRunNewBp(NodeRunner.CurrentBlueprint));
}
else
{
if (!bpLock.IsWriteLockHeld)
{
bpLock.EnterWriteLock();
}
if (InterruptAndRunNewBp(blueprint))
do { } while (InterruptAndRunNewBp(NodeRunner.CurrentBlueprint));
bpLock.ExitWriteLock();
}
}
internal static bool IsInternalCalling
{
get
{
int? id = Task.CurrentId;
if (id is null) return false;
foreach (var item in allAvailableTask)
{
if (item.Id == id)
{
return true;
}
}
return false;
}
}
internal static bool InterruptAndRunNewBp(Blueprint newBp)
{
Debug.Assert(newBp != null);
if (newBp != NodeRunner.CurrentBlueprint && NodeRunner.CurrentBlueprint != null)
{
blueprintInWaiting.Push(NodeRunner.CurrentBlueprint);
}
Interlocked.Exchange(ref NodeRunner.CurrentBlueprint!, newBp);
if (!NodeRunner.CurrentBlueprint.IsRunning)
{
(NodeRunner.CurrentBlueprint.BeginLayer as BeginningNode)!.Run();
}
for (int i = 0; i < allAvailableTask.Length; i++)
{
var item = allAvailableTask[i];
if (item is null || item.IsCompleted)
{
allAvailableTask[i] = NodeRunner.Run();
}
}
BlueprintNode node;
while (!cts.Token.IsCancellationRequested
&& NodeRunner.CurrentBlueprint.RequestNextNativeNode(out node))
{
node.Run();
if (!NodeRunner.CurrentBlueprint.IsRunning)
{
break;
}
}
(NodeRunner.CurrentBlueprint.EndLayer as EndingNode)!.Run();
return blueprintInWaiting.TryPeek(out NodeRunner.CurrentBlueprint);
}
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();
NodeRunner.CurrentBlueprint.Terminate();
foreach (var task in allAvailableTask)
{
task.Cts.Cancel();
}
foreach (var bp in blueprintInWaiting)
{
bp.Terminate();
}
blueprintInWaiting.Clear();
}
}
internal class NodeRunner: CancellableTask
{
private static Blueprint _current;
internal static ref Blueprint CurrentBlueprint
{
get => ref _current;
}
internal NodeRunner() : base()
{
_startup = InternalRun;
}
internal static NodeRunner Run()
{
var ret = new NodeRunner() { _startup = InternalRun};
ret.taskInternal = Task.Run(ret.InternalAction);
return ret;
}
private static void InternalRun(CancellationToken token)
{
//Console.WriteLine("Task Begin");
BlueprintNode item;
while (CurrentBlueprint != null && CurrentBlueprint.IsRunning && !token.IsCancellationRequested
&& CurrentBlueprint.RequestNexGlobalNode(out item))
{
//Console.WriteLine($"Get node of {item.Name}, going to run.");
item.Run();
}
//Console.WriteLine("Task Completed");
}
}//------------------------------------- End of Class ------------------------------------------
}
源代码/StgSharp/Logic/Blueprint/PipeLine.cs
+5 -3
View File
@@ -32,6 +32,7 @@ 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;
@@ -67,17 +68,18 @@ namespace StgSharp.Logic
set => args = value;
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void CompleteAndSkip()
{
//Console.WriteLine($"Complete\t{former.ContextName}\t->\t{after.ContextName}");
//Console.WriteLine($"Complete\t{former.Name}\t->\t{after.Name}");
formerCompleteSemaphore.Release();
}
[MethodImpl(MethodImplOptions.AggressiveInlining)]
public void CompleteAndWriteData(BlueprintPipelineArgs args)
{
this.args = args;
//Console.WriteLine($"Complete\t{former.ContextName}\t->\t{after.ContextName}");
formerCompleteSemaphore.Release();
CompleteAndSkip();
}
public static void SkipAll(Dictionary<string, BlueprintPipeline> ports)
源代码/StgSharp/Logic/Blueprint/RouteSwitch.cs
+2 -2
View File
@@ -41,8 +41,8 @@ namespace StgSharp.Logic
{
private protected Dictionary<string, (BlueprintPipeline pipeline, Func<bool> openingRule)> outputDictionary;
public RouteSwitch(string name, string inputPorts) :
base(name, [inputPorts], [string.Empty])
public RouteSwitch(string name, bool isNative, string inputPorts) :
base(name, isNative, [inputPorts], [string.Empty])
{
outputDictionary = new Dictionary<string, (BlueprintPipeline pipeline, Func<bool> openingRule)>();
}
源代码/StgSharp/Logic/Blueprint/SideNode.cs
+19 -14
View File
@@ -5,16 +5,20 @@ using System.Collections.Generic;
using System.Globalization;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
namespace StgSharp.Logic
{
internal sealed class BeginningNode : BlueprintNode
{
public BeginningNode() : base(name: "BeginningNode", null,
true, ["default"], ["default"]
private Blueprint _bp;
public BeginningNode( Blueprint bp ) : base(name: "BeginningNode",true, ["default"], ["default"]
)
{ }
{
_bp = bp;
}
internal (string name, BlueprintPipelineArgs arg)[] InputData
{
@@ -40,24 +44,22 @@ namespace StgSharp.Logic
_outputInterfaces.Add(name, pipeline);
}
public new void Run()
public override void Run()
{
BlueprintPipeline.SkipAll(_outputInterfaces);
_bp.IsRunning = true;
}
}
internal sealed class EndingNode : BlueprintNode
{
public EndingNode() : base(name: "EndingNode",
operation: (
in Dictionary<string, BlueprintPipeline> input,
in Dictionary<string, BlueprintPipeline> output
) =>
{
},
true, ["default"], ["default"]
private Blueprint _bp;
public EndingNode(Blueprint bp) : base(name: "EndingNode",true, ["default"], ["default"]
)
{ }
{
_bp = bp;
}
internal (string name, BlueprintPipelineArgs arg)[] OutputData
{
@@ -78,9 +80,12 @@ namespace StgSharp.Logic
_inputInterfaces.Add(name, pipeline);
}
public new void Run()
public override void Run()
{
BlueprintPipeline.WaitAll(_inputInterfaces);
Interlocked.Exchange(ref _bp.currentGlobalNodeIndex, 0);
Interlocked.Exchange(ref _bp.currentNativeNodeIndex, 0);
_bp.IsRunning = false;
}
}
}