优化ioswitch

agent/internal/services/grpc/io.go

@@ -0,0 +1,136 @@
+package grpc
+
+import (
+	"fmt"
+	"io"
+
+	"gitlink.org.cn/cloudream/common/pkgs/logger"
+	myio "gitlink.org.cn/cloudream/common/utils/io"
+	agentserver "gitlink.org.cn/cloudream/storage/common/pkgs/grpc/agent"
+	"gitlink.org.cn/cloudream/storage/common/pkgs/ioswitch"
+)
+
+func (s *Service) SendStream(server agentserver.Agent_SendStreamServer) error {
+	msg, err := server.Recv()
+	if err != nil {
+		return fmt.Errorf("recving stream id packet: %w", err)
+	}
+	if msg.Type != agentserver.StreamDataPacketType_SendArgs {
+		return fmt.Errorf("first packet must be a SendArgs packet")
+	}
+
+	logger.
+		WithField("PlanID", msg.PlanID).
+		WithField("StreamID", msg.StreamID).
+		Debugf("receive stream from grpc")
+
+	pr, pw := io.Pipe()
+
+	s.sw.StreamReady(ioswitch.PlanID(msg.PlanID), ioswitch.NewStream(ioswitch.StreamID(msg.StreamID), pr))
+
+	// 然后读取文件数据
+	var recvSize int64
+	for {
+		msg, err := server.Recv()
+
+		// 读取客户端数据失败
+		// 即使err是io.EOF，只要没有收到客户端包含EOF数据包就被断开了连接，就认为接收失败
+		if err != nil {
+			// 关闭文件写入，不需要返回的hash和error
+			pw.CloseWithError(io.ErrClosedPipe)
+			logger.WithField("ReceiveSize", recvSize).
+				Warnf("recv message failed, err: %s", err.Error())
+			return fmt.Errorf("recv message failed, err: %w", err)
+		}
+
+		err = myio.WriteAll(pw, msg.Data)
+		if err != nil {
+			// 关闭文件写入，不需要返回的hash和error
+			pw.CloseWithError(io.ErrClosedPipe)
+			logger.Warnf("write data to file failed, err: %s", err.Error())
+			return fmt.Errorf("write data to file failed, err: %w", err)
+		}
+
+		recvSize += int64(len(msg.Data))
+
+		if msg.Type == agentserver.StreamDataPacketType_EOF {
+			// 客户端明确说明文件传输已经结束，那么结束写入，获得文件Hash
+			err := pw.Close()
+			if err != nil {
+				logger.Warnf("finish writing failed, err: %s", err.Error())
+				return fmt.Errorf("finish writing failed, err: %w", err)
+			}
+
+			// 并将结果返回到客户端
+			err = server.SendAndClose(&agentserver.SendStreamResp{})
+			if err != nil {
+				logger.Warnf("send response failed, err: %s", err.Error())
+				return fmt.Errorf("send response failed, err: %w", err)
+			}
+
+			return nil
+		}
+	}
+}
+
+func (s *Service) FetchStream(req *agentserver.FetchStreamReq, server agentserver.Agent_FetchStreamServer) error {
+	logger.
+		WithField("PlanID", req.PlanID).
+		WithField("StreamID", req.StreamID).
+		Debugf("send stream by grpc")
+
+	strs, err := s.sw.WaitStreams(ioswitch.PlanID(req.PlanID), ioswitch.StreamID(req.StreamID))
+	if err != nil {
+		logger.
+			WithField("PlanID", req.PlanID).
+			WithField("StreamID", req.StreamID).
+			Warnf("watting stream: %s", err.Error())
+		return fmt.Errorf("watting stream: %w", err)
+	}
+
+	reader := strs[0].Stream
+	defer reader.Close()
+
+	buf := make([]byte, 4096)
+	readAllCnt := 0
+	for {
+		readCnt, err := reader.Read(buf)
+
+		if readCnt > 0 {
+			readAllCnt += readCnt
+			err = server.Send(&agentserver.StreamDataPacket{
+				Type: agentserver.StreamDataPacketType_Data,
+				Data: buf[:readCnt],
+			})
+			if err != nil {
+				logger.
+					WithField("PlanID", req.PlanID).
+					WithField("StreamID", req.StreamID).
+					Warnf("send stream data failed, err: %s", err.Error())
+				return fmt.Errorf("send stream data failed, err: %w", err)
+			}
+		}
+
+		// 文件读取完毕
+		if err == io.EOF {
+			logger.
+				WithField("PlanID", req.PlanID).
+				WithField("StreamID", req.StreamID).
+				Debugf("send data size %d", readAllCnt)
+			// 发送EOF消息
+			server.Send(&agentserver.StreamDataPacket{
+				Type: agentserver.StreamDataPacketType_EOF,
+			})
+			return nil
+		}
+
+		// io.ErrUnexpectedEOF没有读满整个buf就遇到了EOF，此时正常发送剩余数据即可。除了这两个错误之外，其他错误都中断操作
+		if err != nil && err != io.ErrUnexpectedEOF {
+			logger.
+				WithField("PlanID", req.PlanID).
+				WithField("StreamID", req.StreamID).
+				Warnf("reading stream data: %s", err.Error())
+			return fmt.Errorf("reading stream data: %w", err)
+		}
+	}
+}

agent/internal/services/grpc/service.go

@@ -8,14 +8,18 @@ import (
 	myio "gitlink.org.cn/cloudream/common/utils/io"
 	stgglb "gitlink.org.cn/cloudream/storage/common/globals"
 	agentserver "gitlink.org.cn/cloudream/storage/common/pkgs/grpc/agent"
+	"gitlink.org.cn/cloudream/storage/common/pkgs/ioswitch"
 )
 
 type Service struct {
 	agentserver.AgentServer
+	sw *ioswitch.Switch
 }
 
-func NewService() *Service {
-	return &Service{}
+func NewService(sw *ioswitch.Switch) *Service {
+	return &Service{
+		sw: sw,
+	}
 }
 
 func (s *Service) SendIPFSFile(server agentserver.Agent_SendIPFSFileServer) error {
@@ -59,7 +63,7 @@ func (s *Service) SendIPFSFile(server agentserver.Agent_SendIPFSFileServer) erro
 
 		recvSize += int64(len(msg.Data))
 
-		if msg.Type == agentserver.FileDataPacketType_EOF {
+		if msg.Type == agentserver.StreamDataPacketType_EOF {
 			// 客户端明确说明文件传输已经结束，那么结束写入，获得文件Hash
 			hash, err := writer.Finish()
 			if err != nil {
@@ -106,7 +110,7 @@ func (s *Service) GetIPFSFile(req *agentserver.GetIPFSFileReq, server agentserve
 		if readCnt > 0 {
 			readAllCnt += readCnt
 			err = server.Send(&agentserver.FileDataPacket{
-				Type: agentserver.FileDataPacketType_Data,
+				Type: agentserver.StreamDataPacketType_Data,
 				Data: buf[:readCnt],
 			})
 			if err != nil {
@@ -121,7 +125,7 @@ func (s *Service) GetIPFSFile(req *agentserver.GetIPFSFileReq, server agentserve
 			log.WithField("FileHash", req.FileHash).Debugf("send data size %d", readAllCnt)
 			// 发送EOF消息
 			server.Send(&agentserver.FileDataPacket{
-				Type: agentserver.FileDataPacketType_EOF,
+				Type: agentserver.StreamDataPacketType_EOF,
 			})
 			return nil
 		}

agent/internal/services/mq/cache.go

@@ -140,7 +140,7 @@ func (svc *Service) WaitCacheMovePackage(msg *agtmq.WaitCacheMovePackage) (*agtm
 		return mq.ReplyOK(agtmq.NewWaitCacheMovePackageResp(true, errMsg, mvPkgTask.ResultCacheInfos))
 
 	} else {
-		if tsk.WaitTimeout(time.Duration(msg.WaitTimeoutMs)) {
+		if tsk.WaitTimeout(time.Duration(msg.WaitTimeoutMs) * time.Millisecond) {
 
 			errMsg := ""
 			if tsk.Error() != nil {

agent/internal/services/mq/io.go

@@ -0,0 +1,65 @@
+package mq
+
+import (
+	"time"
+
+	"gitlink.org.cn/cloudream/common/consts/errorcode"
+	"gitlink.org.cn/cloudream/common/pkgs/logger"
+	"gitlink.org.cn/cloudream/common/pkgs/mq"
+	mytask "gitlink.org.cn/cloudream/storage/agent/internal/task"
+	"gitlink.org.cn/cloudream/storage/common/pkgs/ioswitch"
+	agtmq "gitlink.org.cn/cloudream/storage/common/pkgs/mq/agent"
+)
+
+func (svc *Service) SetupIOPlan(msg *agtmq.SetupIOPlan) (*agtmq.SetupIOPlanResp, *mq.CodeMessage) {
+	err := svc.sw.SetupPlan(msg.Plan)
+	if err != nil {
+		logger.WithField("PlanID", msg.Plan.ID).Warnf("adding plan: %s", err.Error())
+		return nil, mq.Failed(errorcode.OperationFailed, "adding plan failed")
+	}
+
+	return mq.ReplyOK(agtmq.NewSetupIOPlanResp())
+}
+
+func (svc *Service) StartIOPlan(msg *agtmq.StartIOPlan) (*agtmq.StartIOPlanResp, *mq.CodeMessage) {
+	tsk := svc.taskManager.StartNew(mytask.NewExecuteIOPlan(msg.PlanID))
+	return mq.ReplyOK(agtmq.NewStartIOPlanResp(tsk.ID()))
+}
+
+func (svc *Service) WaitIOPlan(msg *agtmq.WaitIOPlan) (*agtmq.WaitIOPlanResp, *mq.CodeMessage) {
+	tsk := svc.taskManager.FindByID(msg.TaskID)
+	if tsk == nil {
+		return nil, mq.Failed(errorcode.TaskNotFound, "task not found")
+	}
+
+	if msg.WaitTimeoutMs == 0 {
+		tsk.Wait()
+
+		errMsg := ""
+		if tsk.Error() != nil {
+			errMsg = tsk.Error().Error()
+		}
+
+		planTsk := tsk.Body().(*mytask.ExecuteIOPlan)
+		return mq.ReplyOK(agtmq.NewWaitIOPlanResp(true, errMsg, planTsk.Result))
+
+	} else {
+		if tsk.WaitTimeout(time.Duration(msg.WaitTimeoutMs) * time.Millisecond) {
+
+			errMsg := ""
+			if tsk.Error() != nil {
+				errMsg = tsk.Error().Error()
+			}
+
+			planTsk := tsk.Body().(*mytask.ExecuteIOPlan)
+			return mq.ReplyOK(agtmq.NewWaitIOPlanResp(true, errMsg, planTsk.Result))
+		}
+
+		return mq.ReplyOK(agtmq.NewWaitIOPlanResp(false, "", ioswitch.PlanResult{}))
+	}
+}
+
+func (svc *Service) CancelIOPlan(msg *agtmq.CancelIOPlan) (*agtmq.CancelIOPlanResp, *mq.CodeMessage) {
+	svc.sw.CancelPlan(msg.PlanID)
+	return mq.ReplyOK(agtmq.NewCancelIOPlanResp())
+}

agent/internal/services/mq/object.go

@@ -43,7 +43,7 @@ func (svc *Service) WaitPinningObject(msg *agtmq.WaitPinningObject) (*agtmq.Wait
 		return mq.ReplyOK(agtmq.NewWaitPinningObjectResp(true, errMsg))
 
 	} else {
-		if tsk.WaitTimeout(time.Duration(msg.WaitTimeoutMs)) {
+		if tsk.WaitTimeout(time.Duration(msg.WaitTimeoutMs) * time.Millisecond) {
 
 			errMsg := ""
 			if tsk.Error() != nil {

agent/internal/services/mq/service.go

@@ -2,14 +2,17 @@ package mq
 
 import (
 	"gitlink.org.cn/cloudream/storage/agent/internal/task"
+	"gitlink.org.cn/cloudream/storage/common/pkgs/ioswitch"
 )
 
 type Service struct {
 	taskManager *task.Manager
+	sw          *ioswitch.Switch
 }
 
-func NewService(taskMgr *task.Manager) *Service {
+func NewService(taskMgr *task.Manager, sw *ioswitch.Switch) *Service {
 	return &Service{
 		taskManager: taskMgr,
+		sw:          sw,
 	}
 }

agent/internal/services/mq/storage.go

@@ -69,7 +69,7 @@ func (svc *Service) WaitStorageLoadPackage(msg *agtmq.WaitStorageLoadPackage) (*
 		return mq.ReplyOK(agtmq.NewWaitStorageLoadPackageResp(true, errMsg, loadTsk.FullPath))
 
 	} else {
-		if tsk.WaitTimeout(time.Duration(msg.WaitTimeoutMs)) {
+		if tsk.WaitTimeout(time.Duration(msg.WaitTimeoutMs) * time.Millisecond) {
 
 			errMsg := ""
 			if tsk.Error() != nil {
@@ -227,7 +227,7 @@ func (svc *Service) WaitStorageCreatePackage(msg *agtmq.WaitStorageCreatePackage
 
 	if msg.WaitTimeoutMs == 0 {
 		tsk.Wait()
-	} else if !tsk.WaitTimeout(time.Duration(msg.WaitTimeoutMs)) {
+	} else if !tsk.WaitTimeout(time.Duration(msg.WaitTimeoutMs) * time.Millisecond) {
 		return mq.ReplyOK(agtmq.NewWaitStorageCreatePackageResp(false, "", 0))
 	}
 

agent/internal/task/execute_io_plan.go

@@ -0,0 +1,45 @@
+package task
+
+import (
+	"fmt"
+	"time"
+
+	"gitlink.org.cn/cloudream/common/pkgs/logger"
+	"gitlink.org.cn/cloudream/common/pkgs/task"
+	"gitlink.org.cn/cloudream/storage/common/pkgs/ioswitch"
+)
+
+// TODO 临时使用Task来等待Plan执行进度
+type ExecuteIOPlan struct {
+	PlanID ioswitch.PlanID
+	Result ioswitch.PlanResult
+}
+
+func NewExecuteIOPlan(planID ioswitch.PlanID) *ExecuteIOPlan {
+	return &ExecuteIOPlan{
+		PlanID: planID,
+	}
+}
+
+func (t *ExecuteIOPlan) Execute(task *task.Task[TaskContext], ctx TaskContext, complete CompleteFn) {
+	log := logger.WithType[ExecuteIOPlan]("Task")
+	log.Debugf("begin with %v", logger.FormatStruct(t))
+	defer log.Debugf("end")
+
+	ret, err := ctx.sw.ExecutePlan(t.PlanID)
+	if err != nil {
+		err := fmt.Errorf("executing io plan: %w", err)
+		log.WithField("PlanID", t.PlanID).Warn(err.Error())
+
+		complete(err, CompleteOption{
+			RemovingDelay: time.Minute,
+		})
+		return
+	}
+
+	t.Result = ret
+
+	complete(nil, CompleteOption{
+		RemovingDelay: time.Minute,
+	})
+}

agent/internal/task/task.go

@@ -3,10 +3,12 @@ package task
 import (
 	"gitlink.org.cn/cloudream/common/pkgs/distlock"
 	"gitlink.org.cn/cloudream/common/pkgs/task"
+	"gitlink.org.cn/cloudream/storage/common/pkgs/ioswitch"
 )
 
 type TaskContext struct {
 	distlock *distlock.Service
+	sw       *ioswitch.Switch
 }
 
 // 需要在Task结束后主动调用，completing函数将在Manager加锁期间被调用，
@@ -21,8 +23,9 @@ type Task = task.Task[TaskContext]
 
 type CompleteOption = task.CompleteOption
 
-func NewManager(distlock *distlock.Service) Manager {
+func NewManager(distlock *distlock.Service, sw *ioswitch.Switch) Manager {
 	return task.NewManager(TaskContext{
 		distlock: distlock,
+		sw:       sw,
 	})
 }

agent/main.go

@@ -13,6 +13,10 @@ import (
 	stgglb "gitlink.org.cn/cloudream/storage/common/globals"
 	"gitlink.org.cn/cloudream/storage/common/pkgs/distlock"
 	agtrpc "gitlink.org.cn/cloudream/storage/common/pkgs/grpc/agent"
+	"gitlink.org.cn/cloudream/storage/common/pkgs/ioswitch"
+
+	// TODO 注册OpUnion，但在mq包中注册会造成循环依赖，所以只能放到这里
+	_ "gitlink.org.cn/cloudream/storage/common/pkgs/ioswitch/ops"
 
 	"google.golang.org/grpc"
 
@@ -55,15 +59,17 @@ func main() {
 		log.Fatalf("new ipfs failed, err: %s", err.Error())
 	}
 
+	sw := ioswitch.NewSwitch()
+
 	//处置协调端、客户端命令（可多建几个）
 	wg := sync.WaitGroup{}
 	wg.Add(5)
 
-	taskMgr := task.NewManager(distlock)
+	taskMgr := task.NewManager(distlock, &sw)
 
 	// 启动命令服务器
 	// TODO 需要设计AgentID持久化机制
-	agtSvr, err := agtmq.NewServer(cmdsvc.NewService(&taskMgr), config.Cfg().ID, &config.Cfg().RabbitMQ)
+	agtSvr, err := agtmq.NewServer(cmdsvc.NewService(&taskMgr, &sw), config.Cfg().ID, &config.Cfg().RabbitMQ)
 	if err != nil {
 		log.Fatalf("new agent server failed, err: %s", err.Error())
 	}
@@ -83,7 +89,7 @@ func main() {
 	}
 
 	s := grpc.NewServer()
-	agtrpc.RegisterAgentServer(s, grpcsvc.NewService())
+	agtrpc.RegisterAgentServer(s, grpcsvc.NewService(&sw))
 	go serveGRPC(s, lis, &wg)
 
 	go serveDistLock(distlock)

common/models/models.go

@@ -31,10 +31,10 @@ func NewRedundancyEcData(ec EC, blocks []ObjectBlockData) ECRedundancyData {
 }
 
 type EC struct {
-	ID   int    `json:"id"`
-	Name string `json:"name"`
-	EcK  int    `json:"ecK"`
-	EcN  int    `json:"ecN"`
+	ID        int64 `json:"id"`
+	K         int   `json:"k"`
+	N         int   `json:"n"`
+	ChunkSize int   `json:"chunkSize"`
 }
 
 type ObjectBlockData struct {
@@ -51,12 +51,12 @@ func NewObjectBlockData(index int, fileHash string, nodeIDs []int64) ObjectBlock
 	}
 }
 
-func NewEc(id int, name string, ecK int, ecN int) EC {
+func NewEc(id int64, k int, n int, chunkSize int) EC {
 	return EC{
-		ID:   id,
-		Name: name,
-		EcK:  ecK,
-		EcN:  ecN,
+		ID:        id,
+		K:         k,
+		N:         n,
+		ChunkSize: chunkSize,
 	}
 }
 

common/pkgs/grpc/agent/agent.pb.go

@@ -22,49 +22,52 @@ const (
 	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
 )
 
-type FileDataPacketType int32
+type StreamDataPacketType int32
 
 const (
-	FileDataPacketType_Data FileDataPacketType = 0
-	FileDataPacketType_EOF  FileDataPacketType = 1
+	StreamDataPacketType_EOF      StreamDataPacketType = 0
+	StreamDataPacketType_Data     StreamDataPacketType = 1
+	StreamDataPacketType_SendArgs StreamDataPacketType = 2
 )
 
-// Enum value maps for FileDataPacketType.
+// Enum value maps for StreamDataPacketType.
 var (
-	FileDataPacketType_name = map[int32]string{
-		0: "Data",
-		1: "EOF",
+	StreamDataPacketType_name = map[int32]string{
+		0: "EOF",
+		1: "Data",
+		2: "SendArgs",
 	}
-	FileDataPacketType_value = map[string]int32{
-		"Data": 0,
-		"EOF":  1,
+	StreamDataPacketType_value = map[string]int32{
+		"EOF":      0,
+		"Data":     1,
+		"SendArgs": 2,
 	}
 )
 
-func (x FileDataPacketType) Enum() *FileDataPacketType {
-	p := new(FileDataPacketType)
+func (x StreamDataPacketType) Enum() *StreamDataPacketType {
+	p := new(StreamDataPacketType)
 	*p = x
 	return p
 }
 
-func (x FileDataPacketType) String() string {
+func (x StreamDataPacketType) String() string {
 	return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x))
 }
 
-func (FileDataPacketType) Descriptor() protoreflect.EnumDescriptor {
+func (StreamDataPacketType) Descriptor() protoreflect.EnumDescriptor {
 	return file_pkgs_grpc_agent_agent_proto_enumTypes[0].Descriptor()
 }
 
-func (FileDataPacketType) Type() protoreflect.EnumType {
+func (StreamDataPacketType) Type() protoreflect.EnumType {
 	return &file_pkgs_grpc_agent_agent_proto_enumTypes[0]
 }
 
-func (x FileDataPacketType) Number() protoreflect.EnumNumber {
+func (x StreamDataPacketType) Number() protoreflect.EnumNumber {
 	return protoreflect.EnumNumber(x)
 }
 
-// Deprecated: Use FileDataPacketType.Descriptor instead.
-func (FileDataPacketType) EnumDescriptor() ([]byte, []int) {
+// Deprecated: Use StreamDataPacketType.Descriptor instead.
+func (StreamDataPacketType) EnumDescriptor() ([]byte, []int) {
 	return file_pkgs_grpc_agent_agent_proto_rawDescGZIP(), []int{0}
 }
 
@@ -74,8 +77,8 @@ type FileDataPacket struct {
 	sizeCache     protoimpl.SizeCache
 	unknownFields protoimpl.UnknownFields
 
-	Type FileDataPacketType `protobuf:"varint,1,opt,name=Type,proto3,enum=FileDataPacketType" json:"Type,omitempty"`
-	Data []byte             `protobuf:"bytes,2,opt,name=Data,proto3" json:"Data,omitempty"`
+	Type StreamDataPacketType `protobuf:"varint,1,opt,name=Type,proto3,enum=StreamDataPacketType" json:"Type,omitempty"`
+	Data []byte               `protobuf:"bytes,2,opt,name=Data,proto3" json:"Data,omitempty"`
 }
 
 func (x *FileDataPacket) Reset() {
@@ -110,11 +113,11 @@ func (*FileDataPacket) Descriptor() ([]byte, []int) {
 	return file_pkgs_grpc_agent_agent_proto_rawDescGZIP(), []int{0}
 }
 
-func (x *FileDataPacket) GetType() FileDataPacketType {
+func (x *FileDataPacket) GetType() StreamDataPacketType {
 	if x != nil {
 		return x.Type
 	}
-	return FileDataPacketType_Data
+	return StreamDataPacketType_EOF
 }
 
 func (x *FileDataPacket) GetData() []byte {
@@ -218,31 +221,218 @@ func (x *GetIPFSFileReq) GetFileHash() string {
 	return ""
 }
 
+// 注：EOF时data也可能有数据
+type StreamDataPacket struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Type     StreamDataPacketType `protobuf:"varint,1,opt,name=Type,proto3,enum=StreamDataPacketType" json:"Type,omitempty"`
+	PlanID   string               `protobuf:"bytes,2,opt,name=PlanID,proto3" json:"PlanID,omitempty"`
+	StreamID string               `protobuf:"bytes,3,opt,name=StreamID,proto3" json:"StreamID,omitempty"`
+	Data     []byte               `protobuf:"bytes,4,opt,name=Data,proto3" json:"Data,omitempty"`
+}
+
+func (x *StreamDataPacket) Reset() {
+	*x = StreamDataPacket{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_pkgs_grpc_agent_agent_proto_msgTypes[3]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *StreamDataPacket) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*StreamDataPacket) ProtoMessage() {}
+
+func (x *StreamDataPacket) ProtoReflect() protoreflect.Message {
+	mi := &file_pkgs_grpc_agent_agent_proto_msgTypes[3]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use StreamDataPacket.ProtoReflect.Descriptor instead.
+func (*StreamDataPacket) Descriptor() ([]byte, []int) {
+	return file_pkgs_grpc_agent_agent_proto_rawDescGZIP(), []int{3}
+}
+
+func (x *StreamDataPacket) GetType() StreamDataPacketType {
+	if x != nil {
+		return x.Type
+	}
+	return StreamDataPacketType_EOF
+}
+
+func (x *StreamDataPacket) GetPlanID() string {
+	if x != nil {
+		return x.PlanID
+	}
+	return ""
+}
+
+func (x *StreamDataPacket) GetStreamID() string {
+	if x != nil {
+		return x.StreamID
+	}
+	return ""
+}
+
+func (x *StreamDataPacket) GetData() []byte {
+	if x != nil {
+		return x.Data
+	}
+	return nil
+}
+
+type SendStreamResp struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+}
+
+func (x *SendStreamResp) Reset() {
+	*x = SendStreamResp{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_pkgs_grpc_agent_agent_proto_msgTypes[4]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *SendStreamResp) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*SendStreamResp) ProtoMessage() {}
+
+func (x *SendStreamResp) ProtoReflect() protoreflect.Message {
+	mi := &file_pkgs_grpc_agent_agent_proto_msgTypes[4]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use SendStreamResp.ProtoReflect.Descriptor instead.
+func (*SendStreamResp) Descriptor() ([]byte, []int) {
+	return file_pkgs_grpc_agent_agent_proto_rawDescGZIP(), []int{4}
+}
+
+type FetchStreamReq struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	PlanID   string `protobuf:"bytes,1,opt,name=PlanID,proto3" json:"PlanID,omitempty"`
+	StreamID string `protobuf:"bytes,2,opt,name=StreamID,proto3" json:"StreamID,omitempty"`
+}
+
+func (x *FetchStreamReq) Reset() {
+	*x = FetchStreamReq{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_pkgs_grpc_agent_agent_proto_msgTypes[5]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *FetchStreamReq) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*FetchStreamReq) ProtoMessage() {}
+
+func (x *FetchStreamReq) ProtoReflect() protoreflect.Message {
+	mi := &file_pkgs_grpc_agent_agent_proto_msgTypes[5]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use FetchStreamReq.ProtoReflect.Descriptor instead.
+func (*FetchStreamReq) Descriptor() ([]byte, []int) {
+	return file_pkgs_grpc_agent_agent_proto_rawDescGZIP(), []int{5}
+}
+
+func (x *FetchStreamReq) GetPlanID() string {
+	if x != nil {
+		return x.PlanID
+	}
+	return ""
+}
+
+func (x *FetchStreamReq) GetStreamID() string {
+	if x != nil {
+		return x.StreamID
+	}
+	return ""
+}
+
 var File_pkgs_grpc_agent_agent_proto protoreflect.FileDescriptor
 
 var file_pkgs_grpc_agent_agent_proto_rawDesc = []byte{
 	0x0a, 0x1b, 0x70, 0x6b, 0x67, 0x73, 0x2f, 0x67, 0x72, 0x70, 0x63, 0x2f, 0x61, 0x67, 0x65, 0x6e,
-	0x74, 0x2f, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0x4d, 0x0a,
+	0x74, 0x2f, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0x4f, 0x0a,
 	0x0e, 0x46, 0x69, 0x6c, 0x65, 0x44, 0x61, 0x74, 0x61, 0x50, 0x61, 0x63, 0x6b, 0x65, 0x74, 0x12,
-	0x27, 0x0a, 0x04, 0x54, 0x79, 0x70, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x13, 0x2e,
-	0x46, 0x69, 0x6c, 0x65, 0x44, 0x61, 0x74, 0x61, 0x50, 0x61, 0x63, 0x6b, 0x65, 0x74, 0x54, 0x79,
-	0x70, 0x65, 0x52, 0x04, 0x54, 0x79, 0x70, 0x65, 0x12, 0x12, 0x0a, 0x04, 0x44, 0x61, 0x74, 0x61,
-	0x18, 0x02, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x04, 0x44, 0x61, 0x74, 0x61, 0x22, 0x2e, 0x0a, 0x10,
-	0x53, 0x65, 0x6e, 0x64, 0x49, 0x50, 0x46, 0x53, 0x46, 0x69, 0x6c, 0x65, 0x52, 0x65, 0x73, 0x70,
+	0x29, 0x0a, 0x04, 0x54, 0x79, 0x70, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x15, 0x2e,
+	0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x44, 0x61, 0x74, 0x61, 0x50, 0x61, 0x63, 0x6b, 0x65, 0x74,
+	0x54, 0x79, 0x70, 0x65, 0x52, 0x04, 0x54, 0x79, 0x70, 0x65, 0x12, 0x12, 0x0a, 0x04, 0x44, 0x61,
+	0x74, 0x61, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x04, 0x44, 0x61, 0x74, 0x61, 0x22, 0x2e,
+	0x0a, 0x10, 0x53, 0x65, 0x6e, 0x64, 0x49, 0x50, 0x46, 0x53, 0x46, 0x69, 0x6c, 0x65, 0x52, 0x65,
+	0x73, 0x70, 0x12, 0x1a, 0x0a, 0x08, 0x46, 0x69, 0x6c, 0x65, 0x48, 0x61, 0x73, 0x68, 0x18, 0x01,
+	0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x46, 0x69, 0x6c, 0x65, 0x48, 0x61, 0x73, 0x68, 0x22, 0x2c,
+	0x0a, 0x0e, 0x47, 0x65, 0x74, 0x49, 0x50, 0x46, 0x53, 0x46, 0x69, 0x6c, 0x65, 0x52, 0x65, 0x71,
 	0x12, 0x1a, 0x0a, 0x08, 0x46, 0x69, 0x6c, 0x65, 0x48, 0x61, 0x73, 0x68, 0x18, 0x01, 0x20, 0x01,
-	0x28, 0x09, 0x52, 0x08, 0x46, 0x69, 0x6c, 0x65, 0x48, 0x61, 0x73, 0x68, 0x22, 0x2c, 0x0a, 0x0e,
-	0x47, 0x65, 0x74, 0x49, 0x50, 0x46, 0x53, 0x46, 0x69, 0x6c, 0x65, 0x52, 0x65, 0x71, 0x12, 0x1a,
-	0x0a, 0x08, 0x46, 0x69, 0x6c, 0x65, 0x48, 0x61, 0x73, 0x68, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09,
-	0x52, 0x08, 0x46, 0x69, 0x6c, 0x65, 0x48, 0x61, 0x73, 0x68, 0x2a, 0x27, 0x0a, 0x12, 0x46, 0x69,
-	0x6c, 0x65, 0x44, 0x61, 0x74, 0x61, 0x50, 0x61, 0x63, 0x6b, 0x65, 0x74, 0x54, 0x79, 0x70, 0x65,
-	0x12, 0x08, 0x0a, 0x04, 0x44, 0x61, 0x74, 0x61, 0x10, 0x00, 0x12, 0x07, 0x0a, 0x03, 0x45, 0x4f,
-	0x46, 0x10, 0x01, 0x32, 0x74, 0x0a, 0x05, 0x41, 0x67, 0x65, 0x6e, 0x74, 0x12, 0x36, 0x0a, 0x0c,
-	0x53, 0x65, 0x6e, 0x64, 0x49, 0x50, 0x46, 0x53, 0x46, 0x69, 0x6c, 0x65, 0x12, 0x0f, 0x2e, 0x46,
-	0x69, 0x6c, 0x65, 0x44, 0x61, 0x74, 0x61, 0x50, 0x61, 0x63, 0x6b, 0x65, 0x74, 0x1a, 0x11, 0x2e,
-	0x53, 0x65, 0x6e, 0x64, 0x49, 0x50, 0x46, 0x53, 0x46, 0x69, 0x6c, 0x65, 0x52, 0x65, 0x73, 0x70,
-	0x22, 0x00, 0x28, 0x01, 0x12, 0x33, 0x0a, 0x0b, 0x47, 0x65, 0x74, 0x49, 0x50, 0x46, 0x53, 0x46,
-	0x69, 0x6c, 0x65, 0x12, 0x0f, 0x2e, 0x47, 0x65, 0x74, 0x49, 0x50, 0x46, 0x53, 0x46, 0x69, 0x6c,
-	0x65, 0x52, 0x65, 0x71, 0x1a, 0x0f, 0x2e, 0x46, 0x69, 0x6c, 0x65, 0x44, 0x61, 0x74, 0x61, 0x50,
+	0x28, 0x09, 0x52, 0x08, 0x46, 0x69, 0x6c, 0x65, 0x48, 0x61, 0x73, 0x68, 0x22, 0x85, 0x01, 0x0a,
+	0x10, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x44, 0x61, 0x74, 0x61, 0x50, 0x61, 0x63, 0x6b, 0x65,
+	0x74, 0x12, 0x29, 0x0a, 0x04, 0x54, 0x79, 0x70, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0e, 0x32,
+	0x15, 0x2e, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x44, 0x61, 0x74, 0x61, 0x50, 0x61, 0x63, 0x6b,
+	0x65, 0x74, 0x54, 0x79, 0x70, 0x65, 0x52, 0x04, 0x54, 0x79, 0x70, 0x65, 0x12, 0x16, 0x0a, 0x06,
+	0x50, 0x6c, 0x61, 0x6e, 0x49, 0x44, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x50, 0x6c,
+	0x61, 0x6e, 0x49, 0x44, 0x12, 0x1a, 0x0a, 0x08, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x49, 0x44,
+	0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x49, 0x44,
+	0x12, 0x12, 0x0a, 0x04, 0x44, 0x61, 0x74, 0x61, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x04,
+	0x44, 0x61, 0x74, 0x61, 0x22, 0x10, 0x0a, 0x0e, 0x53, 0x65, 0x6e, 0x64, 0x53, 0x74, 0x72, 0x65,
+	0x61, 0x6d, 0x52, 0x65, 0x73, 0x70, 0x22, 0x44, 0x0a, 0x0e, 0x46, 0x65, 0x74, 0x63, 0x68, 0x53,
+	0x74, 0x72, 0x65, 0x61, 0x6d, 0x52, 0x65, 0x71, 0x12, 0x16, 0x0a, 0x06, 0x50, 0x6c, 0x61, 0x6e,
+	0x49, 0x44, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x50, 0x6c, 0x61, 0x6e, 0x49, 0x44,
+	0x12, 0x1a, 0x0a, 0x08, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x49, 0x44, 0x18, 0x02, 0x20, 0x01,
+	0x28, 0x09, 0x52, 0x08, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x49, 0x44, 0x2a, 0x37, 0x0a, 0x14,
+	0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x44, 0x61, 0x74, 0x61, 0x50, 0x61, 0x63, 0x6b, 0x65, 0x74,
+	0x54, 0x79, 0x70, 0x65, 0x12, 0x07, 0x0a, 0x03, 0x45, 0x4f, 0x46, 0x10, 0x00, 0x12, 0x08, 0x0a,
+	0x04, 0x44, 0x61, 0x74, 0x61, 0x10, 0x01, 0x12, 0x0c, 0x0a, 0x08, 0x53, 0x65, 0x6e, 0x64, 0x41,
+	0x72, 0x67, 0x73, 0x10, 0x02, 0x32, 0xe1, 0x01, 0x0a, 0x05, 0x41, 0x67, 0x65, 0x6e, 0x74, 0x12,
+	0x36, 0x0a, 0x0c, 0x53, 0x65, 0x6e, 0x64, 0x49, 0x50, 0x46, 0x53, 0x46, 0x69, 0x6c, 0x65, 0x12,
+	0x0f, 0x2e, 0x46, 0x69, 0x6c, 0x65, 0x44, 0x61, 0x74, 0x61, 0x50, 0x61, 0x63, 0x6b, 0x65, 0x74,
+	0x1a, 0x11, 0x2e, 0x53, 0x65, 0x6e, 0x64, 0x49, 0x50, 0x46, 0x53, 0x46, 0x69, 0x6c, 0x65, 0x52,
+	0x65, 0x73, 0x70, 0x22, 0x00, 0x28, 0x01, 0x12, 0x33, 0x0a, 0x0b, 0x47, 0x65, 0x74, 0x49, 0x50,
+	0x46, 0x53, 0x46, 0x69, 0x6c, 0x65, 0x12, 0x0f, 0x2e, 0x47, 0x65, 0x74, 0x49, 0x50, 0x46, 0x53,
+	0x46, 0x69, 0x6c, 0x65, 0x52, 0x65, 0x71, 0x1a, 0x0f, 0x2e, 0x46, 0x69, 0x6c, 0x65, 0x44, 0x61,
+	0x74, 0x61, 0x50, 0x61, 0x63, 0x6b, 0x65, 0x74, 0x22, 0x00, 0x30, 0x01, 0x12, 0x34, 0x0a, 0x0a,
+	0x53, 0x65, 0x6e, 0x64, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x12, 0x11, 0x2e, 0x53, 0x74, 0x72,
+	0x65, 0x61, 0x6d, 0x44, 0x61, 0x74, 0x61, 0x50, 0x61, 0x63, 0x6b, 0x65, 0x74, 0x1a, 0x0f, 0x2e,
+	0x53, 0x65, 0x6e, 0x64, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x52, 0x65, 0x73, 0x70, 0x22, 0x00,
+	0x28, 0x01, 0x12, 0x35, 0x0a, 0x0b, 0x46, 0x65, 0x74, 0x63, 0x68, 0x53, 0x74, 0x72, 0x65, 0x61,
+	0x6d, 0x12, 0x0f, 0x2e, 0x46, 0x65, 0x74, 0x63, 0x68, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x52,
+	0x65, 0x71, 0x1a, 0x11, 0x2e, 0x53, 0x74, 0x72, 0x65, 0x61, 0x6d, 0x44, 0x61, 0x74, 0x61, 0x50,
 	0x61, 0x63, 0x6b, 0x65, 0x74, 0x22, 0x00, 0x30, 0x01, 0x42, 0x09, 0x5a, 0x07, 0x2e, 0x3b, 0x61,
 	0x67, 0x65, 0x6e, 0x74, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
@@ -260,24 +450,32 @@ func file_pkgs_grpc_agent_agent_proto_rawDescGZIP() []byte {
 }
 
 var file_pkgs_grpc_agent_agent_proto_enumTypes = make([]protoimpl.EnumInfo, 1)
-var file_pkgs_grpc_agent_agent_proto_msgTypes = make([]protoimpl.MessageInfo, 3)
+var file_pkgs_grpc_agent_agent_proto_msgTypes = make([]protoimpl.MessageInfo, 6)
 var file_pkgs_grpc_agent_agent_proto_goTypes = []interface{}{
-	(FileDataPacketType)(0),  // 0: FileDataPacketType
-	(*FileDataPacket)(nil),   // 1: FileDataPacket
-	(*SendIPFSFileResp)(nil), // 2: SendIPFSFileResp
-	(*GetIPFSFileReq)(nil),   // 3: GetIPFSFileReq
+	(StreamDataPacketType)(0), // 0: StreamDataPacketType
+	(*FileDataPacket)(nil),    // 1: FileDataPacket
+	(*SendIPFSFileResp)(nil),  // 2: SendIPFSFileResp
+	(*GetIPFSFileReq)(nil),    // 3: GetIPFSFileReq
+	(*StreamDataPacket)(nil),  // 4: StreamDataPacket
+	(*SendStreamResp)(nil),    // 5: SendStreamResp
+	(*FetchStreamReq)(nil),    // 6: FetchStreamReq
 }
 var file_pkgs_grpc_agent_agent_proto_depIdxs = []int32{
-	0, // 0: FileDataPacket.Type:type_name -> FileDataPacketType
-	1, // 1: Agent.SendIPFSFile:input_type -> FileDataPacket
-	3, // 2: Agent.GetIPFSFile:input_type -> GetIPFSFileReq
-	2, // 3: Agent.SendIPFSFile:output_type -> SendIPFSFileResp
-	1, // 4: Agent.GetIPFSFile:output_type -> FileDataPacket
-	3, // [3:5] is the sub-list for method output_type
-	1, // [1:3] is the sub-list for method input_type
-	1, // [1:1] is the sub-list for extension type_name
-	1, // [1:1] is the sub-list for extension extendee
-	0, // [0:1] is the sub-list for field type_name
+	0, // 0: FileDataPacket.Type:type_name -> StreamDataPacketType
+	0, // 1: StreamDataPacket.Type:type_name -> StreamDataPacketType
+	1, // 2: Agent.SendIPFSFile:input_type -> FileDataPacket
+	3, // 3: Agent.GetIPFSFile:input_type -> GetIPFSFileReq
+	4, // 4: Agent.SendStream:input_type -> StreamDataPacket
+	6, // 5: Agent.FetchStream:input_type -> FetchStreamReq
+	2, // 6: Agent.SendIPFSFile:output_type -> SendIPFSFileResp
+	1, // 7: Agent.GetIPFSFile:output_type -> FileDataPacket
+	5, // 8: Agent.SendStream:output_type -> SendStreamResp
+	4, // 9: Agent.FetchStream:output_type -> StreamDataPacket
+	6, // [6:10] is the sub-list for method output_type
+	2, // [2:6] is the sub-list for method input_type
+	2, // [2:2] is the sub-list for extension type_name
+	2, // [2:2] is the sub-list for extension extendee
+	0, // [0:2] is the sub-list for field type_name
 }
 
 func init() { file_pkgs_grpc_agent_agent_proto_init() }
@@ -322,6 +520,42 @@ func file_pkgs_grpc_agent_agent_proto_init() {
 				return nil
 			}
 		}
+		file_pkgs_grpc_agent_agent_proto_msgTypes[3].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*StreamDataPacket); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_pkgs_grpc_agent_agent_proto_msgTypes[4].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*SendStreamResp); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_pkgs_grpc_agent_agent_proto_msgTypes[5].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*FetchStreamReq); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
 	}
 	type x struct{}
 	out := protoimpl.TypeBuilder{
@@ -329,7 +563,7 @@ func file_pkgs_grpc_agent_agent_proto_init() {
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: file_pkgs_grpc_agent_agent_proto_rawDesc,
 			NumEnums:      1,
-			NumMessages:   3,
+			NumMessages:   6,
 			NumExtensions: 0,
 			NumServices:   1,
 		},

common/pkgs/grpc/agent/agent.proto

@@ -5,13 +5,14 @@ syntax = "proto3";
 option go_package = ".;agent";//grpc这里生效了
 
 
-enum FileDataPacketType {
-    Data = 0;
-    EOF = 1;
+enum StreamDataPacketType {
+    EOF = 0;
+    Data = 1;
+    SendArgs = 2;
 }
 // 文件数据。注意：只在Type为Data的时候，Data字段才能有数据
 message FileDataPacket {
-    FileDataPacketType Type = 1;
+    StreamDataPacketType Type = 1;
     bytes Data = 2;
 }
 
@@ -23,8 +24,27 @@ message GetIPFSFileReq {
     string FileHash = 1;
 }
 
+// 注：EOF时data也可能有数据
+message StreamDataPacket {
+    StreamDataPacketType Type = 1;
+    string PlanID = 2;
+    string StreamID = 3;
+    bytes Data = 4;
+}
+
+message SendStreamResp {
+}
+
+message FetchStreamReq {
+    string PlanID = 1;
+    string StreamID = 2;
+}
+
 service Agent {
     rpc SendIPFSFile(stream FileDataPacket)returns(SendIPFSFileResp){}
     rpc GetIPFSFile(GetIPFSFileReq)returns(stream FileDataPacket){}
+
+    rpc SendStream(stream StreamDataPacket)returns(SendStreamResp){}
+    rpc FetchStream(FetchStreamReq)returns(stream StreamDataPacket){}
 }
 

common/pkgs/grpc/agent/agent_grpc.pb.go

@@ -23,6 +23,8 @@ const _ = grpc.SupportPackageIsVersion7
 const (
 	Agent_SendIPFSFile_FullMethodName = "/Agent/SendIPFSFile"
 	Agent_GetIPFSFile_FullMethodName  = "/Agent/GetIPFSFile"
+	Agent_SendStream_FullMethodName   = "/Agent/SendStream"
+	Agent_FetchStream_FullMethodName  = "/Agent/FetchStream"
 )
 
 // AgentClient is the client API for Agent service.
@@ -31,6 +33,8 @@ const (
 type AgentClient interface {
 	SendIPFSFile(ctx context.Context, opts ...grpc.CallOption) (Agent_SendIPFSFileClient, error)
 	GetIPFSFile(ctx context.Context, in *GetIPFSFileReq, opts ...grpc.CallOption) (Agent_GetIPFSFileClient, error)
+	SendStream(ctx context.Context, opts ...grpc.CallOption) (Agent_SendStreamClient, error)
+	FetchStream(ctx context.Context, in *FetchStreamReq, opts ...grpc.CallOption) (Agent_FetchStreamClient, error)
 }
 
 type agentClient struct {
@@ -107,12 +111,80 @@ func (x *agentGetIPFSFileClient) Recv() (*FileDataPacket, error) {
 	return m, nil
 }
 
+func (c *agentClient) SendStream(ctx context.Context, opts ...grpc.CallOption) (Agent_SendStreamClient, error) {
+	stream, err := c.cc.NewStream(ctx, &Agent_ServiceDesc.Streams[2], Agent_SendStream_FullMethodName, opts...)
+	if err != nil {
+		return nil, err
+	}
+	x := &agentSendStreamClient{stream}
+	return x, nil
+}
+
+type Agent_SendStreamClient interface {
+	Send(*StreamDataPacket) error
+	CloseAndRecv() (*SendStreamResp, error)
+	grpc.ClientStream
+}
+
+type agentSendStreamClient struct {
+	grpc.ClientStream
+}
+
+func (x *agentSendStreamClient) Send(m *StreamDataPacket) error {
+	return x.ClientStream.SendMsg(m)
+}
+
+func (x *agentSendStreamClient) CloseAndRecv() (*SendStreamResp, error) {
+	if err := x.ClientStream.CloseSend(); err != nil {
+		return nil, err
+	}
+	m := new(SendStreamResp)
+	if err := x.ClientStream.RecvMsg(m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
+
+func (c *agentClient) FetchStream(ctx context.Context, in *FetchStreamReq, opts ...grpc.CallOption) (Agent_FetchStreamClient, error) {
+	stream, err := c.cc.NewStream(ctx, &Agent_ServiceDesc.Streams[3], Agent_FetchStream_FullMethodName, opts...)
+	if err != nil {
+		return nil, err
+	}
+	x := &agentFetchStreamClient{stream}
+	if err := x.ClientStream.SendMsg(in); err != nil {
+		return nil, err
+	}
+	if err := x.ClientStream.CloseSend(); err != nil {
+		return nil, err
+	}
+	return x, nil
+}
+
+type Agent_FetchStreamClient interface {
+	Recv() (*StreamDataPacket, error)
+	grpc.ClientStream
+}
+
+type agentFetchStreamClient struct {
+	grpc.ClientStream
+}
+
+func (x *agentFetchStreamClient) Recv() (*StreamDataPacket, error) {
+	m := new(StreamDataPacket)
+	if err := x.ClientStream.RecvMsg(m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
+
 // AgentServer is the server API for Agent service.
 // All implementations must embed UnimplementedAgentServer
 // for forward compatibility
 type AgentServer interface {
 	SendIPFSFile(Agent_SendIPFSFileServer) error
 	GetIPFSFile(*GetIPFSFileReq, Agent_GetIPFSFileServer) error
+	SendStream(Agent_SendStreamServer) error
+	FetchStream(*FetchStreamReq, Agent_FetchStreamServer) error
 	mustEmbedUnimplementedAgentServer()
 }
 
@@ -126,6 +198,12 @@ func (UnimplementedAgentServer) SendIPFSFile(Agent_SendIPFSFileServer) error {
 func (UnimplementedAgentServer) GetIPFSFile(*GetIPFSFileReq, Agent_GetIPFSFileServer) error {
 	return status.Errorf(codes.Unimplemented, "method GetIPFSFile not implemented")
 }
+func (UnimplementedAgentServer) SendStream(Agent_SendStreamServer) error {
+	return status.Errorf(codes.Unimplemented, "method SendStream not implemented")
+}
+func (UnimplementedAgentServer) FetchStream(*FetchStreamReq, Agent_FetchStreamServer) error {
+	return status.Errorf(codes.Unimplemented, "method FetchStream not implemented")
+}
 func (UnimplementedAgentServer) mustEmbedUnimplementedAgentServer() {}
 
 // UnsafeAgentServer may be embedded to opt out of forward compatibility for this service.
@@ -186,6 +264,53 @@ func (x *agentGetIPFSFileServer) Send(m *FileDataPacket) error {
 	return x.ServerStream.SendMsg(m)
 }
 
+func _Agent_SendStream_Handler(srv interface{}, stream grpc.ServerStream) error {
+	return srv.(AgentServer).SendStream(&agentSendStreamServer{stream})
+}
+
+type Agent_SendStreamServer interface {
+	SendAndClose(*SendStreamResp) error
+	Recv() (*StreamDataPacket, error)
+	grpc.ServerStream
+}
+
+type agentSendStreamServer struct {
+	grpc.ServerStream
+}
+
+func (x *agentSendStreamServer) SendAndClose(m *SendStreamResp) error {
+	return x.ServerStream.SendMsg(m)
+}
+
+func (x *agentSendStreamServer) Recv() (*StreamDataPacket, error) {
+	m := new(StreamDataPacket)
+	if err := x.ServerStream.RecvMsg(m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
+
+func _Agent_FetchStream_Handler(srv interface{}, stream grpc.ServerStream) error {
+	m := new(FetchStreamReq)
+	if err := stream.RecvMsg(m); err != nil {
+		return err
+	}
+	return srv.(AgentServer).FetchStream(m, &agentFetchStreamServer{stream})
+}
+
+type Agent_FetchStreamServer interface {
+	Send(*StreamDataPacket) error
+	grpc.ServerStream
+}
+
+type agentFetchStreamServer struct {
+	grpc.ServerStream
+}
+
+func (x *agentFetchStreamServer) Send(m *StreamDataPacket) error {
+	return x.ServerStream.SendMsg(m)
+}
+
 // Agent_ServiceDesc is the grpc.ServiceDesc for Agent service.
 // It's only intended for direct use with grpc.RegisterService,
 // and not to be introspected or modified (even as a copy)
@@ -204,6 +329,16 @@ var Agent_ServiceDesc = grpc.ServiceDesc{
 			Handler:       _Agent_GetIPFSFile_Handler,
 			ServerStreams: true,
 		},
+		{
+			StreamName:    "SendStream",
+			Handler:       _Agent_SendStream_Handler,
+			ClientStreams: true,
+		},
+		{
+			StreamName:    "FetchStream",
+			Handler:       _Agent_FetchStream_Handler,
+			ServerStreams: true,
+		},
 	},
 	Metadata: "pkgs/grpc/agent/agent.proto",
 }

common/pkgs/grpc/agent/client.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"io"
 
+	"gitlink.org.cn/cloudream/storage/common/pkgs/ioswitch"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
 )
@@ -37,7 +38,7 @@ func (c *Client) SendIPFSFile(file io.Reader) (string, error) {
 		rd, err := file.Read(buf)
 		if err == io.EOF {
 			err := sendCli.Send(&FileDataPacket{
-				Type: FileDataPacketType_EOF,
+				Type: StreamDataPacketType_EOF,
 				Data: buf[:rd],
 			})
 			if err != nil {
@@ -57,7 +58,7 @@ func (c *Client) SendIPFSFile(file io.Reader) (string, error) {
 		}
 
 		err = sendCli.Send(&FileDataPacket{
-			Type: FileDataPacketType_Data,
+			Type: StreamDataPacketType_Data,
 			Data: buf[:rd],
 		})
 		if err != nil {
@@ -68,7 +69,9 @@ func (c *Client) SendIPFSFile(file io.Reader) (string, error) {
 
 type fileReadCloser struct {
 	io.ReadCloser
-	stream      Agent_GetIPFSFileClient
+	// stream      Agent_GetIPFSFileClient
+	// TODO 临时使用
+	recvFn      func() (*StreamDataPacket, error)
 	cancelFn    context.CancelFunc
 	readingData []byte
 	recvEOF     bool
@@ -76,15 +79,15 @@ type fileReadCloser struct {
 
 func (s *fileReadCloser) Read(p []byte) (int, error) {
 	if len(s.readingData) == 0 && !s.recvEOF {
-		resp, err := s.stream.Recv()
+		resp, err := s.recvFn()
 		if err != nil {
 			return 0, err
 		}
 
-		if resp.Type == FileDataPacketType_Data {
+		if resp.Type == StreamDataPacketType_Data {
 			s.readingData = resp.Data
 
-		} else if resp.Type == FileDataPacketType_EOF {
+		} else if resp.Type == StreamDataPacketType_EOF {
 			s.readingData = resp.Data
 			s.recvEOF = true
 
@@ -121,7 +124,87 @@ func (c *Client) GetIPFSFile(fileHash string) (io.ReadCloser, error) {
 	}
 
 	return &fileReadCloser{
-		stream:   stream,
+		// TODO 临时处理方案
+		recvFn: func() (*StreamDataPacket, error) {
+			pkt, err := stream.Recv()
+			if err != nil {
+				return nil, err
+			}
+
+			return &StreamDataPacket{
+				Type: pkt.Type,
+				Data: pkt.Data,
+			}, nil
+		},
+		cancelFn: cancel,
+	}, nil
+}
+
+func (c *Client) SendStream(planID ioswitch.PlanID, streamID ioswitch.StreamID, file io.Reader) error {
+	sendCli, err := c.cli.SendStream(context.Background())
+	if err != nil {
+		return err
+	}
+
+	err = sendCli.Send(&StreamDataPacket{
+		Type:     StreamDataPacketType_SendArgs,
+		PlanID:   string(planID),
+		StreamID: string(streamID),
+	})
+	if err != nil {
+		return fmt.Errorf("sending stream id packet: %w", err)
+	}
+
+	buf := make([]byte, 4096)
+	for {
+		rd, err := file.Read(buf)
+		if err == io.EOF {
+			err := sendCli.Send(&StreamDataPacket{
+				Type:     StreamDataPacketType_EOF,
+				StreamID: string(streamID),
+				Data:     buf[:rd],
+			})
+			if err != nil {
+				return fmt.Errorf("sending EOF packet: %w", err)
+			}
+
+			_, err = sendCli.CloseAndRecv()
+			if err != nil {
+				return fmt.Errorf("receiving response: %w", err)
+			}
+
+			return nil
+		}
+
+		if err != nil {
+			return fmt.Errorf("reading file data: %w", err)
+		}
+
+		err = sendCli.Send(&StreamDataPacket{
+			Type:     StreamDataPacketType_Data,
+			StreamID: string(streamID),
+			Data:     buf[:rd],
+		})
+		if err != nil {
+			return fmt.Errorf("sending data packet: %w", err)
+		}
+	}
+}
+
+func (c *Client) FetchStream(planID ioswitch.PlanID, streamID ioswitch.StreamID) (io.ReadCloser, error) {
+	ctx, cancel := context.WithCancel(context.Background())
+
+	stream, err := c.cli.FetchStream(ctx, &FetchStreamReq{
+		PlanID:   string(planID),
+		StreamID: string(streamID),
+	})
+	if err != nil {
+		cancel()
+		return nil, fmt.Errorf("request grpc failed, err: %w", err)
+	}
+
+	return &fileReadCloser{
+		recvFn:   stream.Recv,
 		cancelFn: cancel,
 	}, nil
 }

common/pkgs/ioswitch/ioswitch.go

@@ -0,0 +1,35 @@
+package ioswitch
+
+import (
+	"io"
+)
+
+type PlanID string
+
+type StreamID string
+
+type Plan struct {
+	ID  PlanID
+	Ops []Op
+}
+
+type Stream struct {
+	ID     StreamID
+	Stream io.ReadCloser
+}
+
+func NewStream(id StreamID, stream io.ReadCloser) Stream {
+	return Stream{
+		ID:     id,
+		Stream: stream,
+	}
+}
+
+type Op interface {
+	Execute(sw *Switch, planID PlanID) error
+}
+
+type ResultKV struct {
+	Key   string
+	Value any
+}

common/pkgs/ioswitch/ops/ops.go

@@ -0,0 +1,198 @@
+package ops
+
+import (
+	"context"
+	"fmt"
+	"io"
+
+	"gitlink.org.cn/cloudream/common/pkgs/future"
+	"gitlink.org.cn/cloudream/common/pkgs/logger"
+	"gitlink.org.cn/cloudream/common/pkgs/types"
+	myio "gitlink.org.cn/cloudream/common/utils/io"
+	"gitlink.org.cn/cloudream/common/utils/serder"
+	stgglb "gitlink.org.cn/cloudream/storage/common/globals"
+	stgmod "gitlink.org.cn/cloudream/storage/common/models"
+	"gitlink.org.cn/cloudream/storage/common/pkgs/db/model"
+	"gitlink.org.cn/cloudream/storage/common/pkgs/ioswitch"
+)
+
+var _ = serder.UseTypeUnionExternallyTagged(types.Ref(types.NewTypeUnion[ioswitch.Op](
+	(*IPFSRead)(nil),
+	(*IPFSWrite)(nil),
+	(*GRPCSend)(nil),
+	(*GRPCFetch)(nil),
+	(*ECCompute)(nil),
+	(*Combine)(nil),
+)))
+
+type IPFSRead struct {
+	Output   ioswitch.StreamID `json:"output"`
+	FileHash string            `json:"fileHash"`
+}
+
+func (o *IPFSRead) Execute(sw *ioswitch.Switch, planID ioswitch.PlanID) error {
+	logger.
+		WithField("FileHash", o.FileHash).
+		WithField("Output", o.Output).
+		Debugf("ipfs read op")
+	defer logger.Debugf("ipfs read op finished")
+
+	ipfsCli, err := stgglb.IPFSPool.Acquire()
+	if err != nil {
+		return fmt.Errorf("new ipfs client: %w", err)
+	}
+	defer stgglb.IPFSPool.Release(ipfsCli)
+
+	file, err := ipfsCli.OpenRead(o.FileHash)
+	if err != nil {
+		return fmt.Errorf("reading ipfs: %w", err)
+	}
+
+	fut := future.NewSetVoid()
+	file = myio.AfterReadClosed(file, func(closer io.ReadCloser) {
+		fut.SetVoid()
+	})
+
+	sw.StreamReady(planID, ioswitch.NewStream(o.Output, file))
+
+	// TODO context
+	fut.Wait(context.TODO())
+	return nil
+}
+
+type IPFSWrite struct {
+	Input     ioswitch.StreamID `json:"input"`
+	ResultKey string            `json:"resultKey"`
+}
+
+func (o *IPFSWrite) Execute(sw *ioswitch.Switch, planID ioswitch.PlanID) error {
+	logger.
+		WithField("ResultKey", o.ResultKey).
+		WithField("Input", o.Input).
+		Debugf("ipfs write op")
+
+	ipfsCli, err := stgglb.IPFSPool.Acquire()
+	if err != nil {
+		return fmt.Errorf("new ipfs client: %w", err)
+	}
+	defer stgglb.IPFSPool.Release(ipfsCli)
+
+	strs, err := sw.WaitStreams(planID, o.Input)
+	if err != nil {
+		return err
+	}
+	defer strs[0].Stream.Close()
+
+	fileHash, err := ipfsCli.CreateFile(strs[0].Stream)
+	if err != nil {
+		return fmt.Errorf("creating ipfs file: %w", err)
+	}
+
+	if o.ResultKey != "" {
+		sw.AddResultValue(planID, ioswitch.ResultKV{
+			Key:   o.ResultKey,
+			Value: fileHash,
+		})
+	}
+
+	return nil
+}
+
+type GRPCSend struct {
+	StreamID ioswitch.StreamID `json:"streamID"`
+	Node     model.Node        `json:"node"`
+}
+
+func (o *GRPCSend) Execute(sw *ioswitch.Switch, planID ioswitch.PlanID) error {
+	logger.
+		WithField("ioswitch.StreamID", o.StreamID).
+		Debugf("grpc send")
+
+	strs, err := sw.WaitStreams(planID, o.StreamID)
+	if err != nil {
+		return err
+	}
+	defer strs[0].Stream.Close()
+
+	// TODO 根据客户端地址选择IP和端口
+	agtCli, err := stgglb.AgentRPCPool.Acquire(o.Node.ExternalIP, o.Node.ExternalGRPCPort)
+	if err != nil {
+		return fmt.Errorf("new agent rpc client: %w", err)
+	}
+	defer stgglb.AgentRPCPool.Release(agtCli)
+
+	err = agtCli.SendStream(planID, o.StreamID, strs[0].Stream)
+	if err != nil {
+		return fmt.Errorf("sending stream: %w", err)
+	}
+
+	return nil
+}
+
+type GRPCFetch struct {
+	StreamID ioswitch.StreamID `json:"streamID"`
+	Node     model.Node        `json:"node"`
+}
+
+func (o *GRPCFetch) Execute(sw *ioswitch.Switch, planID ioswitch.PlanID) error {
+	// TODO 根据客户端地址选择IP和端口
+	agtCli, err := stgglb.AgentRPCPool.Acquire(o.Node.ExternalIP, o.Node.ExternalGRPCPort)
+	if err != nil {
+		return fmt.Errorf("new agent rpc client: %w", err)
+	}
+	defer stgglb.AgentRPCPool.Release(agtCli)
+
+	str, err := agtCli.FetchStream(planID, o.StreamID)
+	if err != nil {
+		return fmt.Errorf("fetching stream: %w", err)
+	}
+
+	fut := future.NewSetVoid()
+	str = myio.AfterReadClosed(str, func(closer io.ReadCloser) {
+		fut.SetVoid()
+	})
+
+	sw.StreamReady(planID, ioswitch.NewStream(o.StreamID, str))
+
+	// TODO
+	fut.Wait(context.TODO())
+
+	return err
+}
+
+type ECCompute struct {
+	EC                 stgmod.EC           `json:"ec"`
+	InputIDs           []ioswitch.StreamID `json:"inputIDs"`
+	OutputIDs          []ioswitch.StreamID `json:"outputIDs"`
+	InputBlockIndexes  []int               `json:"inputBlockIndexes"`
+	OutputBlockIndexes []int               `json:"outputBlockIndexes"`
+}
+
+func (o *ECCompute) Execute(sw *ioswitch.Switch, planID ioswitch.PlanID) error {
+	// TODO2
+	return nil
+}
+
+type Combine struct {
+	InputIDs []ioswitch.StreamID `json:"inputIDs"`
+	OutputID ioswitch.StreamID   `json:"outputID"`
+}
+
+func (o *Combine) Execute(sw *ioswitch.Switch, planID ioswitch.PlanID) error {
+	strs, err := sw.WaitStreams(planID, o.InputIDs...)
+	if err != nil {
+		return err
+	}
+
+	pr, pw := io.Pipe()
+	sw.StreamReady(planID, ioswitch.NewStream(o.OutputID, pr))
+
+	for _, str := range strs {
+		_, err := io.Copy(pw, str.Stream)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}

common/pkgs/ioswitch/plans/executor.go

@@ -0,0 +1,172 @@
+package plans
+
+import (
+	"errors"
+	"fmt"
+	"io"
+	"sync"
+	"sync/atomic"
+
+	"gitlink.org.cn/cloudream/common/pkgs/future"
+	myio "gitlink.org.cn/cloudream/common/utils/io"
+	stgglb "gitlink.org.cn/cloudream/storage/common/globals"
+	"gitlink.org.cn/cloudream/storage/common/pkgs/ioswitch"
+	agtmq "gitlink.org.cn/cloudream/storage/common/pkgs/mq/agent"
+)
+
+type ExecutorResult struct {
+	ResultValues map[string]any
+}
+
+type Executor struct {
+	plan        ComposedPlan
+	callback    *future.SetValueFuture[ExecutorResult]
+	mqClis      []*agtmq.Client
+	planTaskIDs []string
+}
+
+func Execute(plan ComposedPlan) (*Executor, error) {
+	executor := Executor{
+		plan:     plan,
+		callback: future.NewSetValue[ExecutorResult](),
+	}
+
+	var err error
+	for _, a := range plan.AgentPlans {
+		var cli *agtmq.Client
+		cli, err = stgglb.AgentMQPool.Acquire(a.Node.NodeID)
+		if err != nil {
+			executor.Close()
+			return nil, fmt.Errorf("new mq client for %d: %w", a.Node.NodeID, err)
+		}
+
+		executor.mqClis = append(executor.mqClis, cli)
+	}
+
+	for i, a := range plan.AgentPlans {
+		cli := executor.mqClis[i]
+
+		_, err := cli.SetupIOPlan(agtmq.NewSetupIOPlan(a.Plan))
+		if err != nil {
+			for i -= 1; i >= 0; i-- {
+				executor.mqClis[i].CancelIOPlan(agtmq.NewCancelIOPlan(plan.ID))
+			}
+			executor.Close()
+			return nil, fmt.Errorf("setup plan at %d: %w", a.Node.NodeID, err)
+		}
+	}
+
+	for i, a := range plan.AgentPlans {
+		cli := executor.mqClis[i]
+
+		resp, err := cli.StartIOPlan(agtmq.NewStartIOPlan(a.Plan.ID))
+		if err != nil {
+			executor.cancelAll()
+			executor.Close()
+			return nil, fmt.Errorf("setup plan at %d: %w", a.Node.NodeID, err)
+		}
+
+		executor.planTaskIDs = append(executor.planTaskIDs, resp.TaskID)
+	}
+
+	go executor.pollResult()
+
+	return &executor, nil
+}
+
+func (e *Executor) SendStream(info *FromExecutorStream, stream io.Reader) error {
+	// TODO 根据地域选择IP
+	agtCli, err := stgglb.AgentRPCPool.Acquire(info.toNode.ExternalIP, info.toNode.ExternalGRPCPort)
+	if err != nil {
+		return fmt.Errorf("new agent rpc client: %w", err)
+	}
+	defer stgglb.AgentRPCPool.Release(agtCli)
+
+	return agtCli.SendStream(e.plan.ID, info.info.ID, stream)
+}
+
+func (e *Executor) ReadStream(info *ToExecutorStream) (io.ReadCloser, error) {
+	// TODO 根据地域选择IP
+	agtCli, err := stgglb.AgentRPCPool.Acquire(info.fromNode.ExternalIP, info.fromNode.ExternalGRPCPort)
+	if err != nil {
+		return nil, fmt.Errorf("new agent rpc client: %w", err)
+	}
+
+	str, err := agtCli.FetchStream(e.plan.ID, info.info.ID)
+	if err != nil {
+		return nil, err
+	}
+
+	return myio.AfterReadClosed(str, func(closer io.ReadCloser) {
+		stgglb.AgentRPCPool.Release(agtCli)
+	}), nil
+}
+
+func (e *Executor) cancelAll() {
+	for _, cli := range e.mqClis {
+		cli.CancelIOPlan(agtmq.NewCancelIOPlan(e.plan.ID))
+	}
+}
+
+func (e *Executor) Close() {
+	for _, c := range e.mqClis {
+		stgglb.AgentMQPool.Release(c)
+	}
+}
+
+func (e *Executor) pollResult() {
+	wg := sync.WaitGroup{}
+	anyErr := atomic.Value{}
+	anyErr.Store(nil)
+	rets := make([]*ioswitch.PlanResult, len(e.plan.AgentPlans))
+
+	for i, id := range e.planTaskIDs {
+		idx := i
+		taskID := id
+
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+
+			for {
+				resp, err := e.mqClis[idx].WaitIOPlan(agtmq.NewWaitIOPlan(taskID, 5000))
+				if err != nil {
+					anyErr.Store(err)
+					break
+				}
+
+				if resp.IsComplete {
+					if resp.Error != "" {
+						anyErr.Store(errors.New(resp.Error))
+					} else {
+						rets[idx] = &resp.Result
+					}
+					break
+				}
+
+				if anyErr.Load() != nil {
+					break
+				}
+			}
+		}()
+	}
+
+	wg.Wait()
+
+	err := anyErr.Load().(error)
+	if err != nil {
+		e.callback.SetError(err)
+		return
+	}
+
+	reducedRet := ExecutorResult{
+		ResultValues: make(map[string]any),
+	}
+	for _, ret := range rets {
+		for k, v := range ret.Values {
+			reducedRet.ResultValues[k] = v
+		}
+	}
+
+	e.callback.SetValue(reducedRet)
+}

common/pkgs/ioswitch/plans/plan_builder.go

@@ -0,0 +1,229 @@
+package plans
+
+import (
+	"fmt"
+
+	"github.com/google/uuid"
+	stgmod "gitlink.org.cn/cloudream/storage/common/models"
+	"gitlink.org.cn/cloudream/storage/common/pkgs/db/model"
+	"gitlink.org.cn/cloudream/storage/common/pkgs/ioswitch"
+	"gitlink.org.cn/cloudream/storage/common/pkgs/ioswitch/ops"
+)
+
+type StreamInfo struct {
+	ID ioswitch.StreamID
+}
+
+type PlanBuilder struct {
+	streams    []*StreamInfo
+	agentPlans map[int64]*AgentPlanBuilder
+}
+
+func (b *PlanBuilder) Build() (*ComposedPlan, error) {
+	planID := uuid.NewString()
+
+	var agentPlans []AgentPlan
+	for _, b := range b.agentPlans {
+		plan, err := b.Build(ioswitch.PlanID(planID))
+		if err != nil {
+			return nil, err
+		}
+
+		agentPlans = append(agentPlans, plan)
+	}
+
+	return &ComposedPlan{
+		ID:         ioswitch.PlanID(planID),
+		AgentPlans: agentPlans,
+	}, nil
+}
+
+func (b *PlanBuilder) newStream() *StreamInfo {
+	str := &StreamInfo{
+		ID: ioswitch.StreamID(fmt.Sprintf("%d", len(b.streams)+1)),
+	}
+
+	b.streams = append(b.streams, str)
+
+	return str
+}
+
+func NewPlanBuilder() PlanBuilder {
+	return PlanBuilder{
+		agentPlans: make(map[int64]*AgentPlanBuilder),
+	}
+}
+
+func (b *PlanBuilder) FromExecutor() *FromExecutorStream {
+	return &FromExecutorStream{
+		owner: b,
+		info:  b.newStream(),
+	}
+}
+
+func (b *PlanBuilder) AtAgent(node model.Node) *AgentPlanBuilder {
+	agtPlan, ok := b.agentPlans[node.NodeID]
+	if !ok {
+		agtPlan = &AgentPlanBuilder{
+			owner: b,
+			node:  node,
+		}
+		b.agentPlans[node.NodeID] = agtPlan
+	}
+
+	return agtPlan
+}
+
+type FromExecutorStream struct {
+	owner  *PlanBuilder
+	info   *StreamInfo
+	toNode *model.Node
+}
+
+func (s *FromExecutorStream) ToNode(node model.Node) *AgentStream {
+	s.toNode = &node
+	return &AgentStream{
+		owner: s.owner.AtAgent(node),
+		info:  s.owner.newStream(),
+	}
+}
+
+type ToExecutorStream struct {
+	info     *StreamInfo
+	fromNode *model.Node
+}
+
+type AgentStream struct {
+	owner *AgentPlanBuilder
+	info  *StreamInfo
+}
+
+func (s *AgentStream) IPFSWrite(resultKey string) {
+	s.owner.ops = append(s.owner.ops, &ops.IPFSWrite{
+		Input:     s.info.ID,
+		ResultKey: resultKey,
+	})
+}
+
+func (s *AgentStream) GRPCSend(node model.Node) *AgentStream {
+	agtStr := &AgentStream{
+		owner: s.owner.owner.AtAgent(node),
+		info:  s.owner.owner.newStream(),
+	}
+
+	s.owner.ops = append(s.owner.ops, &ops.GRPCSend{
+		StreamID: s.info.ID,
+		Node:     node,
+	})
+
+	return agtStr
+}
+
+func (s *AgentStream) ToExecutor() *ToExecutorStream {
+	return &ToExecutorStream{
+		info:     s.info,
+		fromNode: &s.owner.node,
+	}
+}
+
+type AgentPlanBuilder struct {
+	owner *PlanBuilder
+	node  model.Node
+	ops   []ioswitch.Op
+}
+
+func (b *AgentPlanBuilder) GRCPFetch(node model.Node) *AgentStream {
+	agtStr := &AgentStream{
+		owner: b,
+		info:  b.owner.newStream(),
+	}
+
+	b.ops = append(b.ops, &ops.GRPCFetch{
+		StreamID: agtStr.info.ID,
+		Node:     node,
+	})
+
+	return agtStr
+}
+
+func (b *AgentPlanBuilder) IPFSRead(fileHash string) *AgentStream {
+	agtStr := &AgentStream{
+		owner: b,
+		info:  b.owner.newStream(),
+	}
+
+	b.ops = append(b.ops, &ops.IPFSRead{
+		Output:   agtStr.info.ID,
+		FileHash: fileHash,
+	})
+
+	return agtStr
+}
+
+func (b *AgentPlanBuilder) ECCompute(ec stgmod.EC, inBlockIndexes []int, outBlockIndexes []int, streams ...*AgentStream) *MultiStream {
+	mstr := &MultiStream{}
+
+	var inputStrIDs []ioswitch.StreamID
+	for _, str := range streams {
+		inputStrIDs = append(inputStrIDs, str.info.ID)
+	}
+
+	var outputStrIDs []ioswitch.StreamID
+	for i := 0; i < ec.N-ec.K; i++ {
+		info := b.owner.newStream()
+		mstr.streams[i] = &AgentStream{
+			owner: b,
+			info:  info,
+		}
+		outputStrIDs = append(outputStrIDs, info.ID)
+	}
+
+	b.ops = append(b.ops, &ops.ECCompute{
+		EC:                 ec,
+		InputIDs:           inputStrIDs,
+		OutputIDs:          outputStrIDs,
+		InputBlockIndexes:  inBlockIndexes,
+		OutputBlockIndexes: outBlockIndexes,
+	})
+
+	return mstr
+}
+
+func (b *AgentPlanBuilder) Combine(streams ...*AgentStream) *AgentStream {
+	agtStr := &AgentStream{
+		owner: b,
+		info:  b.owner.newStream(),
+	}
+
+	var inputStrIDs []ioswitch.StreamID
+	for _, str := range streams {
+		inputStrIDs = append(inputStrIDs, str.info.ID)
+	}
+
+	b.ops = append(b.ops, &ops.Combine{
+		InputIDs: inputStrIDs,
+		OutputID: agtStr.info.ID,
+	})
+
+	return agtStr
+}
+
+func (b *AgentPlanBuilder) Build(planID ioswitch.PlanID) (AgentPlan, error) {
+	plan := ioswitch.Plan{
+		ID:  planID,
+		Ops: b.ops,
+	}
+
+	return AgentPlan{
+		Plan: plan,
+		Node: b.node,
+	}, nil
+}
+
+type MultiStream struct {
+	streams []*AgentStream
+}
+
+func (m *MultiStream) Stream(index int) *AgentStream {
+	return m.streams[index]
+}

common/pkgs/ioswitch/plans/plans.go

@@ -0,0 +1,16 @@
+package plans
+
+import (
+	"gitlink.org.cn/cloudream/storage/common/pkgs/db/model"
+	"gitlink.org.cn/cloudream/storage/common/pkgs/ioswitch"
+)
+
+type AgentPlan struct {
+	Node model.Node
+	Plan ioswitch.Plan
+}
+
+type ComposedPlan struct {
+	ID         ioswitch.PlanID
+	AgentPlans []AgentPlan
+}

common/pkgs/ioswitch/switch.go

@@ -0,0 +1,291 @@
+package ioswitch
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+
+	"gitlink.org.cn/cloudream/common/pkgs/future"
+	"gitlink.org.cn/cloudream/common/pkgs/logger"
+	"gitlink.org.cn/cloudream/common/utils/lo"
+)
+
+var ErrPlanFinished = errors.New("plan is finished")
+
+var ErrPlanNotFound = errors.New("plan not found")
+
+type OpState string
+
+const (
+	OpPending  OpState = "Pending"
+	OpFinished OpState = "Finished"
+)
+
+type Oping struct {
+	State OpState
+}
+
+type PlanResult struct {
+	Values map[string]any `json:"values"`
+}
+
+type Planning struct {
+	plan         Plan
+	opings       []Oping
+	resultValues map[string]any
+	callback     *future.SetValueFuture[PlanResult]
+
+	readys    map[StreamID]Stream
+	waittings []*Watting
+}
+
+func NewPlanning(plan Plan) Planning {
+	planning := Planning{
+		plan:     plan,
+		callback: future.NewSetValue[PlanResult](),
+		readys:   make(map[StreamID]Stream),
+	}
+
+	for _ = range plan.Ops {
+		oping := Oping{
+			State: OpPending,
+		}
+		planning.opings = append(planning.opings, oping)
+	}
+
+	return planning
+}
+
+func (p *Planning) IsCompleted() bool {
+	for _, oping := range p.opings {
+		if oping.State != OpFinished {
+			return false
+		}
+	}
+
+	return true
+}
+
+func (p *Planning) MakeResult() PlanResult {
+	return PlanResult{
+		Values: p.resultValues,
+	}
+}
+
+type Watting struct {
+	WaitIDs  []StreamID
+	Readys   []Stream
+	Callback *future.SetValueFuture[[]Stream]
+}
+
+func (w *Watting) TryReady(str Stream) bool {
+	for i, id := range w.WaitIDs {
+		if id == str.ID {
+			w.Readys[i] = str
+			return true
+		}
+	}
+
+	return false
+}
+
+func (c *Watting) IsAllReady() bool {
+	for _, s := range c.Readys {
+		if s.Stream == nil {
+			return false
+		}
+	}
+
+	return true
+}
+
+func (w *Watting) Complete() {
+	w.Callback.SetValue(w.Readys)
+}
+
+func (w *Watting) Cancel(err error) {
+	w.Callback.SetError(err)
+}
+
+type Switch struct {
+	lock      sync.Mutex
+	plannings map[PlanID]*Planning
+}
+
+func NewSwitch() Switch {
+	return Switch{
+		plannings: make(map[PlanID]*Planning),
+	}
+}
+
+func (s *Switch) SetupPlan(plan Plan) error {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
+	if _, ok := s.plannings[plan.ID]; ok {
+		return fmt.Errorf("plan id exists")
+	}
+
+	planning := NewPlanning(plan)
+	s.plannings[plan.ID] = &planning
+	return nil
+}
+
+func (s *Switch) ExecutePlan(id PlanID) (PlanResult, error) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
+	planning, ok := s.plannings[id]
+	if !ok {
+		return PlanResult{}, fmt.Errorf("plan not found")
+	}
+
+	for i, op := range planning.plan.Ops {
+		idx := i
+		o := op
+		go func() {
+			err := o.Execute(s, id)
+
+			s.lock.Lock()
+			defer s.lock.Unlock()
+
+			if err != nil {
+				logger.Std.Warnf("exeucting op: %s", err.Error())
+				s.cancelPlan(id)
+				return
+			}
+
+			planning.opings[idx].State = OpFinished
+			if planning.IsCompleted() {
+				s.completePlan(id)
+			}
+		}()
+	}
+
+	return planning.callback.WaitValue(context.TODO())
+}
+
+func (s *Switch) CancelPlan(id PlanID) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
+	s.cancelPlan(id)
+}
+
+func (s *Switch) cancelPlan(id PlanID) {
+	plan, ok := s.plannings[id]
+	if !ok {
+		return
+	}
+
+	delete(s.plannings, id)
+
+	for _, s := range plan.readys {
+		s.Stream.Close()
+	}
+
+	for _, c := range plan.waittings {
+		c.Callback.SetError(ErrPlanFinished)
+	}
+
+	plan.callback.SetError(fmt.Errorf("plan cancelled"))
+}
+
+func (s *Switch) completePlan(id PlanID) {
+	plan, ok := s.plannings[id]
+	if !ok {
+		return
+	}
+
+	delete(s.plannings, id)
+
+	for _, s := range plan.readys {
+		s.Stream.Close()
+	}
+
+	for _, c := range plan.waittings {
+		c.Callback.SetError(ErrPlanFinished)
+	}
+
+	plan.callback.SetValue(plan.MakeResult())
+}
+
+func (s *Switch) StreamReady(planID PlanID, stream Stream) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
+	plan, ok := s.plannings[planID]
+	if !ok {
+		//TODO 处理错误
+		return
+	}
+
+	for i, wa := range plan.waittings {
+		if !wa.TryReady(stream) {
+			continue
+		}
+
+		if !wa.IsAllReady() {
+			return
+		}
+
+		plan.waittings = lo.RemoveAt(plan.waittings, i)
+		wa.Complete()
+		return
+	}
+
+	plan.readys[stream.ID] = stream
+}
+
+func (s *Switch) WaitStreams(planID PlanID, streamIDs ...StreamID) ([]Stream, error) {
+
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
+	plan, ok := s.plannings[planID]
+	if !ok {
+		return nil, ErrPlanNotFound
+	}
+
+	allReady := true
+	readys := make([]Stream, len(streamIDs))
+	for i, id := range streamIDs {
+		str, ok := plan.readys[id]
+		if !ok {
+			allReady = false
+			continue
+		}
+
+		readys[i] = str
+		delete(plan.readys, id)
+	}
+
+	if allReady {
+		return readys, nil
+	}
+
+	callback := future.NewSetValue[[]Stream]()
+
+	plan.waittings = append(plan.waittings, &Watting{
+		WaitIDs:  streamIDs,
+		Readys:   readys,
+		Callback: callback,
+	})
+
+	return callback.WaitValue(context.TODO())
+}
+
+func (s *Switch) AddResultValue(planID PlanID, rets ...ResultKV) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
+	plan, ok := s.plannings[planID]
+	if !ok {
+		return
+	}
+
+	for _, ret := range rets {
+		plan.resultValues[ret.Key] = ret.Value
+	}
+}

common/pkgs/mq/agent/io.go

@@ -0,0 +1,120 @@
+package agent
+
+import (
+	"gitlink.org.cn/cloudream/common/pkgs/mq"
+	"gitlink.org.cn/cloudream/storage/common/pkgs/ioswitch"
+)
+
+type IOService interface {
+	SetupIOPlan(msg *SetupIOPlan) (*SetupIOPlanResp, *mq.CodeMessage)
+
+	StartIOPlan(msg *StartIOPlan) (*StartIOPlanResp, *mq.CodeMessage)
+
+	WaitIOPlan(msg *WaitIOPlan) (*WaitIOPlanResp, *mq.CodeMessage)
+
+	CancelIOPlan(msg *CancelIOPlan) (*CancelIOPlanResp, *mq.CodeMessage)
+}
+
+// 设置io计划
+var _ = Register(Service.SetupIOPlan)
+
+type SetupIOPlan struct {
+	mq.MessageBodyBase
+	Plan ioswitch.Plan `json:"plan"`
+}
+type SetupIOPlanResp struct {
+	mq.MessageBodyBase
+}
+
+func NewSetupIOPlan(plan ioswitch.Plan) *SetupIOPlan {
+	return &SetupIOPlan{
+		Plan: plan,
+	}
+}
+func NewSetupIOPlanResp() *SetupIOPlanResp {
+	return &SetupIOPlanResp{}
+}
+func (client *Client) SetupIOPlan(msg *SetupIOPlan, opts ...mq.RequestOption) (*SetupIOPlanResp, error) {
+	return mq.Request(Service.SetupIOPlan, client.rabbitCli, msg, opts...)
+}
+
+// 启动io计划
+var _ = Register(Service.StartIOPlan)
+
+type StartIOPlan struct {
+	mq.MessageBodyBase
+	PlanID ioswitch.PlanID `json:"planID"`
+}
+type StartIOPlanResp struct {
+	mq.MessageBodyBase
+	TaskID string `json:"taskID"`
+}
+
+func NewStartIOPlan(planID ioswitch.PlanID) *StartIOPlan {
+	return &StartIOPlan{
+		PlanID: planID,
+	}
+}
+func NewStartIOPlanResp(taskID string) *StartIOPlanResp {
+	return &StartIOPlanResp{
+		TaskID: taskID,
+	}
+}
+func (client *Client) StartIOPlan(msg *StartIOPlan, opts ...mq.RequestOption) (*StartIOPlanResp, error) {
+	return mq.Request(Service.StartIOPlan, client.rabbitCli, msg, opts...)
+}
+
+// 启动io计划
+var _ = Register(Service.WaitIOPlan)
+
+type WaitIOPlan struct {
+	mq.MessageBodyBase
+	TaskID        string `json:"taskID"`
+	WaitTimeoutMs int64  `json:"waitTimeout"`
+}
+type WaitIOPlanResp struct {
+	mq.MessageBodyBase
+	IsComplete bool                `json:"isComplete"`
+	Error      string              `json:"error"`
+	Result     ioswitch.PlanResult `json:"result"`
+}
+
+func NewWaitIOPlan(taskID string, waitTimeoutMs int64) *WaitIOPlan {
+	return &WaitIOPlan{
+		TaskID:        taskID,
+		WaitTimeoutMs: waitTimeoutMs,
+	}
+}
+func NewWaitIOPlanResp(isComplete bool, err string, result ioswitch.PlanResult) *WaitIOPlanResp {
+	return &WaitIOPlanResp{
+		IsComplete: isComplete,
+		Error:      err,
+		Result:     result,
+	}
+}
+func (client *Client) WaitIOPlan(msg *WaitIOPlan, opts ...mq.RequestOption) (*WaitIOPlanResp, error) {
+	return mq.Request(Service.WaitIOPlan, client.rabbitCli, msg, opts...)
+}
+
+// 取消io计划
+var _ = Register(Service.CancelIOPlan)
+
+type CancelIOPlan struct {
+	mq.MessageBodyBase
+	PlanID ioswitch.PlanID `json:"planID"`
+}
+type CancelIOPlanResp struct {
+	mq.MessageBodyBase
+}
+
+func NewCancelIOPlan(planID ioswitch.PlanID) *CancelIOPlan {
+	return &CancelIOPlan{
+		PlanID: planID,
+	}
+}
+func NewCancelIOPlanResp() *CancelIOPlanResp {
+	return &CancelIOPlanResp{}
+}
+func (client *Client) CancelIOPlan(msg *CancelIOPlan, opts ...mq.RequestOption) (*CancelIOPlanResp, error) {
+	return mq.Request(Service.CancelIOPlan, client.rabbitCli, msg, opts...)
+}

common/pkgs/mq/agent/server.go

@@ -6,6 +6,8 @@ import (
 )
 
 type Service interface {
+	IOService
+
 	ObjectService
 
 	StorageService