如何使处理任务的worker实例和执行RunDag的实例相同，即本机执行任务 about fastflow

[ShiningRush]

目前尚不支持这样的特性，这里有两个解法：

1. 在fastflow层支持 LocalSchedule 特性——任务会指定调度到接收请求的这个节点上
2. 是否可以把创建管道的操作移动到工作流当中，这样在执行工作时即创建了管道

[lxn3642]

感谢回复，有这样一个场景，外部的api与工作流进行交互，如工作流运行到了b节点，运行过程中在进行一个等待操作、比如等待600s，这个等待过程中可以进行跳过等待，需要一个外部的api来触发这个跳过行为，当前是通过管道来实现的，在单副本情况下可以正常运行，但是如果是多副本的情况下，接到api请求的副本可能和实际执行工作流的副本不一致，此时就会交互失败，看上去解法1和2都无法解决这个问题，对此还有什么别的解法吗？

[ShiningRush]

解法1应该可以解决你的诉求？因为如果实现这个特性的话，接收请求的副本即执行工作流的副本
如果对于你这个情况，也可以通过共享存储来解决，比如在数据库写入一个flag，然后等待的这个节点不断轮询数据库，如果flag设置后就跳过后续工作

[lxn3642]

感谢回复~
共享存储的方式有考虑过，等待的节点不断轮询数据库的方式在大量工作流场景下对数据库压力较大，不是一个很完美的手段。解法一是一种本地优先调度的能力，在一些特殊场景下是需要的，对于本次需求，想了一下，框架层面缺乏控制流相关接口，是否可以支持，比如：
框架提供(一些伪代码用来说明相关逻辑)：
1）给task发送消息的接口(暴露给用户)
func CallTask(dagInsID, taskID string, data interface{}) error { // 1. 根据daginsid(taskid) 找到执行该任务的worker // 2. 将 taskid, data 发送给该 worker (实际上由2中的monitoring接收到数据) }
2) worker 处理接收到的(不暴露给用户)
func monitoring(ctx context.Context, req ghttp.Request) ghttp.Response { // 1. 解析接收到的taskid, data数据 // 2. 通过 taskid 找到目标 task // 3. 将data数据发送给与 taskid 通信的管道 ( 这个管道只是一个思路，在服务启动时初始化一个map[string]chain，task等待与外界通信时创建一个chain，并注册在这个map中，用于monitor与其通信) // 4. 返回处理结果 }
3) task 与外界通信 （用户使用）
func (i *Init) Run(ctx run.ExecuteContext, params any) error { // 1. 使用框架提供的方法创建并注册一个管道 c c := ctx.NewChain() // 2. 监听 c 的输出，做不同的业务逻辑 }
用户接口使用1）中的CallTask 方法将数据传递给对应的 task，注意: 使用 CallTask() 的副本和对应执行task的副本可能不是一个，这里需要框架实现调度~

不知道我是否有描述清楚需求和设想的解决方案，辛苦再帮忙看下吧

[ShiningRush]

嗯，你的方案我理解了，其本质就是提供一个渠道给外部，让其可以往正在执行的工作传输数据，技术上是没有问题，就是落地成本会更高，因为工作流执行是分散在各个节点的，而接收请求的节点可以是任意一个，所以这个方案还需要约定一些数据协议，使用pipe可能并不一定合适，因为pipe需要双向响应，反而是事件 or 通知 这样的单向传输会更好一些。
不过不管哪种方法，实现起来成本都不低，对比共享存储的方式其实ROI太低了。
你对共享存储方案的顾虑在于性能，不过共享存储的手段基于引入一些高性能存储来做，比如redis，这样几wqps都不是问题，即便只使用mongo这样的事务db，通过分片水平扩容也能随便支持几万的qps，对于大多数场景都足够了。
技术方案都是在做取舍，虽然在框架做可复用性更高，但是成本也更高。你这边可以做个判断，如果你有时间来贡献这个feature的话，可以考虑在框架上实现，不过在开源项目上的协作会比集中办公效率低一些，你的项目能否接受这样的时间跨度也得慎重考虑。