gc: improve the performance of Juicefs gc command

cmd/gc.go

@@ -70,6 +70,11 @@ $ juicefs gc redis://localhost --delete`,
 				Value:   10,
 				Usage:   "number threads to delete leaked objects",
 			},
+			&cli.IntFlag{
+				Name:  "cleanup-threads",
+				Value: 10,
+				Usage: "number threads for cleaning up pending deleted files",
+			},
 		},
 	}
 }
@@ -79,6 +84,10 @@ func gc(ctx *cli.Context) error {
 	removePassword(ctx.Args().Get(0))
 	metaConf := meta.DefaultConf()
 	metaConf.MaxDeletes = ctx.Int("threads")
+	metaConf.MaxCleanups = ctx.Int("cleanup-threads")
+	if metaConf.MaxCleanups <= 0 {
+		logger.Fatalf("cleanup threads should be greater than 0")
+	}
 	metaConf.NoBGJob = true
 	m := meta.NewClient(ctx.Args().Get(0), metaConf)
 	format, err := m.Load(true)

pkg/meta/config.go

@@ -36,6 +36,7 @@ type Config struct {
 	Strict             bool // update ctime
 	Retries            int
 	MaxDeletes         int
+	MaxCleanups        int // number of threads for cleaning up pending deleted files
 	SkipDirNlink       int
 	CaseInsensi        bool
 	ReadOnly           bool

pkg/meta/redis.go

@@ -3341,36 +3341,59 @@ func (m *redisMeta) scanPendingFiles(ctx Context, scan pendingFileScan) error {
 	visited := make(map[Ino]bool)
 	start := int64(0)
 	const batchSize = 1000
+
+	threads := m.conf.MaxCleanups
+	deleteFileChan := make(chan redis.Z, threads)
+	var wg sync.WaitGroup
+
+	for i := 0; i < threads; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for p := range deleteFileChan {
+				v := p.Member.(string)
+				ps := strings.Split(v, ":")
+				if len(ps) != 2 { // will be cleaned up as legacy
+					continue
+				}
+				inode, _ := strconv.ParseUint(ps[0], 10, 64)
+				if visited[Ino(inode)] {
+					continue
+				}
+				visited[Ino(inode)] = true
+				size, _ := strconv.ParseUint(ps[1], 10, 64)
+				clean, err := scan(Ino(inode), size, int64(p.Score))
+				if err != nil {
+					logger.Errorf("scan pending deleted files: %s", err)
+					continue
+				}
+				if clean {
+					m.doDeleteFileData_(Ino(inode), size, v)
+				}
+			}
+		}()
+	}
+
 	for {
 		pairs, err := m.rdb.ZRangeWithScores(Background(), m.delfiles(), start, start+batchSize).Result()
 		if err != nil {
+			close(deleteFileChan)
+			wg.Wait()
 			return err
 		}
-		start += batchSize
 		for _, p := range pairs {
-			v := p.Member.(string)
-			ps := strings.Split(v, ":")
-			if len(ps) != 2 { // will be cleaned up as legacy
-				continue
-			}
-			inode, _ := strconv.ParseUint(ps[0], 10, 64)
-			if visited[Ino(inode)] {
-				continue
-			}
-			visited[Ino(inode)] = true
-			size, _ := strconv.ParseUint(ps[1], 10, 64)
-			clean, err := scan(Ino(inode), size, int64(p.Score))
-			if err != nil {
-				return err
-			}
-			if clean {
-				m.doDeleteFileData_(Ino(inode), size, v)
-			}
+			deleteFileChan <- p
 		}
+
+		start += batchSize
+
 		if len(pairs) < batchSize {
 			break
 		}
 	}
+
+	close(deleteFileChan)
+	wg.Wait()
 	return nil
 }
 

pkg/meta/sql.go

@@ -3295,15 +3295,34 @@ func (m *dbMeta) scanPendingFiles(ctx Context, scan pendingFileScan) error {
 	if err != nil {
 		return err
 	}
+
+	threads := m.conf.MaxCleanups
+	deleteFileChan := make(chan delfile, threads)
+	var wg sync.WaitGroup
+
+	for i := 0; i < threads; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for ds := range deleteFileChan {
+				clean, err := scan(ds.Inode, ds.Length, ds.Expire)
+				if err != nil {
+					logger.Errorf("scan pending deleted files: %s", err)
+					continue
+				}
+				if clean {
+					m.doDeleteFileData(ds.Inode, ds.Length)
+				}
+			}
+		}()
+	}
+
 	for _, ds := range dfs {
-		clean, err := scan(ds.Inode, ds.Length, ds.Expire)
-		if err != nil {
-			return err
-		}
-		if clean {
-			m.doDeleteFileData(ds.Inode, ds.Length)
-		}
+		deleteFileChan <- ds
 	}
+
+	close(deleteFileChan)
+	wg.Wait()
 	return nil
 }
 

pkg/meta/tkv.go

@@ -2574,28 +2574,65 @@ func (m *kvMeta) scanPendingFiles(ctx Context, scan pendingFileScan) error {
 	}
 	// deleted files: Diiiiiiiissssssss
 	klen := 1 + 8 + 8
-	pairs, err := m.scanValues(m.fmtKey("D"), -1, func(k, v []byte) bool {
-		return len(k) == klen
-	})
-	if err != nil {
-		return err
+	batchSize := 100000
+
+	threads := m.conf.MaxCleanups
+	deleteFileChan := make(chan pair, threads)
+	var wg sync.WaitGroup
+
+	for i := 0; i < threads; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for pair := range deleteFileChan {
+				key, value := pair.key, pair.value
+				if len(key) != klen {
+					logger.Errorf("invalid key %x", key)
+					continue
+				}
+				ino := m.decodeInode([]byte(key)[1:9])
+				size := binary.BigEndian.Uint64([]byte(key)[9:])
+				ts := m.parseInt64(value)
+				clean, err := scan(ino, size, ts)
+				if err != nil {
+					logger.Errorf("scan pending deleted files: %s", err)
+					continue
+				}
+				if clean {
+					m.doDeleteFileData(ino, size)
+				}
+			}
+		}()
 	}
 
-	for key, value := range pairs {
-		if len(key) != klen {
-			return fmt.Errorf("invalid key %x", key)
+	startKey := m.fmtKey("D")
+	endKey := nextKey(startKey)
+	for {
+		keys, values, err := m.scan(startKey, endKey, batchSize, func(k, v []byte) bool {
+			return len(k) == klen
+		})
+		if len(keys) == 0 {
+			break
 		}
-		ino := m.decodeInode([]byte(key)[1:9])
-		size := binary.BigEndian.Uint64([]byte(key)[9:])
-		ts := m.parseInt64(value)
-		clean, err := scan(ino, size, ts)
 		if err != nil {
+			logger.Errorf("scan pending deleted files: %s", err)
+			close(deleteFileChan)
+			wg.Wait()
 			return err
 		}
-		if clean {
-			m.doDeleteFileData(ino, size)
+		startKey = nextKey(keys[len(keys)-1])
+
+		for index, key := range keys {
+			deleteFileChan <- pair{key, values[index]}
+		}
+
+		if len(keys) < batchSize {
+			break
 		}
 	}
+
+	close(deleteFileChan)
+	wg.Wait()
 	return nil
 }