mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-10-22 06:22:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

upload -dir: fetch file ids in batches instead of per file

Browse Source 
Directory upload submitted every file through its own SubmitFiles
call, so each file paid a separate serial assign RPC. Files are now
collected during the walk and submitted in batches, sharing one batch
assign and the parallel uploads.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Chris Lu
2026-10-21 15:10:30 -07:00
co-authored by Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
parent c8002d63bb
commit 84cc179b80
1 changed files with 28 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
weed/command/upload.go
+28 -17
View File
@@ -20,6 +20,10 @@ var (
upload UploadOptions
)
// uploadBatchSize is how many files a -dir upload submits per batch: one
// assign RPC fetches all their file ids, and the batch uploads in parallel.
const uploadBatchSize = 32
type UploadOptions struct {
master *string
dir *string
@@ -85,6 +89,7 @@ func runUpload(cmd *Command, args []string) bool {
if *upload.dir == "" {
return false
}
var paths []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 +98,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
}
paths = append(paths, path)
}
} else {
fmt.Println(err)
@@ -120,6 +109,28 @@ func runUpload(cmd *Command, args []string) bool {
fmt.Println(err.Error())
return false
}
for batchStart := 0; batchStart < len(paths); batchStart += uploadBatchSize {
batchEnd := min(batchStart+uploadBatchSize, len(paths))
parts, e := operation.NewFileParts(paths[batchStart:batchEnd])
if e != nil {
fmt.Println(e.Error())
return false
}
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 {
fmt.Println(e.Error())
return false
}
}
} else {
parts, e := operation.NewFileParts(args)
if e != nil {