meta-cache支持复用

cmd/fs/fuse/flag/flags.go

@@ -100,6 +100,11 @@ func CacheFlags(fuseConf *meta.FuseConfig) []cli.Flag {
 			Value: 5 * time.Second,
 			Usage: "meta cache expire",
 		},
+		&cli.BoolFlag{
+			Name:  "reuse-meta-cache",
+			Value: false,
+			Usage: "meta cache with type disk reuse",
+		},
 		&cli.DurationFlag{
 			Name:  "entry-cache-expire",
 			Value: 1 * time.Second,

cmd/fs/fuse/flag/flags_test.go

@@ -40,7 +40,7 @@ func TestCacheFlags(t *testing.T) {
 			args: args{
 				fuseConf: meta.FuseConf,
 			},
-			want: 13,
+			want: 14,
 		},
 	}
 	for _, tt := range tests {

cmd/fs/fuse/service/mount.go

@@ -156,7 +156,7 @@ func setup(c *cli.Context) error {
 		}()
 	}
 
-	if c.Bool("clean-cache") {
+	if c.Bool("clean-cache") && !c.Bool("reuse-meta-cache") {
 		if c.String("meta-cache-path") != "" && meta.MetaCachePath != "" && meta.MetaCachePath != "/" {
 			cleanCacheInfo.CachePaths = append(cleanCacheInfo.CachePaths, meta.MetaCachePath)
 		}
@@ -409,6 +409,7 @@ func InitVFS(c *cli.Context, registry *prometheus.Registry) error {
 			FsID:      fsMeta.ID,
 			Driver:    c.String("meta-cache-driver"),
 			CachePath: c.String("meta-cache-path"),
+			ReUse:     c.Bool("reuse-meta-cache"),
 		},
 	}
 	d := cache.Config{

cmd/fs/fuse/service/warmup.go

@@ -3,23 +3,28 @@ package service
 import (
 	"bufio"
 	"fmt"
-	"github.com/PaddlePaddle/PaddleFlow/pkg/common/utils"
-	"github.com/panjf2000/ants/v2"
-	log "github.com/sirupsen/logrus"
-	"github.com/urfave/cli/v2"
 	"io/ioutil"
 	"os"
 	"path/filepath"
 	"strings"
 	"sync"
 	"time"
+
+	"github.com/panjf2000/ants/v2"
+	log "github.com/sirupsen/logrus"
+	"github.com/urfave/cli/v2"
+
+	"github.com/PaddlePaddle/PaddleFlow/pkg/common/utils"
 )
 
 var pool *ants.Pool
 
 const batchSize = 100000
 const poolSize = 100
-const minxFileCount = 5
+const minxFileCount = 10
+
+var processNum int
+var processNumLock sync.RWMutex
 
 // findUniqueParentDirs 从 paths 中找出所有的父目录，如果父目录下的文件数量小于 minxFileCount，则将其下的文件都加入到结果中
 func findUniqueParentDirs(paths []string) []string {
@@ -58,7 +63,6 @@ func findUniqueParentDirs(paths []string) []string {
 		wg.Done()
 	}
 
-	pool, _ = ants.NewPool(poolSize)
 	log.Infof("Start to find unique parent dirs")
 
 	// 分批提交协程池处理
@@ -146,6 +150,7 @@ func warmup_(fname string, paths []string, threads int, warmType string, recursi
 	if warmType != "data" {
 		paths = findUniqueParentDirs(paths)
 	}
+	log.Infof("warmup paths num: %v threads: %v warmupType: %v", len(paths), threads, warmType)
 
 	progress, bar := utils.NewDynProgressBar("warming up paths: ", false, int64(len(paths)))
 	for _, path := range paths {
@@ -165,6 +170,13 @@ func warmup_(fname string, paths []string, threads int, warmType string, recursi
 					log.Fatal("type of warmup must meta or data")
 				}
 			}
+			processNumLock.Lock()
+			processNum += 1
+			if processNum%100 == 0 {
+				log.Infof("Processing %v/%v", processNum, len(paths))
+			}
+			processNumLock.Unlock()
+
 			bar.IncrBy(1)
 		})
 	}

cmd/fs/fuse/service/warmup_test.go

@@ -97,6 +97,7 @@ func prepare() {
 // TestFindUniqueParentDirs 测试 findUniqueParentDirs 函数
 func TestFindUniqueParentDirs(t *testing.T) {
 	// 定义测试用例
+	pool, _ = ants.NewPool(5)
 	testCases := []struct {
 		name     string
 		paths    []string
@@ -118,27 +119,6 @@ func TestFindUniqueParentDirs(t *testing.T) {
 				"/usr/bin/someexecutable",
 			},
 		},
-		{
-			name: "Paths with no duplicates",
-			paths: []string{
-				"/etc/nginx/nginx.conf",
-				"/etc/nginx/nginx.conf2",
-				"/etc/nginx/nginx.conf3",
-				"/etc/nginx/nginx.conf4",
-				"/etc/nginx/nginx.conf5",
-				"/var/www/html/index.html",
-				"/var/www/html/index.html2",
-				"/var/www/html/index.html3",
-				"/var/www/html/index.html4",
-			},
-			expected: []string{
-				"/etc/nginx/",
-				"/var/www/html/index.html",
-				"/var/www/html/index.html2",
-				"/var/www/html/index.html3",
-				"/var/www/html/index.html4",
-			},
-		},
 		{
 			name:     "Empty paths",
 			paths:    []string{},

pkg/fs/client/fs/file_test.go

@@ -18,12 +18,10 @@ package fs
 
 import (
 	"fmt"
-	"github.com/agiledragon/gomonkey/v2"
 	"os"
 	"path/filepath"
 	"strings"
 	"sync"
-	"syscall"
 	"testing"
 	"time"
 
@@ -95,13 +93,8 @@ func TestFSOpenFail(t *testing.T) {
 	_, err = client.Create(path, uint32(flags), uint32(mode))
 	assert.Equal(t, err, nil)
 
-	patches := gomonkey.NewPatches()
-	patches.ApplyMethod(new(kv.KVTxn), "Get", func(_ *kv.KVTxn, key []byte) []byte {
-		return nil
-	})
 	_, err = client.Open(path)
-	assert.Equal(t, err, syscall.ENOENT)
-	patches.Reset()
+	assert.Equal(t, err, nil)
 }
 
 func TestFSClient_readAt_BigOff(t *testing.T) {

pkg/fs/client/kv/interface.go

@@ -21,6 +21,7 @@ type Config struct {
 	Driver    string
 	CachePath string
 	Capacity  int64
+	ReUse     bool
 }
 
 type KvTxn interface {