diff --git a/pkgs/ioswitch/dag/graph.go b/pkgs/ioswitch/dag/graph.go new file mode 100644 index 0000000..811a6fe --- /dev/null +++ b/pkgs/ioswitch/dag/graph.go @@ -0,0 +1,74 @@ +package dag + +import ( + "gitlink.org.cn/cloudream/common/utils/lo2" +) + +type Graph[NP any, VP any] struct { + Nodes []*Node[NP, VP] + isWalking bool + nextVarID int +} + +func NewGraph[NP any, VP any]() *Graph[NP, VP] { + return &Graph[NP, VP]{} +} + +func (g *Graph[NP, VP]) NewNode(typ NodeType[NP, VP], props NP) *Node[NP, VP] { + n := &Node[NP, VP]{ + Type: typ, + Props: props, + Graph: g, + } + typ.InitNode(n) + g.Nodes = append(g.Nodes, n) + return n +} + +func (g *Graph[NP, VP]) RemoveNode(node *Node[NP, VP]) { + 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[NP, VP]) Walk(cb func(node *Node[NP, VP]) 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[NP, VP]) genVarID() int { + g.nextVarID++ + return g.nextVarID +} + +func NewNode[NP any, VP any, NT NodeType[NP, VP]](graph *Graph[NP, VP], typ NT, props NP) (*Node[NP, VP], NT) { + return graph.NewNode(typ, props), typ +} + +func WalkOnlyType[N NodeType[NP, VP], NP any, VP any](g *Graph[NP, VP], cb func(node *Node[NP, VP], typ N) bool) { + g.Walk(func(node *Node[NP, VP]) 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..bf3bf24 --- /dev/null +++ b/pkgs/ioswitch/dag/node.go @@ -0,0 +1,75 @@ +package dag + +import ( + "fmt" + + "gitlink.org.cn/cloudream/common/pkgs/ioswitch/exec" +) + +type NodeType[NP any, VP any] interface { + InitNode(node *Node[NP, VP]) + String(node *Node[NP, VP]) string + GenerateOp(node *Node[NP, VP], blder *exec.PlanBuilder) error +} + +type WorkerInfo interface { + // 获取连接到这个worker的GRPC服务的地址 + GetAddress() string + // 判断两个worker是否相同 + Equals(worker WorkerInfo) bool +} + +type NodeEnvType string + +const ( + EnvUnknown NodeEnvType = "" + EnvExecutor NodeEnvType = "Executor" + EnvWorker NodeEnvType = "Worker" +) + +type NodeEnv struct { + Type NodeEnvType + Worker WorkerInfo +} + +func (e *NodeEnv) ToEnvUnknown() { + e.Type = EnvUnknown + e.Worker = nil +} + +func (e *NodeEnv) ToEnvExecutor() { + e.Type = EnvExecutor + e.Worker = nil +} + +func (e *NodeEnv) ToEnvWorker(worker 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[NP any, VP any] struct { + Type NodeType[NP, VP] + Env NodeEnv + Props NP + InputStreams []*StreamVar[NP, VP] + OutputStreams []*StreamVar[NP, VP] + InputValues []*ValueVar[NP, VP] + OutputValues []*ValueVar[NP, VP] + Graph *Graph[NP, VP] +} + +func (n *Node[NP, VP]) String() string { + return fmt.Sprintf("%v", n.Type.String(n)) +} diff --git a/pkgs/ioswitch/dag/var.go b/pkgs/ioswitch/dag/var.go new file mode 100644 index 0000000..e1caa7b --- /dev/null +++ b/pkgs/ioswitch/dag/var.go @@ -0,0 +1,112 @@ +package dag + +import "gitlink.org.cn/cloudream/common/utils/lo2" + +type EndPoint[NP any, VP any] struct { + Node *Node[NP, VP] + SlotIndex int // 所连接的Node的Output或Input数组的索引 +} + +type StreamVar[NP any, VP any] struct { + ID int + From EndPoint[NP, VP] + Toes []EndPoint[NP, VP] + Props VP +} + +func (v *StreamVar[NP, VP]) To(to *Node[NP, VP], slotIdx int) int { + v.Toes = append(v.Toes, EndPoint[NP, VP]{Node: to, SlotIndex: slotIdx}) + to.InputStreams[slotIdx] = v + return len(v.Toes) - 1 +} + +// func (v *StreamVar[NP, VP]) NotTo(toIdx int) EndPoint[NP, VP] { +// ed := v.Toes[toIdx] +// lo2.RemoveAt(v.Toes, toIdx) +// ed.Node.InputStreams[ed.SlotIndex] = nil +// return ed +// } + +func (v *StreamVar[NP, VP]) NotTo(node *Node[NP, VP]) (EndPoint[NP, VP], 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[NP, VP]{}, false +} + +func (v *StreamVar[NP, VP]) NotToWhere(pred func(to EndPoint[NP, VP]) bool) []EndPoint[NP, VP] { + var newToes []EndPoint[NP, VP] + var rmed []EndPoint[NP, VP] + 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[NP, VP]) NotToAll() []EndPoint[NP, VP] { + for _, ed := range v.Toes { + ed.Node.InputStreams[ed.SlotIndex] = nil + } + toes := v.Toes + v.Toes = nil + return toes +} + +func NodeNewOutputStream[NP any, VP any](node *Node[NP, VP], props VP) *StreamVar[NP, VP] { + str := &StreamVar[NP, VP]{ + ID: node.Graph.genVarID(), + From: EndPoint[NP, VP]{Node: node, SlotIndex: len(node.OutputStreams)}, + Props: props, + } + node.OutputStreams = append(node.OutputStreams, str) + return str +} + +func NodeDeclareInputStream[NP any, VP any](node *Node[NP, VP], cnt int) { + node.InputStreams = make([]*StreamVar[NP, VP], cnt) +} + +type ValueVarType int + +const ( + StringValueVar ValueVarType = iota + SignalValueVar +) + +type ValueVar[NP any, VP any] struct { + ID int + From EndPoint[NP, VP] + Toes []EndPoint[NP, VP] + Props VP +} + +func (v *ValueVar[NP, VP]) To(to *Node[NP, VP], slotIdx int) int { + v.Toes = append(v.Toes, EndPoint[NP, VP]{Node: to, SlotIndex: slotIdx}) + to.InputValues[slotIdx] = v + return len(v.Toes) - 1 +} + +func NodeNewOutputValue[NP any, VP any](node *Node[NP, VP], props VP) *ValueVar[NP, VP] { + val := &ValueVar[NP, VP]{ + ID: node.Graph.genVarID(), + From: EndPoint[NP, VP]{Node: node, SlotIndex: len(node.OutputStreams)}, + Props: props, + } + node.OutputValues = append(node.OutputValues, val) + return val +} + +func NodeDeclareInputValue[NP any, VP any](node *Node[NP, VP], cnt int) { + node.InputValues = make([]*ValueVar[NP, VP], cnt) +} diff --git a/pkgs/ioswitch/exec/driver.go b/pkgs/ioswitch/exec/driver.go new file mode 100644 index 0000000..26f8273 --- /dev/null +++ b/pkgs/ioswitch/exec/driver.go @@ -0,0 +1,119 @@ +package exec + +import ( + "context" + "fmt" + "io" + "sync" + + "gitlink.org.cn/cloudream/common/pkgs/future" + "gitlink.org.cn/cloudream/common/utils/io2" + stgglb "gitlink.org.cn/cloudream/storage/common/globals" +) + +type Driver struct { + planID PlanID + planBlder *PlanBuilder + callback *future.SetVoidFuture + 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) { + err := e.callback.Wait(ctx) + if err != nil { + return nil, err + } + + ret := make(map[string]any) + e.planBlder.DriverPlan.StoreMap.Range(func(k, v any) bool { + ret[k.(string)] = v + return true + }) + + return ret, 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 := stgglb.AgentRPCPool.Acquire(stgglb.SelectGRPCAddress(&p.Node)) + if err != nil { + e.stopWith(fmt.Errorf("new agent rpc client of node %v: %w", p.Node.NodeID, err)) + return + } + defer stgglb.AgentRPCPool.Release(cli) + + err = cli.ExecuteIOPlan(e.ctx, plan) + if err != nil { + e.stopWith(fmt.Errorf("execute plan at %v: %w", p.Node.NodeID, err)) + return + } + }(p) + } + + err := e.driverExec.Run(e.ctx) + if err != nil { + e.stopWith(fmt.Errorf("run executor switch: %w", err)) + return + } + + wg.Wait() + + e.callback.SetVoid() +} + +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..d1c3558 --- /dev/null +++ b/pkgs/ioswitch/exec/exec.go @@ -0,0 +1,16 @@ +package exec + +import ( + "context" +) + +type PlanID string + +type Plan struct { + ID PlanID `json:"id"` + Ops []Op `json:"ops"` +} + +type Op interface { + Execute(ctx context.Context, sw *Executor) error +} diff --git a/pkgs/ioswitch/exec/executor.go b/pkgs/ioswitch/exec/executor.go new file mode 100644 index 0000000..da4748d --- /dev/null +++ b/pkgs/ioswitch/exec/executor.go @@ -0,0 +1,151 @@ +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 +} + +func NewExecutor(plan Plan) *Executor { + planning := Executor{ + plan: plan, + vars: make(map[VarID]Var), + } + + return &planning +} + +func (s *Executor) Plan() *Plan { + return &s.plan +} + +func (s *Executor) Run(ctx context.Context) error { + ctx2, cancel := context.WithCancel(ctx) + defer cancel() + + return 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 + }) +} + +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 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..f12d27e --- /dev/null +++ b/pkgs/ioswitch/exec/plan_builder.go @@ -0,0 +1,126 @@ +package exec + +import ( + "context" + "sync" + + "gitlink.org.cn/cloudream/common/pkgs/future" + cdssdk "gitlink.org.cn/cloudream/common/sdks/storage" + "gitlink.org.cn/cloudream/common/utils/lo2" +) + +type PlanBuilder struct { + Vars []Var + WorkerPlans map[cdssdk.NodeID]*WorkerPlanBuilder + DriverPlan DriverPlanBuilder +} + +func NewPlanBuilder() *PlanBuilder { + bld := &PlanBuilder{ + WorkerPlans: make(map[cdssdk.NodeID]*WorkerPlanBuilder), + DriverPlan: DriverPlanBuilder{ + StoreMap: &sync.Map{}, + }, + } + + return bld +} + +func (b *PlanBuilder) AtExecutor() *DriverPlanBuilder { + return &b.DriverPlan +} + +func (b *PlanBuilder) AtAgent(node cdssdk.Node) *WorkerPlanBuilder { + agtPlan, ok := b.WorkerPlans[node.NodeID] + if !ok { + agtPlan = &WorkerPlanBuilder{ + Node: node, + } + b.WorkerPlans[node.NodeID] = agtPlan + } + + return agtPlan +} + +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.NewSetVoid(), + ctx: ctx, + cancel: cancel, + driverExec: NewExecutor(execPlan), + } + go exec.execute() + + return &exec +} + +type WorkerPlanBuilder struct { + Node cdssdk.Node + 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 + StoreMap *sync.Map +} + +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..0a5d9d1 --- /dev/null +++ b/pkgs/ioswitch/exec/worker.go @@ -0,0 +1,85 @@ +package exec + +import ( + "context" + "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 +} diff --git a/pkgs/ioswitch/parser/fromto.go b/pkgs/ioswitch/parser/fromto.go new file mode 100644 index 0000000..5eaa0c6 --- /dev/null +++ b/pkgs/ioswitch/parser/fromto.go @@ -0,0 +1,26 @@ +package parser + +type From interface{} + +type To interface{} + +type FromTos []FromTo + +type FromTo struct { + Froms []From + Toes []To +} + +func NewFromTo() FromTo { + return FromTo{} +} + +func (ft *FromTo) AddFrom(from From) *FromTo { + ft.Froms = append(ft.Froms, from) + return ft +} + +func (ft *FromTo) AddTo(to To) *FromTo { + ft.Toes = append(ft.Toes, to) + return ft +} diff --git a/pkgs/ioswitch/parser/parser.go b/pkgs/ioswitch/parser/parser.go new file mode 100644 index 0000000..9991c96 --- /dev/null +++ b/pkgs/ioswitch/parser/parser.go @@ -0,0 +1,9 @@ +package parser + +import ( + "gitlink.org.cn/cloudream/common/pkgs/ioswitch/exec" +) + +type FromToParser interface { + Parse(ft FromTo, blder *exec.PlanBuilder) error +}