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>
2026-10-22 22:11:39 +02:00 · 2026-10-21 20:08:57 -07:00
1 parent d35f4c1c6b
commit 164a7d8b56
@@ -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
 }
@@ -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()