gc: clean expired trash files, delfiles and delslices

cmd/gc.go

@@ -42,7 +42,7 @@ func cmdGC() *cli.Command {
 		ArgsUsage: "META-URL",
 		Description: `
 It scans all objects in data storage and slices in metadata, comparing them to see if there is any
-leaked object. It can also actively trigger compaction of slices.
+leaked object. It can also actively trigger compaction of slices and the cleanup of delayed deleted slices or files.
 Use this command if you find that data storage takes more than expected.
 
 Examples:
@@ -52,7 +52,7 @@ $ juicefs gc redis://localhost
 # Trigger compaction of all slices
 $ juicefs gc redis://localhost --compact
 
-# Delete leaked objects
+# Delete leaked objects and delayed deleted slices or files
 $ juicefs gc redis://localhost --delete`,
 		Flags: []cli.Flag{
 			&cli.BoolFlag{
@@ -61,7 +61,7 @@ $ juicefs gc redis://localhost --delete`,
 			},
 			&cli.BoolFlag{
 				Name:  "delete",
-				Usage: "delete leaked objects",
+				Usage: "delete leaked objects and delayed deleted slices or files",
 			},
 			&cli.IntFlag{
 				Name:    "threads",
@@ -110,12 +110,69 @@ func gc(ctx *cli.Context) error {
 
 	// Scan all chunks first and do compaction if necessary
 	progress := utils.NewProgress(false, false)
-	if ctx.Bool("compact") {
-		bar := progress.AddCountBar("Compacted chunks", 0)
-		spin := progress.AddDoubleSpinner("Compacted slices")
+	// Delete pending slices while listing all slices
+	delete := ctx.Bool("delete")
+	threads := ctx.Int("threads")
+	compact := ctx.Bool("compact")
+	if (delete || compact) && threads <= 0 {
+		logger.Fatal("threads should be greater than 0 to delete or compact objects")
+	}
+
+	var wg sync.WaitGroup
+	var delSpin *utils.Bar
+	var sliceChan chan *dSlice // pending delete slices
+
+	if delete || compact {
+		delSpin = progress.AddCountSpinner("Deleted pending")
+		sliceChan = make(chan *dSlice, 10240)
 		m.OnMsg(meta.DeleteSlice, func(args ...interface{}) error {
-			return store.Remove(args[0].(uint64), int(args[1].(uint32)))
+			delSpin.Increment()
+			sliceChan <- &dSlice{args[0].(uint64), args[1].(uint32)}
+			return nil
 		})
+		for i := 0; i < threads; i++ {
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				for s := range sliceChan {
+					if err := store.Remove(s.id, int(s.length)); err != nil {
+						logger.Warnf("remove %d_%d: %s", s.id, s.length, err)
+					}
+				}
+			}()
+		}
+	}
+
+	c := meta.WrapContext(ctx.Context)
+	delayedFileSpin := progress.AddDoubleSpinner("Delfiles")
+	cleanedFileSpin := progress.AddDoubleSpinner("Cleaned delfiles")
+	edge := time.Now().Add(-time.Duration(format.TrashDays) * 24 * time.Hour)
+	if delete {
+		cleanTrashSpin := progress.AddCountSpinner("Cleaned trash")
+		m.CleanupTrashBefore(c, edge, cleanTrashSpin.Increment)
+		cleanTrashSpin.Done()
+	}
+	err = m.ScanDeletedObject(
+		c,
+		nil,
+		func(_ meta.Ino, size uint64, ts int64) (bool, error) {
+			delayedFileSpin.IncrInt64(int64(size))
+			if delete {
+				cleanedFileSpin.IncrInt64(int64(size))
+				return true, nil
+			}
+			return false, nil
+		},
+	)
+	if err != nil {
+		logger.Fatalf("scan deleted object: %s", err)
+	}
+	delayedFileSpin.Done()
+	cleanedFileSpin.Done()
+
+	if compact {
+		bar := progress.AddCountBar("Compacted chunks", 0)
+		spin := progress.AddDoubleSpinner("Compacted slices")
 		m.OnMsg(meta.CompactChunk, func(args ...interface{}) error {
 			slices := args[0].([]meta.Slice)
 			err := vfs.Compact(chunkConf, store, slices, args[1].(uint64))
@@ -143,43 +200,38 @@ func gc(ctx *cli.Context) error {
 	// put it above delete count spinner
 	sliceCSpin := progress.AddCountSpinner("Listed slices")
 
-	// Delete pending slices while listing all slices
-	delete := ctx.Bool("delete")
-	threads := ctx.Int("threads")
-	if delete && threads <= 0 {
-		logger.Fatal("threads should be greater than 0 to delete objects")
-	}
-	var delSpin *utils.Bar
-	var sliceChan chan *dSlice // pending delete slices
-	var wg sync.WaitGroup
-	if delete {
-		delSpin = progress.AddCountSpinner("Deleted pending")
-		sliceChan = make(chan *dSlice, 10240)
-		m.OnMsg(meta.DeleteSlice, func(args ...interface{}) error {
-			delSpin.Increment()
-			sliceChan <- &dSlice{args[0].(uint64), args[1].(uint32)}
-			return nil
-		})
-		for i := 0; i < threads; i++ {
-			wg.Add(1)
-			go func() {
-				defer wg.Done()
-				for s := range sliceChan {
-					if err := store.Remove(s.id, int(s.length)); err != nil {
-						logger.Warnf("remove %d_%d: %s", s.id, s.length, err)
-					}
-				}
-			}()
-		}
-	}
-
 	// List all slices in metadata engine
-	var c = meta.NewContext(0, 0, []uint32{0})
 	slices := make(map[meta.Ino][]meta.Slice)
 	r := m.ListSlices(c, slices, delete, sliceCSpin.Increment)
 	if r != 0 {
 		logger.Fatalf("list all slices: %s", r)
 	}
+
+	delayedSliceSpin := progress.AddDoubleSpinner("Delslices")
+	cleanedSliceSpin := progress.AddDoubleSpinner("Cleaned delslices")
+
+	err = m.ScanDeletedObject(
+		c,
+		func(ss []meta.Slice, ts int64) (bool, error) {
+			for _, s := range ss {
+				delayedSliceSpin.IncrInt64(int64(s.Size))
+				if delete && ts < edge.Unix() {
+					cleanedSliceSpin.IncrInt64(int64(s.Size))
+				}
+			}
+			if delete && ts < edge.Unix() {
+				return true, nil
+			}
+			return false, nil
+		},
+		nil,
+	)
+	if err != nil {
+		logger.Fatalf("statistic: %s", err)
+	}
+	delayedSliceSpin.Done()
+	cleanedSliceSpin.Done()
+
 	if delete {
 		close(sliceChan)
 		wg.Wait()
@@ -321,8 +373,10 @@ func gc(ctx *cli.Context) error {
 	cc, cb := compacted.Current()
 	lc, lb := leaked.Current()
 	sc, sb := skipped.Current()
-	logger.Infof("scanned %d objects, %d valid, %d compacted (%d bytes), %d leaked (%d bytes), %d skipped (%d bytes)",
-		bar.Current(), vc, cc, cb, lc, lb, sc, sb)
+	dsc, dsb := cleanedSliceSpin.Current()
+	fc, fb := cleanedFileSpin.Current()
+	logger.Infof("scanned %d objects, %d valid, %d compacted (%d bytes), %d leaked (%d bytes), %d delslices (%d bytes), %d delfiles (%d bytes), %d skipped (%d bytes)",
+		bar.Current(), vc, cc, cb, lc, lb, dsc, dsb, fc, fb, sc, sb)
 	if lc > 0 && !delete {
 		logger.Infof("Please add `--delete` to clean leaked objects")
 	}

cmd/status.go

@@ -120,15 +120,17 @@ func status(ctx *cli.Context) error {
 		fileSpinner := progress.AddDoubleSpinner("Delayed Files")
 		defer fileSpinner.Done()
 
-		err = m.Statistic(
-			ctx.Context,
-			func(s meta.Slice) error {
-				slicesSpinner.IncrInt64(int64(s.Size))
-				return nil
+		err = m.ScanDeletedObject(
+			meta.WrapContext(ctx.Context),
+			func(ss []meta.Slice, _ int64) (bool, error) {
+				for _, s := range ss {
+					slicesSpinner.IncrInt64(int64(s.Size))
+				}
+				return false, nil
 			},
-			func(_ meta.Ino, size uint64) error {
+			func(_ meta.Ino, size uint64, _ int64) (bool, error) {
 				fileSpinner.IncrInt64(int64(size))
-				return nil
+				return false, nil
 			},
 		)
 		if err != nil {

pkg/meta/base.go

@@ -17,7 +17,6 @@
 package meta
 
 import (
-	"context"
 	"encoding/json"
 	"fmt"
 	"path"
@@ -82,12 +81,15 @@ type engine interface {
 	doGetParents(ctx Context, inode Ino) map[Ino]int
 	doRepair(ctx Context, inode Ino, attr *Attr) syscall.Errno
 
-	scanDeletedSlices(ctx context.Context, visitor func(s Slice) error) error
-	scanDeletedFiles(ctx context.Context, visitor func(ino Ino, size uint64) error) error
+	scanDeletedSlices(Context, deletedSliceScan) error
+	scanDeletedFiles(Context, deletedFileScan) error
 
 	GetSession(sid uint64, detail bool) (*Session, error)
 }
 
+type deletedSliceScan func(ss []Slice, ts int64) (clean bool, err error)
+type deletedFileScan func(ino Ino, size uint64, ts int64) (clean bool, err error)
+
 // fsStat aligned for atomic operations
 // nolint:structcheck
 type fsStat struct {
@@ -1363,9 +1365,8 @@ func (m *baseMeta) cleanupTrash() {
 	}
 }
 
-func (m *baseMeta) doCleanupTrash(force bool) {
+func (m *baseMeta) CleanupTrashBefore(ctx Context, edge time.Time, increProgress func()) {
 	logger.Debugf("cleanup trash: started")
-	ctx := Background
 	now := time.Now()
 	var st syscall.Errno
 	var entries []*Entry
@@ -1383,7 +1384,6 @@ func (m *baseMeta) doCleanupTrash(force bool) {
 		}
 	}()
 	batch := 1000000
-	edge := now.Add(-time.Duration(24*m.fmt.TrashDays+1) * time.Hour)
 	for len(entries) > 0 {
 		e := entries[0]
 		ts, err := time.Parse("2006-01-02-15", string(e.Name))
@@ -1392,7 +1392,7 @@ func (m *baseMeta) doCleanupTrash(force bool) {
 			entries = entries[1:]
 			continue
 		}
-		if ts.Before(edge) || force {
+		if ts.Before(edge) {
 			var subEntries []*Entry
 			if st = m.en.doReaddir(ctx, e.Inode, 0, &subEntries, batch); st != 0 {
 				logger.Warnf("readdir subTrash %d: %s", e.Inode, st)
@@ -1411,6 +1411,9 @@ func (m *baseMeta) doCleanupTrash(force bool) {
 				}
 				if st == 0 {
 					count++
+					if increProgress != nil {
+						increProgress()
+					}
 				} else {
 					logger.Warnf("delete from trash %s/%s: %s", e.Name, se.Name, st)
 					rmdir = false
@@ -1431,6 +1434,14 @@ func (m *baseMeta) doCleanupTrash(force bool) {
 	}
 }
 
+func (m *baseMeta) doCleanupTrash(force bool) {
+	edge := time.Now().Add(-time.Duration(24*m.fmt.TrashDays+1) * time.Hour)
+	if force {
+		edge = time.Now()
+	}
+	m.CleanupTrashBefore(Background, edge, nil)
+}
+
 func (m *baseMeta) cleanupDelayedSlices() {
 	now := time.Now()
 	edge := now.Unix() - int64(m.fmt.TrashDays)*24*3600
@@ -1447,16 +1458,16 @@ func (m *baseMeta) cleanupDelayedSlices() {
 	}
 }
 
-func (m *baseMeta) Statistic(ctx context.Context, slicesDeletedScan func(Slice) error, fileDeletedScan func(ino Ino, size uint64) error) error {
+func (m *baseMeta) ScanDeletedObject(ctx Context, sliceScan deletedSliceScan, fileScan deletedFileScan) error {
 	eg := errgroup.Group{}
-	if slicesDeletedScan != nil {
+	if sliceScan != nil {
 		eg.Go(func() error {
-			return m.en.scanDeletedSlices(ctx, slicesDeletedScan)
+			return m.en.scanDeletedSlices(ctx, sliceScan)
 		})
 	}
-	if fileDeletedScan != nil {
+	if fileScan != nil {
 		eg.Go(func() error {
-			return m.en.scanDeletedFiles(ctx, fileDeletedScan)
+			return m.en.scanDeletedFiles(ctx, fileScan)
 		})
 	}
 	return eg.Wait()

pkg/meta/context.go

@@ -40,58 +40,53 @@ type Context interface {
 	Canceled() bool
 }
 
-type emptyContext struct {
-	context.Context
-}
-
-func (ctx *emptyContext) Gid() uint32    { return 0 }
-func (ctx *emptyContext) Gids() []uint32 { return []uint32{0} }
-func (ctx *emptyContext) Uid() uint32    { return 0 }
-func (ctx *emptyContext) Pid() uint32    { return 1 }
-func (ctx *emptyContext) Cancel()        {}
-func (ctx *emptyContext) Canceled() bool { return false }
-func (ctx *emptyContext) WithValue(k, v interface{}) {
-	ctx.Context = context.WithValue(ctx.Context, k, v)
-}
-
-var Background Context = &emptyContext{context.Background()}
+var Background Context = WrapContext(context.Background())
 
-type myContext struct {
+type wrapContext struct {
 	context.Context
-	pid      uint32
-	uid      uint32
-	gids     []uint32
-	canceled bool
+	cancel func()
+	pid    uint32
+	uid    uint32
+	gids   []uint32
 }
 
-func (c *myContext) Uid() uint32 {
+func (c *wrapContext) Uid() uint32 {
 	return c.uid
 }
 
-func (c *myContext) Gid() uint32 {
+func (c *wrapContext) Gid() uint32 {
 	return c.gids[0]
 }
 
-func (c *myContext) Gids() []uint32 {
+func (c *wrapContext) Gids() []uint32 {
 	return c.gids
 }
 
-func (c *myContext) Pid() uint32 {
+func (c *wrapContext) Pid() uint32 {
 	return c.pid
 }
 
-func (c *myContext) Cancel() {
-	c.canceled = true
+func (c *wrapContext) Cancel() {
+	c.cancel()
 }
 
-func (c *myContext) Canceled() bool {
-	return c.canceled
+func (c *wrapContext) Canceled() bool {
+	return c.Err() != nil
 }
 
-func (c *myContext) WithValue(k, v interface{}) {
+func (c *wrapContext) WithValue(k, v interface{}) {
 	c.Context = context.WithValue(c.Context, k, v)
 }
 
 func NewContext(pid, uid uint32, gids []uint32) Context {
-	return &myContext{context.Background(), pid, uid, gids, false}
+	return wrap(context.Background(), pid, uid, gids)
+}
+
+func WrapContext(ctx context.Context) Context {
+	return wrap(ctx, 0, 0, []uint32{0})
+}
+
+func wrap(ctx context.Context, pid, uid uint32, gids []uint32) Context {
+	c, cancel := context.WithCancel(ctx)
+	return &wrapContext{c, cancel, pid, uid, gids}
 }