jhnine
/
JCC-Storage
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

优化PinObject的接口

This commit is contained in:
Sydonian 2023-12-12 16:11:25 +08:00
parent 6d4f4a9ac4
commit 86ea948777
4 changed files with 19 additions and 112 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
agent/internal/services/mq/object.go
View File

@ -1,8 +1,6 @@
package mq
import (
"time"
"gitlink.org.cn/cloudream/common/consts/errorcode"
log "gitlink.org.cn/cloudream/common/pkgs/logger"
"gitlink.org.cn/cloudream/common/pkgs/mq"
@ -10,7 +8,7 @@ import (
agtmq "gitlink.org.cn/cloudream/storage/common/pkgs/mq/agent"
)
func (svc *Service) StartPinningObject(msg *agtmq.StartPinningObject) (*agtmq.StartPinningObjectResp, *mq.CodeMessage) {
func (svc *Service) PinObject(msg *agtmq.PinObject) (*agtmq.PinObjectResp, *mq.CodeMessage) {
log.WithField("FileHash", msg.FileHash).Debugf("pin object")
tsk := svc.taskManager.StartComparable(task.NewIPFSPin(msg.FileHash))
@ -21,38 +19,10 @@ func (svc *Service) StartPinningObject(msg *agtmq.StartPinningObject) (*agtmq.St
return nil, mq.Failed(errorcode.OperationFailed, "pin object failed")
}
return mq.ReplyOK(agtmq.NewStartPinningObjectResp(tsk.ID()))
}
func (svc *Service) WaitPinningObject(msg *agtmq.WaitPinningObject) (*agtmq.WaitPinningObjectResp, *mq.CodeMessage) {
log.WithField("TaskID", msg.TaskID).Debugf("wait pinning object")
tsk := svc.taskManager.FindByID(msg.TaskID)
if tsk == nil {
return nil, mq.Failed(errorcode.TaskNotFound, "task not found")
if msg.Async {
return mq.ReplyOK(agtmq.RespPinObject())
}
if msg.WaitTimeoutMs == 0 {
tsk.Wait()
errMsg := ""
if tsk.Error() != nil {
errMsg = tsk.Error().Error()
}
return mq.ReplyOK(agtmq.NewWaitPinningObjectResp(true, errMsg))
} else {
if tsk.WaitTimeout(time.Duration(msg.WaitTimeoutMs) * time.Millisecond) {
errMsg := ""
if tsk.Error() != nil {
errMsg = tsk.Error().Error()
}
return mq.ReplyOK(agtmq.NewWaitPinningObjectResp(true, errMsg))
}
return mq.ReplyOK(agtmq.NewWaitPinningObjectResp(false, ""))
}
tsk.Wait()
return mq.ReplyOK(agtmq.RespPinObject())
}

18
common/pkgs/cmd/create_package.go
View File

@ -4,7 +4,6 @@ import (
"fmt"
"io"
"math/rand"
"time"
"github.com/samber/lo"
@ -283,25 +282,10 @@ func pinIPFSFile(nodeID cdssdk.NodeID, fileHash string) error {
defer stgglb.AgentMQPool.Release(agtCli)
// 然后让最近节点pin本地上传的文件
pinObjResp, err := agtCli.StartPinningObject(agtmq.NewStartPinningObject(fileHash))
_, err = agtCli.PinObject(agtmq.ReqPinObject(fileHash, false))
if err != nil {
return fmt.Errorf("start pinning object: %w", err)
}
for {
waitResp, err := agtCli.WaitPinningObject(agtmq.NewWaitPinningObject(pinObjResp.TaskID, int64(time.Second)*5))
if err != nil {
return fmt.Errorf("waitting pinning object: %w", err)
}
if waitResp.IsComplete {
if waitResp.Error != "" {
return fmt.Errorf("agent pinning object: %s", waitResp.Error)
}
break
}
}
return nil
}

56
common/pkgs/mq/agent/object.go
View File

@ -3,62 +3,30 @@ package agent
import "gitlink.org.cn/cloudream/common/pkgs/mq"
type ObjectService interface {
StartPinningObject(msg *StartPinningObject) (*StartPinningObjectResp, *mq.CodeMessage)
WaitPinningObject(msg *WaitPinningObject) (*WaitPinningObjectResp, *mq.CodeMessage)
PinObject(msg *PinObject) (*PinObjectResp, *mq.CodeMessage)
}
// 启动Pin对象的任务
var _ = Register(Service.StartPinningObject)
var _ = Register(Service.PinObject)
type StartPinningObject struct {
type PinObject struct {
mq.MessageBodyBase
FileHash string `json:"fileHash"`
Async bool `json:"async"`
}
type StartPinningObjectResp struct {
type PinObjectResp struct {
mq.MessageBodyBase
TaskID string `json:"taskID"`
}
func NewStartPinningObject(fileHash string) *StartPinningObject {
return &StartPinningObject{
func ReqPinObject(fileHash string, async bool) *PinObject {
return &PinObject{
FileHash: fileHash,
Async: async,
}
}
func NewStartPinningObjectResp(taskID string) *StartPinningObjectResp {
return &StartPinningObjectResp{
TaskID: taskID,
}
func RespPinObject() *PinObjectResp {
return &PinObjectResp{}
}
func (client *Client) StartPinningObject(msg *StartPinningObject, opts ...mq.RequestOption) (*StartPinningObjectResp, error) {
return mq.Request(Service.StartPinningObject, client.rabbitCli, msg, opts...)
}
// 等待Pin对象的任务
var _ = Register(Service.WaitPinningObject)
type WaitPinningObject struct {
mq.MessageBodyBase
TaskID string `json:"taskID"`
WaitTimeoutMs int64 `json:"waitTimeout"`
}
type WaitPinningObjectResp struct {
mq.MessageBodyBase
IsComplete bool `json:"isComplete"`
Error string `json:"error"`
}
func NewWaitPinningObject(taskID string, waitTimeoutMs int64) *WaitPinningObject {
return &WaitPinningObject{
TaskID: taskID,
WaitTimeoutMs: waitTimeoutMs,
}
}
func NewWaitPinningObjectResp(isComplete bool, err string) *WaitPinningObjectResp {
return &WaitPinningObjectResp{
IsComplete: isComplete,
Error: err,
}
}
func (client *Client) WaitPinningObject(msg *WaitPinningObject, opts ...mq.RequestOption) (*WaitPinningObjectResp, error) {
return mq.Request(Service.WaitPinningObject, client.rabbitCli, msg, opts...)
func (client *Client) PinObject(msg *PinObject, opts ...mq.RequestOption) (*PinObjectResp, error) {
return mq.Request(Service.PinObject, client.rabbitCli, msg, opts...)
}

17
scanner/internal/event/check_package_redundancy.go
View File

@ -601,26 +601,11 @@ func (t *CheckPackageRedundancy) pinObject(nodeID cdssdk.NodeID, fileHash string
}
defer stgglb.AgentMQPool.Release(agtCli)
pinObjResp, err := agtCli.StartPinningObject(agtmq.NewStartPinningObject(fileHash))
_, err = agtCli.PinObject(agtmq.ReqPinObject(fileHash, false))
if err != nil {
return fmt.Errorf("start pinning object: %w", err)
}
for {
waitResp, err := agtCli.WaitPinningObject(agtmq.NewWaitPinningObject(pinObjResp.TaskID, int64(time.Second)*5))
if err != nil {
return fmt.Errorf("waitting pinning object: %w", err)
}
if waitResp.IsComplete {
if waitResp.Error != "" {
return fmt.Errorf("agent pinning object: %s", waitResp.Error)
}
break
}
}
return nil
}