floraachy
/
pcm-coordinator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

文件断点续传已完成 

Former-commit-id: 8abc72dc7225fc37a456bc08f133a4ff8d41c90f
This commit is contained in:
zhangwei 2023-05-31 08:50:49 +08:00
parent 4b28b15bb1
commit db8acc2e12
5 changed files with 23 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
adaptor/PCM-CORE/api/internal/handler/image/chunkimagehandler.go
View File

@ -6,7 +6,6 @@ import (
"io/ioutil" "io/ioutil"
"net/http" "net/http"
"os" "os"
"path"
"path/filepath" "path/filepath"
"strconv" "strconv"
"strings" "strings"
@ -17,8 +16,7 @@ import (
) )
var dir, _ = os.Getwd() var dir, _ = os.Getwd()
var windowsUploadPath = strings.ReplaceAll(path.Join(dir, "uploads"), "/", "\\") var uploadPath = filepath.Join(dir, "uploads")
var linuxUploadPath = path.Join(dir, "uploads")
func ChunkImageHandler(svcCtx *svc.ServiceContext) http.HandlerFunc { func ChunkImageHandler(svcCtx *svc.ServiceContext) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) { return func(w http.ResponseWriter, r *http.Request) {
@ -26,11 +24,11 @@ func ChunkImageHandler(svcCtx *svc.ServiceContext) http.HandlerFunc {
hash := r.PostFormValue("hash") hash := r.PostFormValue("hash")
name := r.PostFormValue("name") name := r.PostFormValue("name")
// 对比合并请求的文件大小和已上传文件夹大小 // 对比合并请求的文件大小和已上传文件夹大小
toSize, _ := getDirSize(path.Join(linuxUploadTempPath, hash)) toSize, _ := getDirSize(filepath.Join(uploadTempPath, hash))
if size != toSize { if size != toSize {
fmt.Fprintf(w, "文件上传错误") fmt.Fprintf(w, "文件上传错误")
} }
chunksPath := path.Join(linuxUploadTempPath, hash) chunksPath := filepath.Join(uploadTempPath, hash)
files, _ := ioutil.ReadDir(chunksPath) files, _ := ioutil.ReadDir(chunksPath)
// 将文件根据索引序号排序 // 将文件根据索引序号排序
filesSort := make(map[string]string) filesSort := make(map[string]string)
@ -38,7 +36,7 @@ func ChunkImageHandler(svcCtx *svc.ServiceContext) http.HandlerFunc {
nameArr := strings.Split(f.Name(), "-") nameArr := strings.Split(f.Name(), "-")
filesSort[nameArr[1]] = f.Name() filesSort[nameArr[1]] = f.Name()
} }
saveFile := path.Join(linuxUploadPath, name) saveFile := filepath.Join(uploadPath, name)
if exists, _ := PathExists(saveFile); exists { if exists, _ := PathExists(saveFile); exists {
os.Remove(saveFile) os.Remove(saveFile)
} }
@ -51,7 +49,7 @@ func ChunkImageHandler(svcCtx *svc.ServiceContext) http.HandlerFunc {
wg.Add(filesCount) wg.Add(filesCount)
for i := 0; i < filesCount; i++ { for i := 0; i < filesCount; i++ {
// 这里一定要注意按顺序读取不然文件就会损坏 // 这里一定要注意按顺序读取不然文件就会损坏
fileName := path.Join(chunksPath, "/"+filesSort[strconv.Itoa(i)]) fileName := filepath.Join(chunksPath, filesSort[strconv.Itoa(i)])
data, err := ioutil.ReadFile(fileName) data, err := ioutil.ReadFile(fileName)
fmt.Println(err) fmt.Println(err)
fs.Write(data) fs.Write(data)
@ -59,7 +57,7 @@ func ChunkImageHandler(svcCtx *svc.ServiceContext) http.HandlerFunc {
wg.Done() wg.Done()
} }
wg.Wait() wg.Wait()
os.RemoveAll(path.Join(chunksPath, "/")) os.RemoveAll(chunksPath)
defer fs.Close() defer fs.Close()
//// 加载镜像文件到docker //// 加载镜像文件到docker

17
adaptor/PCM-CORE/api/internal/handler/image/uploadimagehandler.go
View File

@ -8,8 +8,8 @@ import (
"net/http" "net/http"
"os" "os"
"path" "path"
"path/filepath"
"strconv" "strconv"
"strings"
"syscall" "syscall"
"PCM/adaptor/PCM-CORE/api/internal/svc" "PCM/adaptor/PCM-CORE/api/internal/svc"
@ -19,8 +19,7 @@ type LoadBody struct {
Stream string `json:"stream"` Stream string `json:"stream"`
} }
var windowsUploadTempPath = strings.ReplaceAll(path.Join(windowsUploadPath, "temp"), "/", "\\") var uploadTempPath = filepath.Join(uploadPath, "temp")
var linuxUploadTempPath = path.Join(linuxUploadPath)
func UploadImageHandler(svcCtx *svc.ServiceContext) http.HandlerFunc { func UploadImageHandler(svcCtx *svc.ServiceContext) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) { return func(w http.ResponseWriter, r *http.Request) {
@ -28,28 +27,30 @@ func UploadImageHandler(svcCtx *svc.ServiceContext) http.HandlerFunc {
index := r.PostFormValue("index") index := r.PostFormValue("index")
hash := r.PostFormValue("hash") hash := r.PostFormValue("hash")
// 合并路径 // 合并路径
chunksPath := path.Join(linuxUploadTempPath, hash) chunksPath := filepath.Join(uploadTempPath, hash)
// 文件路径 // 文件路径
filePath := path.Join(chunksPath + "/" + hash + "-" + index) filePath := filepath.Join(chunksPath, hash+"-"+index)
// 检查临时文件夹是否存在 // 检查临时文件夹是否存在
isPathExists, err := PathExists(chunksPath) isPathExists, err := PathExists(chunksPath)
if !isPathExists { if !isPathExists {
err = os.MkdirAll(chunksPath, os.ModePerm) err = os.MkdirAll(chunksPath, os.ModePerm)
} }
// 检查文件是否存在
exists, err := PathExists(filePath) exists, err := PathExists(filePath)
if exists { if exists {
fileInfo, _ := os.Stat(filePath) fileInfo, _ := os.Stat(filePath)
if fileInfo.Size() == fileHeader.Size { if fileInfo.Size() == fileHeader.Size {
result2.HttpResult(r, w, nil, err)
return return
} }
start := strconv.Itoa(int(fileInfo.Size())) start := strconv.Itoa(int(fileInfo.Size()))
oldfile, _ := os.OpenFile(filePath, os.O_CREATE|os.O_WRONLY, os.ModePerm) oldFile, _ := os.OpenFile(filePath, os.O_CREATE|os.O_WRONLY, os.ModePerm)
defer file.Close() defer file.Close()
count, _ := strconv.ParseInt(start, 10, 64) count, _ := strconv.ParseInt(start, 10, 64)
fmt.Println("已上传:", count) fmt.Println("已上传:", count)
// 设置读，写的偏移量 // 设置读，写的偏移量
file.Seek(count, 0) file.Seek(count, 0)
oldfile.Seek(count, 0) oldFile.Seek(count, 0)
data := make([]byte, 1024, 1024) data := make([]byte, 1024, 1024)
for { for {
total, err := file.Read(data) total, err := file.Read(data)
@ -57,7 +58,7 @@ func UploadImageHandler(svcCtx *svc.ServiceContext) http.HandlerFunc {
fmt.Println("文件复制完毕") fmt.Println("文件复制完毕")
break break
} }
oldfile.Write(data[:total]) oldFile.Write(data[:total])
} }
} else { } else {

1
adaptor/PCM-CORE/api/internal/handler/routes.go
View File

@ -292,6 +292,7 @@ func RegisterHandlers(server *rest.Server, serverCtx *svc.ServiceContext) {
Handler: image.CheckHandler(serverCtx), Handler: image.CheckHandler(serverCtx),
}, },
}, },
rest.WithMaxBytes(111111111),
rest.WithPrefix("/pcm/v1"), rest.WithPrefix("/pcm/v1"),
) )
} }

13
adaptor/PCM-CORE/api/internal/logic/image/checklogic.go
View File

@ -1,14 +1,12 @@
package image package image
import ( import (
"PCM/adaptor/PCM-CORE/api/internal/svc"
"PCM/adaptor/PCM-CORE/api/internal/types"
"context" "context"
"io/ioutil" "io/ioutil"
"os" "os"
"path" "path/filepath"
"strings"
"PCM/adaptor/PCM-CORE/api/internal/svc"
"PCM/adaptor/PCM-CORE/api/internal/types"
"github.com/zeromicro/go-zero/core/logx" "github.com/zeromicro/go-zero/core/logx"
) )
@ -20,8 +18,7 @@ type CheckLogic struct {
} }
var dir, _ = os.Getwd() var dir, _ = os.Getwd()
var windowsUploadPath = strings.ReplaceAll(path.Join(dir, "uploads"), "/", "\\") var uploadPath = filepath.Join(dir, "uploads")
var linuxUploadPath = path.Join(dir, "uploads")
func NewCheckLogic(ctx context.Context, svcCtx *svc.ServiceContext) *CheckLogic { func NewCheckLogic(ctx context.Context, svcCtx *svc.ServiceContext) *CheckLogic {
return &CheckLogic{ return &CheckLogic{
@ -33,7 +30,7 @@ func NewCheckLogic(ctx context.Context, svcCtx *svc.ServiceContext) *CheckLogic
func (l *CheckLogic) Check(req *types.CheckReq) (resp *types.CheckResp, err error) { func (l *CheckLogic) Check(req *types.CheckReq) (resp *types.CheckResp, err error) {
// todo: add your logic here and delete this line // todo: add your logic here and delete this line
files, err := ioutil.ReadDir(windowsUploadPath) files, err := ioutil.ReadDir(uploadPath)
if err != nil { if err != nil {
return nil, err return nil, err
} }

5
adaptor/PCM-HPC/PCM-TH/rpc/internal/logic/cronlogic.go
View File

@ -17,7 +17,6 @@ func InitCron(svc *svc.ServiceContext) {
svc.Cron.Start() svc.Cron.Start()
submitJobLogic := NewSubmitJobLogic(context.Background(), svc) submitJobLogic := NewSubmitJobLogic(context.Background(), svc)
listLogic := NewListJobLogic(context.Background(), svc) listLogic := NewListJobLogic(context.Background(), svc)
historyListLogic := NewListHistoryJobLogic(context.Background(), svc)
svc.Cron.AddFunc("*/5 * * * * ?", func() { svc.Cron.AddFunc("*/5 * * * * ?", func() {
// 查询core端分发下来的任务列表 // 查询core端分发下来的任务列表
@ -31,7 +30,7 @@ func InitCron(svc *svc.ServiceContext) {
return return
} }
// 提交任务 // 提交任务
submitJob(infoList, submitJobLogic, historyListLogic) submitJob(infoList, submitJobLogic)
// 查询运行中的任务列表同步信息 // 查询运行中的任务列表同步信息
listReq := hpcTH.ListJobReq{} listReq := hpcTH.ListJobReq{}
listJob, err := listLogic.ListJob(&listReq) listJob, err := listLogic.ListJob(&listReq)
@ -63,7 +62,7 @@ func InitCron(svc *svc.ServiceContext) {
}) })
} }
func submitJob(infoList *pcmcoreclient.InfoListResp, submitJobLogic *SubmitJobLogic, historyListLogic *ListHistoryJobLogic) { func submitJob(infoList *pcmcoreclient.InfoListResp, submitJobLogic *SubmitJobLogic) {
for index, _ := range infoList.HpcInfoList { for index, _ := range infoList.HpcInfoList {
if infoList.HpcInfoList[index].Status == "Saved" { if infoList.HpcInfoList[index].Status == "Saved" {
submitReq := hpcTH.SubmitJobReq{ submitReq := hpcTH.SubmitJobReq{