✨ up: fsutil/finder - update some worker logic, optimize concurrent processing

Author: inhere

fsutil/find.go

@@ -11,19 +11,19 @@ import (
 	"github.com/gookit/goutil/strutil"
 )
 
-// FilePathInDirs get full file path in dirs.
+// FilePathInDirs get full file path in dirs. return empty string if not found.
 //
 // Params:
 //   - file: can be relative path, file name, full path.
 //   - dirs: dir paths
-func FilePathInDirs(file string, dirs ...string) string {
-	file = comfunc.ExpandHome(file)
-	if FileExists(file) {
-		return file
+func FilePathInDirs(fPath string, dirs ...string) string {
+	fPath = comfunc.ExpandHome(fPath)
+	if FileExists(fPath) {
+		return fPath
 	}
 
 	for _, dirPath := range dirs {
-		fPath := JoinSubPaths(dirPath, file)
+		fPath := JoinSubPaths(dirPath, fPath)
 		if FileExists(fPath) {
 			return fPath
 		}

fsutil/find_test.go

@@ -12,6 +12,16 @@ import (
 	"github.com/gookit/goutil/testutil/fakeobj"
 )
 
+func TestFilePathInDirs(t *testing.T) {
+	result := fsutil.FilePathInDirs("not_existing_file.txt")
+	assert.Empty(t, result)
+	result = fsutil.FilePathInDirs("not_existing_file.txt", "testdata")
+	assert.Empty(t, result)
+
+	result = fsutil.FilePathInDirs("find.go")
+	assert.NotEmpty(t, result)
+}
+
 func TestMatchFirst(t *testing.T) {
 	assert.Eq(t, "testdata", fsutil.MatchFirst([]string{"testdata"}, fsutil.IsDir, ""))
 
@@ -21,6 +31,8 @@ func TestMatchFirst(t *testing.T) {
 
 	ps := fsutil.MatchPaths([]string{"testdata", "testdata/test.jpg"}, fsutil.IsDir)
 	assert.Eq(t, []string{"testdata"}, ps)
+
+	assert.Eq(t, "default_dir", fsutil.MatchFirst([]string{"not_exist_dir"}, fsutil.IsDir, "default_dir"))
 }
 
 func TestSearchNameUp(t *testing.T) {

fsutil/finder/config.go

@@ -56,7 +56,7 @@ type Config struct {
 
 	// ScanDirs scan dir paths for find.
 	ScanDirs []string `json:"scan_dirs"`
-	// FindFlags for find result. default is FlagFile
+	// FindFlags type for find result. default is FlagFile
 	FindFlags FindFlag `json:"find_flags"`
 	// MaxDepth for find result. default is 0 - not limit
 	MaxDepth int `json:"max_depth"`
@@ -383,6 +383,12 @@ func (f *Finder) WithStrFlag(s string) *Finder {
 	return f
 }
 
+// TypeFile only find file.
+func (f *Finder) TypeFile() *Finder { return f.WithFlags(FlagFile) }
+
+// TypeDir only find dir.
+func (f *Finder) TypeDir() *Finder { return f.WithFlags(FlagDir) }
+
 // OnlyFindDir only find dir.
 func (f *Finder) OnlyFindDir() *Finder { return f.WithFlags(FlagDir) }
 

fsutil/finder/finder.go

@@ -119,7 +119,7 @@ func (f *Finder) EachFile(fn func(file *os.File)) {
 		if err == nil {
 			fn(file)
 		} else {
-			f.err = err
+			f.setError(err)
 		}
 	})
 }
@@ -166,8 +166,8 @@ func (f *Finder) prepare() {
 	f.debugf("PREPARE done. type-flag: %s, concurrency: %d", f.c.FindFlags, coNum)
 	f.debugf("config: %+v", f.c)
 	// 创建队列
-	f.ch = make(chan Elem, coNum*8)
-	f.dirQueue = make(chan scanDir, coNum*8)
+	f.ch = make(chan Elem, coNum*8*3)
+	f.dirQueue = make(chan scanDir, coNum*8*2)
 }
 
 // Do finding
@@ -190,14 +190,14 @@ func (f *Finder) find() <-chan Elem {
 
 	f.prepare()
 
+	// 添加初始任务
+	f.addRootDirs()
+
 	// 启动工作goroutine
 	for i := 0; i < f.c.Concurrency; i++ {
 		go f.worker(i)
 	}
 
-	// 添加初始任务
-	f.addRootDirs()
-
 	// 等待所有任务完成并关闭通道
 	go func() {
 		f.debugf("waiting all task complete ...")
@@ -217,13 +217,26 @@ func (f *Finder) find() <-chan Elem {
 
 // worker 处理目录的工作goroutine
 func (f *Finder) worker(index int) {
+	f.debugf("worker#%d STARTING ...", index)
 	for sd := range f.dirQueue {
-		func() {
-			defer f.wg.Done()
-			f.debugf("worker#%d into dir: %s (depth: %d)", index, sd.path, sd.depth)
-			f.findDir(sd.path, sd.depth)
-		}()
+		f.safeFindDir(index, sd.path, sd.depth)
 	}
+	f.debugf("worker#%d DONE.", index)
+}
+
+func (f *Finder) safeFindDir(index int, dirPath string, depth int) {
+	f.debugf("worker#%d into dir: %s (depth: %d)", index, dirPath, depth)
+
+	// recover error and always call wg.Done()
+	defer func() {
+		if err := recover(); err != nil {
+			f.debugf("worker#%d panic in dir: %s, ERROR: %v", index, dirPath, err)
+			f.setError(fmt.Errorf("worker#%d findDir panic, dir: %s, ERROR: %v", index, dirPath, err))
+		}
+		f.wg.Done()
+	}()
+
+	f.findDir(dirPath, depth)
 }
 
 func (f *Finder) addRootDirs() {
@@ -241,6 +254,7 @@ func (f *Finder) addRootDirs() {
 
 		// add task
 		f.debugf("add root-dir: %s", dirPath)
+		f.wg.Add(1)
 		f.dirQueue <- scanDir{path: dirPath}
 	}
 }
@@ -308,7 +322,11 @@ func (f *Finder) findDir(dirPath string, depth int) {
 			// find in sub dir. 添加子目录任务
 			if cfg.MaxDepth == 0 || depth < cfg.MaxDepth {
 				f.debugf("add sub-dir: %s (depth: %d)", fullPath, depth)
-				f.dirQueue <- scanDir{path: fullPath, depth: depth}
+				f.wg.Add(1)
+				// fix: 创建一个 goroutine 添加子目录任务，不然会造成阻塞
+				go func(p string, d int) {
+					f.dirQueue <- scanDir{path: p, depth: d}
+				}(fullPath, depth)
 			}
 			continue
 		}

fsutil/finder/finder_test.go

@@ -26,6 +26,7 @@ func TestFinder_find_file(t *testing.T) {
 	f := finder.EmptyFinder().
 		WithDebug().
 		ScanDir(".").
+		// ScanDir("C:\\Users\\inhere\\workspace\\godev\\gookit\\goutil").
 		NoDotFile().
 		NoDotDir().
 		ExcludeName("*_test.go").
@@ -41,6 +42,7 @@ func TestFinder_find_file(t *testing.T) {
 
 func TestFinder_find_file_withCache(t *testing.T) {
 	f := finder.EmptyFinder().
+		TypeFile().
 		WithDebug().
 		ScanDir("./testdata").
 		NoDotFile().
@@ -102,26 +104,36 @@ func TestFinder_find_file_withCache(t *testing.T) {
 }
 
 func TestFinder_OnlyFindDir(t *testing.T) {
+	// ff := finder.NewFinder("./../").
 	ff := finder.NewFinder("./../../").
+		// ff := finder.NewFinder("C:\\Users\\inhere\\workspace\\godev\\gookit").
+		// ff := finder.EmptyFinder().ScanDir("./../").
+		WithConcurrency(2).
 		OnlyFindDir().
 		UseAbsPath().
 		WithoutDotDir().
-		WithDirName("testdata")
+		WithDirName("testdata").
+		WithDebug()
+
+	fmt.Println(ff.FindPaths())
+	// return
 
-	ff.EachPath(func(filePath string) {
-		fmt.Println(filePath)
+	t.Run("EachPath", func(t *testing.T) {
+		ff.EachPath(func(filePath string) {
+			fmt.Println(filePath)
+		})
+		assert.Gt(t, ff.Num(), 0)
+		assert.Eq(t, 0, ff.CacheNum())
 	})
-	assert.Gt(t, ff.Num(), 0)
-	assert.Eq(t, 0, ff.CacheNum())
 
-	t.Run("each elem", func(t *testing.T) {
-		ff.Each(func(elem finder.Elem) {
-			fmt.Println(elem)
+	t.Run("Each", func(t *testing.T) {
+		ff.Each(func(el finder.Elem) {
+			fmt.Println(el)
 		})
 	})
 
 	ff.ResetResult()
-	assert.Eq(t, 0, ff.Num())
+	assert.Eq(t, uint(0), ff.Num())
 	assert.Eq(t, 0, ff.CacheNum())
 
 	t.Run("max depth", func(t *testing.T) {