zhusenlin/serein-flow
1
0
Fork 0
mirror of https://gitee.com/langsisi_admin/serein-flow synced 2026-04-14 03:46:34 +08:00
Code Issues Packages Projects Releases Wiki Activity

重构了运行逻辑。上下文使用对象池封装，节点方法调用时间传递CancellationTokenSource用来中止任务

Browse Source
This commit is contained in:
fengjiayi
2025-03-20 22:54:10 +08:00
parent 2168c5ec66
commit 9941f24c5d
27 changed files with 830 additions and 621 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
Library/Api/IDynamicContext.cs
View File

@@ -26,12 +26,6 @@ namespace Serein.Library.Api
/// </summary>
RunState RunState { get; }
/// <summary>
/// 用来在当前流程上下文间传递数据
/// </summary>
//Dictionary<string, object> ContextShareData { get; }
object Tag { get; set; }
/// <summary>
/// 下一个要执行的节点类别
@@ -77,6 +71,11 @@ namespace Serein.Library.Api
/// <param name="flowData"></param>
void AddOrUpdate(string nodeGuid, object flowData);
/// <summary>
/// 重置流程状态（用于对象池回收）
/// </summary>
void Reset();
/// <summary>
/// 用以提前结束当前上下文流程的运行
/// </summary>

2
Library/Api/IFlowEnvironment.cs
View File

@@ -693,7 +693,7 @@ namespace Serein.Library.Api
/// <summary>
/// 全局触发器运行状态
/// </summary>
RunState FlipFlopState { get; set; }
//RunState FlipFlopState { get; set; }
/// <summary>
/// 表示当前环境

4
Library/Api/ISereinIoc.cs
View File

@@ -49,7 +49,7 @@ namespace Serein.Library.Api
/// <param name="key">登记使用的名称</param>
/// <param name="instance">实例对象</param>
/// <returns>是否注册成功</returns>
bool RegisterInstance(string key, object instance);
/// bool RegisterInstance(string key, object instance);
/// <summary>
/// 获取类型的实例。如果需要获取的类型以“接口-实现类”的方式注册，请使用接口的类型。
@@ -67,7 +67,7 @@ namespace Serein.Library.Api
/// <typeparam name="T"></typeparam>
/// <param name="key">登记实例时使用的Key</param>
/// <returns></returns>
T Get<T>(string key);
/// T Get<T>(string key);

16
Library/FlowNode/ContainerFlowEnvironment.cs
View File

@@ -295,10 +295,10 @@ namespace Serein.Library
{
return (T)sereinIOC.Get(typeof(T));
}
T ISereinIOC.Get<T>(string key)
{
return sereinIOC.Get<T>(key);
}
//T ISereinIOC.Get<T>(string key)
//{
// return sereinIOC.Get<T>(key);
//}
bool ISereinIOC.RegisterPersistennceInstance(string key, object instance)
@@ -311,10 +311,10 @@ namespace Serein.Library
return sereinIOC.RegisterPersistennceInstance(key, instance);
}
bool ISereinIOC.RegisterInstance(string key, object instance)
{
return sereinIOC.RegisterInstance(key, instance);
}
//bool ISereinIOC.RegisterInstance(string key, object instance)
//{
// return sereinIOC.RegisterInstance(key, instance);
//}
object ISereinIOC.Instantiate(Type type)

42
Library/FlowNode/DynamicContext.cs
View File

@@ -22,7 +22,7 @@ namespace Serein.Library
RunState = RunState.Running;
}
private readonly string _guid = global::System.Guid.NewGuid().ToString();
private string _guid = global::System.Guid.NewGuid().ToString();
string IDynamicContext.Guid => _guid;
/// <summary>
@@ -35,11 +35,6 @@ namespace Serein.Library
/// </summary>
public RunState RunState { get; set; } = RunState.NoStart;
/// <summary>
/// 用来在当前流程上下文间传递数据
/// </summary>
//public Dictionary<string, object> ContextShareData { get; } = new Dictionary<string, object>();
public object Tag { get; set; }
/// <summary>
/// 当前节点执行完成后，设置该属性，让运行环境判断接下来要执行哪个分支的节点。
@@ -133,6 +128,37 @@ namespace Serein.Library
return null;
}
/// <summary>
/// 重置
/// </summary>
public void Reset()
{
//foreach (var nodeObj in dictNodeFlowData.Values)
//{
// if (nodeObj is null)
// {
// }
// else
// {
// if (typeof(IDisposable).IsAssignableFrom(nodeObj?.GetType()) && nodeObj is IDisposable disposable)
// {
// disposable?.Dispose();
// }
// }
//}
//if (Tag != null && typeof(IDisposable).IsAssignableFrom(Tag?.GetType()) && Tag is IDisposable tagDisposable)
//{
// tagDisposable?.Dispose();
//}
this.dictNodeFlowData?.Clear();
ExceptionOfRuning = null;
NextOrientation = ConnectionInvokeType.None;
RunState = RunState.Running;
_guid = global::System.Guid.NewGuid().ToString();
}
/// <summary>
/// 结束当前流程上下文
/// </summary>
@@ -156,9 +182,11 @@ namespace Serein.Library
//{
// tagDisposable?.Dispose();
//}
this.Tag = null;
this.dictNodeFlowData?.Clear();
ExceptionOfRuning = null;
NextOrientation = ConnectionInvokeType.None;
RunState = RunState.Completion;
_guid = global::System.Guid.NewGuid().ToString();
}
private void Dispose(ref IDictionary<string, object> keyValuePairs)

24
Library/FlowNode/MethodDetails.cs
View File

@@ -45,8 +45,8 @@ namespace Serein.Library
/// <summary>
/// 作用实例（多个相同的节点将会共享同一个实例）
/// </summary>
[PropertyInfo]
private object _actingInstance;
// [PropertyInfo]
// private object _actingInstance;
/// <summary>
/// 方法名称
@@ -237,16 +237,16 @@ namespace Serein.Library
// this => 是元数据
var md = new MethodDetails( nodeModel) // 创建新节点时拷贝实例
{
AssemblyName = this.AssemblyName,
ActingInstance = this.ActingInstance,
ActingInstanceType = this.ActingInstanceType,
MethodDynamicType = this.MethodDynamicType,
MethodAnotherName = this.MethodAnotherName,
ReturnType = this.ReturnType,
MethodName = this.MethodName,
MethodLockName = this.MethodLockName,
IsProtectionParameter = this.IsProtectionParameter,
ParamsArgIndex = this.ParamsArgIndex,
AssemblyName = this.AssemblyName, // 拷贝
//ActingInstance = this.ActingInstance,
ActingInstanceType = this.ActingInstanceType, // 拷贝
MethodDynamicType = this.MethodDynamicType, // 拷贝
MethodAnotherName = this.MethodAnotherName, // 拷贝
ReturnType = this.ReturnType, // 拷贝
MethodName = this.MethodName, // 拷贝
MethodLockName = this.MethodLockName, // 拷贝
IsProtectionParameter = this.IsProtectionParameter, // 拷贝
ParamsArgIndex = this.ParamsArgIndex, // 拷贝
ParameterDetailss = this.ParameterDetailss?.Select(p => p?.CloneOfModel(nodeModel)).ToArray(), // 拷贝属于节点方法的新入参描述
};

93
Library/FlowNode/NodeModelBaseFunc.cs
View File

@@ -81,7 +81,7 @@ namespace Serein.Library
}
this.MethodDetails.ParameterDetailss = null;
this.MethodDetails.ActingInstance = null;
//this.MethodDetails.ActingInstance = null;
this.MethodDetails.NodeModel = null;
this.MethodDetails.ReturnType = null;
this.MethodDetails.AssemblyName = null;
@@ -249,43 +249,46 @@ namespace Serein.Library
/// <param name="context"></param>
/// <param name="flowCts"></param>
/// <returns></returns>
public static bool IsBradk(IDynamicContext context, CancellationTokenSource flowCts)
{
// 上下文不再执行
if (context.RunState == RunState.Completion)
{
return true;
}
//public static bool IsBradk(IDynamicContext context)
//{
// // 上下文不再执行
// if (context.RunState == RunState.Completion)
// {
// return true;
// }
// 不存在全局触发器时，流程运行状态被设置为完成，退出执行，用于打断无限循环分支。
if (flowCts is null && context.Env.FlowState == RunState.Completion)
{
return true;
}
// // 不存在全局触发器时，流程运行状态被设置为完成，退出执行，用于打断无限循环分支。
// if (flowCts is null && context.Env.FlowState == RunState.Completion)
// {
// return true;
// }
// 如果存在全局触发器，且触发器的执行任务已经被取消时，退出执行。
if (flowCts != null)
{
if (flowCts.IsCancellationRequested)
return true;
}
return false;
}
// // 如果存在全局触发器，且触发器的执行任务已经被取消时，退出执行。
// if (flowCts != null)
// {
// if (flowCts.IsCancellationRequested)
// return true;
// }
// return false;
//}
/// <summary>
/// 开始执行
/// </summary>
/// <param name="context"></param>
/// <param name="token">流程运行</param>
/// <returns></returns>
public async Task StartFlowAsync(IDynamicContext context)
public async Task StartFlowAsync(IDynamicContext context, CancellationToken token)
{
Stack<NodeModelBase> stack = new Stack<NodeModelBase>();
HashSet<NodeModelBase> processedNodes = new HashSet<NodeModelBase>(); // 用于记录已处理上游节点的节点
stack.Push(this);
var flowCts = context.Env.IOC.Get<CancellationTokenSource>(NodeStaticConfig.FlipFlopCtsName);
bool hasFlipflow = flowCts != null;
while (stack.Count > 0) // 循环中直到栈为空才会退出循环
while (context.RunState != RunState.Completion // 没有完成
&& token.IsCancellationRequested == false // 没有取消
&& stack.Count > 0) // 循环中直到栈为空才会退出循环
{
#if DEBUG
await Task.Delay(1);
#endif
@@ -299,15 +302,12 @@ namespace Serein.Library
object newFlowData;
try
{
newFlowData = await currentNode.ExecutingAsync(context, token);
if (IsBradk(context, flowCts)) break; // 退出执行
newFlowData = await currentNode.ExecutingAsync(context);
if (IsBradk(context, flowCts)) break; // 退出执行
if (context.NextOrientation == ConnectionInvokeType.None) // 没有手动设置时，进行自动设置
{
context.NextOrientation = ConnectionInvokeType.IsSucceed;
}
}
catch (Exception ex)
{
@@ -316,9 +316,7 @@ namespace Serein.Library
context.NextOrientation = ConnectionInvokeType.IsError;
context.ExceptionOfRuning = ex;
}
await RefreshFlowDataAndExpInterrupt(context, currentNode, newFlowData); // 执行当前节点后刷新数据
context.AddOrUpdate(currentNode.Guid, newFlowData); // 上下文中更新数据
#endregion
#region 执行完成
@@ -355,14 +353,10 @@ namespace Serein.Library
/// </summary>
/// <param name="context">流程上下文</param>
/// <returns>节点传回数据对象</returns>
public virtual async Task<object> ExecutingAsync(IDynamicContext context)
public virtual async Task<object> ExecutingAsync(IDynamicContext context, CancellationToken token)
{
#region 调试中断
if(context.NextOrientation == ConnectionInvokeType.IsError)
{
}
// 执行触发检查是否需要中断
if (DebugSetting.IsInterrupt)
{
@@ -370,8 +364,8 @@ namespace Serein.Library
await DebugSetting.GetInterruptTask.Invoke();
//await fit.WaitTriggerAsync(Guid); // 创建一个等待的中断任务
SereinEnv.WriteLine(InfoType.INFO, $"[{this.MethodDetails?.MethodName}]中断已取消，开始执行后继分支");
var flowCts = context.Env.IOC.Get<CancellationTokenSource>(NodeStaticConfig.FlipFlopCtsName);
if (IsBradk(context, flowCts)) return null; // 流程已终止，取消后续的执行
//var flowCts = context.Env.IOC.Get<CancellationTokenSource>(NodeStaticConfig.FlipFlopCtsName);
if (token.IsCancellationRequested) { return null; }
}
#endregion
@@ -385,17 +379,14 @@ namespace Serein.Library
{
throw new Exception($"节点{this.Guid}不存在对应委托");
}
if (md.ActingInstance is null)
var instance = Env.IOC.Get(md.ActingInstanceType);
if(instance == null)
{
md.ActingInstance = context.Env.IOC.Get(md.ActingInstanceType);
if (md.ActingInstance is null)
{
md.ActingInstance = context.Env.IOC.Instantiate(md.ActingInstanceType);
}
Env.IOC.Register(md.ActingInstanceType).Build();
instance = Env.IOC.Get(md.ActingInstanceType);
}
object[] args = await GetParametersAsync(context);
var result = await dd.InvokeAsync(md.ActingInstance, args);
object[] args = await GetParametersAsync(context, token);
var result = await dd.InvokeAsync(instance, args);
return result;
}
@@ -403,7 +394,7 @@ namespace Serein.Library
/// <summary>
/// 获取对应的参数数组
/// </summary>
public async Task<object[]> GetParametersAsync(IDynamicContext context)
public async Task<object[]> GetParametersAsync(IDynamicContext context, CancellationToken token)
{
if (MethodDetails.ParameterDetailss.Length == 0)
{
@@ -454,13 +445,13 @@ namespace Serein.Library
/// <summary>
/// 更新节点数据，并检查监视表达式是否生效
/// 检查监视表达式是否生效
/// </summary>
/// <param name="context">上下文</param>
/// <param name="nodeModel">节点Moel</param>
/// <param name="newData">新的数据</param>
/// <returns></returns>
public static async Task RefreshFlowDataAndExpInterrupt(IDynamicContext context, NodeModelBase nodeModel, object newData = null)
public static async Task CheckExpInterrupt(IDynamicContext context, NodeModelBase nodeModel, object newData = null)
{
string guid = nodeModel.Guid;
context.AddOrUpdate(guid, newData); // 上下文中更新数据

10
Library/NodeStaticConfig.cs
View File

@@ -9,16 +9,6 @@ namespace Serein.Library
{
public static class NodeStaticConfig
{
/// <summary>
/// 全局触发器CTS
/// </summary>
public const string FlipFlopCtsName = "$FlowFlipFlopCts";
/// <summary>
/// 流程运行CTS
/// </summary>
public const string FlowRungCtsName = "$FlowRungCtsName";
/// <summary>
/// 节点的命名空间
/// </summary>

160
Library/Utils/ObjectPool.cs Normal file
View File

@@ -0,0 +1,160 @@
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
namespace Serein.Library.Utils
{
/// <summary>
/// 具有预定义池大小限制的对象池模式的通用实现。其主要目的是将有限数量的经常使用的对象保留在池中，以便进一步回收。
///
/// 注:
/// 1)目标不是保留所有返回的对象。池不是用来存储的。如果池中没有空间，则会丢弃额外返回的对象。
///
/// 2)这意味着如果对象是从池中获得的，调用者将在相对较短的时间内返回它
/// 时间。长时间保持检出对象是可以的，但是会降低池的有用性。你只需要重新开始。
///
/// 不将对象返回给池并不会损害池的工作，但这是一种不好的做法。
/// 基本原理：如果没有重用对象的意图，就不要使用pool——只使用“new”
/// </summary>
public class ObjectPool<T> where T : class
{
[DebuggerDisplay("{Value,nq}")]
private struct Element
{
internal T Value;
}
// 不使用System。Func{T}，因为. net 2.0没有该类型。
public delegate T Factory();
// 池对象的存储。第一个项存储在专用字段中，因为我们希望能够满足来自它的大多数请求。
private T _firstItem;
private readonly Element[] _items;
// 工厂在池的生命周期内被存储。只有当池需要扩展时，我们才调用它。
// 与“new T（）”相比，Func为实现者提供了更多的灵活性，并且比“new T（）”更快。
private readonly Factory _factory;
public ObjectPool(Factory factory)
: this(factory, Environment.ProcessorCount * 2)
{ }
public ObjectPool(Factory factory, int size)
{
Debug.Assert(size >= 1);
_factory = factory;
_items = new Element[size - 1];
}
private T CreateInstance()
{
T inst = _factory();
return inst;
}
/// <summary>
/// 生成实例。
/// </summary>
/// <remarks>
/// 搜索策略是一种简单的线性探测，选择它是为了缓存友好。
/// 请注意，Free会尝试将回收的对象存储在靠近起点的地方，从而在统计上减少我们通常搜索的距离。
/// </remarks>
public T Allocate()
{
/*
* PERF：检查第一个元素。如果失败，AllocateSlow将查看剩余的元素。
* 注意，初始读是乐观地不同步的。
* 这是有意为之。只有了待使用对象，我们才会返回。
* 在最坏的情况下，我们可能会错过一些最近返回的对象。没什么大不了的。
*/
T inst = _firstItem;
if (inst == null || inst != Interlocked.CompareExchange(ref _firstItem, null, inst))
{
inst = AllocateSlow();
}
return inst;
}
private T AllocateSlow()
{
Element[] items = _items;
for (int i = 0; i < items.Length; i++)
{
// 注意，初始读是乐观地不同步的。这是有意为之。只有有了候选人，我们才会联系。在最坏的情况下，我们可能会错过一些最近返回的对象。没什么大不了的。
T inst = items[i].Value;
if (inst != null)
{
if (inst == Interlocked.CompareExchange(ref items[i].Value, null, inst))
{
return inst;
}
}
}
return CreateInstance();
}
/// <summary>
///返回对象到池。
/// </summary>
/// <remarks>
/// 搜索策略是一种简单的线性探测，选择它是因为它具有缓存友好性。
/// 请注意Free会尝试将回收的对象存储在靠近起点的地方，从而在统计上减少我们通常在Allocate中搜索的距离。
/// </remarks>
public void Free(T obj)
{
Validate(obj);
if (_firstItem == null)
{
// 这里故意不使用联锁。在最坏的情况下，两个对象可能存储在同一个槽中。这是不太可能发生的，只意味着其中一个对象将被收集。
_firstItem = obj;
}
else
{
FreeSlow(obj);
}
}
private void FreeSlow(T obj)
{
Element[] items = _items;
for (int i = 0; i < items.Length; i++)
{
if (items[i].Value == null)
{
// 这里故意不使用联锁。在最坏的情况下，两个对象可能存储在同一个槽中。这是不太可能发生的，只意味着其中一个对象将被收集。
items[i].Value = obj;
break;
}
}
}
[Conditional("DEBUG")]
private void Validate(object obj)
{
Debug.Assert(obj != null, "freeing null?");
Debug.Assert(_firstItem != obj, "freeing twice?");
var items = _items;
for (int i = 0; i < items.Length; i++)
{
var value = items[i].Value;
if (value == null)
{
return;
}
Debug.Assert(value != obj, "freeing twice?");
}
}
}
}