Repository: juicedata/juicefs
Pull request #2214 (2022-06-13 14:59:43)

cmd/rmr: remove entries in batch
When removing directories with many files in parallel, they need lots of memory for these entries. 

This PR change to remove them as smaller batches (10000), so it will hold at most 500K entries in memory (100-200MB). 

For smaller directories, it will take one more readdir() to ensure that the directory is empty (better for concurrent adding files).


Patch:
diff --git a/go.mod b/go.mod
index 817ea624..6ad72305 100644
--- a/go.mod
+++ b/go.mod
@@ -229,7 +229,7 @@ require (
 	xorm.io/builder v0.3.7 // indirect
 )
 
-replace github.com/minio/minio v0.0.0-20210206053228-97fe57bba92c => github.com/juicedata/minio v0.0.0-20220324072921-2057662dce6d
+replace github.com/minio/minio v0.0.0-20210206053228-97fe57bba92c => github.com/juicedata/minio v0.0.0-20220613143934-cee0571a1b03
 
 replace github.com/hanwen/go-fuse/v2 v2.1.1-0.20210611132105-24a1dfe6b4f8 => github.com/juicedata/go-fuse/v2 v2.1.1-0.20210926080226-cfe1ec802a7f
 
diff --git a/go.sum b/go.sum
index f3544933..7ccbba25 100644
--- a/go.sum
+++ b/go.sum
@@ -609,8 +609,8 @@ github.com/juicedata/go-fuse/v2 v2.1.1-0.20210926080226-cfe1ec802a7f h1:EhufxJV9
 github.com/juicedata/go-fuse/v2 v2.1.1-0.20210926080226-cfe1ec802a7f/go.mod h1:oRyA5eK+pvJyv5otpO/DgccS8y/RvYMaO00GgRLGryc=
 github.com/juicedata/godaemon v0.0.0-20210629045518-3da5144a127d h1:kpQMvNZJKGY3PTt7OSoahYc4nM0HY67SvK0YyS0GLwA=
 github.com/juicedata/godaemon v0.0.0-20210629045518-3da5144a127d/go.mod h1:dlxKkLh3qAIPtgr2U/RVzsZJDuXA1ffg+Njikfmhvgw=
-github.com/juicedata/minio v0.0.0-20220324072921-2057662dce6d h1:Jkul/jGraZy7U3o8xWztM3cf3e1uiKIiPaCg+cRPaz8=
-github.com/juicedata/minio v0.0.0-20220324072921-2057662dce6d/go.mod h1:ILQUt98Xz3GF9KgJKeGQ81c0RFaRSrUZmZ8oSb/7QxU=
+github.com/juicedata/minio v0.0.0-20220613143934-cee0571a1b03 h1:+0jiEkr6qE5Zm2+LilAv5pV/+V+V1XZ5KUv0pmXZHhQ=
+github.com/juicedata/minio v0.0.0-20220613143934-cee0571a1b03/go.mod h1:ILQUt98Xz3GF9KgJKeGQ81c0RFaRSrUZmZ8oSb/7QxU=
 github.com/juicedata/mpb/v7 v7.0.4-0.20220411070607-092927ed0122 h1:2YZ6nb3kO38/WIG5L5dwrlrElXtCNKRIdrXm2c9meOk=
 github.com/juicedata/mpb/v7 v7.0.4-0.20220411070607-092927ed0122/go.mod h1:NXGsfPGx6G2JssqvEcULtDqUrxuuYs4llpv8W6ZUpzk=
 github.com/juju/ratelimit v1.0.1 h1:+7AIFJVQ0EQgq/K9+0Krm7m530Du7tIz0METWzN0RgY=
diff --git a/pkg/fs/fs.go b/pkg/fs/fs.go
index ad5dc7b0..85da5c61 100644
--- a/pkg/fs/fs.go
+++ b/pkg/fs/fs.go
@@ -438,7 +438,7 @@ func (fs *FileSystem) Rmr(ctx meta.Context, p string) (err syscall.Errno) {
 	if err != 0 {
 		return
 	}
-	err = meta.Remove(fs.m, ctx, parent.inode, path.Base(p), nil)
+	err = fs.m.Remove(ctx, parent.inode, path.Base(p), nil)
 	fs.invalidateEntry(parent.inode, path.Base(p))
 	return
 }
diff --git a/pkg/meta/base.go b/pkg/meta/base.go
index 6ad5a043..92a1b664 100644
--- a/pkg/meta/base.go
+++ b/pkg/meta/base.go
@@ -105,14 +105,14 @@ type baseMeta struct {
 	en engine
 }
 
-func newBaseMeta(addr string, conf *Config) baseMeta {
+func newBaseMeta(addr string, conf *Config) *baseMeta {
 	if conf.Retries == 0 {
 		conf.Retries = 10
 	}
 	if conf.Heartbeat == 0 {
 		conf.Heartbeat = 12 * time.Second
 	}
-	return baseMeta{
+	return &baseMeta{
 		addr:         utils.RemovePassword(addr),
 		conf:         conf,
 		root:         1,
@@ -127,6 +127,10 @@ func newBaseMeta(addr string, conf *Config) baseMeta {
 	}
 }
 
+func (m *baseMeta) getBase() *baseMeta {
+	return m
+}
+
 func (m *baseMeta) checkRoot(inode Ino) Ino {
 	switch inode {
 	case 0:
diff --git a/pkg/meta/interface.go b/pkg/meta/interface.go
index 72154eb6..ca42d873 100644
--- a/pkg/meta/interface.go
+++ b/pkg/meta/interface.go
@@ -337,6 +337,8 @@ type Meta interface {
 	CompactAll(ctx Context, bar *utils.Bar) syscall.Errno
 	// ListSlices returns all slices used by all files.
 	ListSlices(ctx Context, slices map[Ino][]Slice, delete bool, showProgress func()) syscall.Errno
+	// Remove all files and directories recursively.
+	Remove(ctx Context, parent Ino, name string, count *uint64) syscall.Errno
 
 	// OnMsg add a callback for the given message type.
 	OnMsg(mtype uint32, cb MsgCallback)
@@ -344,6 +346,9 @@ type Meta interface {
 	// Dump the tree under root, which may be modified by checkRoot
 	DumpMeta(w io.Writer, root Ino) error
 	LoadMeta(r io.Reader) error
+
+	// getBase return the base engine.
+	getBase() *baseMeta
 }
 
 type Creator func(driver, addr string, conf *Config) (Meta, error)
diff --git a/pkg/meta/redis.go b/pkg/meta/redis.go
index f94ecabc..08d3f08d 100644
--- a/pkg/meta/redis.go
+++ b/pkg/meta/redis.go
@@ -72,7 +72,7 @@ import (
 */
 
 type redisMeta struct {
-	baseMeta
+	*baseMeta
 	rdb        redis.UniversalClient
 	prefix     string
 	shaLookup  string // The SHA returned by Redis for the loaded `scriptLookup`
diff --git a/pkg/meta/sql.go b/pkg/meta/sql.go
index 2975ecf5..e083e5bf 100644
--- a/pkg/meta/sql.go
+++ b/pkg/meta/sql.go
@@ -155,7 +155,7 @@ type delfile struct {
 }
 
 type dbMeta struct {
-	baseMeta
+	*baseMeta
 	db   *xorm.Engine
 	snap *dbSnap
 }
diff --git a/pkg/meta/tkv.go b/pkg/meta/tkv.go
index 91f2fe64..9ff8a970 100644
--- a/pkg/meta/tkv.go
+++ b/pkg/meta/tkv.go
@@ -61,7 +61,7 @@ type tkvClient interface {
 }
 
 type kvMeta struct {
-	baseMeta
+	*baseMeta
 	client tkvClient
 	snap   map[Ino]*DumpedEntry
 }
diff --git a/pkg/meta/utils.go b/pkg/meta/utils.go
index 33f7d149..8e7c80f2 100644
--- a/pkg/meta/utils.go
+++ b/pkg/meta/utils.go
@@ -222,67 +222,72 @@ func updateLocks(ls []plockRecord, nl plockRecord) []plockRecord {
 	return ls
 }
 
-func emptyDir(r Meta, ctx Context, inode Ino, count *uint64, concurrent chan int) syscall.Errno {
-	if st := r.Access(ctx, inode, 3, nil); st != 0 {
+func (m *baseMeta) emptyDir(ctx Context, inode Ino, count *uint64, concurrent chan int) syscall.Errno {
+	if st := m.Access(ctx, inode, 3, nil); st != 0 {
 		return st
 	}
-	var entries []*Entry
-	if st := r.Readdir(ctx, inode, 0, &entries); st != 0 {
-		return st
-	}
-	var wg sync.WaitGroup
-	var status syscall.Errno
-	// try directories first to increase parallel
-	var dirs int
-	for i, e := range entries {
-		if e.Attr.Typ == TypeDirectory {
-			entries[dirs], entries[i] = entries[i], entries[dirs]
-			dirs++
+	for {
+		var entries []*Entry
+		if st := m.en.doReaddir(ctx, inode, 0, &entries, 10000); st != 0 {
+			return st
 		}
-	}
-	for _, e := range entries {
-		if e.Inode == inode || len(e.Name) == 2 && string(e.Name) == ".." {
-			continue
+		if len(entries) == 0 {
+			return 0
+		}
+		var wg sync.WaitGroup
+		var status syscall.Errno
+		// try directories first to increase parallel
+		var dirs int
+		for i, e := range entries {
+			if e.Attr.Typ == TypeDirectory {
+				entries[dirs], entries[i] = entries[i], entries[dirs]
+				dirs++
+			}
 		}
-		if e.Attr.Typ == TypeDirectory {
-			select {
-			case concurrent <- 1:
-				wg.Add(1)
-				go func(child Ino, name string) {
-					defer wg.Done()
-					e := emptyEntry(r, ctx, inode, name, child, count, concurrent)
-					if e != 0 {
-						status = e
+		for i, e := range entries {
+			if e.Attr.Typ == TypeDirectory {
+				select {
+				case concurrent <- 1:
+					wg.Add(1)
+					go func(child Ino, name string) {
+						defer wg.Done()
+						e := m.emptyEntry(ctx, inode, name, child, count, concurrent)
+						if e != 0 {
+							status = e
+						}
+						<-concurrent
+					}(e.Inode, string(e.Name))
+				default:
+					if st := m.emptyEntry(ctx, inode, string(e.Name), e.Inode, count, concurrent); st != 0 {
+						return st
 					}
-					<-concurrent
-				}(e.Inode, string(e.Name))
-			default:
-				if st := emptyEntry(r, ctx, inode, string(e.Name), e.Inode, count, concurrent); st != 0 {
+				}
+			} else {
+				if count != nil {
+					atomic.AddUint64(count, 1)
+				}
+				if st := m.Unlink(ctx, inode, string(e.Name)); st != 0 {
 					return st
 				}
 			}
-		} else {
-			if count != nil {
-				atomic.AddUint64(count, 1)
-			}
-			if st := r.Unlink(ctx, inode, string(e.Name)); st != 0 {
-				return st
+			if ctx.Canceled() {
+				return syscall.EINTR
 			}
+			entries[i] = nil // release memory
 		}
-		if ctx.Canceled() {
-			return syscall.EINTR
+		wg.Wait()
+		if status != 0 {
+			return status
 		}
 	}
-	wg.Wait()
-	return status
 }
 
-func emptyEntry(r Meta, ctx Context, parent Ino, name string, inode Ino, count *uint64, concurrent chan int) syscall.Errno {
-	st := emptyDir(r, ctx, inode, count, concurrent)
+func (m *baseMeta) emptyEntry(ctx Context, parent Ino, name string, inode Ino, count *uint64, concurrent chan int) syscall.Errno {
+	st := m.emptyDir(ctx, inode, count, concurrent)
 	if st == 0 {
-		st = r.Rmdir(ctx, parent, name)
+		st = m.Rmdir(ctx, parent, name)
 		if st == syscall.ENOTEMPTY {
-			st = emptyEntry(r, ctx, parent, name, inode, count, concurrent)
+			st = m.emptyEntry(ctx, parent, name, inode, count, concurrent)
 		} else if count != nil {
 			atomic.AddUint64(count, 1)
 		}
@@ -290,23 +295,24 @@ func emptyEntry(r Meta, ctx Context, parent Ino, name string, inode Ino, count *
 	return st
 }
 
-func Remove(r Meta, ctx Context, parent Ino, name string, count *uint64) syscall.Errno {
-	if st := r.Access(ctx, parent, 3, nil); st != 0 {
+func (m *baseMeta) Remove(ctx Context, parent Ino, name string, count *uint64) syscall.Errno {
+	parent = m.checkRoot(parent)
+	if st := m.Access(ctx, parent, 3, nil); st != 0 {
 		return st
 	}
 	var inode Ino
 	var attr Attr
-	if st := r.Lookup(ctx, parent, name, &inode, &attr); st != 0 {
+	if st := m.Lookup(ctx, parent, name, &inode, &attr); st != 0 {
 		return st
 	}
 	if attr.Typ != TypeDirectory {
 		if count != nil {
 			atomic.AddUint64(count, 1)
 		}
-		return r.Unlink(ctx, parent, name)
+		return m.Unlink(ctx, parent, name)
 	}
 	concurrent := make(chan int, 50)
-	return emptyEntry(r, ctx, parent, name, inode, count, concurrent)
+	return m.emptyEntry(ctx, parent, name, inode, count, concurrent)
 }
 
 func GetSummary(r Meta, ctx Context, inode Ino, summary *Summary, recursive bool) syscall.Errno {
diff --git a/pkg/vfs/internal.go b/pkg/vfs/internal.go
index 9267e51d..eede654d 100644
--- a/pkg/vfs/internal.go
+++ b/pkg/vfs/internal.go
@@ -192,7 +192,7 @@ func (v *VFS) handleInternalMsg(ctx meta.Context, cmd uint32, r *utils.Buffer, d
 		go func() {
 			inode := Ino(r.Get64())
 			name := string(r.Get(int(r.Get8())))
-			st = meta.Remove(v.Meta, ctx, inode, name, &count)
+			st = v.Meta.Remove(ctx, inode, name, &count)
 			if st != 0 {
 				logger.Errorf("remove %d/%s: %s", inode, name, st)
 			}


Test patch:
diff --git a/pkg/meta/base_test.go b/pkg/meta/base_test.go
index 53f90e09..1eab3cb7 100644
--- a/pkg/meta/base_test.go
+++ b/pkg/meta/base_test.go
@@ -55,15 +55,6 @@ func testMeta(t *testing.T, m Meta) {
 	if err := m.Reset(); err != nil {
 		t.Fatalf("reset meta: %s", err)
 	}
-	var base *baseMeta
-	switch m := m.(type) {
-	case *redisMeta:
-		base = &m.baseMeta
-	case *dbMeta:
-		base = &m.baseMeta
-	case *kvMeta:
-		base = &m.baseMeta
-	}
 	testMetaClient(t, m)
 	testTruncateAndDelete(t, m)
 	testTrash(t, m)
@@ -78,6 +69,7 @@ func testMeta(t *testing.T, m Meta) {
 	testCopyFileRange(t, m)
 	testCloseSession(t, m)
 	testConcurrentDir(t, m)
+	base := m.getBase()
 	base.conf.OpenCache = time.Second
 	base.of.expire = time.Second
 	testOpenCache(t, m)
@@ -116,19 +108,9 @@ func testMetaClient(t *testing.T, m Meta) {
 	if err != nil || len(ses) != 1 {
 		t.Fatalf("list sessions %+v: %s", ses, err)
 	}
-	switch r := m.(type) {
-	case *redisMeta:
-		if r.sid != ses[0].Sid {
-			t.Fatalf("my sid %d != registered sid %d", r.sid, ses[0].Sid)
-		}
-	case *dbMeta:
-		if r.sid != ses[0].Sid {
-			t.Fatalf("my sid %d != registered sid %d", r.sid, ses[0].Sid)
-		}
-	case *kvMeta:
-		if r.sid != ses[0].Sid {
-			t.Fatalf("my sid %d != registered sid %d", r.sid, ses[0].Sid)
-		}
+	base := m.getBase()
+	if base.sid != ses[0].Sid {
+		t.Fatalf("my sid %d != registered sid %d", base.sid, ses[0].Sid)
 	}
 	go m.CleanStaleSessions()
 
@@ -737,7 +719,7 @@ func testRemove(t *testing.T, m Meta) {
 	if st := m.Create(ctx, 1, "f", 0644, 0, 0, &inode, attr); st != 0 {
 		t.Fatalf("create f: %s", st)
 	}
-	if st := Remove(m, ctx, 1, "f", nil); st != 0 {
+	if st := m.Remove(ctx, 1, "f", nil); st != 0 {
 		t.Fatalf("rmr f: %s", st)
 	}
 	if st := m.Mkdir(ctx, 1, "d", 0755, 0, 0, &parent, attr); st != 0 {
@@ -766,7 +748,7 @@ func testRemove(t *testing.T, m Meta) {
 	} else if len(entries) != 4099 {
 		t.Fatalf("entries: %d", len(entries))
 	}
-	if st := Remove(m, ctx, 1, "d", nil); st != 0 {
+	if st := m.Remove(ctx, 1, "d", nil); st != 0 {
 		t.Fatalf("rmr d: %s", st)
 	}
 }
@@ -1112,15 +1094,7 @@ func testCloseSession(t *testing.T, m Meta) {
 	if st := m.Unlink(ctx, 1, "f"); st != 0 {
 		t.Fatalf("unlink f: %s", st)
 	}
-	var sid uint64
-	switch m := m.(type) {
-	case *redisMeta:
-		sid = m.sid
-	case *dbMeta:
-		sid = m.sid
-	case *kvMeta:
-		sid = m.sid
-	}
+	sid := m.getBase().sid
 	s, err := m.GetSession(sid, true)
 	if err != nil {
 		t.Fatalf("get session: %s", err)
@@ -1245,14 +1219,7 @@ func testTrash(t *testing.T, m Meta) {
 	if st := m.Rename(ctx2, TrashInode+1, "d", 1, "f", 0, &inode, attr); st != syscall.EPERM {
 		t.Fatalf("rename d -> f: %s", st)
 	}
-	switch bm := m.(type) {
-	case *redisMeta:
-		bm.doCleanupTrash(true)
-	case *dbMeta:
-		bm.doCleanupTrash(true)
-	case *kvMeta:
-		bm.doCleanupTrash(true)
-	}
+	m.getBase().doCleanupTrash(true)
 	if st := m.GetAttr(ctx2, TrashInode+1, attr); st != syscall.ENOENT {
 		t.Fatalf("getattr: %s", st)
 	}
diff --git a/pkg/meta/benchmarks_test.go b/pkg/meta/benchmarks_test.go
index 98d60db5..3b2e03a2 100644
--- a/pkg/meta/benchmarks_test.go
+++ b/pkg/meta/benchmarks_test.go
@@ -116,7 +116,7 @@ func BenchmarkReadSliceBuf(b *testing.B) {
 
 func prepareParent(m Meta, name string, inode *Ino) error {
 	ctx := Background
-	if err := Remove(m, ctx, 1, name, nil); err != 0 && err != syscall.ENOENT {
+	if err := m.Remove(ctx, 1, name, nil); err != 0 && err != syscall.ENOENT {
 		return fmt.Errorf("remove: %s", err)
 	}
 	if err := m.Mkdir(ctx, 1, name, 0755, 0, 0, inode, nil); err != 0 {