Repository: juicedata/juicefs
Pull request #2042 (2022-05-18 08:16:48)

meta: limit the number of goroutines to delete file data


Patch:
diff --git a/.github/workflows/list.yml b/.github/workflows/list.yml
new file mode 100644
index 00000000..b4f3d1ac
--- /dev/null
+++ b/.github/workflows/list.yml
@@ -0,0 +1,206 @@
+name: "list-test"
+
+on:
+  push:
+    branches:
+      - 'release-**'
+    paths-ignore:
+      - 'docs/**'
+  pull_request:
+    #The branches below must be a subset of the branches above
+    branches:
+      - 'release-**'
+    paths-ignore:
+      - 'docs/**'
+  schedule:
+    - cron:  '0 3 * * *'
+  workflow_dispatch:
+
+jobs:
+  pg-list:
+    runs-on: ubuntu-latest
+    services:
+      postgres:
+        image: postgres
+        env:
+          POSTGRES_PASSWORD: postgres
+        options: >-
+          --health-cmd pg_isready
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+        ports:
+          - 5432:5432
+    steps:
+      - uses: actions/checkout@v3
+      - name: Checkout
+        uses: actions/checkout@v2
+        with:
+          fetch-depth: 1
+      - name: Init
+        run: |
+          sudo chmod 777 /var
+          git clone https://github.com/sanwan/actionfile.git
+          cd actionfile
+          cat META_7M_EMPTY_FILE.json.gz* >META_7M_EMPTY_FILE.json.gz
+          gzip -dk  META_7M_EMPTY_FILE.json.gz
+          ls -lh
+          mv META_7M_EMPTY_FILE.json ../
+          cd ../
+          ls -lh
+      - name: Load
+        uses: ./.github/actions/load
+        with:
+          meta_url: "postgres://postgres:postgres@127.0.0.1:5432/sync_test?sslmode=disable" 
+          load_file: "META_7M_EMPTY_FILE.json"
+      - name: List
+        run: |
+          cd /tmp/juicefs-sync-test/createfile/
+          ls -lh |wc -l >/tmp/count.log
+          cat /tmp/count.log | grep 7000001
+          if [ $? != 0 ];then echo "file num is error";exit 1; fi
+      - name: Send Slack Notification
+        if: ${{ failure() }}
+        uses: juicedata/slack-notify-action@main
+        with:
+          channel-id: "${{ secrets.SLACK_CHANNEL_ID_FOR_PR_CHECK_NOTIFY }}"
+          slack_bot_token: "${{ secrets.SLACK_BOT_TOKEN }}"
+      - name: Setup upterm session
+        if: ${{ failure() }}
+        uses: lhotari/action-upterm@v1
+
+
+  mysql-list:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - name: Checkout
+        uses: actions/checkout@v2
+        with:
+          fetch-depth: 1
+      - name: Init
+        run: |
+          sudo chmod 777 /var
+          sudo /etc/init.d/mysql start
+          git clone https://github.com/sanwan/actionfile.git
+          cd actionfile
+          cat META_7M_EMPTY_FILE.json.gz* >META_7M_EMPTY_FILE.json.gz
+          gzip -dk  META_7M_EMPTY_FILE.json.gz
+          ls -lh
+          mv META_7M_EMPTY_FILE.json ../
+          cd ../
+          ls -lh
+      - name: Load
+        uses: ./.github/actions/load
+        with:
+          meta_url: "mysql://root:root@\\(127.0.0.1\\)/sync_test" 
+          load_file: "META_7M_EMPTY_FILE.json"
+      - name: List
+        run: |
+          cd /tmp/juicefs-sync-test/createfile/
+          ls -lh |wc -l >/tmp/count.log
+          cat /tmp/count.log | grep 7000001
+          if [ $? != 0 ];then echo "file num is error";exit 1; fi
+      - name: Send Slack Notification
+        if: ${{ failure() }}
+        uses: juicedata/slack-notify-action@main
+        with:
+          channel-id: "${{ secrets.SLACK_CHANNEL_ID_FOR_PR_CHECK_NOTIFY }}"
+          slack_bot_token: "${{ secrets.SLACK_BOT_TOKEN }}"
+      - name: Setup upterm session
+        if: ${{ failure() }}
+        uses: lhotari/action-upterm@v1
+
+  redis-list:
+    runs-on: ubuntu-latest
+    services:
+      redis:
+        # Docker Hub image
+        image: redis
+        # Set health checks to wait until redis has started
+        options: >-
+          --health-cmd "redis-cli ping"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+        ports:
+          # Maps port 6379 on service container to the host
+          - 6379:6379
+    steps:
+      - uses: actions/checkout@v3
+      - name: Checkout
+        uses: actions/checkout@v2
+        with:
+          fetch-depth: 1
+      - name: Init
+        run: |
+          sudo chmod 777 /var
+          git clone https://github.com/sanwan/actionfile.git
+          cd actionfile
+          cat META_7M_EMPTY_FILE.json.gz* >META_7M_EMPTY_FILE.json.gz
+          gzip -dk  META_7M_EMPTY_FILE.json.gz
+          ls -lh
+          mv META_7M_EMPTY_FILE.json ../
+          cd ../
+          ls -lh
+      - name: Load
+        uses: ./.github/actions/load
+        with:
+          meta_url: redis://127.0.0.1:6379/1 
+          load_file: "META_7M_EMPTY_FILE.json"
+      - name: List
+        run: |
+          cd /tmp/juicefs-sync-test/createfile/
+          ls -lh |wc -l >/tmp/count.log
+          cat /tmp/count.log | grep 7000001
+          if [ $? != 0 ];then echo "file num is error";exit 1; fi
+      - name: Send Slack Notification
+        if: ${{ failure() }}
+        uses: juicedata/slack-notify-action@main
+        with:
+          channel-id: "${{ secrets.SLACK_CHANNEL_ID_FOR_PR_CHECK_NOTIFY }}"
+          slack_bot_token: "${{ secrets.SLACK_BOT_TOKEN }}"
+      - name: Setup upterm session
+        if: ${{ failure() }}
+        uses: lhotari/action-upterm@v1
+
+
+  sqlite3-list:
+    runs-on: ubuntu-latest   
+    steps:
+      - uses: actions/checkout@v3
+      - name: Checkout
+        uses: actions/checkout@v2
+        with:
+          fetch-depth: 1
+      - name: Init
+        run: |
+          sudo chmod 777 /var
+          git clone https://github.com/sanwan/actionfile.git
+          cd actionfile
+          cat META_7M_EMPTY_FILE.json.gz* >META_7M_EMPTY_FILE.json.gz
+          gzip -dk  META_7M_EMPTY_FILE.json.gz
+          ls -lh
+          mv META_7M_EMPTY_FILE.json ../
+          cd ../
+          ls -lh
+      - name: Load
+        uses: ./.github/actions/load
+        with:
+          meta_url: sqlite3://sync-test.db
+          load_file: "META_7M_EMPTY_FILE.json"
+      - name: List
+        run: |
+          cd /tmp/juicefs-sync-test/createfile/
+          ls -lh |wc -l >/tmp/count.log
+          cat /tmp/count.log | grep 7000001
+          if [ $? != 0 ];then echo "file num is error";exit 1; fi
+      - name: Send Slack Notification
+        if: ${{ failure() }}
+        uses: juicedata/slack-notify-action@main
+        with:
+          channel-id: "${{ secrets.SLACK_CHANNEL_ID_FOR_PR_CHECK_NOTIFY }}"
+          slack_bot_token: "${{ secrets.SLACK_BOT_TOKEN }}"
+      - name: Setup upterm session
+        if: ${{ failure() }}
+        uses: lhotari/action-upterm@v1
\ No newline at end of file
diff --git a/cmd/status.go b/cmd/status.go
index 0baec2b4..e3ff4bc3 100644
--- a/cmd/status.go
+++ b/cmd/status.go
@@ -73,7 +73,7 @@ func status(ctx *cli.Context) error {
 	format.RemoveSecret()
 
 	if sid := ctx.Uint64("session"); sid != 0 {
-		s, err := m.GetSession(sid)
+		s, err := m.GetSession(sid, true)
 		if err != nil {
 			logger.Fatalf("get session: %s", err)
 		}
diff --git a/docs/en/getting-started/for_local.md b/docs/en/getting-started/for_local.md
index 42fbe478..071dbc7d 100644
--- a/docs/en/getting-started/for_local.md
+++ b/docs/en/getting-started/for_local.md
@@ -26,7 +26,7 @@ juicefs format [command options] META-URL NAME
 
 As you can see, 3 types of information are required to format a file system.
 
-1. **[command options]**: used to set up the storage medium for the file system; **local disk** will be used by default, and default path is `"$HOME/.juicefs/local"` or `"/var/jfs"`.
+1. **[command options]**: used to set up the storage medium for the file system; **local disk** will be used by default, and default path is `"$HOME/.juicefs/local"`, `"/var/jfs"` or `"C:/jfs/local"`.
 2. **META-URL**: used to set up the metadata engine, usually a URL or the file path of a database.
 3. **NAME**: the name of the file system.
 
diff --git a/docs/zh_cn/getting-started/for_local.md b/docs/zh_cn/getting-started/for_local.md
index 68f28e30..bfc23011 100644
--- a/docs/zh_cn/getting-started/for_local.md
+++ b/docs/zh_cn/getting-started/for_local.md
@@ -26,7 +26,7 @@ juicefs format [command options] META-URL NAME
 
 可见，格式化文件系统需要提供 3 种信息：
 
-1. **[command options]**：设定文件系统的存储介质，留空则**默认使用本地磁盘**作为存储介质，路径为 `"$HOME/.juicefs/local"` 或 `"/var/jfs"`；
+1. **[command options]**：设定文件系统的存储介质，留空则**默认使用本地磁盘**作为存储介质，路径为 `"$HOME/.juicefs/local"`，`"/var/jfs"` 或 `"C:/jfs/local"`；
 2. **META-URL**：用来设置元数据存储，即数据库相关的信息，通常是数据库的 URL 或文件路径；
 3. **NAME**：是文件系统的名称。
 
diff --git a/pkg/meta/base.go b/pkg/meta/base.go
index a742407d..5031ef1e 100644
--- a/pkg/meta/base.go
+++ b/pkg/meta/base.go
@@ -70,6 +70,8 @@ type engine interface {
 	doRename(ctx Context, parentSrc Ino, nameSrc string, parentDst Ino, nameDst string, flags uint32, inode *Ino, attr *Attr) syscall.Errno
 	doSetXattr(ctx Context, inode Ino, name string, value []byte, flags uint32) syscall.Errno
 	doRemoveXattr(ctx Context, inode Ino, name string) syscall.Errno
+
+	GetSession(sid uint64, detail bool) (*Session, error)
 }
 
 type baseMeta struct {
@@ -83,6 +85,7 @@ type baseMeta struct {
 	of           *openfiles
 	removedFiles map[Ino]bool
 	compacting   map[uint64]bool
+	maxDeleting  chan struct{}
 	symlinks     *sync.Map
 	msgCallbacks *msgCallbacks
 	newSpace     int64
@@ -111,6 +114,7 @@ func newBaseMeta(conf *Config) baseMeta {
 		of:           newOpenFiles(conf.OpenCache),
 		removedFiles: make(map[Ino]bool),
 		compacting:   make(map[uint64]bool),
+		maxDeleting:  make(chan struct{}, 100),
 		symlinks:     &sync.Map{},
 		msgCallbacks: &msgCallbacks{
 			callbacks: make(map[uint32]MsgCallback),
@@ -231,7 +235,12 @@ func (m *baseMeta) CleanStaleSessions() {
 		return
 	}
 	for _, sid := range sids {
-		logger.Infof("clean up stale session %d: %s", sid, m.en.doCleanStaleSession(sid))
+		s, err := m.en.GetSession(sid, false)
+		if err != nil {
+			logger.Warnf("Get session info %d: %s", sid, err)
+			s = &Session{Sid: sid}
+		}
+		logger.Infof("clean up stale session %d %+v: %s", sid, s.SessionInfo, m.en.doCleanStaleSession(sid))
 	}
 }
 
@@ -782,9 +791,7 @@ func (m *baseMeta) Close(ctx Context, inode Ino) syscall.Errno {
 		defer m.Unlock()
 		if m.removedFiles[inode] {
 			delete(m.removedFiles, inode)
-			go func() {
-				_ = m.en.doDeleteSustainedInode(m.sid, inode)
-			}()
+			_ = m.en.doDeleteSustainedInode(m.sid, inode)
 		}
 	}
 	return 0
@@ -845,7 +852,19 @@ func (m *baseMeta) fileDeleted(opened bool, inode Ino, length uint64) {
 		m.removedFiles[inode] = true
 		m.Unlock()
 	} else {
-		go m.en.doDeleteFileData(inode, length)
+		m.tryDeleteFileData(inode, length)
+	}
+}
+
+func (m *baseMeta) tryDeleteFileData(inode Ino, length uint64) {
+	select {
+	case m.maxDeleting <- struct{}{}:
+		go func() {
+			m.en.doDeleteFileData(inode, length)
+			<-m.maxDeleting
+		}()
+	default:
+		// will be cleanup later
 	}
 }
 
diff --git a/pkg/meta/interface.go b/pkg/meta/interface.go
index 7664e50a..361851ea 100644
--- a/pkg/meta/interface.go
+++ b/pkg/meta/interface.go
@@ -249,7 +249,7 @@ type Meta interface {
 	// CloseSession does cleanup and close the session.
 	CloseSession() error
 	// GetSession retrieves information of session with sid
-	GetSession(sid uint64) (*Session, error)
+	GetSession(sid uint64, detail bool) (*Session, error)
 	// ListSessions returns all client sessions.
 	ListSessions() ([]*Session, error)
 	// CleanStaleSessions cleans up sessions not active for more than 5 minutes
diff --git a/pkg/meta/redis.go b/pkg/meta/redis.go
index befe405f..a1eeb6e8 100644
--- a/pkg/meta/redis.go
+++ b/pkg/meta/redis.go
@@ -408,7 +408,7 @@ func (m *redisMeta) getSession(sid string, detail bool) (*Session, error) {
 	return &s, nil
 }
 
-func (m *redisMeta) GetSession(sid uint64) (*Session, error) {
+func (m *redisMeta) GetSession(sid uint64, detail bool) (*Session, error) {
 	var legacy bool
 	key := strconv.FormatUint(sid, 10)
 	score, err := m.rdb.ZScore(Background, m.allSessions(), key).Result()
@@ -422,7 +422,7 @@ func (m *redisMeta) GetSession(sid uint64) (*Session, error) {
 	if err != nil {
 		return nil, err
 	}
-	s, err := m.getSession(key, true)
+	s, err := m.getSession(key, detail)
 	if err != nil {
 		return nil, err
 	}
@@ -1874,7 +1874,7 @@ func (m *redisMeta) doDeleteSustainedInode(sid uint64, inode Ino) error {
 	})
 	if err == nil {
 		m.updateStats(newSpace, -1)
-		go m.doDeleteFileData(inode, attr.Length)
+		m.tryDeleteFileData(inode, attr.Length)
 	}
 	return err
 }
diff --git a/pkg/meta/sql.go b/pkg/meta/sql.go
index c8e2d013..2fd41b97 100644
--- a/pkg/meta/sql.go
+++ b/pkg/meta/sql.go
@@ -420,12 +420,12 @@ func (m *dbMeta) getSession(row interface{}, detail bool) (*Session, error) {
 	return &s, nil
 }
 
-func (m *dbMeta) GetSession(sid uint64) (*Session, error) {
+func (m *dbMeta) GetSession(sid uint64, detail bool) (*Session, error) {
 	row := session2{Sid: sid}
 	if ok, err := m.db.Get(&row); err != nil {
 		return nil, err
 	} else if ok {
-		return m.getSession(&row, true)
+		return m.getSession(&row, detail)
 	}
 	if ok, err := m.db.IsTableExist(&session{}); err != nil {
 		return nil, err
@@ -434,7 +434,7 @@ func (m *dbMeta) GetSession(sid uint64) (*Session, error) {
 		if ok, err := m.db.Get(&row); err != nil {
 			return nil, err
 		} else if ok {
-			return m.getSession(&row, true)
+			return m.getSession(&row, detail)
 		}
 	}
 	return nil, fmt.Errorf("session not found: %d", sid)
@@ -1792,7 +1792,7 @@ func (m *dbMeta) doDeleteSustainedInode(sid uint64, inode Ino) error {
 	})
 	if err == nil {
 		m.updateStats(newSpace, -1)
-		go m.doDeleteFileData(inode, n.Length)
+		m.tryDeleteFileData(inode, n.Length)
 	}
 	return err
 }
diff --git a/pkg/meta/tkv.go b/pkg/meta/tkv.go
index 82870701..c21e754a 100644
--- a/pkg/meta/tkv.go
+++ b/pkg/meta/tkv.go
@@ -559,7 +559,7 @@ func (m *kvMeta) getSession(sid uint64, detail bool) (*Session, error) {
 	return &s, nil
 }
 
-func (m *kvMeta) GetSession(sid uint64) (*Session, error) {
+func (m *kvMeta) GetSession(sid uint64, detail bool) (*Session, error) {
 	var legacy bool
 	value, err := m.get(m.sessionKey(sid))
 	if err == nil && value == nil {
@@ -572,7 +572,7 @@ func (m *kvMeta) GetSession(sid uint64) (*Session, error) {
 	if value == nil {
 		return nil, fmt.Errorf("session not found: %d", sid)
 	}
-	s, err := m.getSession(sid, true)
+	s, err := m.getSession(sid, detail)
 	if err != nil {
 		return nil, err
 	}
@@ -1575,7 +1575,7 @@ func (m *kvMeta) doDeleteSustainedInode(sid uint64, inode Ino) error {
 	})
 	if err == nil {
 		m.updateStats(newSpace, -1)
-		go m.doDeleteFileData(inode, attr.Length)
+		m.tryDeleteFileData(inode, attr.Length)
 	}
 	return err
 }
diff --git a/pkg/object/s3.go b/pkg/object/s3.go
index c1c4fd97..9821cd32 100644
--- a/pkg/object/s3.go
+++ b/pkg/object/s3.go
@@ -179,6 +179,9 @@ func (s *s3client) List(prefix, marker string, limit int64) ([]Object, error) {
 	objs := make([]Object, n)
 	for i := 0; i < n; i++ {
 		o := resp.Contents[i]
+		if !strings.HasPrefix(*o.Key, prefix) || *o.Key < marker {
+			return nil, fmt.Errorf("found invalid key %s from List, prefix: %s, marker: %s", *o.Key, prefix, marker)
+		}
 		objs[i] = &obj{
 			*o.Key,
 			*o.Size,
diff --git a/pkg/vfs/backup.go b/pkg/vfs/backup.go
index f542bd29..77021416 100644
--- a/pkg/vfs/backup.go
+++ b/pkg/vfs/backup.go
@@ -104,6 +104,10 @@ func cleanupBackups(blob object.ObjectStorage, now time.Time) {
 	}
 	var objs []string
 	for o := range ch {
+		if o == nil {
+			logger.Warnf("list failed, skip cleanup")
+			return
+		}
 		if !o.IsDir() {
 			objs = append(objs, o.Key())
 		}


Test patch:
diff --git a/pkg/meta/redis_test.go b/pkg/meta/redis_test.go
index ad9ddc97..736a43ca 100644
--- a/pkg/meta/redis_test.go
+++ b/pkg/meta/redis_test.go
@@ -1109,7 +1109,7 @@ func testCloseSession(t *testing.T, m Meta) {
 	case *kvMeta:
 		sid = m.sid
 	}
-	s, err := m.GetSession(sid)
+	s, err := m.GetSession(sid, true)
 	if err != nil {
 		t.Fatalf("get session: %s", err)
 	} else {
@@ -1120,7 +1120,7 @@ func testCloseSession(t *testing.T, m Meta) {
 	if err = m.CloseSession(); err != nil {
 		t.Fatalf("close session: %s", err)
 	}
-	if _, err = m.GetSession(sid); err == nil {
+	if _, err = m.GetSession(sid, true); err == nil {
 		t.Fatalf("get a deleted session: %s", err)
 	}
 	switch m := m.(type) {