Repository: juicedata/juicefs
Pull request #2996 (2022-11-20 07:41:50)

meta/tkv: do compaction during scan

codecov-commenter: # [Codecov](https://codecov.io/gh/juicedata/juicefs/pull/2996?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=juicedata) Report
Base: **58.81**% // Head: **58.63**% // Decreases project coverage by **`-0.18%`** :warning:
> Coverage data is based on head [(`433f96e`)](https://codecov.io/gh/juicedata/juicefs/pull/2996?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=juicedata) compared to base [(`56f2f44`)](https://codecov.io/gh/juicedata/juicefs/commit/56f2f44bd79761255d2da557a38e97be181e206a?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=juicedata).
> Patch coverage: 32.39% of modified lines in pull request are covered.

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##             main    #2996      +/-   ##
==========================================
- Coverage   58.81%   58.63%   -0.19%     
==========================================
  Files         141      141              
  Lines       24613    24716     +103     
==========================================
+ Hits        14476    14491      +15     
- Misses       8254     8332      +78     
- Partials     1883     1893      +10     
```


| [Impacted Files](https://codecov.io/gh/juicedata/juicefs/pull/2996?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=juicedata) | Coverage Δ | |
|---|---|---|
| [pkg/meta/base.go](https://codecov.io/gh/juicedata/juicefs/pull/2996/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=juicedata#diff-cGtnL21ldGEvYmFzZS5nbw==) | `66.79% <0.00%> (-0.86%)` | :arrow_down: |
| [pkg/meta/interface.go](https://codecov.io/gh/juicedata/juicefs/pull/2996/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=juicedata#diff-cGtnL21ldGEvaW50ZXJmYWNlLmdv) | `45.45% <ø> (ø)` | |
| [pkg/meta/tkv\_mem.go](https://codecov.io/gh/juicedata/juicefs/pull/2996/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=juicedata#diff-cGtnL21ldGEvdGt2X21lbS5nbw==) | `87.14% <ø> (+0.52%)` | :arrow_up: |
| [pkg/meta/sql.go](https://codecov.io/gh/juicedata/juicefs/pull/2996/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=juicedata#diff-cGtnL21ldGEvc3FsLmdv) | `63.12% <16.66%> (-0.81%)` | :arrow_down: |
| [pkg/meta/redis.go](https://codecov.io/gh/juicedata/juicefs/pull/2996/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=juicedata#diff-cGtnL21ldGEvcmVkaXMuZ28=) | `69.05% <22.22%> (-0.75%)` | :arrow_down: |
| [cmd/status.go](https://codecov.io/gh/juicedata/juicefs/pull/2996/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=juicedata#diff-Y21kL3N0YXR1cy5nbw==) | `52.85% <30.00%> (-17.88%)` | :arrow_down: |
| [pkg/meta/tkv.go](https://codecov.io/gh/juicedata/juicefs/pull/2996/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=juicedata#diff-cGtnL21ldGEvdGt2Lmdv) | `73.92% <43.33%> (-0.52%)` | :arrow_down: |
| [pkg/meta/sql\_lock.go](https://codecov.io/gh/juicedata/juicefs/pull/2996/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=juicedata#diff-cGtnL21ldGEvc3FsX2xvY2suZ28=) | `71.72% <100.00%> (+1.43%)` | :arrow_up: |
| [pkg/vfs/reader.go](https://codecov.io/gh/juicedata/juicefs/pull/2996/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=juicedata#diff-cGtnL3Zmcy9yZWFkZXIuZ28=) | `79.20% <0.00%> (+0.21%)` | :arrow_up: |
| ... and [2 more](https://codecov.io/gh/juicedata/juicefs/pull/2996/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=juicedata) | |

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=juicedata). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=juicedata)

</details>

[:umbrella: View full report at Codecov](https://codecov.io/gh/juicedata/juicefs/pull/2996?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=juicedata).   
:loudspeaker: Do you have feedback about the report comment? [Let us know in this issue](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=juicedata).

SandyXSD: We call `txn` during the scan process, which might cause problems. However, it seems to be supported by all TKV engines except for the MemKV.
An alternative is to do several scans with a limit for each one.

Patch:
diff --git a/cmd/debug.go b/cmd/debug.go
index d6008636..4a0757b7 100644
--- a/cmd/debug.go
+++ b/cmd/debug.go
@@ -107,13 +107,13 @@ func copyFileOnWindows(srcPath, destPath string) error {
 	return nil
 }
 
-func copyFile(srcPath, destPath string, rootPrivileges bool) error {
+func copyFile(srcPath, destPath string, requireRootPrivileges bool) error {
 	if runtime.GOOS == "windows" {
 		return copyFileOnWindows(srcPath, destPath)
 	}
 
 	var copyArgs []string
-	if rootPrivileges {
+	if requireRootPrivileges {
 		copyArgs = append(copyArgs, "sudo")
 	}
 	copyArgs = append(copyArgs, "/bin/sh", "-c", fmt.Sprintf("cat %s > %s", srcPath, destPath))
@@ -121,7 +121,7 @@ func copyFile(srcPath, destPath string, rootPrivileges bool) error {
 }
 
 func getCmdMount(mp string) (uid, pid, cmd string, err error) {
-	psArgs := []string{"/bin/sh", "-c", "ps -ef | grep -v grep | grep 'juicefs mount' | grep " + mp}
+	psArgs := []string{"/bin/sh", "-c", fmt.Sprintf("ps -ef | grep -v grep | grep -E \"%s mount | /sbin/mount.juicefs\" | grep %s", os.Args[0], mp)}
 	ret, err := exec.Command(psArgs[0], psArgs[1:]...).CombinedOutput()
 	if err != nil {
 		return "", "", "", fmt.Errorf("failed to execute command `%s`: %v", strings.Join(psArgs, " "), err)
@@ -149,16 +149,16 @@ func getCmdMount(mp string) (uid, pid, cmd string, err error) {
 	return uid, pid, cmd, nil
 }
 
-func getDefaultLogDir(rootPrivileges bool) (string, error) {
+func getDefaultLogDir(requireRootPrivileges bool) (string, error) {
 	var defaultLogDir = "/var/log"
 	switch runtime.GOOS {
 	case "linux":
-		if rootPrivileges {
+		if os.Getuid() == 0 {
 			break
 		}
 		fallthrough
 	case "darwin":
-		if rootPrivileges {
+		if requireRootPrivileges {
 			defaultLogDir = "/var/root/.juicefs"
 			break
 		}
@@ -173,13 +173,13 @@ func getDefaultLogDir(rootPrivileges bool) (string, error) {
 
 var logArg = regexp.MustCompile(`--log(\s*=?\s*)(\S+)`)
 
-func getLogPath(cmd string, rootPrivileges bool) (string, error) {
+func getLogPath(cmd string, requireRootPrivileges bool) (string, error) {
 	var logPath string
 	tmp := logArg.FindStringSubmatch(cmd)
 	if len(tmp) == 3 {
 		logPath = tmp[2]
 	} else {
-		defaultLogDir, err := getDefaultLogDir(rootPrivileges)
+		defaultLogDir, err := getDefaultLogDir(requireRootPrivileges)
 		if err != nil {
 			return "", err
 		}
@@ -195,9 +195,9 @@ func closeFile(file *os.File) {
 	}
 }
 
-func copyLogFile(logPath, retLogPath string, limit uint64, rootPrivileges bool) error {
+func copyLogFile(logPath, retLogPath string, limit uint64, requireRootPrivileges bool) error {
 	var copyArgs []string
-	if rootPrivileges {
+	if requireRootPrivileges {
 		copyArgs = append(copyArgs, "sudo")
 	}
 	copyArgs = append(copyArgs, "/bin/sh", "-c")
@@ -209,7 +209,7 @@ func copyLogFile(logPath, retLogPath string, limit uint64, rootPrivileges bool)
 	return exec.Command(copyArgs[0], copyArgs[1:]...).Run()
 }
 
-func getPprofPort(pid, amp string, rootPrivileges bool) (int, error) {
+func getPprofPort(pid, amp string, requireRootPrivileges bool) (int, error) {
 	content, err := os.ReadFile(filepath.Join(amp, ".config"))
 	if err != nil {
 		logger.Warnf("failed to read config file: %v", err)
@@ -229,7 +229,7 @@ func getPprofPort(pid, amp string, rootPrivileges bool) (int, error) {
 	}
 
 	var lsofArgs []string
-	if rootPrivileges {
+	if requireRootPrivileges {
 		lsofArgs = append(lsofArgs, "sudo")
 	}
 	lsofArgs = append(lsofArgs, "/bin/sh", "-c", "lsof -i -nP | grep -v grep | grep LISTEN | grep "+pid)
@@ -395,7 +395,7 @@ func geneZipFile(srcPath, destPath string) error {
 	})
 }
 
-func collectPprof(ctx *cli.Context, cmd string, pid string, amp string, rootPrivileges bool, currDir string, wg *sync.WaitGroup) error {
+func collectPprof(ctx *cli.Context, cmd string, pid string, amp string, requireRootPrivileges bool, currDir string, wg *sync.WaitGroup) error {
 	if !checkAgent(cmd) {
 		logger.Warnf("No agent found, the pprof metrics will not be collected")
 		return nil
@@ -406,7 +406,7 @@ func collectPprof(ctx *cli.Context, cmd string, pid string, amp string, rootPriv
 		return nil
 	}
 
-	port, err := getPprofPort(pid, amp, rootPrivileges)
+	port, err := getPprofPort(pid, amp, requireRootPrivileges)
 	if err != nil {
 		return fmt.Errorf("failed to get pprof port: %v", err)
 	}
@@ -450,12 +450,12 @@ func collectPprof(ctx *cli.Context, cmd string, pid string, amp string, rootPriv
 	return nil
 }
 
-func collectLog(ctx *cli.Context, cmd string, rootPrivileges bool, currDir string) error {
+func collectLog(ctx *cli.Context, cmd string, requireRootPrivileges bool, currDir string) error {
 	if !isUnix() {
 		logger.Warnf("Collecting log currently only support Linux/macOS")
 		return nil
 	}
-	logPath, err := getLogPath(cmd, rootPrivileges)
+	logPath, err := getLogPath(cmd, requireRootPrivileges)
 	if err != nil {
 		return fmt.Errorf("failed to get log path: %v", err)
 	}
@@ -463,7 +463,7 @@ func collectLog(ctx *cli.Context, cmd string, rootPrivileges bool, currDir strin
 	retLogPath := filepath.Join(currDir, "juicefs.log")
 
 	logger.Infof("Log %s is being collected", logPath)
-	return copyLogFile(logPath, retLogPath, limit, rootPrivileges)
+	return copyLogFile(logPath, retLogPath, limit, requireRootPrivileges)
 }
 
 func collectSysInfo(ctx *cli.Context, currDir string) error {
@@ -492,9 +492,9 @@ JuiceFS Version:
 	return nil
 }
 
-func collectSpecialFile(ctx *cli.Context, amp string, currDir string, rootPrivileges bool, wg *sync.WaitGroup) error {
+func collectSpecialFile(ctx *cli.Context, amp string, currDir string, requireRootPrivileges bool, wg *sync.WaitGroup) error {
 	configName := ".config"
-	if err := copyFile(filepath.Join(amp, configName), filepath.Join(currDir, "config.txt"), rootPrivileges); err != nil {
+	if err := copyFile(filepath.Join(amp, configName), filepath.Join(currDir, "config.txt"), requireRootPrivileges); err != nil {
 		return fmt.Errorf("failed to get volume config %s: %v", configName, err)
 	}
 
@@ -505,14 +505,14 @@ func collectSpecialFile(ctx *cli.Context, amp string, currDir string, rootPrivil
 		defer wg.Done()
 		srcPath := filepath.Join(amp, statsName)
 		destPath := filepath.Join(currDir, "stats.txt")
-		if err := copyFile(srcPath, destPath, rootPrivileges); err != nil {
+		if err := copyFile(srcPath, destPath, requireRootPrivileges); err != nil {
 			logger.Errorf("Failed to get volume config %s: %v", statsName, err)
 		}
 
 		logger.Infof("Stats metrics are being sampled, sampling duration: %ds", stats)
 		time.Sleep(time.Second * time.Duration(stats))
 		destPath = filepath.Join(currDir, fmt.Sprintf("stats.%ds.txt", stats))
-		if err := copyFile(srcPath, destPath, rootPrivileges); err != nil {
+		if err := copyFile(srcPath, destPath, requireRootPrivileges); err != nil {
 			logger.Errorf("Failed to get volume config %s: %v", statsName, err)
 		}
 	}()
@@ -552,22 +552,22 @@ func debug(ctx *cli.Context) error {
 	}
 	fmt.Printf("\nMount Command:\n%s\n\n", cmd)
 
-	rootPrivileges := false
+	requireRootPrivileges := false
 	if (uid == "0" || uid == "root") && os.Getuid() != 0 {
 		fmt.Println("Mount point is mounted by the root user, may ask for root privilege...")
-		rootPrivileges = true
+		requireRootPrivileges = true
 	}
 
 	var wg sync.WaitGroup
-	if err := collectSpecialFile(ctx, amp, currDir, rootPrivileges, &wg); err != nil {
+	if err := collectSpecialFile(ctx, amp, currDir, requireRootPrivileges, &wg); err != nil {
 		return err
 	}
 
-	if err := collectLog(ctx, cmd, rootPrivileges, currDir); err != nil {
+	if err := collectLog(ctx, cmd, requireRootPrivileges, currDir); err != nil {
 		return err
 	}
 
-	if err := collectPprof(ctx, cmd, pid, amp, rootPrivileges, currDir, &wg); err != nil {
+	if err := collectPprof(ctx, cmd, pid, amp, requireRootPrivileges, currDir, &wg); err != nil {
 		return err
 	}
 
diff --git a/cmd/gc.go b/cmd/gc.go
index 3eb33d41..78c418fe 100644
--- a/cmd/gc.go
+++ b/cmd/gc.go
@@ -111,7 +111,7 @@ func gc(ctx *cli.Context) error {
 	// Scan all chunks first and do compaction if necessary
 	progress := utils.NewProgress(false, false)
 	if ctx.Bool("compact") {
-		bar := progress.AddCountBar("Scanned chunks", 0)
+		bar := progress.AddCountBar("Compacted chunks", 0)
 		spin := progress.AddDoubleSpinner("Compacted slices")
 		m.OnMsg(meta.DeleteSlice, func(args ...interface{}) error {
 			return store.Remove(args[0].(uint64), int(args[1].(uint32)))
@@ -124,7 +124,7 @@ func gc(ctx *cli.Context) error {
 			}
 			return err
 		})
-		if st := m.CompactAll(meta.Background, bar); st == 0 {
+		if st := m.CompactAll(meta.Background, ctx.Int("threads"), bar); st == 0 {
 			bar.Done()
 			spin.Done()
 			if progress.Quiet {
diff --git a/cmd/mount.go b/cmd/mount.go
index 59c33a3b..60bbe816 100644
--- a/cmd/mount.go
+++ b/cmd/mount.go
@@ -463,7 +463,7 @@ func tryToInstallMountExec() error {
 	if _, err := os.Stat("/sbin/mount.juicefs"); err == nil {
 		return nil
 	}
-	src, err := filepath.Abs(os.Args[0])
+	src, err := os.Executable()
 	if err != nil {
 		return err
 	}
diff --git a/pkg/meta/base.go b/pkg/meta/base.go
index 66e8b35c..e4d6d2ed 100644
--- a/pkg/meta/base.go
+++ b/pkg/meta/base.go
@@ -58,6 +58,8 @@ type engine interface {
 	doFindStaleSessions(limit int) ([]uint64, error) // limit < 0 means all
 	doCleanStaleSession(sid uint64) error
 
+	scanAllChunks(ctx Context, ch chan<- cchunk, bar *utils.Bar) error
+	compactChunk(inode Ino, indx uint32, force bool)
 	doDeleteSustainedInode(sid uint64, inode Ino) error
 	doFindDeletedFiles(ts int64, limit int) (map[Ino]uint64, error) // limit < 0 means all
 	doDeleteFileData(inode Ino, length uint64)
@@ -94,6 +96,13 @@ type fsStat struct {
 	usedInodes int64
 }
 
+// chunk for compaction
+type cchunk struct {
+	inode  Ino
+	indx   uint32
+	slices int
+}
+
 type baseMeta struct {
 	sync.Mutex
 	addr string
@@ -111,6 +120,7 @@ type baseMeta struct {
 	symlinks     *sync.Map
 	msgCallbacks *msgCallbacks
 	umounting    bool
+	sesMu        sync.Mutex
 
 	*fsStat
 
@@ -297,13 +307,14 @@ func (m *baseMeta) expireTime() int64 {
 func (m *baseMeta) refreshSession() {
 	for {
 		utils.SleepWithJitter(m.conf.Heartbeat)
-		m.Lock()
+		m.sesMu.Lock()
 		if m.umounting {
-			m.Unlock()
+			m.sesMu.Unlock()
 			return
 		}
 		m.en.doRefreshSession()
-		m.Unlock()
+		m.sesMu.Unlock()
+
 		old := m.fmt.UUID
 		if _, err := m.Load(false); err != nil {
 			logger.Warnf("reload setting: %s", err)
@@ -343,9 +354,9 @@ func (m *baseMeta) CloseSession() error {
 	if m.conf.ReadOnly {
 		return nil
 	}
-	m.Lock()
+	m.sesMu.Lock()
 	m.umounting = true
-	m.Unlock()
+	m.sesMu.Unlock()
 	logger.Infof("close session %d: %s", m.sid, m.en.doCleanStaleSession(m.sid))
 	return nil
 }
@@ -894,9 +905,12 @@ func (m *baseMeta) NewSlice(ctx Context, id *uint64) syscall.Errno {
 func (m *baseMeta) Close(ctx Context, inode Ino) syscall.Errno {
 	if m.of.Close(inode) {
 		m.Lock()
-		defer m.Unlock()
-		if m.removedFiles[inode] {
+		_, removed := m.removedFiles[inode]
+		if removed {
 			delete(m.removedFiles, inode)
+		}
+		m.Unlock()
+		if removed {
 			_ = m.en.doDeleteSustainedInode(m.sid, inode)
 		}
 	}
@@ -1186,6 +1200,31 @@ func (m *baseMeta) Check(ctx Context, fpath string, repair bool, recursive bool)
 	return
 }
 
+func (m *baseMeta) CompactAll(ctx Context, threads int, bar *utils.Bar) syscall.Errno {
+	var wg sync.WaitGroup
+	ch := make(chan cchunk, 1000000)
+	for i := 0; i < threads; i++ {
+		wg.Add(1)
+		go func() {
+			for c := range ch {
+				logger.Debugf("Compacting chunk %d:%d (%d slices)", c.inode, c.indx, c.slices)
+				m.en.compactChunk(c.inode, c.indx, true)
+				bar.Increment()
+			}
+			wg.Done()
+		}()
+	}
+
+	err := m.en.scanAllChunks(ctx, ch, bar)
+	close(ch)
+	wg.Wait()
+	if err != nil {
+		logger.Warnf("Scan chunks: %s", err)
+		return errno(err)
+	}
+	return 0
+}
+
 func (m *baseMeta) fileDeleted(opened bool, inode Ino, length uint64) {
 	if opened {
 		m.Lock()
diff --git a/pkg/meta/interface.go b/pkg/meta/interface.go
index bdee9181..7b7146a8 100644
--- a/pkg/meta/interface.go
+++ b/pkg/meta/interface.go
@@ -346,7 +346,7 @@ type Meta interface {
 	Setlk(ctx Context, inode Ino, owner uint64, block bool, ltype uint32, start, end uint64, pid uint32) syscall.Errno
 
 	// Compact all the chunks by merge small slices together
-	CompactAll(ctx Context, bar *utils.Bar) syscall.Errno
+	CompactAll(ctx Context, threads int, bar *utils.Bar) syscall.Errno
 	// ListSlices returns all slices used by all files.
 	ListSlices(ctx Context, slices map[Ino][]Slice, delete bool, showProgress func()) syscall.Errno
 	// Remove all files and directories recursively.
@@ -418,7 +418,7 @@ func NewClient(uri string, conf *Config) Meta {
 	}
 	m, err := f(driver, uri[p+3:], conf)
 	if err != nil {
-		logger.Fatalf("Meta %s is not available: %s", uri, err)
+		logger.Fatalf("Meta %s is not available: %s", utils.RemovePassword(uri), err)
 	}
 	return m
 }
diff --git a/pkg/meta/redis.go b/pkg/meta/redis.go
index 01514564..82b91dac 100644
--- a/pkg/meta/redis.go
+++ b/pkg/meta/redis.go
@@ -2638,17 +2638,16 @@ func (m *redisMeta) compactChunk(inode Ino, indx uint32, force bool) {
 	}
 }
 
-func (m *redisMeta) CompactAll(ctx Context, bar *utils.Bar) syscall.Errno {
+func (m *redisMeta) scanAllChunks(ctx Context, ch chan<- cchunk, bar *utils.Bar) error {
 	p := m.rdb.Pipeline()
-	return errno(m.scan(ctx, "c*_*", func(keys []string) error {
-		bar.IncrTotal(int64(len(keys)))
+	return m.scan(ctx, "c*_*", func(keys []string) error {
 		for _, key := range keys {
 			_ = p.LLen(ctx, key)
 		}
 		cmds, err := p.Exec(ctx)
 		if err != nil {
 			logger.Warnf("list slices: %s", err)
-			return errno(err)
+			return err
 		}
 		for i, cmd := range cmds {
 			cnt := cmd.(*redis.IntCmd).Val()
@@ -2657,14 +2656,13 @@ func (m *redisMeta) CompactAll(ctx Context, bar *utils.Bar) syscall.Errno {
 				var indx uint32
 				n, err := fmt.Sscanf(keys[i], m.prefix+"c%d_%d", &inode, &indx)
 				if err == nil && n == 2 {
-					logger.Debugf("compact chunk %d:%d (%d slices)", inode, indx, cnt)
-					m.compactChunk(Ino(inode), indx, true)
+					bar.IncrTotal(1)
+					ch <- cchunk{Ino(inode), indx, int(cnt)}
 				}
 			}
-			bar.Increment()
 		}
 		return nil
-	}))
+	})
 }
 
 func (m *redisMeta) cleanupLeakedInodes(delete bool) {
diff --git a/pkg/meta/sql.go b/pkg/meta/sql.go
index f3410259..4f42048c 100644
--- a/pkg/meta/sql.go
+++ b/pkg/meta/sql.go
@@ -2543,23 +2543,17 @@ func dup(b []byte) []byte {
 	return r
 }
 
-func (m *dbMeta) CompactAll(ctx Context, bar *utils.Bar) syscall.Errno {
-	var cs []chunk
-	err := m.roTxn(func(s *xorm.Session) error {
-		cs = nil
-		return s.Where("length(slices) >= ?", sliceBytes*2).Cols("inode", "indx").Find(&cs)
+func (m *dbMeta) scanAllChunks(ctx Context, ch chan<- cchunk, bar *utils.Bar) error {
+	return m.roTxn(func(s *xorm.Session) error {
+		return s.Table(&chunk{}).Iterate(new(chunk), func(idx int, bean interface{}) error {
+			c := bean.(*chunk)
+			if len(c.Slices) > sliceBytes {
+				bar.IncrTotal(1)
+				ch <- cchunk{c.Inode, c.Indx, len(c.Slices) / sliceBytes}
+			}
+			return nil
+		})
 	})
-	if err != nil {
-		return errno(err)
-	}
-
-	bar.IncrTotal(int64(len(cs)))
-	for _, c := range cs {
-		logger.Debugf("compact chunk %d:%d (%d slices)", c.Inode, c.Indx, len(c.Slices)/sliceBytes)
-		m.compactChunk(c.Inode, c.Indx, true)
-		bar.Increment()
-	}
-	return 0
 }
 
 func (m *dbMeta) ListSlices(ctx Context, slices map[Ino][]Slice, delete bool, showProgress func()) syscall.Errno {
diff --git a/pkg/meta/tkv.go b/pkg/meta/tkv.go
index 64b5d039..6c5add92 100644
--- a/pkg/meta/tkv.go
+++ b/pkg/meta/tkv.go
@@ -2153,27 +2153,19 @@ func (m *kvMeta) compactChunk(inode Ino, indx uint32, force bool) {
 	}
 }
 
-func (r *kvMeta) CompactAll(ctx Context, bar *utils.Bar) syscall.Errno {
+func (m *kvMeta) scanAllChunks(ctx Context, ch chan<- cchunk, bar *utils.Bar) error {
 	// AiiiiiiiiCnnnn     file chunks
 	klen := 1 + 8 + 1 + 4
-	result, err := r.scanValues(r.fmtKey("A"), -1, func(k, v []byte) bool {
-		return len(k) == klen && k[1+8] == 'C' && len(v) > sliceBytes
+	return m.client.scan(m.fmtKey("A"), func(k, v []byte) {
+		if len(k) == klen && k[1+8] == 'C' && len(v) > sliceBytes {
+			bar.IncrTotal(1)
+			ch <- cchunk{
+				inode:  m.decodeInode(k[1:9]),
+				indx:   binary.BigEndian.Uint32(k[10:]),
+				slices: len(v) / sliceBytes,
+			}
+		}
 	})
-	if err != nil {
-		logger.Warnf("scan chunks: %s", err)
-		return errno(err)
-	}
-
-	bar.IncrTotal(int64(len(result)))
-	for k, value := range result {
-		key := []byte(k[1:])
-		inode := r.decodeInode(key[:8])
-		indx := binary.BigEndian.Uint32(key[9:])
-		logger.Debugf("compact chunk %d:%d (%d slices)", inode, indx, len(value)/sliceBytes)
-		r.compactChunk(inode, indx, true)
-		bar.Increment()
-	}
-	return 0
 }
 
 func (m *kvMeta) ListSlices(ctx Context, slices map[Ino][]Slice, delete bool, showProgress func()) syscall.Errno {
diff --git a/pkg/meta/tkv_mem.go b/pkg/meta/tkv_mem.go
index 869554b3..6afcf611 100644
--- a/pkg/meta/tkv_mem.go
+++ b/pkg/meta/tkv_mem.go
@@ -277,7 +277,9 @@ func (c *memKV) scan(prefix []byte, handler func(key []byte, value []byte)) erro
 		if end != "" && it.key >= end {
 			return false
 		}
+		c.Unlock()
 		handler([]byte(it.key), it.value)
+		c.Lock()
 		return true
 	})
 	return nil
diff --git a/sdk/java/src/main/java/io/juicefs/JuiceFileSystemImpl.java b/sdk/java/src/main/java/io/juicefs/JuiceFileSystemImpl.java
index 29def786..0ea020a5 100644
--- a/sdk/java/src/main/java/io/juicefs/JuiceFileSystemImpl.java
+++ b/sdk/java/src/main/java/io/juicefs/JuiceFileSystemImpl.java
@@ -17,7 +17,6 @@ package io.juicefs;
 
 import com.kenai.jffi.internal.StubLoader;
 import io.juicefs.metrics.JuiceFSInstrumentation;
-import io.juicefs.utils.BufferPool;
 import io.juicefs.utils.ConsistentHash;
 import io.juicefs.utils.NodesFetcher;
 import io.juicefs.utils.NodesFetcherBuilder;
@@ -42,6 +41,7 @@ import org.apache.hadoop.io.MD5Hash;
 import org.apache.hadoop.security.AccessControlException;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.util.DataChecksum;
+import org.apache.hadoop.util.DirectBufferPool;
 import org.apache.hadoop.util.Progressable;
 import org.apache.hadoop.util.VersionInfo;
 import org.json.JSONObject;
@@ -93,6 +93,8 @@ public class JuiceFileSystemImpl extends FileSystem {
   private ScheduledExecutorService nodesFetcherThread;
   private ScheduledExecutorService refreshUidThread;
   private Map<String, FileStatus> lastFileStatus = new HashMap<>();
+  private static final DirectBufferPool directBufferPool = new DirectBufferPool();
+
   private boolean metricsEnable = false;
 
   /*
@@ -753,7 +755,7 @@ public class JuiceFileSystemImpl extends FileSystem {
     public FileInputStream(Path f, int fd, int size) throws IOException {
       path = f;
       this.fd = fd;
-      buf = BufferPool.getBuffer(size);
+      buf = directBufferPool.getBuffer(size);
       buf.limit(0);
       position = 0;
     }
@@ -835,12 +837,14 @@ public class JuiceFileSystemImpl extends FileSystem {
       if (!buf.hasRemaining() && b.remaining() <= buf.capacity() && !refill()) {
         return -1;
       }
-      int got = 0;
-      while (b.hasRemaining() && buf.hasRemaining()) {
-        b.put(buf.get());
-        got++;
+      ByteBuffer srcBuf = buf.duplicate();
+      int got = Math.min(b.remaining(), srcBuf.remaining());
+      if (got > 0) {
+        srcBuf.limit(srcBuf.position() + got);
+        b.put(srcBuf);
+        buf.position(srcBuf.position());
+        statistics.incrementBytesRead(got);
       }
-      statistics.incrementBytesRead(got);
       int more = read(position, b);
       if (more <= 0)
         return got > 0 ? got : -1;
@@ -906,7 +910,7 @@ public class JuiceFileSystemImpl extends FileSystem {
       if (buf == null) {
         return; // already closed
       }
-      BufferPool.returnBuffer(buf);
+      directBufferPool.returnBuffer(buf);
       buf = null;
       int r = lib.jfs_close(Thread.currentThread().getId(), fd);
       fd = 0;


Test patch:
diff --git a/pkg/meta/base_test.go b/pkg/meta/base_test.go
index aec7d9a5..1ca41da8 100644
--- a/pkg/meta/base_test.go
+++ b/pkg/meta/base_test.go
@@ -904,7 +904,7 @@ func testCompaction(t *testing.T, m Meta, trash bool) {
 
 	// TODO: check result if that's predictable
 	p, bar := utils.MockProgress()
-	if st := m.CompactAll(ctx, bar); st != 0 {
+	if st := m.CompactAll(ctx, 8, bar); st != 0 {
 		t.Fatalf("compactall: %s", st)
 	}
 	p.Done()
diff --git a/sdk/java/src/test/java/io/juicefs/JuiceFileSystemTest.java b/sdk/java/src/test/java/io/juicefs/JuiceFileSystemTest.java
index 4e238f75..31c826cd 100644
--- a/sdk/java/src/test/java/io/juicefs/JuiceFileSystemTest.java
+++ b/sdk/java/src/test/java/io/juicefs/JuiceFileSystemTest.java
@@ -33,9 +33,9 @@ import java.io.OutputStream;
 import java.net.URI;
 import java.nio.ByteBuffer;
 import java.security.PrivilegedExceptionAction;
-import java.util.Arrays;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
 
 import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_TRASH_CHECKPOINT_INTERVAL_KEY;
 import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_TRASH_INTERVAL_KEY;
@@ -725,4 +725,35 @@ public class JuiceFileSystemTest extends TestCase {
     FileSystem newFs = FileSystem.newInstance(newConf);
     assertEquals(256 << 20, newFs.getDefaultBlockSize(new Path("/")));
   }
+
+  public void testReadSpeed() throws Exception {
+    int read = (128 << 10) ;
+    Path speedFile = new Path("/tmp/speedFile");
+    fs.delete(speedFile, false);
+    FSDataOutputStream ou = fs.create(speedFile);
+    int fileSize = 128 << 20;
+    ou.write(new byte[fileSize]);
+    ou.close();
+    FSDataInputStream open = fs.open(speedFile);
+    AtomicLong counter = new AtomicLong(0L);
+    AtomicBoolean finished = new AtomicBoolean(false);
+    TimerTask timerTask = new TimerTask() {
+      @Override
+      public void run() {
+        System.out.printf("read method calls: %d\n", counter.get());
+        finished.set(true);
+      }
+    };
+    Timer timer = new Timer();
+    timer.schedule(timerTask, 1000);
+
+    ByteBuffer readArray = ByteBuffer.allocateDirect(read);
+    while (!finished.get()) {
+      open.seek(0);
+      readArray.position(0);
+      readArray.limit(read);
+      open.read(readArray);
+      counter.getAndIncrement();
+    }
+  }
 }