meta/sql: optimizate the rmr emptyDir

pkg/meta/base.go

@@ -115,6 +115,7 @@ type engine interface {
 	doLink(ctx Context, inode, parent Ino, name string, attr *Attr) syscall.Errno
 	doUnlink(ctx Context, parent Ino, name string, attr *Attr, skipCheckTrash ...bool) syscall.Errno
 	doRmdir(ctx Context, parent Ino, name string, inode *Ino, attr *Attr, skipCheckTrash ...bool) syscall.Errno
+	doEmptyDir(ctx Context, parent Ino, entries []*Entry, length *int64, space *int64, inodes *int64, userGroupQuotas *[]UserGroupQuotaDelta, skipCheckTrash ...bool) (errno syscall.Errno)
 	doReadlink(ctx Context, inode Ino, noatime bool) (int64, []byte, error)
 	doReaddir(ctx Context, inode Ino, plus uint8, entries *[]*Entry, limit int) syscall.Errno
 	doRename(ctx Context, parentSrc Ino, nameSrc string, parentDst Ino, nameDst string, flags uint32, inode, tinode *Ino, attr, tattr *Attr) syscall.Errno

pkg/meta/interface.go

@@ -327,6 +327,14 @@ type Summary struct {
 	Dirs   uint64
 }
 
+// UserGroupQuotaDelta represents quota changes for a specific user and group.
+type UserGroupQuotaDelta struct {
+	Uid    uint32
+	Gid    uint32
+	Space  int64
+	Inodes int64
+}
+
 type TreeSummary struct {
 	Inode    Ino
 	Path     string

pkg/meta/redis.go

@@ -1666,6 +1666,10 @@ func (m *redisMeta) doUnlink(ctx Context, parent Ino, name string, attr *Attr, s
 	return errno(err)
 }
 
+func (m *redisMeta) doEmptyDir(ctx Context, parent Ino, entries []*Entry, length *int64, space *int64, inodes *int64, userGroupQuotas *[]UserGroupQuotaDelta, skipCheckTrash ...bool) syscall.Errno {
+	return syscall.ENOTSUP
+}
+
 func (m *redisMeta) doRmdir(ctx Context, parent Ino, name string, pinode *Ino, oldAttr *Attr, skipCheckTrash ...bool) syscall.Errno {
 	var trash Ino
 	if !(len(skipCheckTrash) == 1 && skipCheckTrash[0]) {

pkg/meta/sql.go

@@ -2586,6 +2586,234 @@ func (m *dbMeta) doReaddir(ctx Context, inode Ino, plus uint8, entries *[]*Entry
 	}))
 }
 
+func recordDeletionStats(
+	n *node,
+	entrySpace int64,
+	spaceDelta int64, 
+	totalLength *int64,
+	totalSpace *int64,
+	totalInodes *int64,
+	userGroupQuotas *[]UserGroupQuotaDelta,
+	trash Ino,
+) {
+	*totalLength += int64(n.Length)
+	*totalSpace += entrySpace
+	*totalInodes++
+
+	if userGroupQuotas != nil && trash == 0 && n.Uid > 0 {
+		*userGroupQuotas = append(*userGroupQuotas, UserGroupQuotaDelta{
+			Uid:    n.Uid,
+			Gid:    n.Gid,
+			Space:  spaceDelta,
+			Inodes: -1,
+		})
+	}
+}
+
+func (m *dbMeta) doEmptyDir(ctx Context, parent Ino, entries []*Entry, length *int64, space *int64, inodes *int64, userGroupQuotas *[]UserGroupQuotaDelta, skipCheckTrash ...bool) syscall.Errno {
+	if len(entries) == 0 {
+		return 0
+	}
+
+	var trash Ino
+	if len(skipCheckTrash) == 0 || !skipCheckTrash[0] {
+		if st := m.checkTrash(parent, &trash); st != 0 {
+			return st
+		}
+	}
+
+	type entryInfo struct {
+		e          edge
+		n          node
+		opened     bool
+		trash      Ino
+		trashName  string 
+		lastLink   bool   
+	}
+
+	var entryInfos []entryInfo
+	var totalLength, totalSpace, totalInodes int64
+	if userGroupQuotas != nil {
+		*userGroupQuotas = make([]UserGroupQuotaDelta, 0, len(entries))
+	}
+
+	err := m.txn(func(s *xorm.Session) error {
+		pn := node{Inode: parent}
+		ok, err := s.Get(&pn)
+		if err != nil {
+			return err
+		}
+		if !ok {
+			return syscall.ENOENT
+		}
+		if pn.Type != TypeDirectory {
+			return syscall.ENOTDIR
+		}
+		if (pn.Flags&FlagAppend != 0) || (pn.Flags&FlagImmutable) != 0 {
+			return syscall.EPERM
+		}
+
+		entryInfos = make([]entryInfo, 0, len(entries))
+		now := time.Now().UnixNano()
+
+		for _, entry := range entries {
+			e := edge{Parent: parent, Name: entry.Name, Inode: entry.Inode}
+			if entry.Attr != nil {
+				e.Type = entry.Attr.Typ
+			}
+
+			info := entryInfo{e: e, trash: trash}
+			n := node{Inode: e.Inode}
+			ok, err := s.ForUpdate().Get(&n)
+			if err != nil {
+				return err
+			}
+			if !ok {
+				continue
+			}
+			if ctx.Uid() != 0 && pn.Mode&01000 != 0 && ctx.Uid() != pn.Uid && ctx.Uid() != n.Uid {
+				return syscall.EACCES
+			}
+
+			if (n.Flags&FlagAppend) != 0 || (n.Flags&FlagImmutable) != 0 {
+				return syscall.EPERM
+			}
+			if (n.Flags & FlagSkipTrash) != 0 {
+				info.trash = 0
+			}
+
+			if info.trash > 0 {
+				info.trashName = m.trashEntry(parent, e.Inode, string(e.Name))
+				if n.Nlink > 1 {
+					if o, err := s.Get(&edge{Parent: info.trash, Name: []byte(info.trashName), Inode: e.Inode, Type: e.Type}); err == nil && o {
+						info.trash = 0
+					}
+				}
+			}
+
+			n.setCtime(now)
+			if info.trash != 0 && n.Parent > 0 {
+				n.Parent = info.trash
+			}
+
+			info.n = n
+			entryInfos = append(entryInfos, info)
+		}
+
+		seen := make(map[Ino]int)
+		for i := range entryInfos {
+			info := &entryInfos[i]
+			if info.e.Type == TypeDirectory {
+				continue
+			}
+			original := int64(info.n.Nlink)
+			processed := seen[info.e.Inode]
+			finalNlink := original - int64(processed+1)
+			if finalNlink < 0 {
+				finalNlink = 0
+			}
+			// If trash is enabled and this would be the last link, keep one link by moving it into trash.
+			if info.trash > 0 && finalNlink == 0 && info.e.Type != TypeDirectory {
+				finalNlink = 1
+			}
+			info.lastLink = (info.trash == 0 && finalNlink == 0)
+			if info.lastLink && info.e.Type == TypeFile && m.sid > 0 {
+				info.opened = m.of.IsOpen(info.e.Inode)
+			}
+			info.n.Nlink = uint32(finalNlink)
+			seen[info.e.Inode] = processed + 1
+		}
+
+		trashInserted := make(map[Ino]bool)
+		for _, info := range entryInfos {
+			if info.e.Type == TypeDirectory {
+				continue
+			}
+			if _, err := s.Delete(&edge{Parent: parent, Name: info.e.Name}); err != nil {
+				return err
+			}
+
+			if info.n.Nlink > 0 {
+				if _, err := s.Cols("nlink", "ctime", "ctimensec", "parent").Update(&info.n, &node{Inode: info.e.Inode}); err != nil {
+					return err
+				}
+				if info.trash > 0 && !trashInserted[info.e.Inode] {
+					if err := mustInsert(s, &edge{Parent: info.trash, Name: []byte(info.trashName), Inode: info.e.Inode, Type: info.e.Type}); err != nil {
+						return err
+					}
+					trashInserted[info.e.Inode] = true
+				}
+				entrySpace := align4K(info.n.Length)
+				recordDeletionStats(&info.n, entrySpace, 0, &totalLength, &totalSpace, &totalInodes, userGroupQuotas, trash)
+			} else {
+				switch info.e.Type {
+				case TypeFile:
+					entrySpace := align4K(info.n.Length)
+					if info.opened {
+						if err = mustInsert(s, sustained{Sid: m.sid, Inode: info.e.Inode}); err != nil {
+							return err
+						}
+						if _, err := s.Cols("nlink", "ctime", "ctimensec").Update(&info.n, &node{Inode: info.e.Inode}); err != nil {
+							return err
+						}
+						recordDeletionStats(&info.n, entrySpace, 0, &totalLength, &totalSpace, &totalInodes, userGroupQuotas, trash)
+					} else {
+						if err = mustInsert(s, delfile{info.e.Inode, info.n.Length, time.Now().Unix()}); err != nil {
+							return err
+						}
+						if _, err := s.Delete(&node{Inode: info.e.Inode}); err != nil {
+							return err
+						}
+						recordDeletionStats(&info.n, entrySpace, -entrySpace, &totalLength, &totalSpace, &totalInodes, userGroupQuotas, trash)
+					}
+				case TypeSymlink:
+					if _, err := s.Delete(&symlink{Inode: info.e.Inode}); err != nil {
+						return err
+					}
+					if _, err := s.Delete(&node{Inode: info.e.Inode}); err != nil {
+						return err
+					}
+					entrySpace := align4K(0)
+					recordDeletionStats(&info.n, entrySpace, -entrySpace, &totalLength, &totalSpace, &totalInodes, userGroupQuotas, trash)
+				default:
+					if _, err := s.Delete(&node{Inode: info.e.Inode}); err != nil {
+						return err
+					}
+					if info.e.Type != TypeFile {
+						entrySpace := align4K(0)
+						recordDeletionStats(&info.n, entrySpace, -entrySpace, &totalLength, &totalSpace, &totalInodes, userGroupQuotas, trash)
+					}
+				}
+				if _, err := s.Delete(&xattr{Inode: info.e.Inode}); err != nil {
+					return err
+				}
+			}
+			m.of.InvalidateChunk(info.e.Inode, invalidateAttrOnly)
+		}
+
+		return nil
+	})
+
+	if err != nil {
+		return errno(err)
+	}
+
+	if trash == 0 {
+		for _, info := range entryInfos {
+				if info.n.Type == TypeFile && info.lastLink {
+				isTrash := parent.IsTrash()
+				m.fileDeleted(info.opened, isTrash, info.e.Inode, info.n.Length)
+			}
+		}
+		m.updateStats(-totalSpace, -totalInodes)
+	}
+
+	*length = totalLength
+	*space = totalSpace
+	*inodes = totalInodes
+	return 0
+}
+
 func (m *dbMeta) doCleanStaleSession(sid uint64) error {
 	var fail bool
 	// release locks

pkg/meta/tkv.go

@@ -1427,6 +1427,10 @@ func (m *kvMeta) doUnlink(ctx Context, parent Ino, name string, attr *Attr, skip
 	return errno(err)
 }
 
+func (m *kvMeta) doEmptyDir(ctx Context, parent Ino, entries []*Entry, length *int64, space *int64, inodes *int64, userGroupQuotas *[]UserGroupQuotaDelta, skipCheckTrash ...bool) syscall.Errno {
+	return syscall.ENOTSUP
+}
+
 func (m *kvMeta) doRmdir(ctx Context, parent Ino, name string, pinode *Ino, oldAttr *Attr, skipCheckTrash ...bool) syscall.Errno {
 	var trash Ino
 	if !(len(skipCheckTrash) == 1 && skipCheckTrash[0]) {

pkg/meta/utils.go

@@ -277,15 +277,8 @@ func (m *baseMeta) emptyDir(ctx Context, inode Ino, skipCheckTrash bool, count *
 		}
 		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 i, e := range entries {
+		var nonDirEntries []*Entry
+		for _, e := range entries {
 			if e.Attr.Typ == TypeDirectory {
 				select {
 				case concurrent <- 1:
@@ -305,20 +298,51 @@ func (m *baseMeta) emptyDir(ctx Context, inode Ino, skipCheckTrash bool, count *
 					}
 				}
 			} else {
-				if count != nil {
-					atomic.AddUint64(count, 1)
-				}
-				if st := m.Unlink(ctx, inode, string(e.Name), skipCheckTrash); st != 0 && st != syscall.ENOENT {
-					ctx.Cancel()
-					return st
-				}
+				nonDirEntries = append(nonDirEntries, e)
 			}
 			if ctx.Canceled() {
 				return syscall.EINTR
 			}
-			entries[i] = nil // release memory
+
 		}
 		wg.Wait()
+
+		var length int64
+		var space int64
+		var inodes int64
+		var userGroupQuotas []UserGroupQuotaDelta
+		st := m.en.doEmptyDir(ctx, inode, nonDirEntries, &length, &space, &inodes, &userGroupQuotas, skipCheckTrash)
+		if st == 0 {
+			m.updateDirStat(ctx, inode, -length, -space, -inodes)
+			if !inode.IsTrash() {
+				m.updateDirQuota(ctx, inode, -space, -inodes)
+				for _, quota := range userGroupQuotas {
+					m.updateUserGroupQuota(ctx, quota.Uid, quota.Gid, -quota.Space, -quota.Inodes)
+				}
+			}
+			if count != nil && len(nonDirEntries) > 0 {
+				atomic.AddUint64(count, uint64(len(nonDirEntries)))
+			}
+		} else if st == syscall.ENOTSUP {
+			for _, e := range entries {
+				if e.Attr.Typ == TypeDirectory {
+					continue
+				}
+				if ctx.Canceled() {
+					return syscall.EINTR
+				}
+				if st := m.Unlink(ctx, inode, string(e.Name), skipCheckTrash); st != 0 && st != syscall.ENOENT {
+					return st
+				}
+				if count != nil {
+					atomic.AddUint64(count, 1)
+				}
+			}
+		} else if st != 0 {
+			return st
+		}
+		entries = nil
+
 		if status != 0 || inode == TrashInode { // try only once for .trash
 			return status
 		}