Repository: juicedata/juicefs
Pull request #5535 (2025-01-09 11:23:13)

cmd/other: threads option for webdav and libjfs
Introduce threads option for webdav and gateway rmr operation

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/juicedata/juicefs/pull/5535?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=juicedata) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Project coverage is 21.49%. Comparing base [(`52a48a2`)](https://app.codecov.io/gh/juicedata/juicefs/commit/52a48a2989b25833003043af3ea06095faa16249?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=juicedata) to head [(`4c5b634`)](https://app.codecov.io/gh/juicedata/juicefs/commit/4c5b634eb52a7d03cb4ccf8c734bf9772ff6cc3d?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=juicedata).
> Report is 3 commits behind head on main.

<details><summary>Additional details and impacted files</summary>


```diff
@@             Coverage Diff             @@
##             main    #5535       +/-   ##
===========================================
- Coverage   54.81%   21.49%   -33.33%     
===========================================
  Files         165       29      -136     
  Lines       48086    19167    -28919     
===========================================
- Hits        26356     4119    -22237     
+ Misses      18851    14558     -4293     
+ Partials     2879      490     -2389     
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/juicedata/juicefs/pull/5535?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=juicedata).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=juicedata).


Patch:
diff --git a/cmd/webdav.go b/cmd/webdav.go
index d76467e4..e0fd4ac8 100644
--- a/cmd/webdav.go
+++ b/cmd/webdav.go
@@ -64,6 +64,12 @@ func cmdWebDav() *cli.Command {
 			Aliases: []string{"d"},
 			Usage:   "run in background",
 		},
+                &cli.IntFlag{
+                        Name:    "threads",
+                        Aliases: []string{"p"},
+                        Value:   50,
+                        Usage:   "number of threads for delete jobs (max 255)",
+                },
 	}
 
 	return &cli.Command{
@@ -95,6 +101,7 @@ func webdav(c *cli.Context) error {
 		CertFile:        c.String("cert-file"),
 		KeyFile:         c.String("key-file"),
 		EnableProppatch: c.Bool("enable-proppatch"),
+		MaxDeletes:      c.Int("threads"),
 	})
 	return jfs.Meta().CloseSession()
 }
diff --git a/pkg/fs/fs.go b/pkg/fs/fs.go
index 43ed5d24..c800a595 100644
--- a/pkg/fs/fs.go
+++ b/pkg/fs/fs.go
@@ -501,7 +501,7 @@ func (fs *FileSystem) Delete(ctx meta.Context, p string) (err syscall.Errno) {
 	return
 }
 
-func (fs *FileSystem) Rmr(ctx meta.Context, p string) (err syscall.Errno) {
+func (fs *FileSystem) Rmr(ctx meta.Context, p string, numthreads int) (err syscall.Errno) {
 	defer trace.StartRegion(context.TODO(), "fs.Rmr").End()
 	l := vfs.NewLogContext(ctx)
 	defer func() { fs.log(l, "Rmr (%s): %s", p, errstr(err)) }()
@@ -509,7 +509,7 @@ func (fs *FileSystem) Rmr(ctx meta.Context, p string) (err syscall.Errno) {
 	if err != 0 {
 		return
 	}
-	err = fs.m.Remove(ctx, parent.inode, path.Base(p), false, meta.RmrDefaultThreads, nil)
+	err = fs.m.Remove(ctx, parent.inode, path.Base(p), false, numthreads, nil)
 	fs.invalidateEntry(parent.inode, path.Base(p))
 	return
 }
diff --git a/pkg/fs/http.go b/pkg/fs/http.go
index b327b395..69cacea8 100644
--- a/pkg/fs/http.go
+++ b/pkg/fs/http.go
@@ -125,7 +125,7 @@ func (hfs *webdavFS) OpenFile(ctx context.Context, name string, flag int, perm o
 }
 
 func (hfs *webdavFS) RemoveAll(ctx context.Context, name string) error {
-	return econv(hfs.fs.Rmr(hfs.ctx, name))
+	return econv(hfs.fs.Rmr(hfs.ctx, name, hfs.config.MaxDeletes))
 }
 
 func (hfs *webdavFS) Rename(ctx context.Context, oldName, newName string) error {
@@ -252,6 +252,7 @@ type WebdavConfig struct {
 	Password        string
 	CertFile        string
 	KeyFile         string
+	MaxDeletes	int
 }
 
 type indexHandler struct {
diff --git a/pkg/gateway/gateway.go b/pkg/gateway/gateway.go
index 687dd575..5c6f7187 100644
--- a/pkg/gateway/gateway.go
+++ b/pkg/gateway/gateway.go
@@ -1162,7 +1162,7 @@ func (n *jfsObjects) CompleteMultipartUpload(ctx context.Context, bucket, object
 	}
 
 	// remove parts
-	_ = n.fs.Rmr(mctx, n.upath(bucket, uploadID))
+	_ = n.fs.Rmr(mctx, n.upath(bucket, uploadID), meta.RmrDefaultThreads)
 	return minio.ObjectInfo{
 		Bucket:   bucket,
 		Name:     object,
@@ -1179,7 +1179,7 @@ func (n *jfsObjects) AbortMultipartUpload(ctx context.Context, bucket, object, u
 	if err = n.checkUploadIDExists(ctx, bucket, object, uploadID); err != nil {
 		return
 	}
-	eno := n.fs.Rmr(mctx, n.upath(bucket, uploadID))
+	eno := n.fs.Rmr(mctx, n.upath(bucket, uploadID), meta.RmrDefaultThreads)
 	return jfsToObjectErr(ctx, eno, bucket, object, uploadID)
 }
 
@@ -1224,7 +1224,7 @@ func (n *jfsObjects) cleanupDir(dir string) bool {
 			continue
 		}
 		if now.Sub(time.Unix(entry.Attr.Mtime, 0)) > 7*24*time.Hour {
-			if errno = n.fs.Rmr(mctx, dirPath); errno != 0 {
+			if errno = n.fs.Rmr(mctx, dirPath, meta.RmrDefaultThreads); errno != 0 {
 				logger.Errorf("failed to delete expired temporary files path: %s, err: %s", dirPath, errno)
 			} else {
 				deleted += 1
diff --git a/pkg/meta/sql.go b/pkg/meta/sql.go
index 2a1267d0..9f46bf2f 100644
--- a/pkg/meta/sql.go
+++ b/pkg/meta/sql.go
@@ -290,8 +290,8 @@ func retriveUrlConnsOptions(murl string) (string, int, int, int, int) {
 
 	var vOpenConns int = 0
 	var vIdleConns int = runtime.GOMAXPROCS(-1) * 2
-	var vIdleTime  int = 300
-	var vLifeTime  int = 0
+	var vIdleTime int = 300
+	var vLifeTime int = 0
 
 	if optIndex != -1 {
 		baseurl := murl[:optIndex]
@@ -307,11 +307,11 @@ func retriveUrlConnsOptions(murl string) (string, int, int, int, int) {
 			}
 			if vals.Has("max_idle_time") {
 				vIdleTime, _ = strconv.Atoi(vals.Get("max_idle_time"))
-				vals.Del("max_idle_time");
+				vals.Del("max_idle_time")
 			}
 			if vals.Has("max_life_time") {
 				vLifeTime, _ = strconv.Atoi(vals.Get("max_life_time"))
-				vals.Del("max_life_time");
+				vals.Del("max_life_time")
 			}
 			optsurl = vals.Encode()
 		}
diff --git a/pkg/sync/sync.go b/pkg/sync/sync.go
index 7587d7af..e21ece82 100644
--- a/pkg/sync/sync.go
+++ b/pkg/sync/sync.go
@@ -727,7 +727,14 @@ func doCopyMultiple(src, dst object.ObjectStorage, key string, size int64, uploa
 	return chksum, nil
 }
 
-func copyData(src, dst object.ObjectStorage, key string, size int64, calChksum bool) (uint32, error) {
+func InitForCopyData() {
+	concurrent = make(chan int, 10)
+	progress := utils.NewProgress(true)
+	copied = progress.AddCountSpinner("Copied objects")
+	copiedBytes = progress.AddByteSpinner("Copied bytes")
+}
+
+func CopyData(src, dst object.ObjectStorage, key string, size int64, calChksum bool) (uint32, error) {
 	start := time.Now()
 	var err error
 	var srcChksum uint32
@@ -825,7 +832,7 @@ func worker(tasks <-chan object.Object, src, dst object.ObjectStorage, config *C
 					logger.Errorf("copy link failed: %s", err)
 				}
 			} else {
-				srcChksum, err = copyData(src, dst, key, obj.Size(), config.CheckAll || config.CheckNew)
+				srcChksum, err = CopyData(src, dst, key, obj.Size(), config.CheckAll || config.CheckNew)
 			}
 
 			if err == nil && (config.CheckAll || config.CheckNew) {
diff --git a/pkg/vfs/backup.go b/pkg/vfs/backup.go
index 5bf28153..70382823 100644
--- a/pkg/vfs/backup.go
+++ b/pkg/vfs/backup.go
@@ -18,9 +18,13 @@ package vfs
 
 import (
 	"compress/gzip"
+	"errors"
 	"io"
 	"os"
+	"path/filepath"
 	"sort"
+	"strings"
+	"syscall"
 	"time"
 
 	"github.com/juicedata/juicefs/pkg/meta"
@@ -92,7 +96,17 @@ func Backup(m meta.Meta, blob object.ObjectStorage, interval time.Duration, skip
 
 func backup(m meta.Meta, blob object.ObjectStorage, now time.Time, fast, skipTrash bool) (string, error) {
 	name := "dump-" + now.UTC().Format("2006-01-02-150405") + ".json.gz"
-	fp, err := os.CreateTemp("", "juicefs-meta-*")
+	localDir := os.TempDir()
+	if !strings.HasSuffix(localDir, "/") {
+		localDir += "/"
+	}
+	fp, err := os.Create(filepath.Join(localDir, "meta", name))
+	if errors.Is(err, syscall.ENOENT) {
+		if err = os.MkdirAll(filepath.Join(localDir, "meta"), 0755); err != nil {
+			return "", err
+		}
+		fp, err = os.Create(filepath.Join(localDir, "meta", name))
+	}
 	if err != nil {
 		return "", err
 	}
@@ -108,11 +122,19 @@ func backup(m meta.Meta, blob object.ObjectStorage, now time.Time, fast, skipTra
 	if err != nil {
 		return "", err
 	}
-	if _, err = fp.Seek(0, io.SeekStart); err != nil {
+	size, err := fp.Seek(0, io.SeekCurrent)
+	if err != nil {
 		return "", err
 	}
+
 	fpath := "meta/" + name
-	return blob.String() + fpath, blob.Put(fpath, fp)
+	disk, err := object.CreateStorage("file", localDir, "", "", "")
+	if err != nil {
+		return "", err
+	}
+	osync.InitForCopyData()
+	_, err = osync.CopyData(disk, blob, fpath, size, true)
+	return blob.String() + fpath, err
 }
 
 func cleanupBackups(blob object.ObjectStorage, now time.Time) {
diff --git a/sdk/java/libjfs/main.go b/sdk/java/libjfs/main.go
index 08698e6a..3f80091e 100644
--- a/sdk/java/libjfs/main.go
+++ b/sdk/java/libjfs/main.go
@@ -94,6 +94,7 @@ var (
 
 	userGroupCache = make(map[string]map[string][]string) // name -> (user -> groups)
 
+	MaxDeletes = meta.RmrDefaultThreads
 )
 
 const (
@@ -467,6 +468,10 @@ func jfs_init(cname, jsonConf, user, group, superuser, supergroup *C.char) int64
 			utils.SetLogLevel(logrus.WarnLevel)
 		}
 
+		if jConf.MaxDeletes > 0 {
+			MaxDeletes = jConf.MaxDeletes
+		}
+
 		metaConf := meta.DefaultConf()
 		metaConf.Retries = jConf.IORetries
 		metaConf.MaxDeletes = jConf.MaxDeletes
@@ -865,7 +870,7 @@ func jfs_rmr(pid int64, h int64, cpath *C.char) int32 {
 	if w == nil {
 		return EINVAL
 	}
-	return errno(w.Rmr(w.withPid(pid), C.GoString(cpath)))
+	return errno(w.Rmr(w.withPid(pid), C.GoString(cpath), MaxDeletes))
 }
 
 //export jfs_rename


Test patch:
diff --git a/pkg/fs/fs_test.go b/pkg/fs/fs_test.go
index 47bede03..d1d91cbe 100644
--- a/pkg/fs/fs_test.go
+++ b/pkg/fs/fs_test.go
@@ -230,7 +230,7 @@ func TestFileSystem(t *testing.T) {
 	if err := fs.Delete(ctx, "/d/f"); err == 0 || !IsNotExist(err) {
 		t.Fatalf("delete /d/f: %s", err)
 	}
-	if e := fs.Rmr(ctx, "/d"); e != 0 {
+	if e := fs.Rmr(ctx, "/d", meta.RmrDefaultThreads); e != 0 {
 		t.Fatalf("delete /d -r: %s", e)
 	}
 
@@ -261,7 +261,7 @@ func TestFileSystem(t *testing.T) {
 	if err := fs.Rename(ctx, "/ddd/", "/ttt/", 0); err != 0 {
 		t.Fatalf("delete /ddd/: %s", err)
 	}
-	if err := fs.Rmr(ctx, "/ttt/"); err != 0 {
+	if err := fs.Rmr(ctx, "/ttt/", meta.RmrDefaultThreads); err != 0 {
 		t.Fatalf("rmr /ttt/: %s", err)
 	}
 	if _, err := fs.Stat(ctx, "/ttt/"); err != syscall.ENOENT {