cache: speedup scanning (#6549)

Signed-off-by: jiefenghuang <jiefeng@juicedata.io>

Author: jiefenghuang

go.mod

@@ -22,6 +22,7 @@ require (
 	github.com/baidubce/bce-sdk-go v0.9.221
 	github.com/bytedance/mockey v1.2.14
 	github.com/ceph/go-ceph v0.18.0
+	github.com/charlievieth/fastwalk v1.0.14
 	github.com/cloudsoda/go-smb2 v0.0.0-20250228001242-d4c70e6251cc
 	github.com/colinmarc/hdfs/v2 v2.4.0
 	github.com/davies/groupcache v0.0.0-20230821031435-e4e8362f58e1

go.sum

@@ -163,6 +163,8 @@ github.com/ceph/go-ceph v0.18.0/go.mod h1:cflETVTBNAQM6jdr7hpNHHFHKYiJiWWcAeRDrR
 github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/charlievieth/fastwalk v1.0.14 h1:3Eh5uaFGwHZd8EGwTjJnSpBkfwfsak9h6ICgnWlhAyg=
+github.com/charlievieth/fastwalk v1.0.14/go.mod h1:diVcUreiU1aQ4/Wu3NbxxH4/KYdKpLDojrQ1Bb2KgNY=
 github.com/cheggaaa/pb v1.0.29 h1:FckUN5ngEk2LpvuG0fw1GEFx6LtyY2pWI/Z2QgCnEYo=
 github.com/cheggaaa/pb v1.0.29/go.mod h1:W40334L7FMC5JKWldsTWbdGjLo0RxUKK73K+TuPxX30=
 github.com/circonus-labs/circonus-gometrics v2.3.1+incompatible/go.mod h1:nmEj6Dob7S7YxXgwXpfOuvO54S+tGdZdw9fuRZt25Ag=

pkg/chunk/disk_cache.go

@@ -37,6 +37,7 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/charlievieth/fastwalk"
 	"github.com/davies/groupcache/consistenthash"
 	"github.com/dustin/go-humanize"
 	"github.com/google/uuid"
@@ -917,7 +918,8 @@ func (cache *cacheStore) scanCached() {
 
 	cachePrefix := filepath.Join(cache.dir, cacheDir)
 	logger.Debugf("Scan %s to find cached blocks", cachePrefix)
-	_ = filepath.WalkDir(cachePrefix, func(path string, d fs.DirEntry, err error) error {
+	err := fastwalk.Walk(nil, cachePrefix, func(path string, d fs.DirEntry, err error) error {
+		// this func should be concurrent safe
 		if err != nil {
 			return nil
 		}
@@ -953,6 +955,9 @@ func (cache *cacheStore) scanCached() {
 		}
 		return nil
 	})
+	if err != nil {
+		logger.Errorf("Scan cached files in %s failed: %s", cachePrefix, err)
+	}
 
 	cache.Lock()
 	cache.scanned = true
@@ -972,7 +977,8 @@ func (cache *cacheStore) scanStaging() {
 	var count, usage uint64
 	stagingPrefix := filepath.Join(cache.dir, stagingDir)
 	logger.Debugf("Scan %s to find staging blocks", stagingPrefix)
-	_ = filepath.WalkDir(stagingPrefix, func(path string, d fs.DirEntry, err error) error {
+	err := fastwalk.Walk(nil, stagingPrefix, func(path string, d fs.DirEntry, err error) error {
+		// this func should be concurrent safe
 		if err != nil {
 			return nil // ignore it
 		}
@@ -1002,11 +1008,14 @@ func (cache *cacheStore) scanStaging() {
 			cache.m.stageBlocks.Add(1)
 			cache.m.stageBlockBytes.Add(float64(origSize))
 			cache.uploader(key, path, false)
-			count++
-			usage += uint64(origSize)
+			atomic.AddUint64(&count, 1)
+			atomic.AddUint64(&usage, uint64(origSize))
 		}
 		return nil
 	})
+	if err != nil {
+		logger.Errorf("Scan staging files in %s failed: %s", stagingPrefix, err)
+	}
 	if count > 0 {
 		logger.Infof("Found %d staging blocks (%s) in %s with %s", count, humanize.IBytes(usage), cache.dir, time.Since(start))
 	}

pkg/chunk/disk_cache_test.go

@@ -144,6 +144,32 @@ func TestMetrics(t *testing.T) {
 	}
 }
 
+func TestScanCached(t *testing.T) {
+	var err error
+	cfg := defaultConf
+	cfg.CacheEviction = EvictionNone
+	cache := &cacheStore{
+		opTs: make(map[time.Duration]func() error),
+	}
+	cache.state = newDCState(dcUnchanged, cache)
+	cache.keys, err = NewKeyIndex(&cfg)
+	require.NoError(t, err)
+	cache.dir = "/tmp/jfstest_scan"
+	rawDir := filepath.Join(cache.dir, cacheDir)
+	if err := os.MkdirAll(rawDir, 0755); err != nil {
+		t.Fatalf("mkdir %s: %s", rawDir, err)
+	}
+	num := 10
+	for i := 0; i < num; i++ {
+		if f, err := os.Create(filepath.Join(rawDir, fmt.Sprintf("test%d_1024", i))); err == nil {
+			_ = f.Close()
+		}
+	}
+	defer os.RemoveAll(rawDir)
+	cache.scanCached()
+	require.Equal(t, num, cache.keys.len())
+}
+
 func TestChecksum(t *testing.T) {
 	conf := testConf()
 	conf.FreeSpace = 0.01