fileservice: allow loading file infos asynchronously in disk cache

pkg/fileservice/disk_cache.go

@@ -51,6 +51,7 @@ func NewDiskCache(
 	path string,
 	capacity int,
 	perfCounterSets []*perfcounter.CounterSet,
+	asyncLoad bool,
 ) (ret *DiskCache, err error) {
 
 	err = os.MkdirAll(path, 0755)
@@ -80,21 +81,29 @@ func NewDiskCache(
 	ret.updatingPaths.Cond = sync.NewCond(new(sync.Mutex))
 	ret.updatingPaths.m = make(map[string]bool)
 
-	ret.loadCache()
+	if asyncLoad {
+		go ret.loadCache()
+	} else {
+		ret.loadCache()
+	}
 
 	return ret, nil
 }
 
 func (d *DiskCache) loadCache() {
 
+	var numFiles, numCacheFiles int
+
 	_ = filepath.WalkDir(d.path, func(path string, entry os.DirEntry, err error) error {
+		numFiles++
 		if err != nil {
 			return nil //ignore
 		}
 		if entry.IsDir() {
 			// try remove if empty. for cleaning old structure
 			if path != d.path {
-				os.Remove(path)
+				// os.Remove will not delete non-empty directory
+				_ = os.Remove(path)
 			}
 			return nil
 		}
@@ -107,10 +116,16 @@ func (d *DiskCache) loadCache() {
 		}
 
 		d.cache.Set(path, struct{}{}, int(fileSize(info)))
+		numCacheFiles++
 
 		return nil
 	})
 
+	logutil.Info("disk cache info loaded",
+		zap.Any("all files", numFiles),
+		zap.Any("cache files", numCacheFiles),
+	)
+
 }
 
 var _ IOVectorCache = new(DiskCache)

pkg/fileservice/disk_cache_test.go

@@ -41,7 +41,7 @@ func TestDiskCache(t *testing.T) {
 	})
 
 	// new
-	cache, err := NewDiskCache(ctx, dir, 1<<20, nil)
+	cache, err := NewDiskCache(ctx, dir, 1<<20, nil, false)
 	assert.Nil(t, err)
 
 	// update
@@ -125,14 +125,14 @@ func TestDiskCache(t *testing.T) {
 	testRead(cache)
 
 	// new cache instance and read
-	cache, err = NewDiskCache(ctx, dir, 1<<20, nil)
+	cache, err = NewDiskCache(ctx, dir, 1<<20, nil, false)
 	assert.Nil(t, err)
 	testRead(cache)
 
 	assert.Equal(t, 1, numWritten)
 
 	// new cache instance and update
-	cache, err = NewDiskCache(ctx, dir, 1<<20, nil)
+	cache, err = NewDiskCache(ctx, dir, 1<<20, nil, false)
 	assert.Nil(t, err)
 	testUpdate(cache)
 
@@ -150,7 +150,7 @@ func TestDiskCacheWriteAgain(t *testing.T) {
 	var counterSet perfcounter.CounterSet
 	ctx = perfcounter.WithCounterSet(ctx, &counterSet)
 
-	cache, err := NewDiskCache(ctx, dir, 4096, nil)
+	cache, err := NewDiskCache(ctx, dir, 4096, nil, false)
 	assert.Nil(t, err)
 
 	// update
@@ -212,7 +212,7 @@ func TestDiskCacheWriteAgain(t *testing.T) {
 func TestDiskCacheFileCache(t *testing.T) {
 	dir := t.TempDir()
 	ctx := context.Background()
-	cache, err := NewDiskCache(ctx, dir, 1<<20, nil)
+	cache, err := NewDiskCache(ctx, dir, 1<<20, nil, false)
 	assert.Nil(t, err)
 
 	vector := IOVector{
@@ -271,7 +271,7 @@ func TestDiskCacheDirSize(t *testing.T) {
 
 	dir := t.TempDir()
 	capacity := 1 << 20
-	cache, err := NewDiskCache(ctx, dir, capacity, nil)
+	cache, err := NewDiskCache(ctx, dir, capacity, nil, false)
 	assert.Nil(t, err)
 
 	data := bytes.Repeat([]byte("a"), capacity/128)
@@ -330,6 +330,7 @@ func benchmarkDiskCacheWriteThenRead(
 		dir,
 		10<<30,
 		nil,
+		false,
 	)
 	if err != nil {
 		b.Fatal(err)
@@ -421,6 +422,7 @@ func benchmarkDiskCacheReadRandomOffsetAtLargeFile(
 		dir,
 		8<<30,
 		nil,
+		false,
 	)
 	if err != nil {
 		b.Fatal(err)
@@ -488,6 +490,7 @@ func BenchmarkDiskCacheMultipleIOEntries(b *testing.B) {
 		dir,
 		8<<30,
 		nil,
+		false,
 	)
 	if err != nil {
 		b.Fatal(err)

pkg/fileservice/local_fs.go

@@ -150,6 +150,7 @@ func (l *LocalFS) initCaches(ctx context.Context, config CacheConfig) error {
 				*config.DiskPath,
 				int(*config.DiskCapacity),
 				l.perfCounterSets,
+				true,
 			)
 			if err != nil {
 				return err

pkg/fileservice/s3_fs.go

@@ -159,6 +159,7 @@ func (s *S3FS) initCaches(ctx context.Context, config CacheConfig) error {
 			*config.DiskPath,
 			int(*config.DiskCapacity),
 			s.perfCounterSets,
+			true,
 		)
 		if err != nil {
 			return err