From 4fb87e062bd0229785a78cdb7e34b6d65a91f799 Mon Sep 17 00:00:00 2001
From: Sydonian <794346190@qq.com>
Date: Mon, 14 Aug 2023 09:37:57 +0800
Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E4=B8=80=E4=BA=9B=E9=97=AE?=
 =?UTF-8?q?=E9=A2=98=EF=BC=8C=E5=A2=9E=E5=8A=A0TODO?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 internal/config/config.go        |  2 +-
 internal/services/cmd/storage.go | 28 ++++++-------
 internal/task/ec_read.go         | 70 ++++++++++++--------------------
 main.go                          |  2 -
 4 files changed, 41 insertions(+), 61 deletions(-)

diff --git a/internal/config/config.go b/internal/config/config.go
index 1c59be2..47a9a1e 100644
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -12,7 +12,7 @@ type Config struct {
 	ID                int64           `json:"id"`
 	GRPCListenAddress string          `json:"grpcListenAddress"`
 	GRPCPort          int             `json:"grpcPort"`
-	GRPCPacketSize    int64           `json:"grpcPacketSize"`
+	ECPacketSize      int64           `json:"ecPacketSize"`
 	LocalIP           string          `json:"localIP"`
 	ExternalIP        string          `json:"externalIP"`
 	StorageBaseDir    string          `json:"storageBaseDir"`
diff --git a/internal/services/cmd/storage.go b/internal/services/cmd/storage.go
index fab4c6b..608b692 100644
--- a/internal/services/cmd/storage.go
+++ b/internal/services/cmd/storage.go
@@ -37,19 +37,17 @@ func (service *Service) StartStorageMoveObject(msg *agtmsg.StartStorageMoveObjec
 
 		return ramsg.ReplyOK(agtmsg.NewStartStorageMoveObjectResp(taskID))
 
-	} else {
-		// TODO 处理其他备份类型
-		if repRed, ok := msg.Redundancy.(models.ECRedundancyData); ok {
-			taskID, err := service.moveEcObject(msg.FileSize, repRed, outFilePath)
-			if err != nil {
-				logger.Warnf("move ec object as %s failed, err: %s", outFilePath, err.Error())
-				return ramsg.ReplyFailed[agtmsg.StartStorageMoveObjectResp](errorcode.OperationFailed, "move ec object failed")
-			}
-
-			return ramsg.ReplyOK(agtmsg.NewStartStorageMoveObjectResp(taskID))
+	} else if repRed, ok := msg.Redundancy.(models.ECRedundancyData); ok {
+		taskID, err := service.moveEcObject(msg.ObjectID, msg.FileSize, repRed, outFilePath)
+		if err != nil {
+			logger.Warnf("move ec object as %s failed, err: %s", outFilePath, err.Error())
+			return ramsg.ReplyFailed[agtmsg.StartStorageMoveObjectResp](errorcode.OperationFailed, "move ec object failed")
 		}
-		return ramsg.ReplyFailed[agtmsg.StartStorageMoveObjectResp](errorcode.OperationFailed, "not rep or ec object???")
+
+		return ramsg.ReplyOK(agtmsg.NewStartStorageMoveObjectResp(taskID))
 	}
+
+	return ramsg.ReplyFailed[agtmsg.StartStorageMoveObjectResp](errorcode.OperationFailed, "not rep or ec object???")
 }
 
 func (svc *Service) moveRepObject(repData models.RepRedundancyData, outFilePath string) (string, error) {
@@ -57,15 +55,15 @@ func (svc *Service) moveRepObject(repData models.RepRedundancyData, outFilePath
 	return tsk.ID(), nil
 }
 
-func (svc *Service) moveEcObject(fileSize int64, ecData models.ECRedundancyData, outFilePath string) (string, error) {
+func (svc *Service) moveEcObject(objID int64, fileSize int64, ecData models.ECRedundancyData, outFilePath string) (string, error) {
 	ecK := ecData.Ec.EcK
 	blockIDs := make([]int, ecK)
 	hashs := make([]string, ecK)
-	for i:=0 ; i<ecK; i++{
+	for i := 0; i < ecK; i++ {
 		blockIDs[i] = i
-		hashs[i] = ecData.Blocks[i].FileHash 
+		hashs[i] = ecData.Blocks[i].FileHash
 	}
-	tsk := svc.taskManager.StartComparable(task.NewEcRead(fileSize, ecData.Ec, blockIDs, hashs, outFilePath))
+	tsk := svc.taskManager.StartComparable(task.NewEcRead(objID, fileSize, ecData.Ec, blockIDs, hashs, outFilePath))
 	return tsk.ID(), nil
 }
 
diff --git a/internal/task/ec_read.go b/internal/task/ec_read.go
index 7ca10f7..2eeb76e 100644
--- a/internal/task/ec_read.go
+++ b/internal/task/ec_read.go
@@ -6,28 +6,30 @@ import (
 	"os"
 	"path/filepath"
 	"time"
-	
-	"gitlink.org.cn/cloudream/ec"
-	"gitlink.org.cn/cloudream/common/pkg/logger"
-	"gitlink.org.cn/cloudream/common/models"
+
 	"gitlink.org.cn/cloudream/agent/internal/config"
+	"gitlink.org.cn/cloudream/common/models"
+	"gitlink.org.cn/cloudream/common/pkg/logger"
+	"gitlink.org.cn/cloudream/ec"
 )
 
 type EcRead struct {
-	FileSize int64
-	Ec       models.EC
-	BlockIDs    []int
-	BlockHashs  []string
-	LocalPath string
+	objID      int64
+	FileSize   int64
+	Ec         models.EC
+	BlockIDs   []int
+	BlockHashs []string
+	LocalPath  string
 }
 
-func NewEcRead(fileSize int64,ec models.EC, blockIDs []int, blockHashs []string, localPath string) *EcRead {
+func NewEcRead(objID int64, fileSize int64, ec models.EC, blockIDs []int, blockHashs []string, localPath string) *EcRead {
 	return &EcRead{
-		FileSize:  fileSize,
-		Ec:        ec,
-		BlockIDs:  blockIDs,
-		BlockHashs:  blockHashs,
-		LocalPath: localPath,
+		objID:      objID,
+		FileSize:   fileSize,
+		Ec:         ec,
+		BlockIDs:   blockIDs,
+		BlockHashs: blockHashs,
+		LocalPath:  localPath,
 	}
 }
 
@@ -36,28 +38,8 @@ func (t *EcRead) Compare(other *Task) bool {
 	if !ok {
 		return false
 	}
-	checkBLockIDs := true
-	if len(t.BlockIDs) != len(tsk.BlockIDs){
-		checkBLockIDs = false
-	}else{
-		for i:=0; i<len(t.BlockIDs); i++{
-			if t.BlockIDs[i] != tsk.BlockIDs[i]{
-				checkBLockIDs = false
-			}
-		}
-	}
 
-	checkHashs := true
-	if len(t.BlockHashs) != len(tsk.BlockHashs){
-		checkHashs = false
-	}else{
-		for i:=0; i<len(t.BlockHashs); i++{
-			if t.BlockHashs[i] != tsk.BlockHashs[i]{
-				checkHashs = false
-			}
-		}
-	}
-	return t.FileSize == tsk.FileSize && t.Ec == tsk.Ec && checkBLockIDs && checkHashs &&t.LocalPath == tsk.LocalPath
+	return t.objID == tsk.objID && t.LocalPath == tsk.LocalPath
 }
 
 func (t *EcRead) Execute(ctx TaskContext, complete CompleteFn) {
@@ -119,8 +101,10 @@ func (t *EcRead) Execute(ctx TaskContext, complete CompleteFn) {
 	})
 }
 
-func readObject(ctx TaskContext, fileSize int64, ecK int, ecN int, blockIDs []int, hashs []string)(io.ReadCloser, error){
-	numPacket := (fileSize + int64(ecK)*config.Cfg().GRPCPacketSize - 1) / (int64(ecK) * config.Cfg().GRPCPacketSize)
+func readObject(ctx TaskContext, fileSize int64, ecK int, ecN int, blockIDs []int, hashs []string) (io.ReadCloser, error) {
+	// TODO zkx 先使用同步方式实现读取多个block并解码数据的逻辑，做好错误处理
+
+	numPacket := (fileSize + int64(ecK)*config.Cfg().ECPacketSize - 1) / (int64(ecK) * config.Cfg().ECPacketSize)
 	getBufs := make([]chan []byte, ecN)
 	decodeBufs := make([]chan []byte, ecK)
 	for i := 0; i < ecN; i++ {
@@ -134,9 +118,9 @@ func readObject(ctx TaskContext, fileSize int64, ecK int, ecN int, blockIDs []in
 	}
 	//print(numPacket)
 	go decode(getBufs[:], decodeBufs[:], blockIDs, ecK, numPacket)
-	r,w := io.Pipe()
+	r, w := io.Pipe()
 	//这个就是persist函数
-	go func(){
+	go func() {
 		for i := 0; int64(i) < numPacket; i++ {
 			for j := 0; j < len(decodeBufs); j++ {
 				tmp := <-decodeBufs[j]
@@ -159,8 +143,8 @@ func get(ctx TaskContext, blockHash string, getBuf chan []byte, numPacket int64)
 		return fmt.Errorf("read ipfs block failed, err: %w", err)
 	}
 	defer reader.Close()
-	for i:=0; int64(i)<numPacket; i++{
-		buf := make([]byte, config.Cfg().GRPCPacketSize)
+	for i := 0; int64(i) < numPacket; i++ {
+		buf := make([]byte, config.Cfg().ECPacketSize)
 		_, err := io.ReadFull(reader, buf)
 		if err != nil {
 			return fmt.Errorf("read file falied, err:%w", err)
@@ -207,4 +191,4 @@ func decode(inBufs []chan []byte, outBufs []chan []byte, blockSeq []int, ecK int
 	for i := 0; i < len(outBufs); i++ {
 		close(outBufs[i])
 	}
-}
\ No newline at end of file
+}
diff --git a/main.go b/main.go
index 19a6656..4979110 100644
--- a/main.go
+++ b/main.go
@@ -26,8 +26,6 @@ import (
 var AgentIpList []string
 
 func main() {
-	print("!!!!!!!!")
-	//logger.Warnf("worinimab")
 	// TODO 放到配置里读取
 	AgentIpList = []string{"pcm01", "pcm1", "pcm2"}