diff --git a/go.mod b/go.mod index 8207adb..7d1acf5 100644 --- a/go.mod +++ b/go.mod @@ -61,6 +61,7 @@ require ( go.uber.org/zap v1.24.0 // indirect golang.org/x/crypto v0.6.0 // indirect golang.org/x/net v0.8.0 // indirect + golang.org/x/sync v0.1.0 golang.org/x/sys v0.6.0 // indirect golang.org/x/text v0.8.0 // indirect google.golang.org/genproto v0.0.0-20230403163135-c38d8f061ccd // indirect diff --git a/go.sum b/go.sum index 81f0644..68dc7ef 100644 --- a/go.sum +++ b/go.sum @@ -147,6 +147,8 @@ golang.org/x/net v0.8.0/go.mod h1:QVkue5JL9kW//ek3r6jTKnTFis1tRmNAW2P1shuFdJc= golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.1.0 h1:wsuoTGHzEhffawBOhz5CYhcrV4IdKZbEyZjBMuTp12o= +golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= diff --git a/pkgs/ioswitch/dag/graph.go b/pkgs/ioswitch/dag/graph.go new file mode 100644 index 0000000..995ebf2 --- /dev/null +++ b/pkgs/ioswitch/dag/graph.go @@ -0,0 +1,74 @@ +package dag + +import ( + "gitlink.org.cn/cloudream/common/utils/lo2" +) + +type Graph struct { + Nodes []*Node + isWalking bool + nextVarID int +} + +func NewGraph() *Graph { + return &Graph{} +} + +func (g *Graph) NewNode(typ NodeType, props any) *Node { + n := &Node{ + Type: typ, + Props: props, + Graph: g, + } + typ.InitNode(n) + g.Nodes = append(g.Nodes, n) + return n +} + +func (g *Graph) RemoveNode(node *Node) { + for i, n := range g.Nodes { + if n == node { + if g.isWalking { + g.Nodes[i] = nil + } else { + g.Nodes = lo2.RemoveAt(g.Nodes, i) + } + break + } + } +} + +func (g *Graph) Walk(cb func(node *Node) bool) { + g.isWalking = true + for i := 0; i < len(g.Nodes); i++ { + if g.Nodes[i] == nil { + continue + } + + if !cb(g.Nodes[i]) { + break + } + } + g.isWalking = false + + g.Nodes = lo2.RemoveAllDefault(g.Nodes) +} + +func (g *Graph) genVarID() int { + g.nextVarID++ + return g.nextVarID +} + +func NewNode[N NodeType](graph *Graph, typ N, props any) (*Node, N) { + return graph.NewNode(typ, props), typ +} + +func WalkOnlyType[N NodeType](g *Graph, cb func(node *Node, typ N) bool) { + g.Walk(func(node *Node) bool { + typ, ok := node.Type.(N) + if ok { + return cb(node, typ) + } + return true + }) +} diff --git a/pkgs/ioswitch/dag/node.go b/pkgs/ioswitch/dag/node.go new file mode 100644 index 0000000..6298b46 --- /dev/null +++ b/pkgs/ioswitch/dag/node.go @@ -0,0 +1,68 @@ +package dag + +import ( + "fmt" + + "gitlink.org.cn/cloudream/common/pkgs/ioswitch/exec" +) + +type NodeType interface { + InitNode(node *Node) + String(node *Node) string + GenerateOp(node *Node) (exec.Op, error) +} + +type NodeEnvType string + +const ( + EnvUnknown NodeEnvType = "" + EnvDriver NodeEnvType = "Driver" + EnvWorker NodeEnvType = "Worker" +) + +type NodeEnv struct { + Type NodeEnvType + Worker exec.WorkerInfo +} + +func (e *NodeEnv) ToEnvUnknown() { + e.Type = EnvUnknown + e.Worker = nil +} + +func (e *NodeEnv) ToEnvDriver() { + e.Type = EnvDriver + e.Worker = nil +} + +func (e *NodeEnv) ToEnvWorker(worker exec.WorkerInfo) { + e.Type = EnvWorker + e.Worker = worker +} + +func (e *NodeEnv) Equals(other NodeEnv) bool { + if e.Type != other.Type { + return false + } + + if e.Type != EnvWorker { + return true + } + + return e.Worker.Equals(other.Worker) +} + +type Node struct { + Type NodeType + Env NodeEnv + Props any + InputStreams []*StreamVar + OutputStreams []*StreamVar + InputValues []*ValueVar + OutputValues []*ValueVar + Graph *Graph +} + +func (n *Node) String() string { + return fmt.Sprintf("%v", n.Type.String(n)) +} diff --git a/pkgs/ioswitch/dag/utils.go b/pkgs/ioswitch/dag/utils.go new file mode 100644 index 0000000..cdee0ae --- /dev/null +++ b/pkgs/ioswitch/dag/utils.go @@ -0,0 +1,13 @@ +package dag + +func NProps[T any](n *Node) T { + return n.Props.(T) +} + +func SProps[T any](str *StreamVar) T { + return str.Props.(T) +} + +func VProps[T any](v *ValueVar) T { + return v.Props.(T) +} diff --git a/pkgs/ioswitch/dag/var.go b/pkgs/ioswitch/dag/var.go new file mode 100644 index 0000000..48d890b --- /dev/null +++ b/pkgs/ioswitch/dag/var.go @@ -0,0 +1,118 @@ +package dag + +import ( + "gitlink.org.cn/cloudream/common/pkgs/ioswitch/exec" + "gitlink.org.cn/cloudream/common/utils/lo2" +) + +type EndPoint struct { + Node *Node + SlotIndex int // 所连接的Node的Output或Input数组的索引 +} + +type StreamVar struct { + ID int + From EndPoint + Toes []EndPoint + Props any + Var *exec.StreamVar +} + +func (v *StreamVar) To(to *Node, slotIdx int) int { + v.Toes = append(v.Toes, EndPoint{Node: to, SlotIndex: slotIdx}) + to.InputStreams[slotIdx] = v + return len(v.Toes) - 1 +} + +// func (v *StreamVar) NotTo(toIdx int) EndPoint { +// ed := v.Toes[toIdx] +// lo2.RemoveAt(v.Toes, toIdx) +// ed.Node.InputStreams[ed.SlotIndex] = nil +// return ed +// } + +func (v *StreamVar) NotTo(node *Node) (EndPoint, bool) { + for i, ed := range v.Toes { + if ed.Node == node { + v.Toes = lo2.RemoveAt(v.Toes, i) + ed.Node.InputStreams[ed.SlotIndex] = nil + return ed, true + } + } + + return EndPoint{}, false +} + +func (v *StreamVar) NotToWhere(pred func(to EndPoint) bool) []EndPoint { + var newToes []EndPoint + var rmed []EndPoint + for _, ed := range v.Toes { + if pred(ed) { + ed.Node.InputStreams[ed.SlotIndex] = nil + rmed = append(rmed, ed) + } else { + newToes = append(newToes, ed) + } + } + v.Toes = newToes + return rmed +} + +func (v *StreamVar) NotToAll() []EndPoint { + for _, ed := range v.Toes { + ed.Node.InputStreams[ed.SlotIndex] = nil + } + toes := v.Toes + v.Toes = nil + return toes +} + +func NodeNewOutputStream(node *Node, props any) *StreamVar { + str := &StreamVar{ + ID: node.Graph.genVarID(), + From: EndPoint{Node: node, SlotIndex: len(node.OutputStreams)}, + Props: props, + } + node.OutputStreams = append(node.OutputStreams, str) + return str +} + +func NodeDeclareInputStream(node *Node, cnt int) { + node.InputStreams = make([]*StreamVar, cnt) +} + +type ValueVarType int + +const ( + StringValueVar ValueVarType = iota + SignalValueVar +) + +type ValueVar struct { + ID int + Type ValueVarType + From EndPoint + Toes []EndPoint + Props any + Var exec.Var +} + +func (v *ValueVar) To(to *Node, slotIdx int) int { + v.Toes = append(v.Toes, EndPoint{Node: to, SlotIndex: slotIdx}) + to.InputValues[slotIdx] = v + return len(v.Toes) - 1 +} + +func NodeNewOutputValue(node *Node, props any) *ValueVar { + val := &ValueVar{ + ID: node.Graph.genVarID(), + From: EndPoint{Node: node, SlotIndex: len(node.OutputStreams)}, + Props: props, + } + node.OutputValues = append(node.OutputValues, val) + return val +} + +func NodeDeclareInputValue(node *Node, cnt int) { + node.InputValues = make([]*ValueVar, cnt) +} diff --git a/pkgs/ioswitch/exec/driver.go b/pkgs/ioswitch/exec/driver.go new file mode 100644 index 0000000..4b51480 --- /dev/null +++ b/pkgs/ioswitch/exec/driver.go @@ -0,0 +1,112 @@ +package exec + +import ( + "context" + "fmt" + "io" + "sync" + + "gitlink.org.cn/cloudream/common/pkgs/future" + "gitlink.org.cn/cloudream/common/utils/io2" +) + +type Driver struct { + planID PlanID + planBlder *PlanBuilder + callback *future.SetValueFuture[map[string]any] + ctx context.Context + cancel context.CancelFunc + driverExec *Executor +} + +// 开始写入一个流。此函数会将输入视为一个完整的流，因此会给流包装一个Range来获取只需要的部分。 +func (e *Driver) BeginWrite(str io.ReadCloser, handle *DriverWriteStream) { + handle.Var.Stream = io2.NewRange(str, handle.RangeHint.Offset, handle.RangeHint.Length) + e.driverExec.PutVars(handle.Var) +} + +// 开始写入一个流。此函数默认输入流已经是Handle的RangeHint锁描述的范围，因此不会做任何其他处理 +func (e *Driver) BeginWriteRanged(str io.ReadCloser, handle *DriverWriteStream) { + handle.Var.Stream = str + e.driverExec.PutVars(handle.Var) +} + +func (e *Driver) BeginRead(handle *DriverReadStream) (io.ReadCloser, error) { + err := e.driverExec.BindVars(e.ctx, handle.Var) + if err != nil { + return nil, fmt.Errorf("bind vars: %w", err) + } + + return handle.Var.Stream, nil +} + +func (e *Driver) Signal(signal *DriverSignalVar) { + e.driverExec.PutVars(signal.Var) +} + +func (e *Driver) Wait(ctx context.Context) (map[string]any, error) { + stored, err := e.callback.WaitValue(ctx) + if err != nil { + return nil, err + } + + return stored, nil +} + +func (e *Driver) execute() { + wg := sync.WaitGroup{} + + for _, p := range e.planBlder.WorkerPlans { + wg.Add(1) + + go func(p *WorkerPlanBuilder) { + defer wg.Done() + + plan := Plan{ + ID: e.planID, + Ops: p.Ops, + } + + cli, err := p.Worker.NewClient() + if err != nil { + e.stopWith(fmt.Errorf("new client to worker %v: %w", p.Worker, err)) + return + } + defer cli.Close() + + err = cli.ExecutePlan(e.ctx, plan) + if err != nil { + e.stopWith(fmt.Errorf("execute plan at worker %v: %w", p.Worker, err)) + return + } + }(p) + } + + stored, err := e.driverExec.Run(e.ctx) + if err != nil { + e.stopWith(fmt.Errorf("run executor switch: %w", err)) + return + } + + wg.Wait() + + e.callback.SetValue(stored) +} + +func (e *Driver) stopWith(err error) { + e.callback.SetError(err) + e.cancel() +} + +type DriverWriteStream struct { + Var *StreamVar + RangeHint *Range +} + +type DriverReadStream struct { + Var *StreamVar +} + +type DriverSignalVar struct { + Var *SignalVar +} diff --git a/pkgs/ioswitch/exec/exec.go b/pkgs/ioswitch/exec/exec.go new file mode 100644 index 0000000..ec16299 --- /dev/null +++ b/pkgs/ioswitch/exec/exec.go @@ -0,0 +1,26 @@ +package exec + +import ( + "context" + + "gitlink.org.cn/cloudream/common/pkgs/types" + "gitlink.org.cn/cloudream/common/utils/reflect2" + "gitlink.org.cn/cloudream/common/utils/serder" +) + +type PlanID string + +type Plan struct { + ID PlanID `json:"id"` + Ops []Op `json:"ops"` +} + +var opUnion = serder.UseTypeUnionExternallyTagged(types.Ref(types.NewTypeUnion[Op]())) + +type Op interface { + Execute(ctx context.Context, e *Executor) error +} + +func UseOp[T Op]() { + opUnion.Add(reflect2.TypeOf[T]()) +} diff --git a/pkgs/ioswitch/exec/executor.go b/pkgs/ioswitch/exec/executor.go new file mode 100644 index 0000000..dcd087e --- /dev/null +++ b/pkgs/ioswitch/exec/executor.go @@ -0,0 +1,165 @@ +package exec + +import ( + "context" + "fmt" + "sync" + + "gitlink.org.cn/cloudream/common/pkgs/future" + "gitlink.org.cn/cloudream/common/utils/lo2" + "gitlink.org.cn/cloudream/common/utils/sync2" +) + +type bindingVars struct { + Waittings []Var + Bindeds []Var + Callback *future.SetVoidFuture +} + +type Executor struct { + plan Plan + vars map[VarID]Var + bindings []*bindingVars + lock sync.Mutex + store map[string]any +} + +func NewExecutor(plan Plan) *Executor { + planning := Executor{ + plan: plan, + vars: make(map[VarID]Var), + store: make(map[string]any), + } + + return &planning +} + +func (s *Executor) Plan() *Plan { + return &s.plan +} + +func (s *Executor) Run(ctx context.Context) (map[string]any, error) { + ctx2, cancel := context.WithCancel(ctx) + defer cancel() + + err := sync2.ParallelDo(s.plan.Ops, func(o Op, idx int) error { + err := o.Execute(ctx2, s) + + s.lock.Lock() + defer s.lock.Unlock() + + if err != nil { + cancel() + return fmt.Errorf("%T: %w", o, err) + } + + return nil + }) + if err != nil { + return nil, err + } + + return s.store, nil +} + +func (s *Executor) BindVars(ctx context.Context, vs ...Var) error { + s.lock.Lock() + + callback := future.NewSetVoid() + binding := &bindingVars{ + Callback: callback, + } + + for _, v := range vs { + v2 := s.vars[v.GetID()] + if v2 == nil { + binding.Waittings = append(binding.Waittings, v) + continue + } + + if err := AssignVar(v2, v); err != nil { + s.lock.Unlock() + return fmt.Errorf("assign var %v to %v: %w", v2.GetID(), v.GetID(), err) + } + + binding.Bindeds = append(binding.Bindeds, v) + } + + if len(binding.Waittings) == 0 { + s.lock.Unlock() + return nil + } + + s.bindings = append(s.bindings, binding) + s.lock.Unlock() + + err := callback.Wait(ctx) + + s.lock.Lock() + defer s.lock.Unlock() + + s.bindings = lo2.Remove(s.bindings, binding) + + return err +} + +func (s *Executor) PutVars(vs ...Var) { + s.lock.Lock() + defer s.lock.Unlock() + +loop: + for _, v := range vs { + for ib, b := range s.bindings { + for iw, w := range b.Waittings { + if w.GetID() != v.GetID() { + continue + } + + if err := AssignVar(v, w); err != nil { + b.Callback.SetError(fmt.Errorf("assign var %v to %v: %w", v.GetID(), w.GetID(), err)) + // 绑定类型不对，说明生成的执行计划有问题，怎么处理都可以，因为最终会执行失败 + continue loop + } + + b.Bindeds = append(b.Bindeds, w) + b.Waittings = lo2.RemoveAt(b.Waittings, iw) + if len(b.Waittings) == 0 { + b.Callback.SetVoid() + s.bindings = lo2.RemoveAt(s.bindings, ib) + } + + // 绑定成功，继续最外层循环 + continue loop + } + + } + + // 如果没有绑定，则直接放入变量表中 + s.vars[v.GetID()] = v + } +} + +func (s *Executor) Store(key string, val any) { + s.lock.Lock() + defer s.lock.Unlock() + + s.store[key] = val +} + +func BindArrayVars[T Var](sw *Executor, ctx context.Context, vs []T) error { + var vs2 []Var + for _, v := range vs { + vs2 = append(vs2, v) + } + + return sw.BindVars(ctx, vs2...) +} + +func PutArrayVars[T Var](sw *Executor, vs []T) { + var vs2 []Var + for _, v := range vs { + vs2 = append(vs2, v) + } + + sw.PutVars(vs2...) +} diff --git a/pkgs/ioswitch/exec/plan_builder.go b/pkgs/ioswitch/exec/plan_builder.go new file mode 100644 index 0000000..24b0525 --- /dev/null +++ b/pkgs/ioswitch/exec/plan_builder.go @@ -0,0 +1,123 @@ +package exec + +import ( + "context" + + "gitlink.org.cn/cloudream/common/pkgs/future" + "gitlink.org.cn/cloudream/common/utils/lo2" +) + +type PlanBuilder struct { + Vars []Var + WorkerPlans []*WorkerPlanBuilder + DriverPlan DriverPlanBuilder +} + +func NewPlanBuilder() *PlanBuilder { + bld := &PlanBuilder{ + DriverPlan: DriverPlanBuilder{}, + } + + return bld +} + +func (b *PlanBuilder) AtDriver() *DriverPlanBuilder { + return &b.DriverPlan +} + +func (b *PlanBuilder) AtWorker(worker WorkerInfo) *WorkerPlanBuilder { + for _, p := range b.WorkerPlans { + if p.Worker.Equals(worker) { + return p + } + } + + p := &WorkerPlanBuilder{ + Worker: worker, + } + b.WorkerPlans = append(b.WorkerPlans, p) + + return p +} + +func (b *PlanBuilder) NewStreamVar() *StreamVar { + v := &StreamVar{ + ID: VarID(len(b.Vars)), + } + b.Vars = append(b.Vars, v) + + return v +} + +func (b *PlanBuilder) NewIntVar() *IntVar { + v := &IntVar{ + ID: VarID(len(b.Vars)), + } + b.Vars = append(b.Vars, v) + + return v +} + +func (b *PlanBuilder) NewStringVar() *StringVar { + v := &StringVar{ + ID: VarID(len(b.Vars)), + } + b.Vars = append(b.Vars, v) + + return v +} +func (b *PlanBuilder) NewSignalVar() *SignalVar { + v := &SignalVar{ + ID: VarID(len(b.Vars)), + } + b.Vars = append(b.Vars, v) + + return v +} + +func (b *PlanBuilder) Execute() *Driver { + ctx, cancel := context.WithCancel(context.Background()) + planID := genRandomPlanID() + + execPlan := Plan{ + ID: planID, + Ops: b.DriverPlan.Ops, + } + + exec := Driver{ + planID: planID, + planBlder: b, + callback: future.NewSetValue[map[string]any](), + ctx: ctx, + cancel: cancel, + driverExec: NewExecutor(execPlan), + } + go exec.execute() + + return &exec +} + +type WorkerPlanBuilder struct { + Worker WorkerInfo + Ops []Op +} + +func (b *WorkerPlanBuilder) AddOp(op Op) { + b.Ops = append(b.Ops, op) +} + +func (b *WorkerPlanBuilder) RemoveOp(op Op) { + b.Ops = lo2.Remove(b.Ops, op) +} + +type DriverPlanBuilder struct { + Ops []Op +} + +func (b *DriverPlanBuilder) AddOp(op Op) { + b.Ops = append(b.Ops, op) +} + +func (b *DriverPlanBuilder) RemoveOp(op Op) { + b.Ops = lo2.Remove(b.Ops, op) +} diff --git a/pkgs/ioswitch/exec/utils.go b/pkgs/ioswitch/exec/utils.go new file mode 100644 index 0000000..4112669 --- /dev/null +++ b/pkgs/ioswitch/exec/utils.go @@ -0,0 +1,98 @@ +package exec + +import ( + "fmt" + "reflect" + + "github.com/google/uuid" + "gitlink.org.cn/cloudream/common/utils/math2" +) + +func genRandomPlanID() PlanID { + return PlanID(uuid.NewString()) +} + +func AssignVar(from Var, to Var) error { + if reflect.TypeOf(from) != reflect.TypeOf(to) { + return fmt.Errorf("cannot assign %T to %T", from, to) + } + + switch from := from.(type) { + case *StreamVar: + to.(*StreamVar).Stream = from.Stream + case *IntVar: + to.(*IntVar).Value = from.Value + case *StringVar: + to.(*StringVar).Value = from.Value + case *SignalVar: + } + + return nil +} + +type Range struct { + Offset int64 + Length *int64 +} + +func (r *Range) Extend(other Range) { + newOffset := math2.Min(r.Offset, other.Offset) + + if r.Length == nil { + r.Offset = newOffset + return + } + + if other.Length == nil { + r.Offset = newOffset + r.Length = nil + return + } + + otherEnd := other.Offset + *other.Length + rEnd := r.Offset + *r.Length + + newEnd := math2.Max(otherEnd, rEnd) + r.Offset = newOffset + *r.Length = newEnd - newOffset +} + +func (r *Range) ExtendStart(start int64) { + r.Offset = math2.Min(r.Offset, start) +} + +func (r *Range) ExtendEnd(end int64) { + if r.Length == nil { + return + } + + rEnd := r.Offset + *r.Length + newLen := math2.Max(end, rEnd) - r.Offset + r.Length = &newLen +} + +func (r *Range) Fix(maxLength int64) { + if r.Length != nil { + return + } + + len := maxLength - r.Offset + r.Length = &len +} + +func (r *Range) ToStartEnd(maxLen int64) (start int64, end int64) { + if r.Length == nil { + return r.Offset, maxLen + } + + end = r.Offset + *r.Length + return r.Offset, end +} + +func (r *Range) ClampLength(maxLen int64) { + if r.Length == nil { + return + } + + *r.Length = math2.Min(*r.Length, maxLen-r.Offset) +} diff --git a/pkgs/ioswitch/exec/var.go b/pkgs/ioswitch/exec/var.go new file mode 100644 index 0000000..62e76ad --- /dev/null +++ b/pkgs/ioswitch/exec/var.go @@ -0,0 +1,57 @@ +package exec + +import ( + "io" + + "gitlink.org.cn/cloudream/common/pkgs/types" + "gitlink.org.cn/cloudream/common/utils/serder" +) + +type VarID int + +type Var interface { + GetID() VarID +} + +var VarUnion = types.NewTypeUnion[Var]( + (*IntVar)(nil), + (*StringVar)(nil), + (*SignalVar)(nil), + (*StreamVar)(nil), +) +var _ = serder.UseTypeUnionExternallyTagged(&VarUnion) + +type StreamVar struct { + ID VarID `json:"id"` + Stream io.ReadCloser `json:"-"` +} + +func (v *StreamVar) GetID() VarID { + return v.ID +} + +type IntVar struct { + ID VarID `json:"id"` + Value string `json:"value"` +} + +func (v *IntVar) GetID() VarID { + return v.ID +} + +type StringVar struct { + ID VarID `json:"id"` + Value string `json:"value"` +} + +func (v *StringVar) GetID() VarID { + return v.ID +} + +type SignalVar struct { + ID VarID `json:"id"` +} + +func (v *SignalVar) GetID() VarID { + return v.ID +} diff --git a/pkgs/ioswitch/exec/worker.go b/pkgs/ioswitch/exec/worker.go new file mode 100644 index 0000000..b9bf4f8 --- /dev/null +++ b/pkgs/ioswitch/exec/worker.go @@ -0,0 +1,103 @@ +package exec + +import ( + "context" + "io" + "sync" + + "github.com/samber/lo" + "gitlink.org.cn/cloudream/common/pkgs/future" + "gitlink.org.cn/cloudream/common/utils/lo2" +) + +type finding struct { + PlanID PlanID + Callback *future.SetValueFuture[*Executor] +} + +type Worker struct { + lock sync.Mutex + executors map[PlanID]*Executor + findings []*finding +} + +func NewWorker() Worker { + return Worker{ + executors: make(map[PlanID]*Executor), + } +} + +func (s *Worker) Add(exe *Executor) { + s.lock.Lock() + defer s.lock.Unlock() + + s.executors[exe.Plan().ID] = exe + + s.findings = lo.Reject(s.findings, func(f *finding, idx int) bool { + if f.PlanID != exe.Plan().ID { + return false + } + + f.Callback.SetValue(exe) + return true + }) +} + +func (s *Worker) Remove(sw *Executor) { + s.lock.Lock() + defer s.lock.Unlock() + + delete(s.executors, sw.Plan().ID) +} + +func (s *Worker) FindByID(id PlanID) *Executor { + s.lock.Lock() + defer s.lock.Unlock() + + return s.executors[id] +} + +func (s *Worker) FindByIDContexted(ctx context.Context, id PlanID) *Executor { + s.lock.Lock() + + sw := s.executors[id] + if sw != nil { + s.lock.Unlock() + return sw + } + + cb := future.NewSetValue[*Executor]() + f := &finding{ + PlanID: id, + Callback: cb, + } + s.findings = append(s.findings, f) + + s.lock.Unlock() + + sw, _ = cb.WaitValue(ctx) + + s.lock.Lock() + defer s.lock.Unlock() + + s.findings = lo2.Remove(s.findings, f) + + return sw +} + +type WorkerInfo interface { + NewClient() (WorkerClient, error) + // 判断两个worker是否相同 + Equals(worker WorkerInfo) bool + // Worker信息，比如ID、地址等 + String() string +} + +type WorkerClient interface { + ExecutePlan(ctx context.Context, plan Plan) error + SendStream(ctx context.Context, planID PlanID, v *StreamVar, str io.ReadCloser) error + SendVar(ctx context.Context, planID PlanID, v Var) error + GetStream(ctx context.Context, planID PlanID, v *StreamVar, signal *SignalVar) (io.ReadCloser, error) + GetVar(ctx context.Context, planID PlanID, v Var, signal *SignalVar) error + Close() error +} diff --git a/pkgs/ioswitch/plan/generate.go b/pkgs/ioswitch/plan/generate.go new file mode 100644 index 0000000..6a7362b --- /dev/null +++ b/pkgs/ioswitch/plan/generate.go @@ -0,0 +1,157 @@ +package plan + +import ( + "gitlink.org.cn/cloudream/common/pkgs/ioswitch/dag" + "gitlink.org.cn/cloudream/common/pkgs/ioswitch/exec" + "gitlink.org.cn/cloudream/common/pkgs/ioswitch/plan/ops" + "gitlink.org.cn/cloudream/storage/common/pkgs/ioswitch" +) + +func Generate(graph *dag.Graph, planBld *exec.PlanBuilder) error { + generateSend(graph) + return buildPlan(graph, planBld) +} + +// 生成Send指令 +func generateSend(graph *dag.Graph) { + graph.Walk(func(node *dag.Node) bool { + for _, out := range node.OutputStreams { + to := out.Toes[0] + if to.Node.Env.Equals(node.Env) { + continue + } + + switch to.Node.Env.Type { + case dag.EnvDriver: + // // 如果是要送到Driver，则只能由Driver主动去拉取 + getNode := graph.NewNode(&ops.GetStreamType{}, ioswitch.NodeProps{}) + getNode.Env.ToEnvDriver() + + // // 同时需要对此变量生成HoldUntil指令，避免Plan结束时Get指令还未到达 + holdNode := graph.NewNode(&ops.HoldUntilType{}, ioswitch.NodeProps{}) + holdNode.Env = node.Env + + // 将Get指令的信号送到Hold指令 + getNode.OutputValues[0].To(holdNode, 0) + // 将Get指令的输出送到目的地 + getNode.OutputStreams[0].To(to.Node, to.SlotIndex) + out.Toes = nil + // 将源节点的输出送到Hold指令 + out.To(holdNode, 0) + // 将Hold指令的输出送到Get指令 + holdNode.OutputStreams[0].To(getNode, 0) + + case dag.EnvWorker: + // 如果是要送到Agent，则可以直接发送 + n := graph.NewNode(&ops.SendStreamType{}, ioswitch.NodeProps{}) + n.Env = node.Env + n.OutputStreams[0].To(to.Node, to.SlotIndex) + out.Toes = nil + out.To(n, 0) + } + } + + for _, out := range node.OutputValues { + to := out.Toes[0] + if to.Node.Env.Equals(node.Env) { + continue + } + + switch to.Node.Env.Type { + case dag.EnvDriver: + // // 如果是要送到Driver，则只能由Driver主动去拉取 + getNode := graph.NewNode(&ops.GetVaType{}, ioswitch.NodeProps{}) + getNode.Env.ToEnvDriver() + + // // 同时需要对此变量生成HoldUntil指令，避免Plan结束时Get指令还未到达 + holdNode := graph.NewNode(&ops.HoldUntilType{}, ioswitch.NodeProps{}) + holdNode.Env = node.Env + + // 将Get指令的信号送到Hold指令 + getNode.OutputValues[0].To(holdNode, 0) + // 将Get指令的输出送到目的地 + getNode.OutputValues[1].To(to.Node, to.SlotIndex) + out.Toes = nil + // 将源节点的输出送到Hold指令 + out.To(holdNode, 0) + // 将Hold指令的输出送到Get指令 + holdNode.OutputValues[0].To(getNode, 0) + + case dag.EnvWorker: + // 如果是要送到Agent，则可以直接发送 + n := graph.NewNode(&ops.SendVarType{}, ioswitch.NodeProps{}) + n.Env = node.Env + n.OutputValues[0].To(to.Node, to.SlotIndex) + out.Toes = nil + out.To(n, 0) + } + } + + return true + }) +} + +// 生成Plan +func buildPlan(graph *dag.Graph, blder *exec.PlanBuilder) error { + var retErr error + graph.Walk(func(node *dag.Node) bool { + for _, out := range node.OutputStreams { + if out.Var != nil { + continue + } + + out.Var = blder.NewStreamVar() + } + + for _, in := range node.InputStreams { + if in.Var != nil { + continue + } + + in.Var = blder.NewStreamVar() + } + + for _, out := range node.OutputValues { + if out.Var != nil { + continue + } + + switch out.Type { + case dag.StringValueVar: + out.Var = blder.NewStringVar() + case dag.SignalValueVar: + out.Var = blder.NewSignalVar() + } + } + + for _, in := range node.InputValues { + if in.Var != nil { + continue + } + + switch in.Type { + case dag.StringValueVar: + in.Var = blder.NewStringVar() + case dag.SignalValueVar: + in.Var = blder.NewSignalVar() + } + } + + op, err := node.Type.GenerateOp(node) + if err != nil { + retErr = err + return false + } + + switch node.Env.Type { + case dag.EnvDriver: + blder.AtDriver().AddOp(op) + case dag.EnvWorker: + blder.AtWorker(node.Env.Worker).AddOp(op) + } + + return true + }) + + return retErr +} diff --git a/pkgs/ioswitch/plan/ops/driver.go b/pkgs/ioswitch/plan/ops/driver.go new file mode 100644 index 0000000..214f333 --- /dev/null +++ b/pkgs/ioswitch/plan/ops/driver.go @@ -0,0 +1,43 @@ +package ops + +import ( + "fmt" + + "gitlink.org.cn/cloudream/common/pkgs/ioswitch/dag" + "gitlink.org.cn/cloudream/common/pkgs/ioswitch/exec" +) + +type FromDriverType struct { + Handle *exec.DriverWriteStream +} + +func (t *FromDriverType) InitNode(node *dag.Node) { + dag.NodeNewOutputStream(node, nil) +} + +func (t *FromDriverType) GenerateOp(op *dag.Node) (exec.Op, error) { + t.Handle.Var = op.OutputStreams[0].Var + return nil, nil +} + +func (t *FromDriverType) String(node *dag.Node) string { + return fmt.Sprintf("FromDriver[]%v%v", formatStreamIO(node), formatValueIO(node)) +} + +type ToDriverType struct { + Handle *exec.DriverReadStream + Range exec.Range +} + +func (t *ToDriverType) InitNode(node *dag.Node) { + dag.NodeDeclareInputStream(node, 1) +} + +func (t *ToDriverType) GenerateOp(op *dag.Node) (exec.Op, error) { + t.Handle.Var = op.InputStreams[0].Var + return nil, nil +} + +func (t *ToDriverType) String(node *dag.Node) string { + return fmt.Sprintf("ToDriver[%v+%v]%v%v", t.Range.Offset, t.Range.Length, formatStreamIO(node), formatValueIO(node)) +} diff --git a/pkgs/ioswitch/plan/ops/drop.go b/pkgs/ioswitch/plan/ops/drop.go new file mode 100644 index 0000000..22f2d5d --- /dev/null +++ b/pkgs/ioswitch/plan/ops/drop.go @@ -0,0 +1,52 @@ +package ops + +import ( + "context" + "fmt" + "io" + + "gitlink.org.cn/cloudream/common/pkgs/ioswitch/dag" + "gitlink.org.cn/cloudream/common/pkgs/ioswitch/exec" +) + +func init() { + exec.UseOp[*DropStream]() +} + +type DropStream struct { + Input *exec.StreamVar `json:"input"` +} + +func (o *DropStream) Execute(ctx context.Context, e *exec.Executor) error { + err := e.BindVars(ctx, o.Input) + if err != nil { + return err + } + + for { + buf := make([]byte, 1024*8) + _, err = o.Input.Stream.Read(buf) + if err == io.EOF { + return nil + } + if err != nil { + return err + } + } +} + +type DropType struct{} + +func (t *DropType) InitNode(node *dag.Node) { + dag.NodeDeclareInputStream(node, 1) +} + +func (t *DropType) GenerateOp(op *dag.Node) (exec.Op, error) { + return &DropStream{ + Input: op.InputStreams[0].Var, + }, nil +} + +func (t *DropType) String(node *dag.Node) string { + return fmt.Sprintf("Drop[]%v%v", formatStreamIO(node), formatValueIO(node)) +} diff --git a/pkgs/ioswitch/plan/ops/send.go b/pkgs/ioswitch/plan/ops/send.go new file mode 100644 index 0000000..148938a --- /dev/null +++ b/pkgs/ioswitch/plan/ops/send.go @@ -0,0 +1,224 @@ +package ops + +import ( + "context" + "fmt" + "io" + + "gitlink.org.cn/cloudream/common/pkgs/future" + "gitlink.org.cn/cloudream/common/pkgs/ioswitch/dag" + "gitlink.org.cn/cloudream/common/pkgs/ioswitch/exec" + "gitlink.org.cn/cloudream/common/pkgs/logger" + "gitlink.org.cn/cloudream/common/utils/io2" +) + +func init() { + exec.UseOp[*SendStream]() + exec.UseOp[*GetStream]() + exec.UseOp[*SendVar]() + exec.UseOp[*GetVar]() +} + +type SendStream struct { + Input *exec.StreamVar `json:"input"` + Send *exec.StreamVar `json:"send"` + Worker exec.WorkerInfo `json:"worker"` +} + +func (o *SendStream) Execute(ctx context.Context, e *exec.Executor) error { + err := e.BindVars(ctx, o.Input) + if err != nil { + return err + } + defer o.Input.Stream.Close() + + cli, err := o.Worker.NewClient() + if err != nil { + return fmt.Errorf("new worker %v client: %w", o.Worker, err) + } + defer cli.Close() + + logger.Debugf("sending stream %v as %v to worker %v", o.Input.ID, o.Send.ID, o.Worker) + + // 发送后流的ID不同 + err = cli.SendStream(ctx, e.Plan().ID, o.Send, o.Input.Stream) + if err != nil { + return fmt.Errorf("sending stream: %w", err) + } + + return nil +} + +type GetStream struct { + Signal *exec.SignalVar `json:"signal"` + Target *exec.StreamVar `json:"target"` + Output *exec.StreamVar `json:"output"` + Worker exec.WorkerInfo `json:"worker"` +} + +func (o *GetStream) Execute(ctx context.Context, e *exec.Executor) error { + cli, err := o.Worker.NewClient() + if err != nil { + return fmt.Errorf("new worker %v client: %w", o.Worker, err) + } + defer cli.Close() + + logger.Debugf("getting stream %v as %v from worker %v", o.Target.ID, o.Output.ID, o.Worker) + + str, err := cli.GetStream(ctx, e.Plan().ID, o.Target, o.Signal) + if err != nil { + return fmt.Errorf("getting stream: %w", err) + } + + fut := future.NewSetVoid() + // 获取后送到本地的流ID是不同的 + o.Output.Stream = io2.AfterReadClosedOnce(str, func(closer io.ReadCloser) { + fut.SetVoid() + }) + e.PutVars(o.Output) + + return fut.Wait(ctx) +} + +type SendVar struct { + Input exec.Var `json:"input"` + Send exec.Var `json:"send"` + Worker exec.WorkerInfo `json:"worker"` +} + +func (o *SendVar) Execute(ctx context.Context, e *exec.Executor) error { + err := e.BindVars(ctx, o.Input) + if err != nil { + return err + } + + cli, err := o.Worker.NewClient() + if err != nil { + return fmt.Errorf("new worker %v client: %w", o.Worker, err) + } + defer cli.Close() + + logger.Debugf("sending var %v as %v to worker %v", o.Input.GetID(), o.Send.GetID(), o.Worker) + + exec.AssignVar(o.Input, o.Send) + err = cli.SendVar(ctx, e.Plan().ID, o.Send) + if err != nil { + return fmt.Errorf("sending var: %w", err) + } + + return nil +} + +type GetVar struct { + Signal *exec.SignalVar `json:"signal"` + Target exec.Var `json:"target"` + Output exec.Var `json:"output"` + Worker exec.WorkerInfo `json:"worker"` +} + +func (o *GetVar) Execute(ctx context.Context, e *exec.Executor) error { + cli, err := o.Worker.NewClient() + if err != nil { + return fmt.Errorf("new worker %v client: %w", o.Worker, err) + } + defer cli.Close() + + logger.Debugf("getting var %v as %v from worker %v", o.Target.GetID(), o.Output.GetID(), o.Worker) + + err = cli.GetVar(ctx, e.Plan().ID, o.Target, o.Signal) + if err != nil { + return fmt.Errorf("getting var: %w", err) + } + exec.AssignVar(o.Target, o.Output) + e.PutVars(o.Output) + + return nil +} + +type SendStreamType struct { + ToWorker exec.WorkerInfo +} + +func (t *SendStreamType) InitNode(node *dag.Node) { + dag.NodeDeclareInputStream(node, 1) + dag.NodeNewOutputStream(node, nil) +} + +func (t *SendStreamType) GenerateOp(op *dag.Node) (exec.Op, error) { + return &SendStream{ + Input: op.InputStreams[0].Var, + Send: op.OutputStreams[0].Var, + Worker: t.ToWorker, + }, nil +} + +func (t *SendStreamType) String(node *dag.Node) string { + return fmt.Sprintf("SendStream[]%v%v", formatStreamIO(node), formatValueIO(node)) +} + +type SendVarType struct { + ToWorker exec.WorkerInfo +} + +func (t *SendVarType) InitNode(node *dag.Node) { + dag.NodeDeclareInputValue(node, 1) + dag.NodeNewOutputValue(node, nil) +} + +func (t *SendVarType) GenerateOp(op *dag.Node) (exec.Op, error) { + return &SendVar{ + Input: op.InputValues[0].Var, + Send: op.OutputValues[0].Var, + Worker: t.ToWorker, + }, nil +} + +func (t *SendVarType) String(node *dag.Node) string { + return fmt.Sprintf("SendVar[]%v%v", formatStreamIO(node), formatValueIO(node)) +} + +type GetStreamType struct { + FromWorker exec.WorkerInfo +} + +func (t *GetStreamType) InitNode(node *dag.Node) { + dag.NodeDeclareInputStream(node, 1) + dag.NodeNewOutputValue(node, nil) + dag.NodeNewOutputStream(node, nil) +} + +func (t *GetStreamType) GenerateOp(op *dag.Node) (exec.Op, error) { + return &GetStream{ + Signal: op.OutputValues[0].Var.(*exec.SignalVar), + Output: op.OutputStreams[0].Var, + Target: op.InputStreams[0].Var, + Worker: t.FromWorker, + }, nil +} + +func (t *GetStreamType) String(node *dag.Node) string { + return fmt.Sprintf("GetStream[]%v%v", formatStreamIO(node), formatValueIO(node)) +} + +type GetVaType struct { + FromWorker exec.WorkerInfo +} + +func (t *GetVaType) InitNode(node *dag.Node) { + dag.NodeDeclareInputValue(node, 1) + dag.NodeNewOutputValue(node, nil) + dag.NodeNewOutputValue(node, nil) +} + +func (t *GetVaType) GenerateOp(op *dag.Node) (exec.Op, error) { + return &GetVar{ + Signal: op.OutputValues[0].Var.(*exec.SignalVar), + Output: op.OutputValues[1].Var, + Target: op.InputValues[0].Var, + Worker: t.FromWorker, + }, nil +} + +func (t *GetVaType) String(node *dag.Node) string { + return fmt.Sprintf("GetVar[]%v%v", formatStreamIO(node), formatValueIO(node)) +} diff --git a/pkgs/ioswitch/plan/ops/store.go b/pkgs/ioswitch/plan/ops/store.go new file mode 100644 index 0000000..08886bd --- /dev/null +++ b/pkgs/ioswitch/plan/ops/store.go @@ -0,0 +1,49 @@ +package ops + +import ( + "context" + "fmt" + + "gitlink.org.cn/cloudream/common/pkgs/ioswitch/dag" + "gitlink.org.cn/cloudream/common/pkgs/ioswitch/exec" +) + +type Store struct { + Var exec.Var + Key string +} + +func (o *Store) Execute(ctx context.Context, e *exec.Executor) error { + err := e.BindVars(ctx, o.Var) + if err != nil { + return err + } + + switch v := o.Var.(type) { + case *exec.IntVar: + e.Store(o.Key, v.Value) + case *exec.StringVar: + e.Store(o.Key, v.Value) + } + + return nil +} + +type StoreType struct { + StoreKey string +} + +func (t *StoreType) InitNode(node *dag.Node) { + dag.NodeDeclareInputValue(node, 1) +} + +func (t *StoreType) GenerateOp(op *dag.Node) (exec.Op, error) { + return &Store{ + Var: op.InputValues[0].Var, + Key: t.StoreKey, + }, nil +} + +func (t *StoreType) String(node *dag.Node) string { + return fmt.Sprintf("Store[%s]%v%v", t.StoreKey, formatStreamIO(node), formatValueIO(node)) +} diff --git a/pkgs/ioswitch/plan/ops/sync.go b/pkgs/ioswitch/plan/ops/sync.go new file mode 100644 index 0000000..26a98ab --- /dev/null +++ b/pkgs/ioswitch/plan/ops/sync.go @@ -0,0 +1,172 @@ +package ops + +import ( + "context" + "fmt" + "io" + + "gitlink.org.cn/cloudream/common/pkgs/future" + "gitlink.org.cn/cloudream/common/pkgs/ioswitch/dag" + "gitlink.org.cn/cloudream/common/pkgs/ioswitch/exec" +) + +func init() { + exec.UseOp[*OnStreamBegin]() + exec.UseOp[*OnStreamEnd]() + exec.UseOp[*HoldUntil]() + exec.UseOp[*HangUntil]() + exec.UseOp[*Broadcast]() +} + +type OnStreamBegin struct { + Raw *exec.StreamVar `json:"raw"` + New *exec.StreamVar `json:"new"` + Signal *exec.SignalVar `json:"signal"` +} + +func (o *OnStreamBegin) Execute(ctx context.Context, e *exec.Executor) error { + err := e.BindVars(ctx, o.Raw) + if err != nil { + return err + } + + o.New.Stream = o.Raw.Stream + + e.PutVars(o.New, o.Signal) + return nil +} + +type OnStreamEnd struct { + Raw *exec.StreamVar `json:"raw"` + New *exec.StreamVar `json:"new"` + Signal *exec.SignalVar `json:"signal"` +} + +type onStreamEnd struct { + inner io.ReadCloser + callback *future.SetVoidFuture +} + +func (o *onStreamEnd) Read(p []byte) (n int, err error) { + n, err = o.inner.Read(p) + if err == io.EOF { + o.callback.SetVoid() + } else if err != nil { + o.callback.SetError(err) + } + return n, err +} + +func (o *onStreamEnd) Close() error { + o.callback.SetError(fmt.Errorf("stream closed early")) + return o.inner.Close() +} + +func (o *OnStreamEnd) Execute(ctx context.Context, e *exec.Executor) error { + err := e.BindVars(ctx, o.Raw) + if err != nil { + return err + } + + cb := future.NewSetVoid() + + o.New.Stream = &onStreamEnd{ + inner: o.Raw.Stream, + callback: cb, + } + e.PutVars(o.New) + + err = cb.Wait(ctx) + if err != nil { + return err + } + + e.PutVars(o.Signal) + return nil +} + +type HoldUntil struct { + Waits []*exec.SignalVar `json:"waits"` + Holds []exec.Var `json:"holds"` + Emits []exec.Var `json:"emits"` +} + +func (w *HoldUntil) Execute(ctx context.Context, e *exec.Executor) error { + err := e.BindVars(ctx, w.Holds...) + if err != nil { + return err + } + + err = exec.BindArrayVars(e, ctx, w.Waits) + if err != nil { + return err + } + + for i := 0; i < len(w.Holds); i++ { + err := exec.AssignVar(w.Holds[i], w.Emits[i]) + if err != nil { + return err + } + } + + e.PutVars(w.Emits...) + return nil +} + +type HangUntil struct { + Waits []*exec.SignalVar `json:"waits"` + Op exec.Op `json:"op"` +} + +func (h *HangUntil) Execute(ctx context.Context, e *exec.Executor) error { + err := exec.BindArrayVars(e, ctx, h.Waits) + if err != nil { + return err + } + + return h.Op.Execute(ctx, e) +} + +type Broadcast struct { + Source *exec.SignalVar `json:"source"` + Targets []*exec.SignalVar `json:"targets"` +} + +func (b *Broadcast) Execute(ctx context.Context, e *exec.Executor) error { + err := e.BindVars(ctx, b.Source) + if err != nil { + return err + } + + exec.PutArrayVars(e, b.Targets) + return nil +} + +type HoldUntilType struct { +} + +func (t *HoldUntilType) InitNode(node *dag.Node) { + dag.NodeDeclareInputValue(node, 1) +} + +func (t *HoldUntilType) GenerateOp(op *dag.Node) (exec.Op, error) { + o := &HoldUntil{ + Waits: []*exec.SignalVar{op.InputValues[0].Var.(*exec.SignalVar)}, + } + + for i := 0; i < len(op.OutputValues); i++ { + o.Holds = append(o.Holds, op.InputValues[i+1].Var) + o.Emits = append(o.Emits, op.OutputValues[i].Var) + } + + for i := 0; i < len(op.OutputStreams); i++ { + o.Holds = append(o.Holds, op.InputStreams[i].Var) + o.Emits = append(o.Emits, op.OutputStreams[i].Var) + } + + return o, nil +} + +func (t *HoldUntilType) String(node *dag.Node) string { + return fmt.Sprintf("HoldUntil[]%v%v", formatStreamIO(node), formatValueIO(node)) +} diff --git a/pkgs/ioswitch/plan/ops/utils.go b/pkgs/ioswitch/plan/ops/utils.go new file mode 100644 index 0000000..0a32ac9 --- /dev/null +++ b/pkgs/ioswitch/plan/ops/utils.go @@ -0,0 +1,75 @@ +package ops + +import ( + "fmt" + + "gitlink.org.cn/cloudream/common/pkgs/ioswitch/dag" +) + +func formatStreamIO(node *dag.Node) string { + is := "" + for i, in := range node.InputStreams { + if i > 0 { + is += "," + } + + if in == nil { + is += "." + } else { + is += fmt.Sprintf("%v", in.ID) + } + } + + os := "" + for i, out := range node.OutputStreams { + if i > 0 { + os += "," + } + + if out == nil { + os += "." + } else { + os += fmt.Sprintf("%v", out.ID) + } + } + + if is == "" && os == "" { + return "" + } + + return fmt.Sprintf("S{%s>%s}", is, os) +} + +func formatValueIO(node *dag.Node) string { + is := "" + for i, in := range node.InputValues { + if i > 0 { + is += "," + } + + if in == nil { + is += "." + } else { + is += fmt.Sprintf("%v", in.ID) + } + } + + os := "" + for i, out := range node.OutputValues { + if i > 0 { + os += "," + } + + if out == nil { + os += "." + } else { + os += fmt.Sprintf("%v", out.ID) + } + } + + if is == "" && os == "" { + return "" + } + + return fmt.Sprintf("V{%s>%s}", is, os) +} diff --git a/pkgs/ioswitch/plan/ops/var.go b/pkgs/ioswitch/plan/ops/var.go new file mode 100644 index 0000000..d2a7fe6 --- /dev/null +++ b/pkgs/ioswitch/plan/ops/var.go @@ -0,0 +1,20 @@ +package ops + +import ( + "context" + + "gitlink.org.cn/cloudream/common/pkgs/ioswitch/exec" +) + +func init() { + exec.UseOp[*ConstVar]() +} + +type ConstVar struct { + Var *exec.StringVar `json:"var"` +} + +func (o *ConstVar) Execute(ctx context.Context, e *exec.Executor) error { + e.PutVars(o.Var) + return nil +}