count space and inodes usage for each directory

pkg/meta/base.go

@@ -17,6 +17,7 @@
 package meta
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"os"
@@ -43,6 +44,8 @@ const (
 	sliceIdBatch  = 1000
 	minUpdateTime = time.Millisecond * 10
 	nlocks        = 1024
+
+	dirUsageBatchSize = 1000
 )
 
 type engine interface {
@@ -73,16 +76,19 @@ type engine interface {
 	doLookup(ctx Context, parent Ino, name string, inode *Ino, attr *Attr) syscall.Errno
 	doMknod(ctx Context, parent Ino, name string, _type uint8, mode, cumask uint16, rdev uint32, path string, inode *Ino, attr *Attr) syscall.Errno
 	doLink(ctx Context, inode, parent Ino, name string, attr *Attr) syscall.Errno
-	doUnlink(ctx Context, parent Ino, name string) syscall.Errno
+	doUnlink(ctx Context, parent Ino, name string, attr *Attr) syscall.Errno
 	doRmdir(ctx Context, parent Ino, name string) syscall.Errno
 	doReadlink(ctx Context, inode Ino) ([]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 *Ino, attr *Attr) syscall.Errno
 	doSetXattr(ctx Context, inode Ino, name string, value []byte, flags uint32) syscall.Errno
 	doRemoveXattr(ctx Context, inode Ino, name string) syscall.Errno
-	doGetParents(ctx Context, inode Ino) map[Ino]int
 	doRepair(ctx Context, inode Ino, attr *Attr) syscall.Errno
 
+	doGetParents(ctx Context, inode Ino) map[Ino]int
+	doIncreDirUsage(ctx Context, ino Ino, space int64, inodes int64) error
+	doGetDirUsage(ctx Context, ino Ino) (space, inodes uint64, err error)
+
 	scanTrashSlices(Context, trashSliceScan) error
 	scanPendingSlices(Context, pendingSliceScan) error
 	scanPendingFiles(Context, pendingFileScan) error
@@ -111,6 +117,13 @@ type cchunk struct {
 	slices int
 }
 
+// event to update dir usage
+type dirUsageEvent struct {
+	ino    Ino
+	space  int64
+	inodes int64
+}
+
 type baseMeta struct {
 	sync.Mutex
 	addr string
@@ -132,6 +145,8 @@ type baseMeta struct {
 	umounting    bool
 	sesMu        sync.Mutex
 
+	dirUsageEvents chan *dirUsageEvent
+
 	*fsStat
 
 	freeMu     sync.Mutex
@@ -149,16 +164,17 @@ type baseMeta struct {
 
 func newBaseMeta(addr string, conf *Config) *baseMeta {
 	return &baseMeta{
-		addr:         utils.RemovePassword(addr),
-		conf:         conf,
-		root:         RootInode,
-		of:           newOpenFiles(conf.OpenCache, conf.OpenCacheLimit),
-		removedFiles: make(map[Ino]bool),
-		compacting:   make(map[uint64]bool),
-		maxDeleting:  make(chan struct{}, 100),
-		dslices:      make(chan Slice, conf.MaxDeletes*10240),
-		symlinks:     &sync.Map{},
-		fsStat:       new(fsStat),
+		addr:           utils.RemovePassword(addr),
+		conf:           conf,
+		root:           RootInode,
+		of:             newOpenFiles(conf.OpenCache, conf.OpenCacheLimit),
+		removedFiles:   make(map[Ino]bool),
+		compacting:     make(map[uint64]bool),
+		maxDeleting:    make(chan struct{}, 100),
+		dslices:        make(chan Slice, conf.MaxDeletes*10240),
+		symlinks:       &sync.Map{},
+		fsStat:         new(fsStat),
+		dirUsageEvents: make(chan *dirUsageEvent, dirUsageBatchSize),
 		msgCallbacks: &msgCallbacks{
 			callbacks: make(map[uint32]MsgCallback),
 		},
@@ -230,6 +246,100 @@ func (m *baseMeta) checkRoot(inode Ino) Ino {
 	}
 }
 
+func (m *baseMeta) countDirUsage(ctx Context, ino Ino) (space, inodes uint64, err syscall.Errno) {
+	var entries []*Entry
+	if err = m.en.doReaddir(ctx, ino, 1, &entries, -1); err != 0 {
+		return
+	}
+	for _, e := range entries {
+		inodes += 1
+		var newSpace uint64
+		if e.Attr.Typ == TypeFile {
+			newSpace = e.Attr.Length
+		}
+		space += uint64(align4K(newSpace))
+	}
+	return
+}
+
+func (m *baseMeta) GetDirUsage(ctx Context, inode Ino) (space, inodes uint64, err error) {
+	return m.en.doGetDirUsage(ctx, m.checkRoot(inode))
+}
+
+func (m *baseMeta) tryIncreDirUsage(ctx Context, ino Ino, space int64, inodes int64) bool {
+	select {
+	case m.dirUsageEvents <- &dirUsageEvent{ino, space, inodes}:
+		return true
+	default:
+		return false
+	}
+}
+
+func (m *baseMeta) mustIncreDirUsage(ctx Context, ino Ino, space int64, inodes int64) {
+	for i := 0; i < 50; i++ {
+		select {
+		case m.dirUsageEvents <- &dirUsageEvent{ino, space, inodes}:
+			return
+		default:
+			time.Sleep(time.Second * 1)
+		}
+	}
+	logger.Warn("already tried 50 times, dirUsageEvents channel is full, try to update dir usage directly")
+	err := m.en.doIncreDirUsage(ctx, ino, space, inodes)
+	if err != nil {
+		logger.Errorf("update dir usage directly failed: %v", err)
+	}
+}
+
+func (m *baseMeta) increParentUsage(ctx Context, inode, parent Ino, space int64) {
+	if parent > 0 {
+		if !m.tryIncreDirUsage(ctx, parent, space, 0) {
+			go m.mustIncreDirUsage(ctx, parent, space, 0)
+		}
+	} else {
+		go func() {
+			for p := range m.en.doGetParents(ctx, inode) {
+				m.mustIncreDirUsage(ctx, p, space, 0)
+			}
+		}()
+	}
+}
+
+func (m *baseMeta) dirUsageBatch() {
+	commitChan := make(chan map[Ino]*dirUsageEvent, 16)
+	go func() {
+		for records := range commitChan {
+			for _, r := range records {
+				err := m.en.doIncreDirUsage(WrapContext(context.TODO()), r.ino, r.space, r.inodes)
+				if err != nil {
+					logger.Errorf("update dir usage failed: %v", err)
+				}
+			}
+
+		}
+	}()
+
+	for {
+		count := 0
+		records := make(map[Ino]*dirUsageEvent)
+		for count <= dirUsageBatchSize {
+			count++
+			e := <-m.dirUsageEvents
+			record, exist := records[e.ino]
+			if !exist {
+				records[e.ino] = e
+			} else {
+				record.space += e.space
+				record.inodes += e.inodes
+			}
+			if len(m.dirUsageEvents) == 0 {
+				break
+			}
+		}
+		commitChan <- records
+	}
+}
+
 func (r *baseMeta) txLock(idx uint) {
 	r.txlocks[idx%nlocks].Lock()
 }
@@ -317,6 +427,7 @@ func (m *baseMeta) NewSession() error {
 		go m.cleanupDeletedFiles()
 		go m.cleanupSlices()
 		go m.cleanupTrash()
+		go m.dirUsageBatch()
 	}
 	return nil
 }
@@ -769,7 +880,12 @@ func (m *baseMeta) Mknod(ctx Context, parent Ino, name string, _type uint8, mode
 	if m.checkQuota(4<<10, 1) {
 		return syscall.ENOSPC
 	}
-	return m.en.doMknod(ctx, m.checkRoot(parent), name, _type, mode, cumask, rdev, path, inode, attr)
+	err := m.en.doMknod(ctx, m.checkRoot(parent), name, _type, mode, cumask, rdev, path, inode, attr)
+	newSpace := align4K(0)
+	if err == 0 && !m.tryIncreDirUsage(ctx, parent, newSpace, 1) {
+		go m.mustIncreDirUsage(ctx, parent, newSpace, 1)
+	}
+	return err
 }
 
 func (m *baseMeta) Create(ctx Context, parent Ino, name string, mode uint16, cumask uint16, flags uint32, inode *Ino, attr *Attr) syscall.Errno {
@@ -812,7 +928,15 @@ func (m *baseMeta) Link(ctx Context, inode, parent Ino, name string, attr *Attr)
 	defer m.timeit(time.Now())
 	parent = m.checkRoot(parent)
 	defer func() { m.of.InvalidateChunk(inode, invalidateAttrOnly) }()
-	return m.en.doLink(ctx, inode, parent, name, attr)
+	if attr == nil {
+		attr = &Attr{}
+	}
+	err := m.en.doLink(ctx, inode, parent, name, attr)
+	newSpace := align4K(attr.Length)
+	if err == 0 && !m.tryIncreDirUsage(ctx, parent, newSpace, 1) {
+		go m.mustIncreDirUsage(ctx, parent, newSpace, 1)
+	}
+	return err
 }
 
 func (m *baseMeta) ReadLink(ctx Context, inode Ino, path *[]byte) syscall.Errno {
@@ -842,7 +966,18 @@ func (m *baseMeta) Unlink(ctx Context, parent Ino, name string) syscall.Errno {
 	}
 
 	defer m.timeit(time.Now())
-	return m.en.doUnlink(ctx, m.checkRoot(parent), name)
+	var attr Attr
+	err := m.en.doUnlink(ctx, m.checkRoot(parent), name, &attr)
+	if err == 0 {
+		newSpace := -align4K(0)
+		if attr.Typ == TypeFile {
+			newSpace = -align4K(attr.Length)
+		}
+		if !m.tryIncreDirUsage(ctx, parent, newSpace, -1) {
+			go m.mustIncreDirUsage(ctx, parent, newSpace, -1)
+		}
+	}
+	return err
 }
 
 func (m *baseMeta) Rmdir(ctx Context, parent Ino, name string) syscall.Errno {
@@ -860,7 +995,12 @@ func (m *baseMeta) Rmdir(ctx Context, parent Ino, name string) syscall.Errno {
 	}
 
 	defer m.timeit(time.Now())
-	return m.en.doRmdir(ctx, m.checkRoot(parent), name)
+	err := m.en.doRmdir(ctx, m.checkRoot(parent), name)
+	newSpace := -align4K(0)
+	if err == 0 && !m.tryIncreDirUsage(ctx, parent, newSpace, -1) {
+		go m.mustIncreDirUsage(ctx, parent, newSpace, -1)
+	}
+	return err
 }
 
 func (m *baseMeta) Rename(ctx Context, parentSrc Ino, nameSrc string, parentDst Ino, nameDst string, flags uint32, inode *Ino, attr *Attr) syscall.Errno {
@@ -885,7 +1025,25 @@ func (m *baseMeta) Rename(ctx Context, parentSrc Ino, nameSrc string, parentDst
 	}
 
 	defer m.timeit(time.Now())
-	return m.en.doRename(ctx, m.checkRoot(parentSrc), nameSrc, m.checkRoot(parentDst), nameDst, flags, inode, attr)
+	if attr == nil {
+		attr = &Attr{}
+	}
+	err := m.en.doRename(ctx, m.checkRoot(parentSrc), nameSrc, m.checkRoot(parentDst), nameDst, flags, inode, attr)
+	if err == 0 {
+		diffSpace := align4K(0)
+		if attr.Typ == TypeFile {
+			diffSpace = align4K(attr.Length)
+		}
+		if !m.tryIncreDirUsage(ctx, parentSrc, -diffSpace, -1) {
+			go m.mustIncreDirUsage(ctx, parentSrc, -diffSpace, -1)
+
+		}
+		if !m.tryIncreDirUsage(ctx, parentDst, diffSpace, 1) {
+			go m.mustIncreDirUsage(ctx, parentDst, diffSpace, 1)
+
+		}
+	}
+	return err
 }
 
 func (m *baseMeta) Open(ctx Context, inode Ino, flags uint32, attr *Attr) syscall.Errno {
@@ -1444,7 +1602,7 @@ func (m *baseMeta) CleanupTrashBefore(ctx Context, edge time.Time, increProgress
 				if se.Attr.Typ == TypeDirectory {
 					st = m.en.doRmdir(ctx, e.Inode, string(se.Name))
 				} else {
-					st = m.en.doUnlink(ctx, e.Inode, string(se.Name))
+					st = m.en.doUnlink(ctx, e.Inode, string(se.Name), nil)
 				}
 				if st == 0 {
 					count++