From 164a7d8b56042254da174d2e5ac5796276aece9b Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Wed, 21 Oct 2026 20:08:57 -0700 Subject: [PATCH] upload: upload -dir files concurrently behind a new -c flag (#11918) * upload: upload -dir files concurrently behind a new -c flag Each file under -dir was submitted serially inside filepath.Walk, so every small file paid a sequential master-assign plus upload round trip. A bounded pool now runs those submissions concurrently while keeping results in walk order; failures are reported per file and the command exits non-zero when any upload failed. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * submit: close opened readers when a batch assign fails SubmitFiles returns early when the assign RPC fails, before FilePart.Upload gets a chance to close each opened file. Close the readers on that path so repeated assign failures do not leak handles. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * upload: stream -dir results as each file completes Batching the output until every upload finished meant an interrupted run lost the file ids of completed uploads. A per-index ready channel lets a printer emit each result in walk order as soon as it lands. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * submit: close the opened file when Stat fails newFilePart returned the still-open handle with the error, and the concurrent upload path records the error without ever reaching SubmitFiles, leaking one descriptor per failing file. * upload: still upload files collected before a walk error A walk that aborts partway used to leave the already-uploaded files behind; collecting paths first must not drop them, so upload what was gathered and still exit non-zero. * upload: submit batched files concurrently SubmitFiles fetched all file ids in one assign RPC but then uploaded the files one after another, so a multi-file list still spent its time in serial per-file round trips even with -dir concurrency. Uploads now run through a small worker pool while results keep their input order. Verified end to end: a 5-file list uploads with consecutive file keys and each stored needle matches its source bytes. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * upload: print per-file results before exiting on a partial batch failure --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- weed/command/upload.go | 92 +++++++++++++++++++++++++++++++--------- weed/operation/submit.go | 43 ++++++++++++++++--- 2 files changed, 110 insertions(+), 25 deletions(-) diff --git a/weed/command/upload.go b/weed/command/upload.go index bb542a030c..def64adc77 100644 --- a/weed/command/upload.go +++ b/weed/command/upload.go @@ -6,6 +6,7 @@ import ( "fmt" "os" "path/filepath" + "sync" "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" @@ -31,6 +32,7 @@ type UploadOptions struct { diskType *string maxMB *int usePublicUrl *bool + concurrency *int } func init() { @@ -46,6 +48,7 @@ func init() { upload.ttl = cmdUpload.Flag.String("ttl", "", "time to live, e.g.: 1m, 1h, 1d, 1M, 1y") upload.maxMB = cmdUpload.Flag.Int("maxMB", 4, "split files larger than the limit") upload.usePublicUrl = cmdUpload.Flag.Bool("usePublicUrl", false, "upload to public url from volume server") + upload.concurrency = cmdUpload.Flag.Int("c", 8, "concurrent file upload goroutines when uploading a folder") } var cmdUpload = &Command{ @@ -85,6 +88,7 @@ func runUpload(cmd *Command, args []string) bool { if *upload.dir == "" { return false } + var filePaths []string err = filepath.Walk(util.ResolvePath(*upload.dir), func(path string, info os.FileInfo, err error) error { if err == nil { if !info.IsDir() { @@ -93,23 +97,7 @@ func runUpload(cmd *Command, args []string) bool { return nil } } - parts, e := operation.NewFileParts([]string{path}) - if e != nil { - return e - } - results, e := operation.SubmitFiles(func(_ context.Context) pb.ServerAddress { return pb.ServerAddress(*upload.master) }, grpcDialOption, parts, operation.StoragePreference{ - Replication: *upload.replication, - Collection: *upload.collection, - DataCenter: *upload.dataCenter, - Ttl: *upload.ttl, - DiskType: *upload.diskType, - MaxMB: *upload.maxMB, - }, *upload.usePublicUrl) - bytes, _ := json.Marshal(results) - fmt.Println(string(bytes)) - if e != nil { - return e - } + filePaths = append(filePaths, path) } } else { fmt.Println(err) @@ -118,6 +106,72 @@ func runUpload(cmd *Command, args []string) bool { }) if err != nil { fmt.Println(err.Error()) + } + + concurrency := *upload.concurrency + if concurrency < 1 { + concurrency = 1 + } + masterFn := func(_ context.Context) pb.ServerAddress { return pb.ServerAddress(*upload.master) } + pref := operation.StoragePreference{ + Replication: *upload.replication, + Collection: *upload.collection, + DataCenter: *upload.dataCenter, + Ttl: *upload.ttl, + DiskType: *upload.diskType, + MaxMB: *upload.maxMB, + } + results := make([][]operation.SubmitResult, len(filePaths)) + ready := make([]chan struct{}, len(filePaths)) + for index := range ready { + ready[index] = make(chan struct{}) + } + printDone := make(chan bool, 1) + go func() { + failed := false + for index := range ready { + <-ready[index] + bytes, _ := json.Marshal(results[index]) + fmt.Println(string(bytes)) + for _, r := range results[index] { + if r.Error != "" { + failed = true + } + } + } + printDone <- failed + }() + + sem := make(chan struct{}, concurrency) + var wg sync.WaitGroup + for index, filePath := range filePaths { + sem <- struct{}{} + wg.Add(1) + go func(index int, filePath string) { + defer close(ready[index]) + defer wg.Done() + defer func() { <-sem }() + parts, e := operation.NewFileParts([]string{filePath}) + if e != nil { + results[index] = []operation.SubmitResult{{FileName: filepath.Base(filePath), Error: e.Error()}} + return + } + fileResults, e := operation.SubmitFiles(masterFn, grpcDialOption, parts, pref, *upload.usePublicUrl) + if e != nil { + for i := range fileResults { + if fileResults[i].Error == "" { + fileResults[i].Error = e.Error() + } + } + } + results[index] = fileResults + }(index, filePath) + } + wg.Wait() + if <-printDone { + return false + } + if err != nil { return false } } else { @@ -134,12 +188,12 @@ func runUpload(cmd *Command, args []string) bool { DiskType: *upload.diskType, MaxMB: *upload.maxMB, }, *upload.usePublicUrl) + bytes, _ := json.Marshal(results) + fmt.Println(string(bytes)) if err != nil { fmt.Println(err.Error()) return false } - bytes, _ := json.Marshal(results) - fmt.Println(string(bytes)) } return true } diff --git a/weed/operation/submit.go b/weed/operation/submit.go index 1d94f5bab3..347b1647fd 100644 --- a/weed/operation/submit.go +++ b/weed/operation/submit.go @@ -10,6 +10,7 @@ import ( "path" "strconv" "strings" + "sync" "github.com/seaweedfs/seaweedfs/weed/pb" @@ -50,6 +51,10 @@ type StoragePreference struct { type GetMasterFn func(ctx context.Context) pb.ServerAddress +// submitUploadWorkers bounds how many files one SubmitFiles call uploads at +// once; the batch assign above already fetched all their file ids. +const submitUploadWorkers = 4 + func SubmitFiles(masterFn GetMasterFn, grpcDialOption grpc.DialOption, files []*FilePart, pref StoragePreference, usePublicUrl bool) ([]SubmitResult, error) { results := make([]SubmitResult, len(files)) var totalBytes int64 @@ -72,8 +77,11 @@ func SubmitFiles(masterFn GetMasterFn, grpcDialOption grpc.DialOption, files []* } ret, err := Assign(context.Background(), masterFn, grpcDialOption, ar) if err != nil { - for index := range files { + for index, file := range files { results[index].Error = err.Error() + if closer, ok := file.Reader.(io.Closer); ok { + closer.Close() + } } return results, err } @@ -87,14 +95,35 @@ func SubmitFiles(masterFn GetMasterFn, grpcDialOption grpc.DialOption, files []* file.Server = ret.PublicUrl } file.Pref = pref - results[index].Size, err = file.Upload(pref.MaxMB, masterFn, usePublicUrl, ret.Auth, grpcDialOption) - if err != nil { - results[index].Error = err.Error() - } results[index].Fid = file.Fid results[index].FileUrl = ret.PublicUrl + "/" + file.Fid } - return results, nil + + var wg sync.WaitGroup + var lastErrLock sync.Mutex + jobs := make(chan int) + for w := 0; w < min(submitUploadWorkers, len(files)); w++ { + wg.Add(1) + go func() { + defer wg.Done() + for index := range jobs { + size, uploadErr := files[index].Upload(pref.MaxMB, masterFn, usePublicUrl, ret.Auth, grpcDialOption) + results[index].Size = size + if uploadErr != nil { + results[index].Error = uploadErr.Error() + lastErrLock.Lock() + err = uploadErr + lastErrLock.Unlock() + } + } + }() + } + for index := range files { + jobs <- index + } + close(jobs) + wg.Wait() + return results, err } func NewFileParts(fullPathFilenames []string) (ret []*FilePart, err error) { @@ -118,6 +147,8 @@ func newFilePart(fullPathFilename string) (ret *FilePart, err error) { fi, fiErr := fh.Stat() if fiErr != nil { glog.V(0).Info("Failed to stat file:", fullPathFilename) + fh.Close() + ret.Reader = nil return ret, fiErr } ret.ModTime = fi.ModTime().UTC().Unix()