Repository: juicedata/juicefs
Pull request #2038 (2022-05-18 04:23:11)

rm millions of files


Patch:
diff --git a/.github/actions/load/action.yml b/.github/actions/load/action.yml
index f907ea7f..3332743a 100644
--- a/.github/actions/load/action.yml
+++ b/.github/actions/load/action.yml
@@ -37,9 +37,9 @@ runs:
           printf "\set AUTOCOMMIT on\ndrop database if exists $db_name; create database $db_name; " |  psql -U postgres -h localhost
         fi
         echo "start load..."
-        ./juicefs load $meta_url $load_file
+        sudo ./juicefs load $meta_url $load_file
         echo "finish load..."
-        ./juicefs mount $meta_url $mount_point -d 
+        sudo ./juicefs mount $meta_url $mount_point -d
         if [ -d data/ ]; then
           rm data/{*,.*} -rf || true
         else
diff --git a/.github/workflows/codeql-analysis.yml b/.github/workflows/codeql-analysis.yml
index dbfd0286..8ccf5291 100644
--- a/.github/workflows/codeql-analysis.yml
+++ b/.github/workflows/codeql-analysis.yml
@@ -47,7 +47,7 @@ jobs:
 
     # Initializes the CodeQL tools for scanning.
     - name: Initialize CodeQL
-      uses: github/codeql-action/init@v1
+      uses: github/codeql-action/init@v2
       with:
         languages: ${{ matrix.language }}
         # If you wish to specify custom queries, you can do so here or in a config file.
@@ -58,7 +58,7 @@ jobs:
     # Autobuild attempts to build any compiled languages  (C/C++, C#, or Java).
     # If this step fails, then you should remove it and run the build manually (see below)
     #- name: Autobuild
-    #  uses: github/codeql-action/autobuild@v1
+    #  uses: github/codeql-action/autobuild@v2
 
     # ℹ️ Command-line programs to run using the OS shell.
     # 📚 https://git.io/JvXDl
@@ -68,7 +68,7 @@ jobs:
     #    uses a compiled language
     - if: matrix.language == 'go'
       name: Autobuild
-      uses: github/codeql-action/autobuild@v1
+      uses: github/codeql-action/autobuild@v2
 
     - if: matrix.language == 'java'
       name: build-java
@@ -79,11 +79,11 @@ jobs:
     #   make release
 
     - name: Perform CodeQL Analysis
-      uses: github/codeql-action/analyze@v1
+      uses: github/codeql-action/analyze@v2
 
     - 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 }}"  
\ No newline at end of file
+        slack_bot_token: "${{ secrets.SLACK_BOT_TOKEN }}"
diff --git a/.github/workflows/config.yml b/.github/workflows/config.yml
new file mode 100644
index 00000000..a2ca50b5
--- /dev/null
+++ b/.github/workflows/config.yml
@@ -0,0 +1,93 @@
+name: "config-test"
+
+on:
+  push:
+    branches: 
+      - main
+      - release-**
+    paths-ignore:
+      - 'docs/**'
+      - '**.md'
+  pull_request:
+    #The branches below must be a subset of the branches above
+    branches: 
+      - main
+      - release-**
+    paths-ignore:
+      - 'docs/**'
+      - '**.md'
+  workflow_dispatch:
+  
+jobs:
+  config-test:
+    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:
+      - name: Set up Go
+        uses: actions/setup-go@v2
+        with:
+          go-version: '1.17.x'
+
+      - name: Checkout
+        uses: actions/checkout@v2
+        with:
+          fetch-depth: 1
+
+      - name: Build linux target
+        run: make juicefs
+
+      - name: Setup minio
+        run: |
+            docker run -d -p 9000:9000 --name minio \
+                     -e "MINIO_ACCESS_KEY=minioadmin" \
+                     -e "MINIO_SECRET_KEY=minioadmin" \
+                     -v /tmp/data:/data \
+                     -v /tmp/config:/root/.minio \
+                     minio/minio server /data     
+                     
+      - name: Download tool
+        run: | 
+          wget -q https://dl.minio.io/client/mc/release/linux-amd64/mc
+          chmod +x mc 
+          
+      - name: Test
+        run: |
+          meta_url=redis://127.0.0.1/1
+          mount_point=/tmp/sync-test/
+          ./mc config host add minio http://127.0.0.1:9000 minioadmin minioadmin
+          ./mc admin user add minio juicedata juicedata
+          ./mc admin policy set minio consoleAdmin user=juicedata
+          ./juicefs format --storage minio --bucket http://localhost:9000/jfs-test --access-key juicedata --secret-key juicedata $meta_url myjfs
+          ./juicefs mount $meta_url $mount_point -d --io-retries 1
+
+          ./mc admin user remove minio juicedata
+          ./mc admin user add minio juicedata1 juicedata1
+          ./mc admin policy set minio consoleAdmin user=juicedata1
+          ./juicefs config $meta_url --access-key juicedata1 --secret-key juicedata1
+          sleep 60
+          echo abc > abc.txt
+          cp abc.txt $mount_point/abc.txt && echo "write success"
+          cat $mount_point/abc.txt | grep abc && echo "read success"
+
+      - 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/.github/workflows/fio_benchmark.yml b/.github/workflows/fio_benchmark.yml
index 1287c76b..36c1c719 100644
--- a/.github/workflows/fio_benchmark.yml
+++ b/.github/workflows/fio_benchmark.yml
@@ -52,7 +52,10 @@ jobs:
         run: |
           export PYROSCOPE_AUTH_TOKEN="${{ secrets.PYROSCOPE_AUTH_TOKEN }}"  
           size=${{ github.event.inputs.size }}
+          test -z "$size" && size='1G' 
           nrfiles=${{ github.event.inputs.nrfiles }}
+          test -z "$nrfiles" && nrfiles=10000
+
           disk_warning_value=90
           decrease_ratio=0.8
           fio_jobs=(
diff --git a/.github/workflows/gateway.yml b/.github/workflows/gateway.yml
new file mode 100644
index 00000000..034cd94e
--- /dev/null
+++ b/.github/workflows/gateway.yml
@@ -0,0 +1,155 @@
+name: "gateway-test"
+
+on:
+  push:
+    branches: 
+      - main
+      - release-**
+    paths-ignore:
+      - 'docs/**'
+      - '**.md'
+  pull_request:
+    #The branches below must be a subset of the branches above
+    branches: 
+      - main
+      - release-**
+    paths-ignore:
+      - 'docs/**'
+      - '**.md'
+  workflow_dispatch:
+
+jobs:
+  upload-small-files:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v2
+        with:
+          fetch-depth: 1
+
+      - name: Build linux target
+        run: |
+          make juicefs
+
+      - name: Test copy small files with gateway
+        run: | 
+          wget -q https://dl.minio.io/client/mc/release/linux-amd64/mc
+          chmod +x mc 
+          sudo apt install md5deep
+          sudo /etc/init.d/mysql start
+
+          meta_url="mysql://root:root@(127.0.0.1)/test_gateway"
+          db_name=$(basename $meta_url)
+          volume=myjfs
+          mp=/tmp/myjfs
+          mysql -uroot -proot -e "drop database if exists $db_name; create database $db_name;" 
+          export MINIO_ROOT_USER=minioadmin
+          export MINIO_ROOT_PASSWORD=minioadmin
+          ./juicefs format $meta_url $volume
+          ./juicefs gateway $meta_url localhost:9000 &
+
+          dd if=/dev/urandom of=file iflag=fullblock,count_bytes bs=4k count='400K' > /dev/null
+          mkdir data
+          for i in {1..1000}; do
+            cp file data/file$i
+          done
+          md5deep -lr data/ > sum.md5
+
+          ./juicefs sync --dirs data/  s3://minioadmin:minioadmin@localhost:9000/$volume/data/ --no-https -p 50
+          #./mc alias set minio http://localhost:9000 minioadmin minioadmin --api S3v4
+          #./mc mb minio/$volume
+          #./mc cp --recursive data/  minio/$volume/data
+
+          ./juicefs mount -d $meta_url $mp
+          count=$(md5deep -lrX sum.md5 || wc -l)
+          [ $count = 0 ] && echo "check md5 succeed" || echo "check md5 failed"
+
+      - name: Setup upterm session
+        if: ${{ failure() }}
+        uses: lhotari/action-upterm@v1
+
+  upload-big-files:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v2
+        with:
+          fetch-depth: 1
+
+      - name: Build linux target
+        run: |
+          make juicefs
+
+      - name: Test copy big files with gateway
+        run: | 
+          wget -q https://dl.minio.io/client/mc/release/linux-amd64/mc
+          chmod +x mc 
+          sudo apt install md5deep
+          sudo /etc/init.d/mysql start
+
+          meta_url="mysql://root:root@(127.0.0.1)/test_gateway"
+          db_name=$(basename $meta_url)
+          volume=myjfs
+          mp=/tmp/myjfs
+          mysql -uroot -proot -e "drop database if exists $db_name; create database $db_name;" 
+          export MINIO_ROOT_USER=minioadmin
+          export MINIO_ROOT_PASSWORD=minioadmin
+          ./juicefs format $meta_url $volume
+          ./juicefs gateway $meta_url localhost:9000 &
+
+          dd if=/dev/urandom of=file iflag=fullblock,count_bytes bs=4k count='1G' > /dev/null
+          mkdir data
+          for i in {1..5}; do
+            cp file data/file_$i
+          done          
+          md5deep -lr data/ > sum.md5
+
+          ./juicefs sync --dirs data/  s3://minioadmin:minioadmin@localhost:9000/$volume/data/ --no-https -p 50
+          #./mc alias set minio http://localhost:9000 minioadmin minioadmin --api S3v4
+          #./mc mb minio/$volume
+          #./mc cp --recursive data/  minio/$volume/data
+          
+          ./juicefs mount -d $meta_url $mp
+          count=$(md5deep -lrX sum.md5 || wc -l)
+          [ $count = 0 ] && echo "check md5 succeed" || echo "check md5 failed"
+
+      - name: Setup upterm session
+        if: ${{ failure() }}
+        uses: lhotari/action-upterm@v1
+
+  sync-with-multithread:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v2
+        with:
+          fetch-depth: 1
+
+      - name: Build linux target
+        run: |
+          make juicefs
+
+      - name: Test sync to gateway with multithread
+        run: | 
+          #test for PR: https://github.com/juicedata/juicefs/pull/2024
+          wget -q https://dl.minio.io/client/mc/release/linux-amd64/mc
+          chmod +x mc 
+          sudo apt install md5deep
+          sudo /etc/init.d/mysql start
+
+          meta_url="mysql://root:root@(127.0.0.1)/test_gateway"
+          volume=myjfs
+          mp=/tmp/myjfs
+          db_name=$(basename $meta_url)
+          mysql -uroot -proot -e "drop database if exists $db_name; create database $db_name;" 
+          export MINIO_ROOT_USER=minioadmin
+          export MINIO_ROOT_PASSWORD=minioadmin
+          ./juicefs format $meta_url $volume
+          ./juicefs gateway $meta_url localhost:9000 &
+          ./juicefs sync --dirs ./  s3://minioadmin:minioadmin@localhost:9000/$volume/source_dir/ --no-https -p 20
+          ./juicefs mount -d $meta_url $mp
+          rm -rf $mp/source_dir
+
+      - name: Setup upterm session
+        if: ${{ failure() }}
+        uses: lhotari/action-upterm@v1
\ No newline at end of file
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/.github/workflows/rmfiles.yml b/.github/workflows/rmfiles.yml
new file mode 100644
index 00000000..576b7688
--- /dev/null
+++ b/.github/workflows/rmfiles.yml
@@ -0,0 +1,283 @@
+name: "rm-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 20 * * *'
+  workflow_dispatch:
+
+jobs:
+  pg-rm:
+    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_4M_EMPTY_FILE_RM.json.gz* >META_4M_EMPTY_FILE_RM.json.gz
+          gzip -dk  META_4M_EMPTY_FILE_RM.json.gz
+          ls -lh
+          mv META_4M_EMPTY_FILE_RM.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_4M_EMPTY_FILE_RM.json"
+      - name: Rm 100w Files in Depth of 1 Directory
+        run: |
+          cd /tmp/juicefs-sync-test/mkdir/
+          sudo chmod 777 ../mkdir
+          echo "rm 100w files in depth of 1 directory"
+          git clone https://github.com/juicedata/juicefs.git
+          cd juicefs
+          make
+          sudo ./juicefs rmr ../aaa/
+      - name: Rm 100w Files in Depth of 10 Directory
+        run: |
+          cd /tmp/juicefs-sync-test/mkdir/
+          echo "rm 100w files in depth of 10 directory"
+          cd juicefs
+          sudo ./juicefs rmr ../aaa0/
+      - name: Rm 200w Files in Depth of 20 Directory
+        run: |
+          cd /tmp/juicefs-sync-test/mkdir/
+          echo "rm 200w files in depth of 20 directory"
+          cd juicefs
+          sudo ./juicefs rmr ../aaa2/
+      - name: Log
+        if: ${{ failure() }}
+        run: |
+          sudo tail -n 100 /var/log/juicefs.log
+      - 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-rm:
+    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_4M_EMPTY_FILE_RM.json.gz* >META_4M_EMPTY_FILE_RM.json.gz
+          gzip -dk  META_4M_EMPTY_FILE_RM.json.gz
+          ls -lh
+          mv META_4M_EMPTY_FILE_RM.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_4M_EMPTY_FILE_RM.json"
+      - name: Rm 100w Files in Depth of 1 Directory
+        run: |
+          cd /tmp/juicefs-sync-test/mkdir/
+          sudo chmod 777 ../mkdir
+          echo "rm 100w files in depth of 1 directory"
+          git clone https://github.com/juicedata/juicefs.git
+          cd juicefs
+          make
+          sudo ./juicefs rmr ../aaa/
+      - name: Rm 100w Files in Depth of 10 Directory
+        run: |
+          cd /tmp/juicefs-sync-test/mkdir/
+          echo "rm 100w files in depth of 10 directory"
+          cd juicefs
+          sudo ./juicefs rmr ../aaa0/
+      - name: Rm 200w Files in Depth of 20 Directory
+        run: |
+          cd /tmp/juicefs-sync-test/mkdir/
+          echo "rm 200w files in depth of 20 directory"
+          cd juicefs
+          sudo ./juicefs rmr ../aaa2/
+      - name: Log
+        if: ${{ failure() }}
+        run: |
+          sudo tail -n 100 /var/log/juicefs.log
+      - 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-rm:
+    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_4M_EMPTY_FILE_RM.json.gz* >META_4M_EMPTY_FILE_RM.json.gz
+          gzip -dk  META_4M_EMPTY_FILE_RM.json.gz
+          ls -lh
+          mv META_4M_EMPTY_FILE_RM.json ../
+          cd ../
+          ls -lh
+      - name: Load
+        uses: ./.github/actions/load
+        with:
+          meta_url: redis://127.0.0.1:6379/1
+          load_file: "META_4M_EMPTY_FILE_RM.json"
+      - name: Rm 100w Files in Depth of 1 Directory
+        run: |
+          cd /tmp/juicefs-sync-test/mkdir/
+          sudo chmod 777 ../mkdir
+          echo "rm 100w files in depth of 1 directory"
+          git clone https://github.com/juicedata/juicefs.git
+          cd juicefs
+          make
+          sudo ./juicefs rmr ../aaa/
+      - name: Rm 100w Files in Depth of 10 Directory
+        run: |
+          cd /tmp/juicefs-sync-test/mkdir/
+          echo "rm 100w files in depth of 10 directory"
+          cd juicefs
+          sudo ./juicefs rmr ../aaa0/
+      - name: Rm 200w Files in Depth of 20 Directory
+        run: |
+          cd /tmp/juicefs-sync-test/mkdir/
+          echo "rm 200w files in depth of 20 directory"
+          cd juicefs
+          sudo ./juicefs rmr ../aaa2/
+      - name: Log
+        if: ${{ failure() }}
+        run: |
+          sudo tail -n 100 /var/log/juicefs.log
+      - 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-rm:
+    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_4M_EMPTY_FILE_RM.json.gz* >META_4M_EMPTY_FILE_RM.json.gz
+          gzip -dk  META_4M_EMPTY_FILE_RM.json.gz
+          ls -lh
+          mv META_4M_EMPTY_FILE_RM.json ../
+          cd ../
+          ls -lh
+      - name: Load
+        uses: ./.github/actions/load
+        with:
+          meta_url: sqlite3://sync-test.db
+          load_file: "META_4M_EMPTY_FILE_RM.json"
+      - name: Rm 100w Files in Depth of 1 Directory
+        run: |
+          cd /tmp/juicefs-sync-test/mkdir/
+          sudo chmod 777 ../mkdir
+          echo "rm 100w files in depth of 1 directory"
+          git clone https://github.com/juicedata/juicefs.git
+          cd juicefs
+          make
+          sudo ./juicefs rmr ../aaa/
+      - name: Rm 100w Files in Depth of 10 Directory
+        run: |
+          cd /tmp/juicefs-sync-test/mkdir/
+          echo "rm 100w files in depth of 10 directory"
+          cd juicefs
+          sudo ./juicefs rmr ../aaa0/
+      - name: Rm 200w Files in Depth of 20 Directory
+        run: |
+          cd /tmp/juicefs-sync-test/mkdir/
+          echo "rm 200w files in depth of 20 directory"
+          cd juicefs
+          sudo ./juicefs rmr ../aaa2/
+      - name: Log
+        if: ${{ failure() }}
+        run: |
+          sudo tail -n 100 /var/log/juicefs.log
+      - 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/.github/workflows/tpch_clickhouse.yml b/.github/workflows/tpch_clickhouse.yml
index 3f51f13c..9ab20e26 100644
--- a/.github/workflows/tpch_clickhouse.yml
+++ b/.github/workflows/tpch_clickhouse.yml
@@ -44,7 +44,7 @@ jobs:
 
       - name: Juicefs Format
         run: |
-          sudo ./juicefs format redis://127.0.0.1:6379/1 pics
+          sudo ./juicefs format --trash-days 0 redis://127.0.0.1:6379/1 pics
 
       - name: Juicefs Mount
         run: |
@@ -72,14 +72,16 @@ jobs:
       - name: Generate Data
         run: |
           cd /data/jfs/
-          git clone https://github.com/sanwan/ssb-dbgen.git
-          cd ssb-dbgen
-          make
-          ./dbgen -s 10 -T l -q
+          git clone https://github.com/sanwan/actiondata.git
+          cd actiondata
+          cat lineorder.tbl.gz* >lineorder.tbl.gz
+          gzip -dk lineorder.tbl.gz
           git clone https://github.com/sanwan/actionfile.git
           mv actionfile/*.tbl .
+          rm lineorder.tbl.gz*
           ls -lrth
           sudo du -sh /data/jfs/
+          df -lh
 
       - name: Create Table
         run: |
@@ -144,11 +146,14 @@ jobs:
 
       - name: Import Data
         run: |
-          cd /data/jfs/ssb-dbgen/
+          cd /data/jfs/actiondata/
           clickhouse-client --query "INSERT INTO customer FORMAT CSV" < customer.tbl
           clickhouse-client --query "INSERT INTO part FORMAT CSV" < part.tbl
           clickhouse-client --query "INSERT INTO supplier FORMAT CSV" < supplier.tbl
           time clickhouse-client --query "INSERT INTO lineorder FORMAT CSV" < lineorder.tbl
+          sudo du -sh /data/jfs/
+          sudo du -sh /var/jfsCache/
+          df -lh
 
 
       - name: ClickHouse Test
@@ -250,7 +255,9 @@ jobs:
                                      ORDER BY
                                          year ASC,
                                          C_NATION ASC;"
-
+          sudo du -sh /var/jfsCache/
+          sudo du -sh /data/jfs/
+          df -lh
 
 
       - name: Send Slack Notification
diff --git a/.github/workflows/tpch_clickhouse_mode.yml b/.github/workflows/tpch_clickhouse_limitcache.yml
similarity index 95%
rename from .github/workflows/tpch_clickhouse_mode.yml
rename to .github/workflows/tpch_clickhouse_limitcache.yml
index a0f45bb3..bd32c08e 100644
--- a/.github/workflows/tpch_clickhouse_mode.yml
+++ b/.github/workflows/tpch_clickhouse_limitcache.yml
@@ -1,4 +1,4 @@
-name: "tpch_clickhouse_mode"
+name: "tpch_clickhouse_limitcache"
 
 on:
   push:
@@ -18,7 +18,7 @@ on:
   workflow_dispatch:
 
 jobs:
-  tpchchmode:
+  tpchchlimitcache:
     runs-on: ubuntu-latest
     steps:
       - name: Set up Go
@@ -44,11 +44,11 @@ jobs:
 
       - name: Juicefs Format
         run: |
-          sudo ./juicefs format redis://127.0.0.1:6379/1 pics
+          sudo ./juicefs format --trash-days 0 redis://127.0.0.1:6379/1 pics
 
       - name: Juicefs Mount
         run: |
-          sudo ./juicefs mount -d redis://127.0.0.1:6379/1 /data/jfs --no-usage-report &
+          sudo ./juicefs mount -d redis://127.0.0.1:6379/1 /data/jfs  --cache-size 3072 --no-usage-report &
 
       - name: Install ClickHouse
         run: |
@@ -72,14 +72,16 @@ jobs:
       - name: Generate Data
         run: |
           cd /data/jfs/
-          git clone https://github.com/sanwan/ssb-dbgen.git
-          cd ssb-dbgen
-          make
-          ./dbgen -s 10 -T l -q
+          git clone https://github.com/sanwan/actiondata.git
+          cd actiondata
+          cat lineorder.tbl.gz* >lineorder.tbl.gz
+          gzip -dk lineorder.tbl.gz
           git clone https://github.com/sanwan/actionfile.git
           mv actionfile/*.tbl .
+          rm lineorder.tbl.gz*
           ls -lrth
           sudo du -sh /data/jfs/
+          df -lh
 
       - name: Create Table
         run: |
@@ -144,12 +146,15 @@ jobs:
 
       - name: Import Data
         run: |
-          cd /data/jfs/ssb-dbgen/
+          df -lh
+          cd /data/jfs/actiondata/
           clickhouse-client --query "INSERT INTO customer FORMAT CSV" < customer.tbl
           clickhouse-client --query "INSERT INTO part FORMAT CSV" < part.tbl
           clickhouse-client --query "INSERT INTO supplier FORMAT CSV" < supplier.tbl
           time clickhouse-client --query "INSERT INTO lineorder FORMAT CSV" < lineorder.tbl
-
+          sudo du -sh /data/jfs/
+          sudo du -sh /var/jfsCache/
+          df -lh
 
       - name: ClickHouse Test
         run: |
@@ -250,7 +255,9 @@ jobs:
                                      ORDER BY
                                          year ASC,
                                          C_NATION ASC;"
-
+          sudo du -sh /var/jfsCache/
+          sudo du -sh /data/jfs/
+          df -lh
 
 
       - name: Send Slack Notification
diff --git a/.github/workflows/tpch_clickhouse_mode_local.yml b/.github/workflows/tpch_clickhouse_mode_local.yml
index 41023c55..f92178c4 100644
--- a/.github/workflows/tpch_clickhouse_mode_local.yml
+++ b/.github/workflows/tpch_clickhouse_mode_local.yml
@@ -74,14 +74,16 @@ jobs:
       - name: Generate Data
         run: |
           cd /home/runner/work/
-          git clone https://github.com/sanwan/ssb-dbgen.git
-          cd ssb-dbgen
-          make
-          ./dbgen -s 10 -T l -q
+          git clone https://github.com/sanwan/actiondata.git
+          cd actiondata
+          cat lineorder.tbl.gz* >lineorder.tbl.gz
+          gzip -dk lineorder.tbl.gz
           git clone https://github.com/sanwan/actionfile.git
           mv actionfile/*.tbl .
+          rm lineorder.tbl.gz*
           ls -lrth
-          sudo du -sh /home/runner/work/ssb-dbgen/
+          sudo du -sh /home/runner/work/actiondata/
+          df -lh
 
 
       - name: Create Table
@@ -145,7 +147,7 @@ jobs:
                                      ENGINE = MergeTree ORDER BY S_SUPPKEY;"
       - name: Import Data
         run: |
-          cd /home/runner/work/ssb-dbgen/
+          cd /home/runner/work/actiondata/
           clickhouse-client --query "INSERT INTO customer FORMAT CSV" < customer.tbl
           clickhouse-client --query "INSERT INTO part FORMAT CSV" < part.tbl
           clickhouse-client --query "INSERT INTO supplier FORMAT CSV" < supplier.tbl
@@ -250,6 +252,7 @@ jobs:
                                      ORDER BY
                                          year ASC,
                                          C_NATION ASC;"
+          df -lh
 
       - name: Send Slack Notification
         if: ${{ failure() }}
diff --git a/README.md b/README.md
index f0f140f5..f820ce5e 100644
--- a/README.md
+++ b/README.md
@@ -1,7 +1,7 @@
 <p align="center"><a href="https://github.com/juicedata/juicefs"><img alt="JuiceFS Logo" src="docs/en/images/juicefs-logo.png" width="50%" /></a></p>
 <p align="center">
     <a href="https://travis-ci.com/juicedata/juicefs"><img alt="Build Status" src="https://travis-ci.com/juicedata/juicefs.svg?token=jKSPwswpc2ph4uMtwpHa&branch=main" /></a>
-    <a href="https://join.slack.com/t/juicefs/shared_invite/zt-n9h5qdxh-0bJojPaql8cfFgwerDQJgA"><img alt="Join Slack" src="https://badgen.net/badge/Slack/Join%20JuiceFS/0abd59?icon=slack" /></a>
+    <a href="https://join.slack.com/t/juicefs/shared_invite/zt-n9h5qdxh-YD7e0JxWdesSEa9vY_f_DA"><img alt="Join Slack" src="https://badgen.net/badge/Slack/Join%20JuiceFS/0abd59?icon=slack" /></a>
     <a href="https://goreportcard.com/report/github.com/juicedata/juicefs"><img alt="Go Report" src="https://goreportcard.com/badge/github.com/juicedata/juicefs" /></a>
     <a href="https://juicefs.com/docs/community/introduction/"><img alt="English doc" src="https://img.shields.io/badge/docs-Doc%20Center-brightgreen" /></a>
     <a href="README_CN.md"><img alt="中文手册" src="https://img.shields.io/badge/docs-%E4%B8%AD%E6%96%87%E6%89%8B%E5%86%8C-brightgreen" /></a>
@@ -191,7 +191,7 @@ Thank you for your contribution! Please refer to the [CONTRIBUTING.md](CONTRIBUT
 
 ## Community
 
-Welcome to join the [Discussions](https://github.com/juicedata/juicefs/discussions) and the [Slack channel](https://join.slack.com/t/juicefs/shared_invite/zt-n9h5qdxh-0bJojPaql8cfFgwerDQJgA) to connect with JuiceFS team members and other users.
+Welcome to join the [Discussions](https://github.com/juicedata/juicefs/discussions) and the [Slack channel](https://join.slack.com/t/juicefs/shared_invite/zt-n9h5qdxh-YD7e0JxWdesSEa9vY_f_DA) to connect with JuiceFS team members and other users.
 
 ## Usage Tracking
 
diff --git a/README_CN.md b/README_CN.md
index 2d79e465..98316afb 100644
--- a/README_CN.md
+++ b/README_CN.md
@@ -1,7 +1,7 @@
 <p align="center"><a href="https://github.com/juicedata/juicefs"><img alt="JuiceFS Logo" src="docs/zh_cn/images/juicefs-logo.png" width="50%" /></a></p>
 <p align="center">
     <a href="https://travis-ci.com/juicedata/juicefs"><img alt="Build Status" src="https://travis-ci.com/juicedata/juicefs.svg?token=jKSPwswpc2ph4uMtwpHa&branch=main" /></a>
-    <a href="https://join.slack.com/t/juicefs/shared_invite/zt-n9h5qdxh-0bJojPaql8cfFgwerDQJgA"><img alt="Join Slack" src="https://badgen.net/badge/Slack/Join%20JuiceFS/0abd59?icon=slack" /></a>
+    <a href="https://join.slack.com/t/juicefs/shared_invite/zt-n9h5qdxh-YD7e0JxWdesSEa9vY_f_DA"><img alt="Join Slack" src="https://badgen.net/badge/Slack/Join%20JuiceFS/0abd59?icon=slack" /></a>
     <a href="https://goreportcard.com/report/github.com/juicedata/juicefs"><img alt="Go Report" src="https://goreportcard.com/badge/github.com/juicedata/juicefs" /></a>
     <a href="README.md"><img alt="English Docs" src="https://img.shields.io/badge/docs-English-informational" /></a>
 </p>
@@ -187,7 +187,7 @@ JuiceFS 目前是 beta 状态，核心的存储格式还没有完全确定，如
 
 ## 社区
 
-欢迎加入 [Discussions](https://github.com/juicedata/juicefs/discussions) 和 [Slack 频道](https://join.slack.com/t/juicefs/shared_invite/zt-n9h5qdxh-0bJojPaql8cfFgwerDQJgA) 跟我们的团队和其他社区成员交流。
+欢迎加入 [Discussions](https://github.com/juicedata/juicefs/discussions) 和 [Slack 频道](https://join.slack.com/t/juicefs/shared_invite/zt-n9h5qdxh-YD7e0JxWdesSEa9vY_f_DA) 跟我们的团队和其他社区成员交流。
 
 ## 使用量收集
 
diff --git a/cmd/bench.go b/cmd/bench.go
index db2091a1..e410dbf7 100644
--- a/cmd/bench.go
+++ b/cmd/bench.go
@@ -248,23 +248,28 @@ func (bm *benchmark) colorize(item string, value, cost float64, prec int) (strin
 	return svalue, scost
 }
 
-func (bm *benchmark) printResult(result [][3]string) {
-	var rawmax, max [3]int
+func printResult(result [][]string, isatty bool) {
+	if len(result) < 2 {
+		logger.Fatalf("result must not be empty")
+	}
+	colNum := len(result[0])
+	rawmax, max := make([]int, colNum), make([]int, colNum)
 	for _, l := range result {
-		for i := 0; i < 3; i++ {
+		for i := 0; i < colNum; i++ {
 			if len(l[i]) > rawmax[i] {
 				rawmax[i] = len(l[i])
 			}
 		}
 	}
-	max = rawmax
-	if bm.tty {
-		max[1] -= 11 // no color chars
-		max[2] -= 11
+	copy(max, rawmax)
+	if isatty {
+		for i := 1; i < colNum; i++ {
+			max[i] -= 11
+		}
 	}
 
 	var b strings.Builder
-	for i := 0; i < 3; i++ {
+	for i := 0; i < colNum; i++ {
 		b.WriteByte('+')
 		b.WriteString(strings.Repeat("-", max[i]+2))
 	}
@@ -273,8 +278,8 @@ func (bm *benchmark) printResult(result [][3]string) {
 	fmt.Println(divider)
 
 	b.Reset()
-	header := []string{"ITEM", "VALUE", "COST"}
-	for i := 0; i < 3; i++ {
+	header := result[0]
+	for i := 0; i < colNum; i++ {
 		b.WriteString(" | ")
 		b.WriteString(padding(header[i], max[i], ' '))
 	}
@@ -282,9 +287,9 @@ func (bm *benchmark) printResult(result [][3]string) {
 	fmt.Println(b.String()[1:])
 	fmt.Println(divider)
 
-	for _, l := range result {
+	for _, l := range result[1:] {
 		b.Reset()
-		for i := 0; i < 3; i++ {
+		for i := 0; i < colNum; i++ {
 			b.WriteString(" | ")
 			if spaces := rawmax[i] - len(l[i]); spaces > 0 {
 				b.WriteString(strings.Repeat(" ", spaces))
@@ -371,10 +376,12 @@ func bench(ctx *cli.Context) error {
 	if statsPath != "" {
 		stats = readStats(statsPath)
 	}
-	var result [][3]string
+	var result [][]string
+	result = append(result, []string{"ITEM", "VALUE", "COST"})
 	if b := bm.big; b != nil {
 		cost := b.run("write")
-		line := [3]string{"Write big file"}
+		line := make([]string, 3)
+		line[0] = "Write big file"
 		line[1], line[2] = bm.colorize("bigwr", float64((b.fsize>>20)*b.fcount*bm.threads)/cost, cost/float64(b.fcount), 2)
 		line[1] += " MiB/s"
 		line[2] += " s/file"
@@ -390,7 +397,8 @@ func bench(ctx *cli.Context) error {
 	}
 	if s := bm.small; s != nil {
 		cost := s.run("write")
-		line := [3]string{"Write small file"}
+		line := make([]string, 3)
+		line[0] = "Write small file"
 		line[1], line[2] = bm.colorize("smallwr", float64(s.fcount*bm.threads)/cost, cost*1000/float64(s.fcount), 1)
 		line[1] += " files/s"
 		line[2] += " ms/file"
@@ -434,7 +442,8 @@ func bench(ctx *cli.Context) error {
 			if count > 0 {
 				cost = diff(item+"_sum") * 1000 / count
 			}
-			line := [3]string{title}
+			line := make([]string, 3)
+			line[0] = title
 			line[1], line[2] = bm.colorize(nick, count, cost, 0)
 			line[1] += " operations"
 			line[2] += " ms/op"
@@ -457,6 +466,6 @@ func bench(ctx *cli.Context) error {
 		}
 		fmt.Printf(fmtString, diff("uptime"), diff("cpu_usage")*100/diff("uptime"), stats2["juicefs_memory"]/1024/1024)
 	}
-	bm.printResult(result)
+	printResult(result, bm.tty)
 	return nil
 }
diff --git a/cmd/main.go b/cmd/main.go
index 2802dd33..031d174c 100644
--- a/cmd/main.go
+++ b/cmd/main.go
@@ -65,6 +65,7 @@ func Main(args []string) error {
 			cmdGateway(),
 			cmdWebDav(),
 			cmdBench(),
+			cmdObjbench(),
 			cmdWarmup(),
 			cmdRmr(),
 			cmdSync(),
@@ -77,7 +78,6 @@ func Main(args []string) error {
 	}
 
 	return app.Run(reorderOptions(app, args))
-
 }
 
 func handleSysMountArgs(args []string) []string {
@@ -219,7 +219,7 @@ func reorderOptions(app *cli.App, args []string) []string {
 // Check number of positional arguments, set logger level and setup agent if needed
 func setup(c *cli.Context, n int) {
 	if c.NArg() < n {
-		logger.Errorf("This command requires at least %d arguments", n)
+		fmt.Printf("ERROR: This command requires at least %d arguments\n", n)
 		fmt.Printf("USAGE:\n   juicefs %s [command options] %s\n", c.Command.Name, c.Command.ArgsUsage)
 		os.Exit(1)
 	}
diff --git a/cmd/mount.go b/cmd/mount.go
index c954d97c..3c506ce4 100644
--- a/cmd/mount.go
+++ b/cmd/mount.go
@@ -91,10 +91,12 @@ func installHandler(mp string) {
 	signal.Notify(signalChan, syscall.SIGTERM, syscall.SIGINT, syscall.SIGHUP)
 	go func() {
 		for {
-			<-signalChan
+			sig := <-signalChan
+			logger.Infof("Received signal %s, exiting...", sig.String())
 			go func() { _ = doUmount(mp, true) }()
 			go func() {
 				time.Sleep(time.Second * 3)
+				logger.Warnf("Umount not finished after 3 seconds, force exit")
 				os.Exit(1)
 			}()
 		}
diff --git a/cmd/objbench.go b/cmd/objbench.go
new file mode 100644
index 00000000..60fb920c
--- /dev/null
+++ b/cmd/objbench.go
@@ -0,0 +1,954 @@
+/*
+ * JuiceFS, Copyright 2022 Juicedata, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package cmd
+
+import (
+	"bytes"
+	"errors"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"math"
+	"math/rand"
+	"os"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/juicedata/juicefs/pkg/object"
+	osync "github.com/juicedata/juicefs/pkg/sync"
+	"github.com/juicedata/juicefs/pkg/utils"
+	"github.com/mattn/go-isatty"
+	"github.com/urfave/cli/v2"
+)
+
+func cmdObjbench() *cli.Command {
+	return &cli.Command{
+		Name:      "objbench",
+		Action:    objbench,
+		Category:  "TOOL",
+		Usage:     "Run benchmark on an object storage",
+		ArgsUsage: "BUCKET",
+		Description: `
+Run basic benchmark on the target object storage to test if it works as expected.
+
+Examples:
+# Run benchmark on S3
+$ ACCESS_KEY=myAccessKey SECRET_KEY=mySecretKey juicefs objbench --storage s3 --bucket https://mybucket.s3.us-east-2.amazonaws.com -p 4`,
+		Flags: []cli.Flag{
+			&cli.StringFlag{
+				Name:  "storage",
+				Value: "file",
+				Usage: "object storage type (e.g. s3, gcs, oss, cos)",
+			},
+			&cli.StringFlag{
+				Name:  "access-key",
+				Usage: "access key for object storage (env ACCESS_KEY)",
+			},
+			&cli.StringFlag{
+				Name:  "secret-key",
+				Usage: "secret key for object storage (env SECRET_KEY)",
+			},
+			&cli.UintFlag{
+				Name:  "block-size",
+				Value: 4096,
+				Usage: "size of each IO block in KiB",
+			},
+			&cli.UintFlag{
+				Name:  "big-object-size",
+				Value: 1024,
+				Usage: "size of each big object in MiB",
+			},
+			&cli.UintFlag{
+				Name:  "small-object-size",
+				Value: 128,
+				Usage: "size of each small object in KiB",
+			},
+			&cli.UintFlag{
+				Name:  "small-objects",
+				Value: 100,
+				Usage: "number of small object",
+			},
+			&cli.BoolFlag{
+				Name:  "skip-functional-tests",
+				Usage: "skip functional tests",
+			},
+			&cli.UintFlag{
+				Name:    "threads",
+				Aliases: []string{"p"},
+				Value:   4,
+				Usage:   "number of concurrent threads",
+			},
+		},
+	}
+}
+
+var (
+	nspt    = "not support"
+	pass    = "pass"
+	skipped = "skipped"
+	failed  = "failed"
+)
+
+func objbench(ctx *cli.Context) error {
+	setup(ctx, 1)
+	for _, name := range []string{"big-object-size", "small-object-size", "block-size", "small-objects", "threads"} {
+		if ctx.Uint(name) == 0 {
+			logger.Fatalf("%s should not be set to zero", name)
+		}
+	}
+	ak, sk := ctx.String("access-key"), ctx.String("secret-key")
+	if ak == "" {
+		ak = os.Getenv("ACCESS_KEY")
+	}
+	if sk == "" {
+		sk = os.Getenv("SECRET_KEY")
+	}
+	blobOrigin, err := object.CreateStorage(strings.ToLower(ctx.String("storage")), ctx.Args().First(), ak, sk)
+	if err != nil {
+		logger.Fatalf("create storage failed: %v", err)
+	}
+
+	prefix := fmt.Sprintf("__juicefs_benchmark_%d__/", time.Now().UnixNano())
+	blob := object.WithPrefix(blobOrigin, prefix)
+	defer func() {
+		_ = blobOrigin.Delete(prefix)
+	}()
+	bSize := int(ctx.Uint("block-size")) << 10
+	fsize := int(ctx.Uint("big-object-size")) << 20
+	smallBSize := int(ctx.Uint("small-object-size")) << 10
+	bCount := int(math.Ceil(float64(fsize) / float64(bSize)))
+	sCount := int(ctx.Uint("small-objects"))
+	threads := int(ctx.Uint("threads"))
+	tty := isatty.IsTerminal(os.Stdout.Fd())
+	progress := utils.NewProgress(!tty, false)
+	if tty {
+		nspt = fmt.Sprintf("%s%dm%s%s", COLOR_SEQ, YELLOW, nspt, RESET_SEQ)
+		skipped = fmt.Sprintf("%s%dm%s%s", COLOR_SEQ, YELLOW, skipped, RESET_SEQ)
+		pass = fmt.Sprintf("%s%dm%s%s", COLOR_SEQ, GREEN, pass, RESET_SEQ)
+		failed = fmt.Sprintf("%s%dm%s%s", COLOR_SEQ, RED, failed, RESET_SEQ)
+	}
+	fmt.Println("Start Functional Testing ...")
+
+	var result [][]string
+	result = append(result, []string{"CATEGORY", "TEST", "RESULT"})
+	if !ctx.IsSet("skip-functional-tests") {
+		functionalTesting(blob, &result, tty)
+	}
+	printResult(result, tty)
+	fmt.Println("\nStart Performance Testing ...")
+	var pResult [][]string
+	pResult = append(pResult, []string{"ITEM", "VALUE", "COST"})
+
+	apis := []apiInfo{
+		{
+			name:  "smallput",
+			count: sCount,
+			title: "put small objects",
+			getResult: func(cost float64) []string {
+				line := []string{"", nspt, nspt}
+				if cost > 0 {
+					line[1], line[2] = colorize("smallput", float64(sCount)/cost, cost*1000/float64(sCount), 1, tty)
+					line[1] += " objects/s"
+					line[2] += " ms/object"
+				}
+				return line
+			},
+		}, {
+			name:  "smallget",
+			count: sCount,
+			title: "get small objects",
+			getResult: func(cost float64) []string {
+				line := []string{"", nspt, nspt}
+				if cost > 0 {
+					line[1], line[2] = colorize("smallget", float64(sCount)/cost, cost*1000/float64(sCount), 1, tty)
+					line[1] += " objects/s"
+					line[2] += " ms/object"
+				}
+				return line
+			},
+		}, {
+			name:     "put",
+			count:    bCount,
+			title:    "upload objects",
+			startKey: sCount,
+			getResult: func(cost float64) []string {
+				line := []string{"", nspt, nspt}
+				if cost > 0 {
+					line[1], line[2] = colorize("put", float64(bSize>>20*bCount)/cost, cost*1000/float64(bCount), 2, tty)
+					line[1] += " MiB/s"
+					line[2] += " ms/object"
+				}
+				return line
+			},
+		}, {
+			name:     "get",
+			count:    bCount,
+			title:    "download objects",
+			startKey: sCount,
+			after: func(blob object.ObjectStorage) {
+				for i := sCount; i < sCount+bCount; i++ {
+					_ = blob.Delete(strconv.Itoa(i))
+				}
+			},
+			getResult: func(cost float64) []string {
+				line := []string{"", nspt, nspt}
+				if cost > 0 {
+					line[1], line[2] = colorize("get", float64(bSize>>20*bCount)/cost, cost*1000/float64(bCount), 2, tty)
+					line[1] += " MiB/s"
+					line[2] += " ms/object"
+				}
+				return line
+			},
+		}, {
+			name:  "list",
+			title: "list objects",
+			count: 100,
+			getResult: func(cost float64) []string {
+				line := []string{"", nspt, nspt}
+				if cost > 0 {
+					line[1], line[2] = colorize("list", float64(sCount)*100/cost, cost*10, 2, tty)
+					line[1] += " objects/s"
+					line[2] += " ms/op"
+				}
+				return line
+			},
+		}, {
+			name:  "head",
+			count: sCount,
+			title: "head objects",
+			getResult: func(cost float64) []string {
+				line := []string{"", nspt, nspt}
+				if cost > 0 {
+					line[1], line[2] = colorize("head", float64(sCount)/cost, cost*1000/float64(sCount), 1, tty)
+					line[1] += " objects/s"
+					line[2] += " ms/object"
+				}
+				return line
+			},
+		}, {
+			name:  "chtimes",
+			count: sCount,
+			title: "update mtime",
+			getResult: func(cost float64) []string {
+				line := []string{"", nspt, nspt}
+				if cost > 0 {
+					line[1], line[2] = colorize("chtimes", float64(sCount)/cost, cost*1000/float64(sCount), 1, tty)
+					line[1] += " objects/s"
+					line[2] += " ms/object"
+				}
+				return line
+			},
+		}, {
+			name:  "chmod",
+			count: sCount,
+			title: "change permissions",
+			getResult: func(cost float64) []string {
+				line := []string{"", nspt, nspt}
+				if cost > 0 {
+					line[1], line[2] = colorize("chmod", float64(sCount)/cost, cost*1000/float64(sCount), 1, tty)
+					line[1] += " objects/s"
+					line[2] += " ms/object"
+				}
+				return line
+			},
+		}, {
+			name:  "chown",
+			count: sCount,
+			title: "change owner/group",
+			getResult: func(cost float64) []string {
+				line := []string{"", nspt, nspt}
+				if cost > 0 {
+					line[1], line[2] = colorize("chown", float64(sCount)/cost, cost*1000/float64(sCount), 1, tty)
+					line[1] += " objects/s"
+					line[2] += " ms/object"
+				}
+				return line
+			},
+		}, {
+			name:  "delete",
+			count: sCount,
+			title: "delete objects",
+			getResult: func(cost float64) []string {
+				line := []string{"", nspt, nspt}
+				if cost > 0 {
+					line[1], line[2] = colorize("delete", float64(sCount)/cost, cost*1000/float64(sCount), 1, tty)
+					line[1] += " objects/s"
+					line[2] += " ms/object"
+				}
+				return line
+			},
+		},
+	}
+
+	bm := &benchMarkObj{
+		blob:        blob,
+		progressBar: progress,
+		threads:     threads,
+		seed:        make([]byte, bSize),
+		smallSeed:   make([]byte, smallBSize),
+	}
+	rand.Read(bm.seed)
+	rand.Read(bm.smallSeed)
+
+	for _, api := range apis {
+		pResult = append(pResult, bm.run(api))
+	}
+	progress.Done()
+
+	for i := bCount; i < bCount*2; i++ {
+		_ = bm.delete(strconv.Itoa(i), 0)
+	}
+
+	fmt.Printf("Benchmark finished! block-size: %d KiB, big-object-size: %d MiB, small-object-size: %d KiB, small-objects: %d, NumThreads: %d\n",
+		ctx.Uint("block-size"), ctx.Uint("big-object-size"), ctx.Uint("small-object-size"), sCount, threads)
+
+	// adjust the print order
+	pResult[1], pResult[3] = pResult[3], pResult[1]
+	pResult[2], pResult[4] = pResult[4], pResult[2]
+	pResult[7], pResult[10] = pResult[10], pResult[7]
+	printResult(pResult, tty)
+	return nil
+}
+
+var resultRangeForObj = map[string][4]float64{
+	"put":          {100, 150, 50, 150},
+	"get":          {100, 150, 50, 150},
+	"smallput":     {10, 30, 30, 100},
+	"smallget":     {10, 30, 30, 100},
+	"multi-upload": {100, 150, 20, 50},
+	"list":         {1000, 10000, 100, 200},
+	"head":         {10, 30, 30, 100},
+	"delete":       {10, 30, 30, 100},
+	"chmod":        {10, 30, 30, 100},
+	"chown":        {10, 30, 30, 100},
+	"chtimes":      {10, 30, 30, 100},
+}
+
+func colorize(item string, value, cost float64, prec int, isatty bool) (string, string) {
+	svalue := strconv.FormatFloat(value, 'f', prec, 64)
+	scost := strconv.FormatFloat(cost, 'f', 2, 64)
+	if isatty {
+		r, ok := resultRangeForObj[item]
+		if !ok {
+			logger.Fatalf("Invalid item: %s", item)
+		}
+		var color int
+		if value > r[1] { // max
+			color = GREEN
+		} else if value > r[0] { // min
+			color = YELLOW
+		} else {
+			color = RED
+		}
+		svalue = fmt.Sprintf("%s%dm%s%s", COLOR_SEQ, color, svalue, RESET_SEQ)
+		if cost < r[2] { // min
+			color = GREEN
+		} else if cost < r[3] { // max
+			color = YELLOW
+		} else {
+			color = RED
+		}
+		scost = fmt.Sprintf("%s%dm%s%s", COLOR_SEQ, color, scost, RESET_SEQ)
+	}
+	return svalue, scost
+}
+
+type apiInfo struct {
+	name      string
+	title     string
+	count     int
+	startKey  int
+	getResult func(cost float64) []string
+	after     func(blob object.ObjectStorage)
+}
+
+type benchMarkObj struct {
+	progressBar     *utils.Progress
+	blob            object.ObjectStorage
+	threads         int
+	seed, smallSeed []byte
+}
+
+func (bm *benchMarkObj) run(api apiInfo) []string {
+	if api.name == "chown" || api.name == "chmod" || api.name == "chtimes" {
+		if err := bm.chmod("not_exists", 0); err == utils.ENOTSUP {
+			line := api.getResult(-1)
+			line[0] = api.title
+			return line
+		}
+		if api.name == "chown" && strings.HasPrefix(bm.blob.String(), "file://") && os.Getuid() != 0 {
+			logger.Warnf("chown test should be run by root")
+			return []string{api.title, skipped, skipped}
+		}
+	}
+	var fn func(key string, startKey int) error
+	switch api.name {
+	case "put":
+		fn = bm.put
+	case "get":
+		fn = bm.get
+	case "smallput":
+		fn = bm.smallPut
+	case "smallget":
+		fn = bm.smallGet
+	case "delete":
+		fn = bm.delete
+	case "head":
+		fn = bm.head
+	case "list":
+		fn = bm.list
+	case "chown":
+		fn = bm.chown
+	case "chmod":
+		fn = bm.chmod
+	case "chtimes":
+		fn = bm.chtimes
+	}
+
+	var wg sync.WaitGroup
+	pool := make(chan struct{}, bm.threads)
+	count := api.count
+	if api.count != 0 {
+		count = api.count
+	}
+	bar := bm.progressBar.AddCountBar(api.title, int64(count))
+	var err error
+	start := time.Now()
+	for i := api.startKey; i < api.startKey+count; i++ {
+		pool <- struct{}{}
+		wg.Add(1)
+		go func(key int) {
+			defer func() {
+				<-pool
+				wg.Done()
+			}()
+			if e := fn(strconv.Itoa(key), api.startKey); e != nil {
+				err = e
+			}
+			bar.Increment()
+		}(i)
+	}
+	wg.Wait()
+	bar.Done()
+	line := api.getResult(time.Since(start).Seconds())
+	if api.after != nil {
+		api.after(bm.blob)
+	}
+	if err != nil {
+		logger.Errorf("%s test failed: %s", api.name, err)
+		return []string{api.title, failed, failed}
+	}
+	line[0] = api.title
+	return line
+}
+
+func getMockData(seed []byte, idx int) []byte {
+	size := len(seed)
+	if size == 0 {
+		return nil
+	}
+	content := make([]byte, size)
+	if idx == 0 {
+		content = seed
+	} else {
+		i := idx % size
+		copy(content[:size-i], seed[i:size])
+		copy(content[size-i:size], seed[:i])
+	}
+	return content
+}
+
+func (bm *benchMarkObj) put(key string, startKey int) error {
+	idx, _ := strconv.Atoi(key)
+	return bm.blob.Put(key, bytes.NewReader(getMockData(bm.seed, idx-startKey)))
+}
+
+func (bm *benchMarkObj) smallPut(key string, startKey int) error {
+	idx, _ := strconv.Atoi(key)
+	return bm.blob.Put(key, bytes.NewReader(getMockData(bm.smallSeed, idx)))
+}
+
+func getAndCheckN(blob object.ObjectStorage, key string, seed []byte, getOrgIdx func(idx int) int) error {
+	idx, _ := strconv.Atoi(key)
+	r, err := blob.Get(key, 0, -1)
+	if err != nil {
+		return err
+	}
+	defer r.Close()
+	content, err := io.ReadAll(r)
+	if err != nil {
+		return err
+	}
+	orgIdx := getOrgIdx(idx)
+	checkN := 10
+	l := len(seed)
+	if l < checkN {
+		checkN = l
+	}
+	if len(content) != len(seed) || !bytes.Equal(content[:checkN], getMockData(seed, orgIdx)[:checkN]) {
+		return fmt.Errorf("the downloaded content is incorrect")
+	}
+	return nil
+}
+
+func (bm *benchMarkObj) get(key string, startKey int) error {
+	return getAndCheckN(bm.blob, key, bm.seed, func(idx int) int {
+		return idx - startKey
+	})
+}
+
+func (bm *benchMarkObj) smallGet(key string, startKey int) error {
+	return getAndCheckN(bm.blob, key, bm.smallSeed, func(idx int) int {
+		return idx
+	})
+}
+
+func (bm *benchMarkObj) delete(key string, startKey int) error {
+	return bm.blob.Delete(key)
+}
+
+func (bm *benchMarkObj) head(key string, startKey int) error {
+	_, err := bm.blob.Head(key)
+	return err
+}
+
+func (bm *benchMarkObj) list(key string, startKey int) error {
+	result, err := osync.ListAll(bm.blob, "", "")
+	for range result {
+	}
+	return err
+}
+
+func (bm *benchMarkObj) chown(key string, startKey int) error {
+	return bm.blob.(object.FileSystem).Chown(key, "nobody", "nogroup")
+}
+
+func (bm *benchMarkObj) chmod(key string, startKey int) error {
+	return bm.blob.(object.FileSystem).Chmod(key, 0755)
+}
+
+func (bm *benchMarkObj) chtimes(key string, startKey int) error {
+	return bm.blob.(object.FileSystem).Chtimes(key, time.Now())
+}
+
+func listAll(s object.ObjectStorage, prefix, marker string, limit int64) ([]object.Object, error) {
+	r, err := s.List(prefix, marker, limit)
+	if err == nil {
+		return r, nil
+	}
+	ch, err := s.ListAll(prefix, marker)
+	if err == nil {
+		objs := make([]object.Object, 0)
+		for obj := range ch {
+			if len(objs) < int(limit) {
+				objs = append(objs, obj)
+			}
+		}
+		return objs, nil
+	}
+	return nil, err
+}
+
+var syncTests = map[string]bool{
+	"put a big object":    true,
+	"put an empty object": true,
+	"multipart upload":    true,
+}
+
+func functionalTesting(blob object.ObjectStorage, result *[][]string, tty bool) {
+	runCase := func(title string, fn func(blob object.ObjectStorage) error) {
+		r := pass
+		if err := fn(blob); err == utils.ENOTSUP {
+			r = nspt
+		} else if err != nil {
+			r = err.Error()
+			if tty {
+				r = fmt.Sprintf("%s%dm%s%s", COLOR_SEQ, RED, r, RESET_SEQ)
+			}
+			logger.Debug(err.Error())
+		}
+
+		category := "basic"
+		if syncTests[title] || strings.HasPrefix(title, "change") {
+			category = "sync"
+		}
+
+		if tty {
+			title = fmt.Sprintf("%s%sm%s%s", COLOR_SEQ, DEFAULT, title, RESET_SEQ)
+		}
+
+		*result = append(*result, []string{category, title, r})
+	}
+	isFileSystem := true
+	fi, ok := blob.(object.FileSystem)
+	if ok {
+		if err := fi.Chmod("not_exists_file", 0755); err == utils.ENOTSUP {
+			isFileSystem = false
+		}
+	}
+
+	get := func(s object.ObjectStorage, k string, off, limit int64) (string, error) {
+		r, err := s.Get(k, off, limit)
+		if err != nil {
+			return "", err
+		}
+		data, err := ioutil.ReadAll(r)
+		if err != nil {
+			return "", err
+		}
+		return string(data), nil
+	}
+	key := "put_test_file"
+
+	funFSCase := func(name string, fn func() error) {
+		runCase(name, func(blob object.ObjectStorage) error {
+			if !isFileSystem {
+				return utils.ENOTSUP
+			}
+			br := []byte("hello")
+			if err := blob.Put(key, bytes.NewReader(br)); err != nil {
+				return fmt.Errorf("put object failed: %s", err)
+			}
+			defer blob.Delete(key) //nolint:errcheck
+			return fn()
+		})
+	}
+
+	runCase("create a bucket", func(blob object.ObjectStorage) error {
+		created := true
+		if err := blob.Put(key, bytes.NewReader(nil)); err != nil {
+			created = false
+		}
+		defer blob.Delete(key) //nolint:errcheck
+
+		if !created {
+			if err := blob.Create(); err != nil {
+				return fmt.Errorf("can't create bucket: %s", err)
+			}
+		}
+		if err := blob.Create(); err != nil {
+			return fmt.Errorf("creating a bucket that already exists returns an error")
+		}
+		return nil
+	})
+
+	runCase("put an object", func(blob object.ObjectStorage) error {
+		br := []byte("hello")
+		if err := blob.Put(key, bytes.NewReader(br)); err != nil {
+			return fmt.Errorf("put object failed: %s", err)
+		}
+		defer blob.Delete(key) //nolint:errcheck
+		return nil
+	})
+
+	runCase("get an object", func(blob object.ObjectStorage) error {
+		br := []byte("hello")
+		if err := blob.Put(key, bytes.NewReader(br)); err != nil {
+			return fmt.Errorf("put object failed: %s", err)
+		}
+		defer blob.Delete(key) //nolint:errcheck
+		if s, err := get(blob, key, 0, -1); err != nil && s != string(br) {
+			return fmt.Errorf("get object failed: %s", err)
+		}
+		return nil
+	})
+
+	runCase("get non-exist", func(blob object.ObjectStorage) error {
+		if _, err := blob.Get("not_exists_file", 0, -1); err == nil {
+			return fmt.Errorf("get not exists object should failed: %s", err)
+		}
+		return nil
+	})
+
+	runCase("get partial object", func(blob object.ObjectStorage) error {
+		br := []byte("hello")
+		if err := blob.Put(key, bytes.NewReader(br)); err != nil {
+			return fmt.Errorf("put object failed: %s", err)
+		}
+		defer blob.Delete(key) //nolint:errcheck
+		if d, e := get(blob, key, 0, -1); d != "hello" {
+			return fmt.Errorf("expect hello, but got %v, error: %s", d, e)
+		}
+		if d, e := get(blob, key, 2, -1); d != "llo" {
+			return fmt.Errorf("expect llo, but got %v, error: %s", d, e)
+		}
+		if d, e := get(blob, key, 2, 3); d != "llo" {
+			return fmt.Errorf("expect llo, but got %v, error: %s", d, e)
+		}
+		if d, e := get(blob, key, 2, 2); d != "ll" {
+			return fmt.Errorf("expect ll, but got %v, error: %s", d, e)
+		}
+		if d, e := get(blob, key, 4, 2); d != "o" {
+			return fmt.Errorf("out-of-range get: 'o', but got %v, error: %s", len(d), e)
+		}
+		if d, e := get(blob, key, 6, 2); d != "" {
+			return fmt.Errorf("out-of-range get: '', but got %v, error: %s", len(d), e)
+		}
+		return nil
+	})
+
+	runCase("head an object", func(blob object.ObjectStorage) error {
+		br := []byte("hello")
+		if err := blob.Put(key, bytes.NewReader(br)); err != nil {
+			return fmt.Errorf("put object failed: %s", err)
+		}
+		defer blob.Delete(key) //nolint:errcheck
+		if h, err := blob.Head(key); err != nil {
+			return fmt.Errorf("failed to head object %s", err)
+		} else {
+			if h.Key() != key {
+				return fmt.Errorf("expected get key is test but get %s", h.Key())
+			}
+		}
+		return nil
+	})
+
+	runCase("delete an object", func(blob object.ObjectStorage) error {
+		br := []byte("hello")
+		if err := blob.Put(key, bytes.NewReader(br)); err != nil {
+			return fmt.Errorf("put object failed: %s", err)
+		}
+		if err := blob.Delete(key); err != nil {
+			return fmt.Errorf("delete failed: %s", err)
+		}
+		if _, err := blob.Head(key); err == nil {
+			return fmt.Errorf("expect err is not nil")
+		}
+
+		if err := blob.Delete(key); err != nil {
+			return fmt.Errorf("delete not exists %v", err)
+		}
+		return nil
+	})
+
+	runCase("delete non-exist", func(blob object.ObjectStorage) error {
+		if err := blob.Delete(key); err != nil {
+			return fmt.Errorf("deleting a non-existent object returns an error %v", err)
+		}
+		return nil
+	})
+
+	runCase("list objects", func(blob object.ObjectStorage) error {
+		br := []byte("hello")
+		if err := blob.Put(key, bytes.NewReader(br)); err != nil {
+			return fmt.Errorf("put object failed: %s", err)
+		}
+		defer blob.Delete(key) //nolint:errcheck
+		if isFileSystem {
+			objs, err := listAll(blob, "", "", 2)
+			if err == nil {
+				if len(objs) != 2 {
+					return fmt.Errorf("list should return 2 keys, but got %d", len(objs))
+				}
+				if objs[0].Key() != "" {
+					return fmt.Errorf("first key should be empty string, but got %s", objs[0].Key())
+				}
+				if objs[0].Size() != 0 {
+					return fmt.Errorf("first object size should be 0, but got %d", objs[0].Size())
+				}
+				if objs[1].Key() != key {
+					return fmt.Errorf("first key should be test, but got %s", objs[1].Key())
+				}
+				if objs[1].Size() != 5 {
+					return fmt.Errorf("size of first key shold be 5, but got %v", objs[1].Size())
+				}
+				now := time.Now()
+				if objs[1].Mtime().Before(now.Add(-30*time.Second)) || objs[1].Mtime().After(now.Add(time.Second*30)) {
+					return fmt.Errorf("mtime of key should be within 10 seconds, but got %s", objs[1].Mtime().Sub(now))
+				}
+			} else {
+				return fmt.Errorf("list failed: %s", err)
+			}
+
+			objs, err = listAll(blob, "", "test2", 1)
+			if err != nil {
+				return fmt.Errorf("list failed: %s", err)
+			} else if len(objs) != 0 {
+				return fmt.Errorf("list should not return anything, but got %d", len(objs))
+			}
+		} else {
+			objs, err2 := listAll(blob, "", "", 1)
+			if err2 == nil {
+				if len(objs) != 1 {
+					return fmt.Errorf("list should return 1 keys, but got %d", len(objs))
+				}
+				if objs[0].Key() != key {
+					return fmt.Errorf("first key should be test, but got %s", objs[0].Key())
+				}
+				if objs[0].Size() != 5 {
+					return fmt.Errorf("size of first key shold be 5, but got %v", objs[0].Size())
+				}
+				now := time.Now()
+				if objs[0].Mtime().Before(now.Add(-30*time.Second)) || objs[0].Mtime().After(now.Add(time.Second*30)) {
+					return fmt.Errorf("mtime of key should be within 10 seconds, but got %s", objs[0].Mtime().Sub(now))
+				}
+			} else {
+				return fmt.Errorf("list failed: %s", err2)
+			}
+
+			objs, err2 = listAll(blob, "", "test2", 1)
+			if err2 != nil {
+				return fmt.Errorf("list failed: %s", err2)
+			} else if len(objs) != 0 {
+				return fmt.Errorf("list should not return anything, but got %d", len(objs))
+			}
+		}
+		return nil
+	})
+
+	runCase("put a big object", func(blob object.ObjectStorage) error {
+		fsize := 256 << 20
+		buffL := 4 << 20
+		buff := make([]byte, buffL)
+		rand.Read(buff)
+		count := int(math.Floor(float64(fsize) / float64(buffL)))
+		content := make([]byte, fsize)
+		for i := 0; i < count; i++ {
+			copy(content[i*buffL:(i+1)*buffL], buff)
+		}
+		if err := blob.Put(key, bytes.NewReader(content)); err != nil {
+			return fmt.Errorf("put big object failed: %s", err)
+		}
+		defer blob.Delete(key) //nolint:errcheck
+		return nil
+	})
+
+	runCase("put an empty object", func(blob object.ObjectStorage) error {
+		// Copy empty objects
+		defer blob.Delete("empty_test_file") //nolint:errcheck
+		if err := blob.Put("empty_test_file", bytes.NewReader([]byte{})); err != nil {
+			return fmt.Errorf("put empty object failed: %s", err)
+		}
+
+		// Copy `/` suffixed object
+		defer blob.Delete("slash_test_file/") //nolint:errcheck
+		if err := blob.Put("slash_test_file/", bytes.NewReader([]byte{})); err != nil {
+			return fmt.Errorf("put `/` suffixed object failed: %s", err)
+		}
+		return nil
+	})
+
+	runCase("multipart upload", func(blob object.ObjectStorage) error {
+		key := "multi_test_file"
+		if err := blob.CompleteUpload(key, "notExistsUploadId", []*object.Part{}); err != utils.ENOTSUP {
+			defer blob.Delete(key) //nolint:errcheck
+			partSize := 5 << 20
+			total := 10
+			seed := make([]byte, partSize)
+			rand.Read(seed)
+			upload, err := blob.CreateMultipartUpload(key)
+			if err != nil {
+				return fmt.Errorf("create multipart upload failed: %s", err)
+			}
+			parts := make([]*object.Part, total)
+			content := make([][]byte, total)
+			for i := 0; i < total; i++ {
+				content[i] = getMockData(seed, i)
+			}
+			pool := make(chan struct{}, 4)
+			var wg sync.WaitGroup
+			for i := 1; i <= total; i++ {
+				pool <- struct{}{}
+				wg.Add(1)
+				num := i
+				go func() {
+					defer func() {
+						<-pool
+						wg.Done()
+					}()
+					parts[num-1], err = blob.UploadPart(key, upload.UploadID, num, content[num-1])
+					if err != nil {
+						logger.Fatalf("multipart upload error: %v", err)
+					}
+				}()
+			}
+			wg.Wait()
+			// overwrite part
+			if parts[total-1], err = blob.UploadPart(key, upload.UploadID, total, getMockData(seed, total)); err != nil {
+				logger.Fatalf("multipart upload error: %v", err)
+			}
+			if err = blob.CompleteUpload(key, upload.UploadID, parts); err != nil {
+				logger.Fatalf("failed to complete multipart upload: %v", err)
+			}
+			r, err := blob.Get(key, 0, -1)
+			if err != nil {
+				logger.Fatalf("failed to get multipart upload file: %v", err)
+			}
+			cnt, err := io.ReadAll(r)
+			if err != nil {
+				logger.Fatalf("failed to get multipart upload file: %v", err)
+			}
+			content[total-1] = getMockData(seed, total)
+			if !bytes.Equal(cnt, bytes.Join(content, nil)) {
+				logger.Fatal("the content of the multipart upload file is incorrect")
+			}
+			return nil
+		}
+		return utils.ENOTSUP
+	})
+
+	funFSCase("change owner/group", func() error {
+		if strings.HasPrefix(blob.String(), "file://") && os.Getuid() != 0 {
+			return errors.New("root required")
+		}
+		if err := fi.Chown(key, "nobody", "nogroup"); err != nil {
+			return fmt.Errorf("failed to chown object %s", err)
+		}
+		if objInfo, err := blob.Head(key); err != nil {
+			return fmt.Errorf("failed to head object %s", err)
+		} else if info, ok := objInfo.(object.File); ok {
+			if info.Owner() != "nobody" {
+				return fmt.Errorf("expect owner nobody but got %s", info.Owner())
+			}
+			if info.Group() != "nogroup" {
+				return fmt.Errorf("expect group nogroup but got %s", info.Group())
+			}
+		}
+		return nil
+	})
+
+	funFSCase("change permission", func() error {
+		if err := fi.Chmod(key, 0777); err != nil {
+			return err
+		}
+		if objInfo, err := blob.Head(key); err != nil {
+			return fmt.Errorf("failed to head object %s", err)
+		} else if info, ok := objInfo.(object.File); ok {
+			if info.Mode()&0xFFF != 0777 {
+				return fmt.Errorf("expect mode %o but got %o", 0777, info.Mode())
+			}
+		}
+		return nil
+	})
+
+	funFSCase("change mtime", func() error {
+		mtime := time.Now().Add(-10 * time.Minute)
+		if err := fi.Chtimes(key, mtime); err != nil {
+			return fmt.Errorf("failed to chtimes %s", err)
+		}
+		if objInfo, err := blob.Head(key); err != nil {
+			return fmt.Errorf("failed to head object %s", err)
+		} else {
+			if objInfo.Mtime().Before(mtime.Add(-2*time.Second)) || objInfo.Mtime().After(mtime.Add(2*time.Second)) {
+				return fmt.Errorf("mtime deviation is too large, the actual mtime is %s but got %s", mtime.Format(time.RFC3339), objInfo.Mtime().Format(time.RFC3339))
+			}
+		}
+		return nil
+	})
+}
diff --git a/cmd/rmr.go b/cmd/rmr.go
index ef1bbc0f..d8df9978 100644
--- a/cmd/rmr.go
+++ b/cmd/rmr.go
@@ -25,7 +25,6 @@ import (
 
 	"github.com/juicedata/juicefs/pkg/meta"
 	"github.com/juicedata/juicefs/pkg/utils"
-	"github.com/pkg/errors"
 	"github.com/urfave/cli/v2"
 )
 
@@ -44,16 +43,22 @@ $ juicefs rmr /mnt/jfs/foo`,
 	}
 }
 
-func openController(path string) *os.File {
-	f, err := os.OpenFile(filepath.Join(path, ".control"), os.O_RDWR, 0)
-	if err != nil && !os.IsNotExist(err) && !errors.Is(err, syscall.ENOTDIR) {
-		logger.Errorf("%s", err)
-		return nil
+func openController(mp string) *os.File {
+	st, err := os.Stat(mp)
+	if err != nil {
+		logger.Fatal(err)
+	}
+	if !st.IsDir() {
+		mp = filepath.Dir(mp)
 	}
-	if err != nil && path != "/" {
-		return openController(filepath.Dir(path))
+	for ; mp != "/"; mp = filepath.Dir(mp) {
+		f, err := os.OpenFile(filepath.Join(mp, ".control"), os.O_RDWR, 0)
+		if err == nil {
+			return f
+		}
 	}
-	return f
+	logger.Fatalf("Path %s is not inside JuiceFS", mp)
+	panic("unreachable")
 }
 
 func rmr(ctx *cli.Context) error {
diff --git a/cmd/stats.go b/cmd/stats.go
index dcf933c5..b567fb00 100644
--- a/cmd/stats.go
+++ b/cmd/stats.go
@@ -76,6 +76,7 @@ const (
 	MAGENTA
 	CYAN
 	WHITE
+	DEFAULT = "00"
 )
 
 const (
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/cmd/warmup.go b/cmd/warmup.go
index 6abb7eb1..01f437aa 100644
--- a/cmd/warmup.go
+++ b/cmd/warmup.go
@@ -18,6 +18,7 @@ package cmd
 
 import (
 	"bufio"
+	"fmt"
 	"io"
 	"os"
 	"path/filepath"
@@ -86,8 +87,8 @@ func readControl(cf *os.File, resp []byte) int {
 }
 
 // send fill-cache command to controller file
-func sendCommand(cf *os.File, batch []string, count int, threads uint, background bool) {
-	paths := strings.Join(batch[:count], "\n")
+func sendCommand(cf *os.File, batch []string, threads uint, background bool) {
+	paths := strings.Join(batch, "\n")
 	var back uint8
 	if background {
 		back = 1
@@ -103,7 +104,7 @@ func sendCommand(cf *os.File, batch []string, count int, threads uint, backgroun
 		logger.Fatalf("Write message: %s", err)
 	}
 	if background {
-		logger.Infof("Warm-up cache for %d paths in background", count)
+		logger.Infof("Warm-up cache for %d paths in background", len(batch))
 		return
 	}
 	var errs = make([]byte, 1)
@@ -150,59 +151,52 @@ func warmup(ctx *cli.Context) error {
 
 	// find mount point
 	first := paths[0]
-	st, err := os.Stat(first)
-	if err != nil {
-		logger.Fatalf("Failed to stat path %s: %s", first, err)
-	}
-	var mp string
-	if st.IsDir() {
-		mp = first
-	} else {
-		mp = filepath.Dir(first)
+	controller := openController(first)
+	if controller == nil {
+		logger.Fatalf("open control file for %s", first)
 	}
+	defer controller.Close()
+
+	mp := first
 	for ; mp != "/"; mp = filepath.Dir(mp) {
 		inode, err := utils.GetFileInode(mp)
 		if err != nil {
-			logger.Fatalf("Failed to lookup inode for %s: %s", mp, err)
+			logger.Fatalf("lookup inode for %s: %s", mp, err)
 		}
 		if inode == 1 {
 			break
 		}
 	}
-	if mp == "/" {
-		logger.Fatalf("Path %s is not inside JuiceFS", first)
-	}
-
-	controller := openController(mp)
-	if controller == nil {
-		logger.Fatalf("Failed to open control file under %s", mp)
-	}
-	defer controller.Close()
 
 	threads := ctx.Uint("threads")
 	background := ctx.Bool("background")
 	start := len(mp)
-	batch := make([]string, batchMax)
+	batch := make([]string, 0, batchMax)
 	progress := utils.NewProgress(background, false)
 	bar := progress.AddCountBar("Warmed up paths", int64(len(paths)))
-	var index int
 	for _, path := range paths {
-		if strings.HasPrefix(path, mp) {
-			batch[index] = path[start:]
-			index++
+		if mp == "/" {
+			inode, err := utils.GetFileInode(path)
+			if err != nil {
+				logger.Errorf("lookup inode for %s: %s", mp, err)
+				continue
+			}
+			batch = append(batch, fmt.Sprintf("inode:%d", inode))
+		} else if strings.HasPrefix(path, mp) {
+			batch = append(batch, path[start:])
 		} else {
-			logger.Warnf("Path %s is not under mount point %s", path, mp)
+			logger.Errorf("Path %s is not under mount point %s", path, mp)
 			continue
 		}
-		if index >= batchMax {
-			sendCommand(controller, batch, index, threads, background)
-			bar.IncrBy(index)
-			index = 0
+		if len(batch) >= batchMax {
+			sendCommand(controller, batch, threads, background)
+			bar.IncrBy(len(batch))
+			batch = batch[0:]
 		}
 	}
-	if index > 0 {
-		sendCommand(controller, batch, index, threads, background)
-		bar.IncrBy(index)
+	if len(batch) > 0 {
+		sendCommand(controller, batch, threads, background)
+		bar.IncrBy(len(batch))
 	}
 	progress.Done()
 	if !background {
diff --git a/docs/en/README.md b/docs/en/README.md
index 8183425e..a45c7369 100644
--- a/docs/en/README.md
+++ b/docs/en/README.md
@@ -1,6 +1,6 @@
 # JuiceFS User Manual
 
-[![license](https://img.shields.io/badge/license-Apache%20v2.0-blue)](https://github.com/juicedata/juicefs/blob/main/LICENSE) [![Go Report](https://img.shields.io/badge/go%20report-A+-brightgreen.svg?style=flat)](https://goreportcard.com/badge/github.com/juicedata/juicefs) [![Join Slack](https://badgen.net/badge/Slack/Join%20JuiceFS/0abd59?icon=slack)](https://join.slack.com/t/juicefs/shared_invite/zt-n9h5qdxh-0bJojPaql8cfFgwerDQJgA)
+[![license](https://img.shields.io/badge/license-Apache%20v2.0-blue)](https://github.com/juicedata/juicefs/blob/main/LICENSE) [![Go Report](https://img.shields.io/badge/go%20report-A+-brightgreen.svg?style=flat)](https://goreportcard.com/badge/github.com/juicedata/juicefs) [![Join Slack](https://badgen.net/badge/Slack/Join%20JuiceFS/0abd59?icon=slack)](https://join.slack.com/t/juicefs/shared_invite/zt-n9h5qdxh-YD7e0JxWdesSEa9vY_f_DA)
 
 ![JuiceFS LOGO](images/juicefs-logo.png)
 
diff --git a/docs/en/administration/metadata/postgresql_best_practices.md b/docs/en/administration/metadata/postgresql_best_practices.md
index ae5c551a..9b4a39a6 100644
--- a/docs/en/administration/metadata/postgresql_best_practices.md
+++ b/docs/en/administration/metadata/postgresql_best_practices.md
@@ -18,19 +18,18 @@ It is recommended to configure and always enable SSL encryption on the database
 
 ## Passing sensitive information via environment variables
 
-Although it is easy and convenient to set the database password directly in the metadata URL, the password may be leaked in logs or program output, and for data security, the database password should always be passed through an environment variable.
+Although it is easy and convenient to set the database password directly in the metadata URL, JuiceFS hides it from logging and process list, but it's still possible to leak in other ways. For data security, it's better to pass the database password through an environment variable.
 
-Environment variable names can be freely defined, e.g.
+`META_PASSWORD` is a predefined environment variable for it:
 
 ```shell
-export $PG_PASSWD=mypassword
+export META_PASSWORD=mypassword
+juicefs mount -d "postgres://user@192.168.1.6:5432/juicefs" /mnt/jfs
 ```
 
-Passing the database password in the metadata URL via environment variables.
+## Authentication methods
 
-```shell
-juicefs mount -d "postgres://user:$PG_PASSWD@192.168.1.6:5432/juicefs" /mnt/jfs
-```
+Authentication with PostgreSQL is currently possible with "md5" method.
 
 ## Backup periodically
 
diff --git a/docs/en/deployment/s3_gateway.md b/docs/en/deployment/s3_gateway.md
index 72433092..da451dde 100644
--- a/docs/en/deployment/s3_gateway.md
+++ b/docs/en/deployment/s3_gateway.md
@@ -5,9 +5,9 @@ slug: /s3_gateway
 ---
 # Deploy JuiceFS S3 Gateway
 
-JuiceFS has introduced S3 gateway since v0.11, a feature implemented through the [MinIO S3 Gateway](https://docs.min.io/docs/minio-gateway-for-s3.html). It provides an S3-compatible RESTful API for files on JuiceFS, enabling the management of files stored on JuiceFS with tools such as s3cmd, AWS CLI, MinIO Client (mc), etc. in cases where mounting is not convenient. In addition, S3 gateway also provides a web-based file manager that allows users to use a browser to manage the files on JuiceFS.
+JuiceFS has introduced S3 gateway since v0.11. The feature is implemented based on the [MinIO S3 Gateway](https://docs.min.io/docs/minio-gateway-for-s3.html). It provides an S3-compatible RESTful API for files on JuiceFS, enabling the management of files stored on JuiceFS with tools such as s3cmd, AWS CLI, and MinIO Client (mc) in cases where mounting is not convenient. In addition, S3 gateway also provides a web-based file manager that allows users to manage files in web browsers.
 
-Because JuiceFS stores files in chunks, the files cannot be accessed directly using the interface of the underlying object storage. The S3 gateway provides similar access to the underlying object storage, with the following architecture diagram.
+Since JuiceFS stores files in chunks, the files cannot be accessed directly through the interfaces of the underlying object storage. The S3 gateway accesses the underlying object storage in a similar way, shown in the following architecture diagram.
 
 ![](../images/juicefs-s3-gateway-arch.png)
 
@@ -15,11 +15,11 @@ Because JuiceFS stores files in chunks, the files cannot be accessed directly us
 
 The S3 gateway is a feature built on top of the JuiceFS file system. If you do not have a JuiceFS file system, please refer to the [quick start guide](../getting-started/for_local.md) to create one first.
 
-JuiceFS S3 gateway is a feature introduced in v0.11, please make sure you have the latest version of JuiceFS.
+JuiceFS S3 gateway is a feature introduced since v0.11. Please make sure you have the latest version of JuiceFS.
 
 ## Quickstart
 
-The S3 gateway can be enabled on the current host using the `gateway` subcommand of JuiceFS. Before enabling the feature, you need to set `MINIO_ROOT_USER` and `MINIO_ROOT_PASSWORD` environment variables, which are the Access Key and Secret Key used to authenticate when accessing the S3 API, and can be simply considered as the username and password of the S3 gateway. For example.
+The S3 gateway can be enabled on the current host using the `gateway` subcommand of JuiceFS. Before enabling the feature, you need to set the environment variables `MINIO_ROOT_USER` and `MINIO_ROOT_PASSWORD`. These are the Access Key and Secret Key for authenticating when accessing the S3 API, and can be simply considered as the username and password of the S3 gateway. For example.
 
 ```shell
 $ export MINIO_ROOT_USER=admin
@@ -27,19 +27,19 @@ $ export MINIO_ROOT_PASSWORD=12345678
 $ juicefs gateway redis://localhost:6379 localhost:9000
 ```
 
-Among the above three commands, the first two commands are used to set environment variables. Note that the length of `MINIO_ROOT_USER` is at least 3 characters, and the length of `MINIO_ROOT_PASSWORD` is at least 8 characters (Windows users should set the environment variable with the `set` command instead, e.g., `set MINIO_ROOT_USER=admin`).
+The first two commands of the above three are used to set environment variables. Note that the length of `MINIO_ROOT_USER` is at least 3 characters, and the length of `MINIO_ROOT_PASSWORD` is at least 8 characters. If you are a Windows user, replace `export` with `set` in the above commands to set the environment variable. i.e., `set MINIO_ROOT_USER=admin`.
 
-The last command is used to enable the S3 gateway. The `gateway` subcommand requires at least two parameters, the first is the URL of the database where the metadata is stored, and the second is the address and port on which the S3 gateway is listening. You can add [other options](../reference/command_reference.md#juicefs-gateway) to the `gateway` subcommand to optimize the S3 gateway as needed, for example, to set the default local cache to 20 GiB.
+The last command is used to enable the S3 gateway. The `gateway` subcommand requires at least two parameters. The first is the URL of the database where the metadata is stored, and the second is the address and port on which the S3 gateway is listening. You can add [other options](../reference/command_reference.md#juicefs-gateway) to the `gateway` subcommand to optimize the S3 gateway as needed, for example, to set the default local cache to 20 GiB.
 
 ```shell
 $ juicefs gateway --cache-size 20480 redis://localhost:6379 localhost:9000
 ```
 
-In this example, we assume that the JuiceFS file system is using a local Redis database. When the S3 gateway is enabled, the administrative interface of the S3 gateway can be accessed on the **current host** using the address `http://localhost:9000`.
+In this example, we assume that the JuiceFS file system is using a local Redis database. When the S3 gateway is enabled, the administrative interface of the S3 gateway can be accessed from the **current host** using the address `http://localhost:9000`.
 
 ![](../images/s3-gateway-file-manager.jpg)
 
-If you want to access the S3 gateway through other hosts on the LAN or the Internet, you need to adjust the listening address, e.g.
+If you want to access the S3 gateway from other hosts on the LAN or over the Internet, you need to change the listening address, e.g.
 
 ```shell
 $ juicefs gateway redis://localhost:6379 0.0.0.0:9000
@@ -47,11 +47,11 @@ $ juicefs gateway redis://localhost:6379 0.0.0.0:9000
 
 In this way, the S3 gateway will accept all network requests by default. S3 clients in different locations can access the S3 gateway using different addresses, e.g.
 
-- Third-party clients in the host where the S3 gateway is located can use `http://127.0.0.1:9000` or `http://localhost:9000` for access.
+- A third-party client in the host where the S3 gateway is located can use `http://127.0.0.1:9000` or `http://localhost:9000` for access.
 - A third-party client on the same LAN as the host where the S3 gateway is located can access it using `http://192.168.1.8:9000` (assuming the intranet IP address of the S3 gateway-enabled host is 192.168.1.8).
-- The S3 gateway can be accessed via the Internet using `http://110.220.110.220:9000` (assuming that the public IP address of the S3 gateway-enabled host is 110.220.110.220).
+- The S3 gateway can be accessed over the Internet using `http://110.220.110.220:9000` (assuming that the public IP address of the S3 gateway-enabled host is 110.220.110.220).
 
-## Accessing S3 gateway
+## Access S3 gateway
 
 The JuiceFS S3 gateway can be accessed by various clients, desktop applications, web applications, etc. that support the S3 API. Please note the address and port that the S3 gateway listens on when using it.
 
@@ -59,7 +59,7 @@ The JuiceFS S3 gateway can be accessed by various clients, desktop applications,
 The following examples are for using a third-party client to access the S3 gateway running on the local host. In specific scenarios, please adjust the address to access the S3 gateway according to the actual situation.
 :::
 
-### Using the AWS CLI
+### Use the AWS CLI
 
 Download and install the AWS CLI from [https://aws.amazon.com/cli](https://aws.amazon.com/cli), then configure:
 
@@ -71,7 +71,7 @@ Default region name [None]:
 Default output format [None]:
 ```
 
-The program will guide you through adding the new configuration interactively, where `Access Key ID` is the same as `MINIO_ROOT_USER` and `Secret Access Key` is the same as `MINIO_ROOT_PASSWORD`, leave the region name and output format blank.
+The program will guide you interactively to add the new configuration, where `Access Key ID` and `Secret Access Key` are the same as `MINIO_ROOT_USER` and `MINIO_ROOT_PASSWORD`, respectively. The region name and output format should be left blank.
 
 After that, you can access the JuiceFS storage using the `aws s3` command, for example:
 
@@ -83,7 +83,7 @@ $ aws --endpoint-url http://localhost:9000 s3 ls
 $ aws --endpoint-url http://localhost:9000 s3 ls s3://<bucket>
 ```
 
-### Using the MinIO client
+### Use the MinIO client
 
 First install `mc` by referring to the [MinIO download page](https://min.io/download), then add a new alias:
 
@@ -91,9 +91,9 @@ First install `mc` by referring to the [MinIO download page](https://min.io/down
 $ mc alias set juicefs http://localhost:9000 admin 12345678 --api S3v4
 ```
 
-Following the mc command format, the above command creates a configuration with the alias `juicefs`. In particular, note that the API version must be specified in the command, `-api "s3v4"`.
+Following the mc command format, the above command creates a configuration with the alias `juicefs`. Note that the API version `-api S3v4` must be specified in the command.
 
-Then, you can freely manage the copying, moving, adding and deleting of files and folders between your local disk and JuiceFS storage as well as other cloud storage via the mc client.
+Then, you can freely manage files and folders by copying, moving, adding and deleting between your local disk and JuiceFS storage as well as other cloud storage via the mc client.
 
 ```shell
 $ mc ls juicefs/jfs
@@ -128,21 +128,23 @@ kubectl -n ${NAMESPACE} create secret generic juicefs-secret \
     --from-literal=secret-key=<SECRET_KEY>
 ```
 
-- `name`: The JuiceFS file system name.
-- `metaurl`: Connection URL for metadata engine (e.g. Redis). Read [this document](../reference/how_to_setup_metadata_engine.md) for more information.
-- `storage`: Object storage type, such as `s3`, `gs`, `oss`. Read [this document](../reference/how_to_setup_object_storage.md) for the full supported list.
+Here we have:
+
+- `name`: name of the JuiceFS file system.
+- `metaurl`: URL of the metadata engine (e.g. Redis). Read [this document](../reference/how_to_setup_metadata_engine.md) for more information.
+- `storage`: Object storage type, such as `s3`, `gs`, `oss`. Read [this document](../reference/how_to_setup_object_storage.md) to find all supported object storages.
 - `bucket`: Bucket URL. Read [this document](../reference/how_to_setup_object_storage.md) to learn how to setup different object storage.
 - `access-key`: Access key of object storage. Read [this document](../reference/how_to_setup_object_storage.md) for more information.
 - `secret-key`: Secret key of object storage. Read [this document](../reference/how_to_setup_object_storage.md) for more information.
 
-Then download the S3 gateway [deployment YAML](https://github.com/juicedata/juicefs/blob/main/deploy/juicefs-s3-gateway.yaml) and create the `Deployment` and `Service` resources via `kubectl`. The following points require special attention:
+Then download the S3 gateway [deployment YAML](https://github.com/juicedata/juicefs/blob/main/deploy/juicefs-s3-gateway.yaml) and create the `Deployment` and `Service` resources with `kubectl`. The following points require special attention:
 
 - Please replace `${NAMESPACE}` in the following command with the Kubernetes namespace of the actual S3 gateway deployment, which defaults to `kube-system`.
-- The `replicas` for `Deployment` defaults to 1, please adjust as appropriate.
-- The latest version of `juicedata/juicefs-csi-driver` image is used by default, which already integrates the latest version of JuiceFS client, please check [here](https://github.com/juicedata/juicefs-csi-driver/releases) for the specific integrated JuiceFS client version.
+- The `replicas` for `Deployment` defaults to 1. Please adjust as needed.
+- The latest version of `juicedata/juicefs-csi-driver` image is used by default, which has already integrated the latest version of JuiceFS client. Please check [here](https://github.com/juicedata/juicefs-csi-driver/releases) for the specific integrated JuiceFS client version.
 - The `initContainers` of `Deployment` will first try to format the JuiceFS file system, if you have already formatted it in advance, this step will not affect the existing JuiceFS file system.
-- The default port number that the S3 gateway listens on is 9000
-- The [startup options](../reference/command_reference.md#juicefs-gateway) of S3 gateway are the default values, please adjust them according to your actual needs.
+  - The default port number that the S3 gateway listens on is 9000
+- The [startup options](../reference/command_reference.md#juicefs-gateway) of S3 gateway will use default values if not specified.
 - The value of `MINIO_ROOT_USER` environment variable is `access-key` in Secret, and the value of `MINIO_ROOT_PASSWORD` environment variable is `secret-key` in Secret.
 
 ```shell
@@ -197,7 +199,7 @@ There are some differences between the various versions of Ingress. For more usa
 
 1. Prepare a YAML file
 
-   Create a configuration file, for example: `values.yaml`, copy and complete the following configuration information. Among them, the `secret` part is the information related to the JuiceFS file system, you can refer to [JuiceFS Quick Start Guide](../getting-started/for_local.md) for more information.
+   Create a configuration file, for example: `values.yaml`. Copy and fill in the following configuration information. Among them, the `secret` part is the information related to the JuiceFS file system, and you can refer to [JuiceFS Quick Start Guide](../getting-started/for_local.md) for more information.
 
    ```yaml
    secret:
@@ -209,7 +211,7 @@ There are some differences between the various versions of Ingress. For more usa
      bucket: "<bucket>"
    ```
 
-   If you want to deploy Ingress, add these in `values.yaml`:
+   If you want to deploy Ingress, add the following snippet into `values.yaml`:
 
    ```yaml
    ingress:
@@ -218,7 +220,7 @@ There are some differences between the various versions of Ingress. For more usa
 
 2. Deploy
 
-   Execute the following three commands in sequence to deploy the JuiceFS S3 gateway via Helm (note that the following example is deployed to the `kube-system` namespace).
+   Execute the following three commands in sequence to deploy the JuiceFS S3 gateway with Helm (note that the following example is deployed to the `kube-system` namespace).
 
    ```sh
    helm repo add juicefs-s3-gateway https://juicedata.github.io/charts/
@@ -250,14 +252,14 @@ Please see the ["Monitoring"](../administration/monitoring.md) documentation to
 
 ## Use a full-featured S3 gateway
 
-If you need to use some advanced features of the MinIO S3 gateway, you can pull [the gateway branch of this repository](https://github.com/juicedata/minio/tree/gateway) and compile MinIO yourself. This branch is developed based on the [MinIO RELEASE.2022-03-05T06-32-39Z](https://github.com/minio/minio/tree/RELEASE.2022-03-05T06-32-39Z) version and adds JuiceFS gateway support, which supports full functionality of MinIO gateways such as [multi-user management](https://docs.min.io/docs/minio-multi-user-quickstart-guide.html) while using JuiceFS as a backend.
+If you need to use some advanced features of the MinIO S3 gateway, you can pull [the gateway branch of this repository](https://github.com/juicedata/minio/tree/gateway) and compile MinIO yourself. This branch is developed based on [MinIO RELEASE.2022-03-05T06-32-39Z](https://github.com/minio/minio/tree/RELEASE.2022-03-05T06-32-39Z) with JuiceFS gateway supported, which supports full functionality of MinIO gateways such as [multi-user management](https://docs.min.io/docs/minio-multi-user-quickstart-guide.html) while using JuiceFS as a backend.
 
 ### Compile
 
 :::tip
 This branch relies on a newer version of JuiceFS. Please refer to the [`go.mod`](https://github.com/juicedata/minio/blob/gateway/go.mod) file for the specific JuiceFS version.
 
-Similar to [manually compile JuiceFS client](../getting-started/installation.md#manually-compiling), you need to install some dependencies in advance to compile S3 gateway normally.
+Similar to [manually compiling JuiceFS client](../getting-started/installation.md#manually-compiling), you need to install some dependencies in advance to compile S3 gateway.
 :::
 
 ```shell
@@ -279,6 +281,6 @@ $ export MINIO_ROOT_PASSWORD=12345678
 $ ./minio gateway juicefs --console-address ':59001' redis://localhost:6379
 ```
 
-The port number of the S3 gateway console is explicitly specified here as 59001. If not specified, a port will be randomly selected. According to the command line prompt, open the [http://127.0.0.1:59001](http://127.0.0.1:59001) address in the browser to access the console, as shown in the following figure:
+The port number of the S3 gateway console is explicitly specified here as 59001. If not specified, a port will be randomly assigned. According to the command line prompt, open the address [http://127.0.0.1:59001](http://127.0.0.1:59001) in the browser to access the console, as shown in the following snapshot:
 
 ![](../images/s3-gateway-console.png)
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/en/introduction/_case.md b/docs/en/introduction/_case.md
index 56e8f502..5dda7a62 100644
--- a/docs/en/introduction/_case.md
+++ b/docs/en/introduction/_case.md
@@ -35,5 +35,5 @@ You can create a branch of this repository through GitHub, add the titile and UR
 
 ### 2. Social media
 
-You can join JuiceFS official [Slack channel](https://juicefs.slack.com/), any staff member can contact contribution matters.
+You can join JuiceFS official [Slack channel](https://join.slack.com/t/juicefs/shared_invite/zt-n9h5qdxh-YD7e0JxWdesSEa9vY_f_DA), any staff member can contact contribution matters.
 
diff --git a/docs/en/release_notes.md b/docs/en/release_notes.md
index 33656605..1ee9f644 100644
--- a/docs/en/release_notes.md
+++ b/docs/en/release_notes.md
@@ -6,14 +6,14 @@ For all versions, please see [GitHub Releases](https://github.com/juicedata/juic
 
 ## Upgrade to JuiceFS v1.0.0 Beta3
 
-JuiceFS client has only one binary file, so when upgrading, you only need to replace the old version with the new one. 
+JuiceFS client has only one binary file, so you only need to replace the old version with the new one when upgrading JuiceFS. 
 
 ### SQL: Update table schema to support encoding other than UTF-8
 
-JuiceFS v1.0.0 Beta3 changed the table schema to support encoding other than UTF-8. For existing file systems, we need to upgrade the table schema to support that. It's recommanded to upgrade all clients first, then update the table schema.
+JuiceFS v1.0.0 Beta3 has changed the table schema to support encoding other than UTF-8. For existing file systems, you need to upgrade the table schema manually to support that. It's recommended to upgrade all clients first and then the table schema.
 
 :::note
-Table schema upgrades are not mandatory and are only required if you need to use non-UTF-8 characters. In addition, metadatabase performance may degrade when upgrading SQL table schemas, affecting running services.
+Table schema upgrades are optional, and they are required only if you need to use non-UTF-8 characters. In addition, database performance may degrade when upgrading SQL table schemas, affecting running services.
 :::
 
 #### MySQL/MariaDB
@@ -36,8 +36,8 @@ alter table jfs_symlink
 
 #### SQLite
 
-Since SQLite does not support modifying columns, you can migrate them by `dump` and `load` commands, refer to: [JuiceFS Metadata Backup and Recovery](administration/metadata_dump_load.md) for details.
+SQLite does not support modifying columns, but you can migrate columns by `dump` and `load` commands, refer to [JuiceFS Metadata Backup and Recovery](administration/metadata_dump_load.md) for details.
 
 ### New session management format
 
-JuiceFS v1.0.0 Beta3 uses a new session management format, previous versions of clients will not be able to see the sessions generated by v1.0.0 Beta3 clients via `juicefs status` or `juicefs destroy`, new versions of clients can see all sessions.
+JuiceFS v1.0.0 Beta3 uses a new session management format. The previous versions of clients cannot see the sessions generated by v1.0.0 Beta3 clients via `juicefs status` or `juicefs destroy`, whereas the new versions are able to see all the sessions.
diff --git a/docs/en/tutorials/juicefs_on_colab.md b/docs/en/tutorials/juicefs_on_colab.md
new file mode 100644
index 00000000..a3835133
--- /dev/null
+++ b/docs/en/tutorials/juicefs_on_colab.md
@@ -0,0 +1,153 @@
+---
+sidebar_label: Use JuiceFS on Colab with Google CloudSQL and GCS
+sidebar_position: 4
+slug: /juicefs_on_colab
+---
+
+# Use JuiceFS on Colab with Google CloudSQL and GCS
+
+[Colaboratory](https://colab.research.google.com), or “Colab” for short, is a product from Google Research. Colab allows
+anybody to write and execute arbitrary python code through the browser, and is especially well suited to machine
+learning, data analysis and education.
+
+Colab supports Google Drive for upload files to or download files from Colab instances. However, in some cases, Google
+Drive might not be that convenient to use with Colab. This is where JuiceFS could be a helpful tool to allow easily sync
+files between Colab instances, or between Colab instance and a local or on-premises machine.
+
+[A demo Colab notebook using JuiceFS can be seen here](https://colab.research.google.com/drive/1wA8vRwqiihXkI6ViDU8Ud868UeYtmCo5)
+.
+
+We illustrate the needed steps for using JuiceFS in Colab environment. We use Google CloudSQL as the JuiceFS metadata
+engine, and Google Cloud Storage (GCS) as the JuiceFS object storage.
+
+For other types of metadata engines or object storages, please refer
+to [the metadata engine reference doc](../reference/how_to_setup_metadata_engine.md)
+and [the object storage reference doc](../reference/how_to_setup_object_storage.md).
+
+Many of the steps mentioned here will be quite similar with
+the [getting started doc](../getting-started/for_distributed.md)
+which you can also use for reference.
+
+## Summary of steps:
+
+1. Format a `juicefs` file system from any machine/instance which having access to Google cloud resources
+2. Mount `juicefs` file system in a Colab Notebook
+3. Happily storing sharing files cross machines and platforms
+
+## Prerequisite
+
+In this demo we will use Google Cloud Platform's CloudSQL and Google Cloud Storage (GCS) as the backbone for creating a
+massive high performance file storage system via `juicefs`. It is required that you have a Google Cloud Platform account
+to follow this demo document.
+
+Otherwise, if you have another cloud vendors' resources (for example AWS's RDBS and S3), you can still use this guide as
+a reference and with other reference docs provided by `juicefs` to achieve a similar solution.
+
+You might also want the Colab instance is in the same zone or close to the region where CloudSQL and GCS are deployed,
+to make JuiceFS reaching the best performance. The tutorial should work for a random hosted Colab instance but you might
+notice a slow performance due to the latency between the Colab instance and the CloudSQL/GCS regions. To start Colab
+instaces in a specific region,
+see [instructions for starting a GCE VM on Colab via GCP Marketplace](https://research.google.com/colaboratory/marketplace.html)
+.
+
+So in order to follow this guide you will need to have these resources ready:
+
+* A Google Cloud Platform account ready and also a *porject* created. In our case we will use GCP project
+  as `juicefs-learning` for the demo
+* A CloudSQL (Postgres) ready to be used. In this demo uses instance
+  `juicefs-learning:europe-west1:juicefs-sql-example-1` as the metadata service
+* A GCS bucket created as the object storage service. In this demo we use
+  `gs://juicefs-bucket-example-1` as the bucket to store file chunks.
+* An IAM ServiceAccount or an authrized user account that has write access to the posgtes server and GCS buckets
+
+## Detailed Steps
+
+### Step 1 - Format and mount a juicefs file system folder
+
+This step is only need to be done once, and you can choose to do this at any machine/instance where you have good
+connection and access to your Google Cloud resources.
+
+In this example I am doing this on my local machine. Firstly you can use
+`gcloud auth application-default login` to get a local credential ready or you can also
+use `GOOGLE_APPLICATION_CREDENTIALS` to setup as json key file.
+
+Then you can use [`cloud_sql_proxy`](https://cloud.google.com/sql/docs/mysql/connect-admin-proxy) to open a port (in
+this case 5432) locally so to expose your cloud Postgres service to your local machine:
+
+```
+gcloud auth application-default login
+
+# Or set up the json key file via GOOGLE_APPLICATION_CREDENTIALS=/path/to/key
+
+cloud_sql_proxy -instances=juicefs-learning:europe-west1:juicefs-sql-example-1=tcp:0.0.0.0:5432
+```
+
+Then use the following command to create a new file system with a name as `myvolume` by using `juicefs format` command.
+Then later we can mount this filesystem in any other machines/instances where you have access to your cloud resources.
+
+You can download `juicefs` [here](https://github.com/juicedata/juicefs/releases).
+
+```
+juicefs format \
+    --storage gs \
+    --bucket gs://juicefs-bucket-example-1 \
+    "postgres://postgres:mushroom1@localhost:5432/juicefs?sslmode=disable" \
+    myvolume
+```
+
+Noticing again, this step above is only need to be done once at any machine you feel like to work on.
+
+### Step 2 - Mount juicefs file system on colab
+
+After you have finished with Step 1 above, then it means you already have a JuiceFS filesystem (called `myvolume` in
+this case) is defined and ready to be used.
+
+So here we open a Colab page and run those command to mount our filesystem into a folder called `mnt`.
+
+Firstly we download the `juicefs` binary and do the same as Step 1 to get GCP credentials and open cloudsql proxy.
+
+Note that the follow commands are run in Colab environment so there is a
+`!` mark in the beginning for running shell command.
+
+1. Download `juicefs` to colab runtime instance
+
+```
+! curl -L -o juicefs.tar.gz https://github.com/juicedata/juicefs/releases/download/v1.0.0-beta2/juicefs-1.0.0-beta2-linux-amd64.tar.gz 
+! tar -xf juicefs.tar.gz
+```
+
+1. Set up Google Cloud credentials
+
+```
+! gcloud auth application-default login
+```
+
+1. Open cloud_sql_proxy
+
+```
+! wget https://dl.google.com/cloudsql/cloud_sql_proxy.linux.amd64 -O cloud_sql_proxy
+! chmod +x cloud_sql_proxy
+! GOOGLE_APPLICATION_CREDENTIALS=/content/.config/application_default_credentials.json nohup ./cloud_sql_proxy -instances=juicefs-learning:europe-west1:juicefs-sql-example-1=tcp:0.0.0.0:5432 >> cloud_sql_proxy.log &
+```
+
+1. Mount JuiceFS filesystem `myvolumn` onto folder `mnt`
+
+```
+! GOOGLE_APPLICATION_CREDENTIALS=/content/.config/application_default_credentials.json nohup ./juicefs mount  "postgres://postgres:mushroom1@localhost:5432/juicefs?sslmode=disable" mnt > juicefs.log &
+```
+
+Now you should be able to use the folder `mnt` as if it is a local filesystem folder to write and read folders and files
+in it.
+
+### Step 3 - Loading data at another time or another instance
+
+Now as you have data stored in Step 2 in your JuiceFS file system, you can repeat all the operations mentioned in Step 2
+anytime in any other machines so to have access again to the data previously stored or to store more data into it.
+
+Congratulations! Now you have learned how to use JuiceFS and specifically how to use it with Google Colab to
+conveniently sharing and storing data files in a distributed fashion.
+
+[A demo Colab notebook using JuiceFS can be seen here](https://colab.research.google.com/drive/1wA8vRwqiihXkI6ViDU8Ud868UeYtmCo5)
+.
+
+Happy coding :)
diff --git a/docs/en/tutorials/juicefs_on_k3s.md b/docs/en/tutorials/juicefs_on_k3s.md
index 6eaffc7c..11a486a8 100644
--- a/docs/en/tutorials/juicefs_on_k3s.md
+++ b/docs/en/tutorials/juicefs_on_k3s.md
@@ -5,9 +5,9 @@ slug: /juicefs_on_k3s
 ---
 # Use JuiceFS on K3s
 
-[K3s](https://k3s.io/) is a functionally optimized lightweight Kubernetes distribution that is fully compatible with Kubernetes, that is, almost all operations on Kubernetes can be performed on K3s. K3s has packaged the entire container orchestration system into a binary program with a capacity of less than 100MB, which greatly reduces the environment dependency and installation of deploying Kubernetes production clusters. In contrast, K3s has lower performance requirements for the operating system, and ARM devices such as Raspberry Pi can be used to form a cluster.
+[K3s](https://k3s.io/) is a functionally optimized lightweight Kubernetes distribution that is fully compatible with Kubernetes, that is, almost all operations on Kubernetes can be performed on K3s. K3s has packaged the entire container orchestration system into a binary program with a capacity of less than 100MB, which greatly reduces the environment dependencies and steps for installation of deploying Kubernetes production clusters. Compared to Kubernetes, K3s has lower performance requirements for the operating system, and ARM devices such as Raspberry Pi can be used to form a cluster.
 
-In this article, we will build a K3s cluster with two nodes, install and configure [JuiceFS CSI Driver](https://github.com/juicedata/juicefs-csi-driver) for the cluster, and finally create a Nginx Pod for verification.
+In this article, we will build a K3s cluster with two nodes, install and configure [JuiceFS CSI Driver](https://github.com/juicedata/juicefs-csi-driver) for the cluster, and lastly create an Nginx Pod for verification.
 
 ## Deploy a K3s cluster
 
@@ -16,21 +16,21 @@ K3s has very low **minimum requirements** for hardware:
 - **Memory**: 512MB+ (recommend 1GB+)
 - **CPU**: 1 core
 
-When deploying a production cluster, you can usually use the Raspberry Pi 4B (4 CPU cores, 8G memory) as the starting point for the hardware of a node. For details, see [Hardware Requirements](https://rancher.com/docs/k3s/latest/en/installation/installation-requirements/#hardware).
+When deploying a production cluster, usually you can start with the Raspberry Pi 4B (4 CPU cores, 8G memory) for the hardware of a node. For details, see [Hardware Requirements](https://rancher.com/docs/k3s/latest/en/installation/installation-requirements/#hardware).
 
 ### K3s server node
 
 The IP address of the server node is: `192.168.1.35`
 
-Use the script officially provided by K3s to deploy the server node on a regular Linux distribution.
+You can use the official script provided by K3s to deploy the server node on a regular Linux distribution.
 
 ```shell
 $ curl -sfL https://get.k3s.io | sh -
 ```
 
-After the deployment is successful, the K3s service will automatically start, and kubectl and other tools will also be installed.
+After the deployment is successful, the K3s service will automatically start, and kubectl and other tools will also be installed at the same time.
 
-Execute the command to view the status of the node:
+You can execute the following command to view the status of the node:
 
 ```shell
 $ sudo kubectl get nodes
@@ -49,7 +49,7 @@ K1041f7c4fabcdefghijklmnopqrste2ec338b7300674f::server:3d0ab12800000000000000006
 
 The IP address of the worker node is: `192.168.1.36`
 
-Execute the following command and change the value of `K3S_URL` to the IP or domain name of the server node, the default port is `6443`. Replace the value of `K3S_TOKEN` with the `node-token` obtained from the server node.
+Execute the following command and change the value of `K3S_URL` to the IP or domain name of the server node (the default port is `6443`). Replace the value of `K3S_TOKEN` with the `node-token` obtained from the server node.
 
 ```shell
 $ curl -sfL https://get.k3s.io | K3S_URL=http://192.168.1.35:6443 K3S_TOKEN=K1041f7c4fabcdefghijklmnopqrste2ec338b7300674f::server:3d0ab12800000000000000006328bbd80 sh -
@@ -66,9 +66,9 @@ k3s-n1   Ready    <none>                 28h   v1.21.4+k3s1
 
 ## Install CSI Driver
 
-It is consistent with the method of [Use JuiceFS on Kubernetes](../deployment/how_to_use_on_kubernetes.md), you can install it through Helm or kubectl.
+It is consistent with the method of [Use JuiceFS on Kubernetes](../deployment/how_to_use_on_kubernetes.md). Therefore, you can install CSI Driver through Helm or kubectl.
 
-Here we use kubectl, execute the following command to install the CSI Driver:
+Here we use kubectl as an example. Execute the following command to install the CSI Driver:
 
 ```shell
 $ kubectl apply -f https://raw.githubusercontent.com/juicedata/juicefs-csi-driver/master/deploy/k8s.yaml
@@ -124,11 +124,11 @@ local-path (default)   rancher.io/local-path   Delete          WaitForFirstConsu
 juicefs-sc             csi.juicefs.com         Retain          Immediate              false                  28h
 ```
 
-> **Note**: A storage class is associated with a JuiceFS file system. You can create as many storage classes as you need. But you need to pay attention to modifying the storage class name in the configuration file to avoid conflicts with the same name.
+> **Note**: A storage class is associated with a JuiceFS file system. You can create as many storage classes as you need, but be aware of the storage class name in the configuration file as the same name can cause conflicts.
 
 ## Use JuiceFS to persist Nginx data
 
-Next, deploy an Nginx Pod, using the persistent storage declared by the JuiceFS storage class.
+Next, deploy an Nginx Pod using a persistent storage declared by the JuiceFS storage class.
 
 ### Depolyment
 
@@ -238,11 +238,11 @@ $ sudo kubectl apply -f ingress.yaml
 
 ### Visit
 
-After the deployment is complete, use the host on the same LAN to access any cluster node, and you can see the Nginx welcome page.
+After the deployment is completed, use the host on the same LAN to access any cluster node, and then you will see the Nginx welcome page.
 
 ![](../images/k3s-nginx-welcome.png)
 
-Next, check whether the container has successfully mounted JuiceFS, and execute the command to check the pod status:
+Next, check whether the container has successfully mounted JuiceFS, and execute the following command to check the Pod status:
 
 ```shell
 $ sudo kubectl get pods
@@ -251,7 +251,7 @@ nginx-run-7d6fb7d6df-qhr2m   1/1     Running   0          28h
 nginx-run-7d6fb7d6df-5hpv7   1/1     Running   0          24h
 ```
 
-Execute the command to view the file system mount status of any pods:
+Executing the following command will show the file system mount status of any Pod:
 
 ```shell
 $ sudo kubectl exec nginx-run-7d6fb7d6df-qhr2m -- df -Th
diff --git a/docs/zh_cn/README.md b/docs/zh_cn/README.md
index c446044c..c2311da3 100644
--- a/docs/zh_cn/README.md
+++ b/docs/zh_cn/README.md
@@ -1,6 +1,6 @@
 # JuiceFS 用户手册
 
-[![license](https://img.shields.io/badge/license-Apache%20v2.0-blue)](https://github.com/juicedata/juicefs/blob/main/LICENSE) [![Go Report](https://img.shields.io/badge/go%20report-A+-brightgreen.svg?style=flat)](https://goreportcard.com/badge/github.com/juicedata/juicefs) [![Join Slack](https://badgen.net/badge/Slack/Join%20JuiceFS/0abd59?icon=slack)](https://join.slack.com/t/juicefs/shared_invite/zt-n9h5qdxh-0bJojPaql8cfFgwerDQJgA)
+[![license](https://img.shields.io/badge/license-Apache%20v2.0-blue)](https://github.com/juicedata/juicefs/blob/main/LICENSE) [![Go Report](https://img.shields.io/badge/go%20report-A+-brightgreen.svg?style=flat)](https://goreportcard.com/badge/github.com/juicedata/juicefs) [![Join Slack](https://badgen.net/badge/Slack/Join%20JuiceFS/0abd59?icon=slack)](https://join.slack.com/t/juicefs/shared_invite/zt-n9h5qdxh-YD7e0JxWdesSEa9vY_f_DA)
 
 ![JuiceFS LOGO](images/juicefs-logo.png)
 
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/docs/zh_cn/introduction/_case.md b/docs/zh_cn/introduction/_case.md
index 8468b5da..80227d2e 100644
--- a/docs/zh_cn/introduction/_case.md
+++ b/docs/zh_cn/introduction/_case.md
@@ -42,4 +42,4 @@ JuiceFS 广泛适用于各种数据存储和共享场景，本页汇总来自世
 
 ### 2. 社交媒体投稿
 
-你可以加入 JuiceFS 官方的 [Slack 频道](https://juicefs.slack.com/)，任何一位工作人员都可以接洽案例投稿事宜。
+你可以加入 JuiceFS 官方的 [Slack 频道](https://join.slack.com/t/juicefs/shared_invite/zt-n9h5qdxh-YD7e0JxWdesSEa9vY_f_DA)，任何一位工作人员都可以接洽案例投稿事宜。
diff --git a/docs/zh_cn/reference/how_to_setup_metadata_engine.md b/docs/zh_cn/reference/how_to_setup_metadata_engine.md
index ab3ef952..bea68ea2 100644
--- a/docs/zh_cn/reference/how_to_setup_metadata_engine.md
+++ b/docs/zh_cn/reference/how_to_setup_metadata_engine.md
@@ -352,6 +352,52 @@ $ juicefs format --storage s3 \
 sudo juicefs mount -d "tikv://192.168.1.6:2379,192.168.1.7:2379,192.168.1.8:2379/jfs" /mnt/jfs
 ```
 
+## etcd
+
+[etcd](https://etcd.io) 是一个高可用高可靠的小规模键值数据库，可以用作 JuiceFS 的元数据存储。
+
+### 创建文件系统
+
+使用 etcd 作为元数据引擎时，需要使用如下格式来指定 `Meta-URL` 参数：
+
+```shell
+etcd://[user:password@]<addr>[,<addr>...]/<prefix>
+```
+
+其中 `user` 和 `password` 是当 `etcd` 开启了用户认证时需要的。 `prefix` 是一个用户自定义的字符串，当多个文件系统或者应用共用一个 etcd 集群时，设置前缀可以避免混淆和冲突。示例如下：
+
+```bash
+$ juicefs format etcd://user:password@192.168.1.6:2379,192.168.1.7:2379,192.168.1.8:2379/jfs pics
+```
+
+### 设置 TLS
+
+如果需要开启 TLS，可以通过在 Meta-URL 后以添加 query 参数的形式设置 TLS 的配置项，目前支持的配置项：
+
+| 配置项             | 值                    |
+|-------------------|-----------------------|
+| cacert            | CA 根证书              |
+| cert              | 证书文件路径            |
+| key               | 私钥文件路径            |
+| server-name       | 服务器名称              |
+| insecure-skip-verify | 1                  |
+
+例子：
+```bash
+$ juicefs format --storage s3 \
+    ...
+    "etcd://192.168.1.6:2379,192.168.1.7:2379,192.168.1.8:2379/jfs?ca=/path/to/ca.pem&cacert=/path/to/etcd-server.pem&key=/path/to/etcd-key.pem&server-name=etcd" \
+    pics
+```
+
+### 挂载文件系统
+
+```shell
+sudo juicefs mount -d "etcd://192.168.1.6:2379,192.168.1.7:2379,192.168.1.8:2379/jfs" /mnt/jfs
+```
+
+注意：挂载到后台时，证书的路径需要使用绝对路径。
+
 ## FoundationDB
 
 即将推出......
diff --git a/docs/zh_cn/tutorials/juicefs_on_k3s.md b/docs/zh_cn/tutorials/juicefs_on_k3s.md
index 353b1f27..8df224ab 100644
--- a/docs/zh_cn/tutorials/juicefs_on_k3s.md
+++ b/docs/zh_cn/tutorials/juicefs_on_k3s.md
@@ -242,7 +242,7 @@ $ sudo kubectl apply -f ingress.yaml
 
 ![](../images/k3s-nginx-welcome.png)
 
-接下来查看一下容器是否成功挂载了 JuiceFS，执行命令查看 pod 状态：
+接下来查看一下容器是否成功挂载了 JuiceFS，执行命令查看 Pod 状态：
 
 ```shell
 $ sudo kubectl get pods
@@ -251,7 +251,7 @@ nginx-run-7d6fb7d6df-qhr2m   1/1     Running   0          28h
 nginx-run-7d6fb7d6df-5hpv7   1/1     Running   0          24h
 ```
 
-执行命令，查看任何一个 pods 的文件系统挂载情况：
+执行命令，查看任何一个 Pod 的文件系统挂载情况：
 
 ```shell
 $ sudo kubectl exec nginx-run-7d6fb7d6df-qhr2m -- df -Th
diff --git a/go.mod b/go.mod
index a8ec8085..3f565d7e 100644
--- a/go.mod
+++ b/go.mod
@@ -57,7 +57,7 @@ require (
 	github.com/sirupsen/logrus v1.8.1
 	github.com/smartystreets/goconvey v1.7.2
 	github.com/tencentyun/cos-go-sdk-v5 v0.7.34
-	github.com/tikv/client-go/v2 v2.0.0
+	github.com/tikv/client-go/v2 v2.0.1
 	github.com/upyun/go-sdk/v3 v3.0.2
 	github.com/urfave/cli/v2 v2.4.0
 	github.com/vbauerster/mpb/v7 v7.0.3
diff --git a/go.sum b/go.sum
index 7d00d915..8b803268 100644
--- a/go.sum
+++ b/go.sum
@@ -791,8 +791,8 @@ github.com/tidwall/pretty v1.2.0 h1:RWIZEg2iJ8/g6fDDYzMpobmaoGh5OLl4AXtGUGPcqCs=
 github.com/tidwall/pretty v1.2.0/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU=
 github.com/tidwall/sjson v1.0.4 h1:UcdIRXff12Lpnu3OLtZvnc03g4vH2suXDXhBwBqmzYg=
 github.com/tidwall/sjson v1.0.4/go.mod h1:bURseu1nuBkFpIES5cz6zBtjmYeOQmEESshn7VpF15Y=
-github.com/tikv/client-go/v2 v2.0.0 h1:XsYR7TwtyJbF6l6kyFG0RSAR7g2ftyl8LWbOAmq+CVI=
-github.com/tikv/client-go/v2 v2.0.0/go.mod h1:gaHSp8rnxZ0w36qb6QPPNPh9P0Mu5vAEwCQcc0Brni4=
+github.com/tikv/client-go/v2 v2.0.1 h1:+K/VvVOxEOXKMtR83bs5Aj3lrYdTdTZdvH0apfAWW10=
+github.com/tikv/client-go/v2 v2.0.1/go.mod h1:gaHSp8rnxZ0w36qb6QPPNPh9P0Mu5vAEwCQcc0Brni4=
 github.com/tikv/pd/client v0.0.0-20220216070739-26c668271201 h1:7h/Oi4Zw6eGCeXh4Q4ZvKI4k7nBJVUq0c29YCcLwKPM=
 github.com/tikv/pd/client v0.0.0-20220216070739-26c668271201/go.mod h1:fEvI5fhAuJn1Fn87VJF8ByE9Vc16EzWGoePZB21/nL8=
 github.com/tinylib/msgp v1.1.3 h1:3giwAkmtaEDLSV0MdO1lDLuPgklgPzmk8H9+So2BVfA=
diff --git a/pkg/gateway/gateway.go b/pkg/gateway/gateway.go
index 0a12823f..13a00aa6 100644
--- a/pkg/gateway/gateway.go
+++ b/pkg/gateway/gateway.go
@@ -30,6 +30,7 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/google/uuid"
 	"github.com/minio/minio-go/pkg/s3utils"
 	minio "github.com/minio/minio/cmd"
 
@@ -60,7 +61,9 @@ func NewJFSGateway(conf *vfs.Config, m meta.Meta, store chunk.ChunkStore, gConf
 		return nil, fmt.Errorf("Initialize failed: %s", err)
 	}
 	mctx = meta.NewContext(uint32(os.Getpid()), uint32(os.Getuid()), []uint32{uint32(os.Getgid())})
-	return &jfsObjects{fs: jfs, conf: conf, listPool: minio.NewTreeWalkPool(time.Minute * 30), gConf: gConf}, nil
+	jfsObj := &jfsObjects{fs: jfs, conf: conf, listPool: minio.NewTreeWalkPool(time.Minute * 30), gConf: gConf}
+	go jfsObj.cleanup()
+	return jfsObj, nil
 }
 
 type jfsObjects struct {
@@ -914,3 +917,43 @@ func (n *jfsObjects) AbortMultipartUpload(ctx context.Context, bucket, object, u
 	eno := n.fs.Rmr(mctx, n.upath(bucket, uploadID))
 	return jfsToObjectErr(ctx, eno, bucket, object, uploadID)
 }
+
+func (n *jfsObjects) cleanup() {
+	for t := range time.Tick(24 * time.Hour) {
+		var tmpDirs []string
+		if n.gConf.MultiBucket {
+			buckets, err := n.ListBuckets(context.Background())
+			if err != nil {
+				logger.Errorf("list buckets error: %v", err)
+				continue
+			}
+			for _, bucket := range buckets {
+				tmpDirs = append(tmpDirs, fmt.Sprintf(".sys/%s/tmp", bucket.Name))
+				tmpDirs = append(tmpDirs, fmt.Sprintf(".sys/%s/uploads", bucket.Name))
+			}
+		} else {
+			tmpDirs = append(tmpDirs, ".sys/tmp/")
+			tmpDirs = append(tmpDirs, ".sys/uploads/")
+		}
+		for _, dir := range tmpDirs {
+			f, errno := n.fs.Open(mctx, dir, 0)
+			if errno != 0 {
+				continue
+			}
+			entries, _ := f.ReaddirPlus(mctx, 0)
+			for _, entry := range entries {
+				if _, err := uuid.Parse(string(entry.Name)); err != nil {
+					continue
+				}
+				if t.Sub(time.Unix(entry.Attr.Mtime, 0)) > 7*24*time.Hour {
+					p := n.path(dir, string(entry.Name))
+					if errno := n.fs.Rmr(mctx, p); errno != 0 {
+						logger.Errorf("failed to delete expired temporary files path: %s,", p)
+					} else {
+						logger.Infof("delete expired temporary files path: %s, mtime: %s", p, time.Unix(entry.Attr.Mtime, 0).Format(time.RFC3339))
+					}
+				}
+			}
+		}
+	}
+}
diff --git a/pkg/meta/base.go b/pkg/meta/base.go
index 0e6e9626..ac0fdbf6 100644
--- a/pkg/meta/base.go
+++ b/pkg/meta/base.go
@@ -28,6 +28,7 @@ import (
 	"time"
 
 	"github.com/juicedata/juicefs/pkg/utils"
+	"github.com/juicedata/juicefs/pkg/version"
 )
 
 const (
@@ -65,10 +66,13 @@ type engine interface {
 	doUnlink(ctx Context, parent Ino, name string) syscall.Errno
 	doRmdir(ctx Context, parent Ino, name string) syscall.Errno
 	doReadlink(ctx Context, inode Ino) ([]byte, error)
-	doReaddir(ctx Context, inode Ino, plus uint8, entries *[]*Entry) syscall.Errno
+	doReaddir(ctx Context, inode Ino, plus uint8, entries *[]*Entry, limit int) syscall.Errno
 	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
+	doGetParents(ctx Context, inode Ino) map[Ino]int
+
+	GetSession(sid uint64, detail bool) (*Session, error)
 }
 
 type baseMeta struct {
@@ -82,6 +86,7 @@ type baseMeta struct {
 	of           *openfiles
 	removedFiles map[Ino]bool
 	compacting   map[uint64]bool
+	maxDeleting  chan struct{}
 	symlinks     *sync.Map
 	msgCallbacks *msgCallbacks
 	newSpace     int64
@@ -110,6 +115,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),
@@ -166,6 +172,7 @@ func (m *baseMeta) Load(checkVersion bool) (*Format, error) {
 func (m *baseMeta) NewSession() error {
 	go m.refreshUsage()
 	if m.conf.ReadOnly {
+		logger.Infof("Create read-only session OK with version: %s", version.Version())
 		return nil
 	}
 
@@ -183,7 +190,7 @@ func (m *baseMeta) NewSession() error {
 	if err = m.en.doNewSession(data); err != nil {
 		return fmt.Errorf("create session: %s", err)
 	}
-	logger.Infof("create session %d OK", m.sid)
+	logger.Infof("Create session %d OK with version: %s", m.sid, version.Version())
 
 	go m.refreshSession()
 	if !m.conf.NoBGJob {
@@ -229,7 +236,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))
 	}
 }
 
@@ -378,7 +390,7 @@ func (m *baseMeta) StatFS(ctx Context, totalspace, availspace, iused, iavail *ui
 
 func (m *baseMeta) resolveCase(ctx Context, parent Ino, name string) *Entry {
 	var entries []*Entry
-	_ = m.en.doReaddir(ctx, parent, 0, &entries)
+	_ = m.en.doReaddir(ctx, parent, 0, &entries, -1)
 	for _, e := range entries {
 		n := string(e.Name)
 		if strings.EqualFold(name, n) {
@@ -780,9 +792,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
@@ -810,7 +820,7 @@ func (m *baseMeta) Readdir(ctx Context, inode Ino, plus uint8, entries *[]*Entry
 		Name:  []byte(".."),
 		Attr:  &Attr{Typ: TypeDirectory},
 	})
-	return m.en.doReaddir(ctx, inode, plus, entries)
+	return m.en.doReaddir(ctx, inode, plus, entries, -1)
 }
 
 func (m *baseMeta) SetXattr(ctx Context, inode Ino, name string, value []byte, flags uint32) syscall.Errno {
@@ -837,17 +847,48 @@ func (m *baseMeta) RemoveXattr(ctx Context, inode Ino, name string) syscall.Errn
 	return m.en.doRemoveXattr(ctx, m.checkRoot(inode), name)
 }
 
+func (m *baseMeta) GetParents(ctx Context, inode Ino) map[Ino]int {
+	if inode == 1 {
+		return map[Ino]int{1: 1}
+	}
+	var attr Attr
+	if st := m.GetAttr(ctx, inode, &attr); st != 0 {
+		logger.Warnf("GetAttr inode %d: %s", inode, st)
+		return nil
+	}
+	if attr.Parent > 0 {
+		return map[Ino]int{attr.Parent: 1}
+	} else {
+		return m.en.doGetParents(ctx, inode)
+	}
+}
+
 func (m *baseMeta) fileDeleted(opened bool, inode Ino, length uint64) {
 	if opened {
 		m.Lock()
 		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
 	}
 }
 
 func (m *baseMeta) deleteSlice(chunkid uint64, size uint32) {
+	if chunkid == 0 {
+		return
+	}
 	if err := m.newMsg(DeleteChunk, chunkid, size); err == nil || strings.Contains(err.Error(), "NoSuchKey") || strings.Contains(err.Error(), "not found") {
 		if err = m.en.doDeleteSlice(chunkid, size); err != nil {
 			logger.Errorf("delete slice %d: %s", chunkid, err)
@@ -893,6 +934,15 @@ func (m *baseMeta) checkTrash(parent Ino, trash *Ino) syscall.Errno {
 	return st
 }
 
+func (m *baseMeta) trashEntry(parent, inode Ino, name string) string {
+	s := fmt.Sprintf("%d-%d-%s", parent, inode, name)
+	if len(s) > MaxName {
+		s = s[:MaxName]
+		logger.Warnf("File name is too long as a trash entry, truncating it: %s -> %s", name, s)
+	}
+	return s
+}
+
 func (m *baseMeta) cleanupTrash() {
 	for {
 		utils.SleepWithJitter(time.Hour)
@@ -917,7 +967,7 @@ func (m *baseMeta) doCleanupTrash(force bool) {
 	now := time.Now()
 	var st syscall.Errno
 	var entries []*Entry
-	if st = m.en.doReaddir(ctx, TrashInode, 0, &entries); st != 0 {
+	if st = m.en.doReaddir(ctx, TrashInode, 0, &entries, -1); st != 0 {
 		logger.Warnf("readdir trash %d: %s", TrashInode, st)
 		return
 	}
@@ -930,9 +980,10 @@ func (m *baseMeta) doCleanupTrash(force bool) {
 			logger.Debugf("cleanup trash: nothing to delete")
 		}
 	}()
-
+	batch := 1000000
 	edge := now.Add(-time.Duration(24*m.fmt.TrashDays+1) * time.Hour)
-	for _, e := range entries {
+	for len(entries) > 0 {
+		e := entries[0]
 		ts, err := time.Parse("2006-01-02-15", string(e.Name))
 		if err != nil {
 			logger.Warnf("bad entry as a subTrash: %s", e.Name)
@@ -940,11 +991,14 @@ func (m *baseMeta) doCleanupTrash(force bool) {
 		}
 		if ts.Before(edge) || force {
 			var subEntries []*Entry
-			if st = m.en.doReaddir(ctx, e.Inode, 0, &subEntries); st != 0 {
+			if st = m.en.doReaddir(ctx, e.Inode, 0, &subEntries, batch); st != 0 {
 				logger.Warnf("readdir subTrash %d: %s", e.Inode, st)
 				continue
 			}
-			rmdir := true
+			rmdir := len(subEntries) < batch
+			if rmdir {
+				entries = entries[1:]
+			}
 			for _, se := range subEntries {
 				if se.Attr.Typ == TypeDirectory {
 					st = m.en.doRmdir(ctx, e.Inode, string(se.Name))
diff --git a/pkg/meta/dump.go b/pkg/meta/dump.go
index 08d5e6b3..224ebef1 100644
--- a/pkg/meta/dump.go
+++ b/pkg/meta/dump.go
@@ -88,7 +88,7 @@ type DumpedXattr struct {
 
 type DumpedEntry struct {
 	Name    string                  `json:"-"`
-	Parent  Ino                     `json:"-"`
+	Parents []Ino                   `json:"-"`
 	Attr    *DumpedAttr             `json:"attr,omitempty"`
 	Symlink string                  `json:"symlink,omitempty"`
 	Xattrs  []*DumpedXattr          `json:"xattrs,omitempty"`
@@ -317,8 +317,10 @@ func collectEntry(e *DumpedEntry, entries map[Ino]*DumpedEntry, showProgress fun
 			return fmt.Errorf("inode conflict: %d", inode)
 		}
 		eattr.Nlink++
+		exist.Parents = append(exist.Parents, e.Parents...)
 		if eattr.Ctime*1e9+int64(eattr.Ctimensec) < attr.Ctime*1e9+int64(attr.Ctimensec) {
 			attr.Nlink = eattr.Nlink
+			e.Parents = exist.Parents
 			entries[inode] = e
 		}
 		return nil
@@ -329,12 +331,12 @@ func collectEntry(e *DumpedEntry, entries map[Ino]*DumpedEntry, showProgress fun
 		e.Attr.Nlink = 1 // reset
 	} else if typ == TypeDirectory {
 		if inode == 1 || inode == TrashInode { // root or trash inode
-			e.Parent = 1
+			e.Parents = []Ino{1}
 		}
 		e.Attr.Nlink = 2
 		for name, child := range e.Entries {
 			child.Name = name
-			child.Parent = inode
+			child.Parents = []Ino{inode}
 			if child.Attr == nil {
 				logger.Warnf("ignore empty entry: %s/%s", inode, name)
 				continue
diff --git a/pkg/meta/info.go b/pkg/meta/info.go
index babaee99..0cb72d94 100644
--- a/pkg/meta/info.go
+++ b/pkg/meta/info.go
@@ -85,7 +85,7 @@ func checkRedisInfo(rawInfo string) (info redisInfo, err error) {
 		case "maxmemory_policy":
 			info.maxMemoryPolicy = val
 			if val != "noeviction" {
-				logger.Warnf("maxmemory_policy is %q, please set it to 'noeviction'.", val)
+				logger.Warnf("maxmemory_policy is %q,  we will try to reconfigure it to 'noeviction'.", val)
 			}
 		case "redis_version":
 			info.redisVersion = val
diff --git a/pkg/meta/interface.go b/pkg/meta/interface.go
index 7664e50a..2e7a3289 100644
--- a/pkg/meta/interface.go
+++ b/pkg/meta/interface.go
@@ -73,6 +73,7 @@ const (
 	SetAttrMtimeNow
 )
 
+const MaxName = 255
 const TrashInode = 0x7FFFFFFF10000000 // larger than vfs.minInternalNode
 const TrashName = ".trash"
 
@@ -105,7 +106,7 @@ type Attr struct {
 	Nlink     uint32 // number of links (sub-directories or hardlinks)
 	Length    uint64 // length of regular file
 
-	Parent    Ino  // inode of parent, only for Directory
+	Parent    Ino  // inode of parent; 0 means tracked by parentKey (for hardlinks)
 	Full      bool // the attributes are completed or not
 	KeepCache bool // whether to keep the cached page or not
 }
@@ -249,7 +250,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
@@ -310,6 +311,8 @@ type Meta interface {
 	InvalidateChunkCache(ctx Context, inode Ino, indx uint32) syscall.Errno
 	// CopyFileRange copies part of a file to another one.
 	CopyFileRange(ctx Context, fin Ino, offIn uint64, fout Ino, offOut uint64, size uint64, flags uint32, copied *uint64) syscall.Errno
+	// GetParents returns a map of node parents (> 1 parents if hardlinked)
+	GetParents(ctx Context, inode Ino) map[Ino]int
 
 	// GetXattr returns the value of extended attribute for given name.
 	GetXattr(ctx Context, inode Ino, name string, vbuff *[]byte) syscall.Errno
diff --git a/pkg/meta/redis.go b/pkg/meta/redis.go
index befe405f..02a106d7 100644
--- a/pkg/meta/redis.go
+++ b/pkg/meta/redis.go
@@ -49,6 +49,7 @@ import (
 /*
 	Node: i$inode -> Attribute{type,mode,uid,gid,atime,mtime,ctime,nlink,length,rdev}
 	Dir:   d$inode -> {name -> {inode,type}}
+	Parent: p$inode -> {parent -> count} // for hard links
 	File:  c$inode_$indx -> [Slice{pos,id,length,off,len}]
 	Symlink: s$inode -> target
 	Xattr: x$inode -> {name -> value}
@@ -408,7 +409,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 +423,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
 	}
@@ -487,6 +488,10 @@ func (m *redisMeta) entryKey(parent Ino) string {
 	return m.prefix + "d" + parent.String()
 }
 
+func (m *redisMeta) parentKey(inode Ino) string {
+	return m.prefix + "p" + inode.String()
+}
+
 func (m *redisMeta) chunkKey(inode Ino, indx uint32) string {
 	return m.prefix + "c" + inode.String() + "_" + strconv.FormatInt(int64(indx), 10)
 }
@@ -1245,13 +1250,14 @@ func (m *redisMeta) doUnlink(ctx Context, parent Ino, name string) syscall.Errno
 				if _type == TypeFile && attr.Nlink == 0 {
 					opened = m.of.IsOpen(inode)
 				}
-			} else if attr.Nlink == 1 { // don't change parent if it has hard links
+			} else if attr.Parent > 0 {
 				attr.Parent = trash
 			}
 		} else {
 			logger.Warnf("no attribute for inode %d (%d, %s)", inode, parent, name)
 			trash = 0
 		}
+
 		_, err = tx.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
 			pipe.HDel(ctx, m.entryKey(parent), name)
 			if updateParent {
@@ -1260,7 +1266,13 @@ func (m *redisMeta) doUnlink(ctx Context, parent Ino, name string) syscall.Errno
 			if attr.Nlink > 0 {
 				pipe.Set(ctx, m.inodeKey(inode), m.marshal(&attr), 0)
 				if trash > 0 {
-					pipe.HSet(ctx, m.entryKey(trash), fmt.Sprintf("%d-%d-%s", parent, inode, name), buf)
+					pipe.HSet(ctx, m.entryKey(trash), m.trashEntry(parent, inode, name), buf)
+					if attr.Parent == 0 {
+						pipe.HIncrBy(ctx, m.parentKey(inode), trash.String(), 1)
+					}
+				}
+				if attr.Parent == 0 {
+					pipe.HIncrBy(ctx, m.parentKey(inode), parent.String(), -1)
 				}
 			} else {
 				switch _type {
@@ -1285,6 +1297,9 @@ func (m *redisMeta) doUnlink(ctx Context, parent Ino, name string) syscall.Errno
 					pipe.Decr(ctx, m.totalInodesKey())
 				}
 				pipe.Del(ctx, m.xattrKey(inode))
+				if attr.Parent == 0 {
+					pipe.Del(ctx, m.parentKey(inode))
+				}
 			}
 			return nil
 		})
@@ -1372,7 +1387,7 @@ func (m *redisMeta) doRmdir(ctx Context, parent Ino, name string) syscall.Errno
 			}
 			if trash > 0 {
 				pipe.Set(ctx, m.inodeKey(inode), m.marshal(&attr), 0)
-				pipe.HSet(ctx, m.entryKey(trash), fmt.Sprintf("%d-%d-%s", parent, inode, name), buf)
+				pipe.HSet(ctx, m.entryKey(trash), m.trashEntry(parent, inode, name), buf)
 			} else {
 				pipe.Del(ctx, m.inodeKey(inode))
 				pipe.Del(ctx, m.xattrKey(inode))
@@ -1480,11 +1495,15 @@ func (m *redisMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentD
 			tattr.Ctime = now.Unix()
 			tattr.Ctimensec = uint32(now.Nanosecond())
 			if exchange {
-				tattr.Parent = parentSrc
-				if dtyp == TypeDirectory && parentSrc != parentDst {
-					dattr.Nlink--
-					sattr.Nlink++
-					supdate, dupdate = true, true
+				if parentSrc != parentDst {
+					if dtyp == TypeDirectory {
+						tattr.Parent = parentSrc
+						dattr.Nlink--
+						sattr.Nlink++
+						supdate, dupdate = true, true
+					} else if tattr.Parent > 0 {
+						tattr.Parent = parentSrc
+					}
 				}
 			} else {
 				if dtyp == TypeDirectory {
@@ -1507,7 +1526,7 @@ func (m *redisMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentD
 							opened = m.of.IsOpen(dino)
 						}
 						defer func() { m.of.InvalidateChunk(dino, 0xFFFFFFFE) }()
-					} else if tattr.Nlink == 1 {
+					} else if tattr.Parent > 0 {
 						tattr.Parent = trash
 					}
 				}
@@ -1525,10 +1544,15 @@ func (m *redisMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentD
 			return syscall.EACCES
 		}
 
-		if typ == TypeDirectory && parentSrc != parentDst {
-			sattr.Nlink--
-			dattr.Nlink++
-			supdate, dupdate = true, true
+		if parentSrc != parentDst {
+			if typ == TypeDirectory {
+				iattr.Parent = parentDst
+				sattr.Nlink--
+				dattr.Nlink++
+				supdate, dupdate = true, true
+			} else if iattr.Parent > 0 {
+				iattr.Parent = parentDst
+			}
 		}
 		if supdate || now.Sub(time.Unix(sattr.Mtime, int64(sattr.Mtimensec))) >= minUpdateTime {
 			sattr.Mtime = now.Unix()
@@ -1544,7 +1568,6 @@ func (m *redisMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentD
 			dattr.Ctimensec = uint32(now.Nanosecond())
 			dupdate = true
 		}
-		iattr.Parent = parentDst
 		iattr.Ctime = now.Unix()
 		iattr.Ctimensec = uint32(now.Nanosecond())
 		if inode != nil {
@@ -1557,14 +1580,25 @@ func (m *redisMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentD
 			if exchange { // dbuf, tattr are valid
 				pipe.HSet(ctx, m.entryKey(parentSrc), nameSrc, dbuf)
 				pipe.Set(ctx, m.inodeKey(dino), m.marshal(&tattr), 0)
+				if parentSrc != parentDst && tattr.Parent == 0 {
+					pipe.HIncrBy(ctx, m.parentKey(dino), parentSrc.String(), 1)
+					pipe.HIncrBy(ctx, m.parentKey(dino), parentDst.String(), -1)
+				}
 			} else {
 				pipe.HDel(ctx, m.entryKey(parentSrc), nameSrc)
 				if dino > 0 {
 					if trash > 0 {
 						pipe.Set(ctx, m.inodeKey(dino), m.marshal(&tattr), 0)
-						pipe.HSet(ctx, m.entryKey(trash), fmt.Sprintf("%d-%d-%s", parentDst, dino, nameDst), dbuf)
+						pipe.HSet(ctx, m.entryKey(trash), m.trashEntry(parentDst, dino, nameDst), dbuf)
+						if tattr.Parent == 0 {
+							pipe.HIncrBy(ctx, m.parentKey(dino), trash.String(), 1)
+							pipe.HIncrBy(ctx, m.parentKey(dino), parentDst.String(), -1)
+						}
 					} else if dtyp != TypeDirectory && tattr.Nlink > 0 {
 						pipe.Set(ctx, m.inodeKey(dino), m.marshal(&tattr), 0)
+						if tattr.Parent == 0 {
+							pipe.HIncrBy(ctx, m.parentKey(dino), parentDst.String(), -1)
+						}
 					} else {
 						if dtyp == TypeFile {
 							if opened {
@@ -1587,11 +1621,20 @@ func (m *redisMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentD
 							pipe.Decr(ctx, m.totalInodesKey())
 						}
 						pipe.Del(ctx, m.xattrKey(dino))
+						if tattr.Parent == 0 {
+							pipe.Del(ctx, m.parentKey(dino))
+						}
 					}
 				}
 			}
-			if parentDst != parentSrc && !isTrash(parentSrc) && supdate {
-				pipe.Set(ctx, m.inodeKey(parentSrc), m.marshal(&sattr), 0)
+			if parentDst != parentSrc {
+				if !isTrash(parentSrc) && supdate {
+					pipe.Set(ctx, m.inodeKey(parentSrc), m.marshal(&sattr), 0)
+				}
+				if iattr.Parent == 0 {
+					pipe.HIncrBy(ctx, m.parentKey(ino), parentDst.String(), 1)
+					pipe.HIncrBy(ctx, m.parentKey(ino), parentSrc.String(), -1)
+				}
 			}
 			pipe.Set(ctx, m.inodeKey(ino), m.marshal(&iattr), 0)
 			pipe.HSet(ctx, m.entryKey(parentDst), nameDst, buf)
@@ -1638,6 +1681,8 @@ func (m *redisMeta) doLink(ctx Context, inode, parent Ino, name string, attr *At
 		if iattr.Typ == TypeDirectory {
 			return syscall.EPERM
 		}
+		oldParent := iattr.Parent
+		iattr.Parent = 0
 		iattr.Ctime = now.Unix()
 		iattr.Ctimensec = uint32(now.Nanosecond())
 		iattr.Nlink++
@@ -1657,6 +1702,10 @@ func (m *redisMeta) doLink(ctx Context, inode, parent Ino, name string, attr *At
 				pipe.Set(ctx, m.inodeKey(parent), m.marshal(&pattr), 0)
 			}
 			pipe.Set(ctx, m.inodeKey(inode), m.marshal(&iattr), 0)
+			if oldParent > 0 {
+				pipe.HIncrBy(ctx, m.parentKey(inode), oldParent.String(), 1)
+			}
+			pipe.HIncrBy(ctx, m.parentKey(inode), parent.String(), 1)
 			return nil
 		})
 		if err == nil && attr != nil {
@@ -1666,7 +1715,8 @@ func (m *redisMeta) doLink(ctx Context, inode, parent Ino, name string, attr *At
 	}, m.inodeKey(inode), m.entryKey(parent), m.inodeKey(parent)))
 }
 
-func (m *redisMeta) doReaddir(ctx Context, inode Ino, plus uint8, entries *[]*Entry) syscall.Errno {
+func (m *redisMeta) doReaddir(ctx Context, inode Ino, plus uint8, entries *[]*Entry, limit int) syscall.Errno {
+	var stop = errors.New("stop")
 	err := m.hscan(ctx, m.entryKey(inode), func(keys []string) error {
 		newEntries := make([]Entry, len(keys)/2)
 		newAttrs := make([]Attr, len(keys)/2)
@@ -1682,9 +1732,15 @@ func (m *redisMeta) doReaddir(ctx Context, inode Ino, plus uint8, entries *[]*En
 			ent.Attr = &newAttrs[i/2]
 			ent.Attr.Typ = typ
 			*entries = append(*entries, ent)
+			if limit > 0 && len(*entries) >= limit {
+				return stop
+			}
 		}
 		return nil
 	})
+	if errors.Is(err, stop) {
+		err = nil
+	}
 	if err != nil {
 		return errno(err)
 	}
@@ -1874,7 +1930,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
 }
@@ -2083,6 +2139,22 @@ func (m *redisMeta) CopyFileRange(ctx Context, fin Ino, offIn uint64, fout Ino,
 	return errno(err)
 }
 
+func (m *redisMeta) doGetParents(ctx Context, inode Ino) map[Ino]int {
+	vals, err := m.rdb.HGetAll(ctx, m.parentKey(inode)).Result()
+	if err != nil {
+		logger.Warnf("Scan parent key of inode %d: %s", inode, err)
+		return nil
+	}
+	ps := make(map[Ino]int)
+	for k, v := range vals {
+		if n, _ := strconv.Atoi(v); n > 0 {
+			ino, _ := strconv.ParseUint(k, 10, 64)
+			ps[Ino(ino)] = n
+		}
+	}
+	return ps
+}
+
 // For now only deleted files
 func (m *redisMeta) cleanupLegacies() {
 	for {
@@ -2245,19 +2317,23 @@ func (m *redisMeta) deleteChunk(inode Ino, indx uint32) error {
 		var slices []*slice
 		var rs []*redis.IntCmd
 		err := m.txn(ctx, func(tx *redis.Tx) error {
-			slices = nil
 			vals, err := tx.LRange(ctx, key, 0, 100).Result()
 			if err == redis.Nil {
 				return nil
 			}
+			slices = nil
+			rs = nil
 			_, err = tx.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
 				for _, v := range vals {
+					pipe.LPop(ctx, key)
 					rb := utils.ReadBuffer([]byte(v))
 					_ = rb.Get32() // pos
 					chunkid := rb.Get64()
+					if chunkid == 0 {
+						continue
+					}
 					size := rb.Get32()
 					slices = append(slices, &slice{chunkid: chunkid, size: size})
-					pipe.LPop(ctx, key)
 					rs = append(rs, pipe.HIncrBy(ctx, m.sliceRefs(), m.sliceKey(chunkid, size), -1))
 				}
 				return nil
@@ -2435,10 +2511,11 @@ func (m *redisMeta) compactChunk(inode Ino, indx uint32, force bool) {
 		for _, s := range ss {
 			buf = append(buf, m.encodeDelayedSlice(s.chunkid, s.size)...)
 		}
+	} else {
+		rs = make([]*redis.IntCmd, len(ss))
 	}
 	key := m.chunkKey(inode, indx)
 	errno := errno(m.txn(ctx, func(tx *redis.Tx) error {
-		rs = nil
 		vals2, err := tx.LRange(ctx, key, 0, int64(len(vals)-1)).Result()
 		if err != nil {
 			return err
@@ -2462,8 +2539,10 @@ func (m *redisMeta) compactChunk(inode Ino, indx uint32, force bool) {
 			if trash {
 				pipe.HSet(ctx, m.delSlices(), fmt.Sprintf("%d_%d", chunkid, time.Now().Unix()), buf)
 			} else {
-				for _, s := range ss {
-					rs = append(rs, pipe.HIncrBy(ctx, m.sliceRefs(), m.sliceKey(s.chunkid, s.size), -1))
+				for i, s := range ss {
+					if s.chunkid > 0 {
+						rs[i] = pipe.HIncrBy(ctx, m.sliceRefs(), m.sliceKey(s.chunkid, s.size), -1)
+					}
 				}
 			}
 			return nil
@@ -2488,7 +2567,7 @@ func (m *redisMeta) compactChunk(inode Ino, indx uint32, force bool) {
 		m.cleanupZeroRef(m.sliceKey(chunkid, size))
 		if !trash {
 			for i, s := range ss {
-				if rs[i].Err() == nil && rs[i].Val() < 0 {
+				if s.chunkid > 0 && rs[i].Err() == nil && rs[i].Val() < 0 {
 					m.deleteSlice(s.chunkid, s.size)
 				}
 			}
@@ -2769,11 +2848,21 @@ func (m *redisMeta) checkServerConfig() {
 		logger.Warnf("parse info: %s", err)
 		return
 	}
-	_, err = checkRedisInfo(rawInfo)
+	rInfo, err := checkRedisInfo(rawInfo)
 	if err != nil {
 		logger.Warnf("parse info: %s", err)
 	}
-
+	if rInfo.maxMemoryPolicy != "noeviction" {
+		if _, err := m.rdb.ConfigSet(Background, "maxmemory-policy", "noeviction").Result(); err != nil {
+			logger.Errorf("try to reconfigure maxmemory-policy to 'noeviction' failed: %s", err)
+		} else if result, err := m.rdb.ConfigGet(Background, "maxmemory-policy").Result(); err != nil {
+			logger.Warnf("get config maxmemory-policy failed: %s", err)
+		} else if len(result) == 2 && result[1] != "noeviction" {
+			logger.Warnf("reconfigured maxmemory-policy to 'noeviction', but it's still %s", result[1])
+		} else {
+			logger.Infof("set maxmemory-policy to 'noeviction' successfully")
+		}
+	}
 	start := time.Now()
 	_ = m.rdb.Ping(Background)
 	logger.Infof("Ping redis: %s", time.Since(start))
@@ -3179,7 +3268,14 @@ func (m *redisMeta) loadEntry(e *DumpedEntry, p redis.Pipeliner, tryExec func(),
 	logger.Debugf("Loading entry inode %d name %s", inode, unescape(e.Name))
 	ctx := Background
 	attr := loadAttr(e.Attr)
-	attr.Parent = e.Parent
+	numParents := len(e.Parents)
+	if numParents == 1 {
+		attr.Parent = e.Parents[0]
+	} else if numParents == 0 {
+		logger.Fatalf("No parent for inode: %d", inode)
+	} else if attr.Typ == TypeDirectory {
+		logger.Fatalf("Too many parents for directory inode %d: %v", inode, e.Parents)
+	}
 	batch := 100
 	if attr.Typ == TypeFile {
 		attr.Length = e.Attr.Length
@@ -3241,6 +3337,11 @@ func (m *redisMeta) loadEntry(e *DumpedEntry, p redis.Pipeliner, tryExec func(),
 		}
 		p.HSet(ctx, m.xattrKey(inode), xattrs)
 	}
+	if numParents > 1 {
+		for _, parent := range e.Parents {
+			p.HIncrBy(ctx, m.parentKey(inode), parent.String(), 1)
+		}
+	}
 	p.Set(ctx, m.inodeKey(inode), m.marshal(attr), 0)
 }
 
diff --git a/pkg/meta/sql.go b/pkg/meta/sql.go
index c8e2d013..c9e42beb 100644
--- a/pkg/meta/sql.go
+++ b/pkg/meta/sql.go
@@ -61,6 +61,12 @@ type edge struct {
 	Type   uint8  `xorm:"notnull"`
 }
 
+type linkParent struct {
+	Id     int64 `xorm:"pk bigserial"`
+	Inode  Ino   `xorm:"index notnull"`
+	Parent Ino   `xorm:"notnull"`
+}
+
 type node struct {
 	Inode  Ino    `xorm:"pk"`
 	Type   uint8  `xorm:"notnull"`
@@ -224,28 +230,40 @@ func (m *dbMeta) doDeleteSlice(chunkid uint64, size uint32) error {
 	})
 }
 
+func (m *dbMeta) syncTable(beans ...interface{}) error {
+	err := m.db.Sync2(beans...)
+	if err != nil && strings.Contains(err.Error(), "Duplicate key") {
+		err = nil
+	}
+	return err
+}
+
 func (m *dbMeta) Init(format Format, force bool) error {
-	if err := m.db.Sync2(new(setting), new(counter)); err != nil {
+	if err := m.syncTable(new(setting), new(counter)); err != nil {
 		return fmt.Errorf("create table setting, counter: %s", err)
 	}
-	if err := m.db.Sync2(new(edge)); err != nil && !strings.Contains(err.Error(), "Duplicate entry") {
-		return fmt.Errorf("create table edge: %s", err)
+	if err := m.syncTable(new(edge), new(linkParent)); err != nil {
+		return fmt.Errorf("create table edge, linkParent: %s", err)
 	}
-	if err := m.db.Sync2(new(node), new(symlink), new(xattr)); err != nil {
+	if err := m.syncTable(new(node), new(symlink), new(xattr)); err != nil {
 		return fmt.Errorf("create table node, symlink, xattr: %s", err)
 	}
-	if err := m.db.Sync2(new(chunk), new(chunkRef), new(delslices)); err != nil {
+	if err := m.syncTable(new(chunk), new(chunkRef), new(delslices)); err != nil {
 		return fmt.Errorf("create table chunk, chunk_ref, delslices: %s", err)
 	}
-	if err := m.db.Sync2(new(session2), new(sustained), new(delfile)); err != nil {
+	if err := m.syncTable(new(session2), new(sustained), new(delfile)); err != nil {
 		return fmt.Errorf("create table session2, sustaind, delfile: %s", err)
 	}
-	if err := m.db.Sync2(new(flock), new(plock)); err != nil {
+	if err := m.syncTable(new(flock), new(plock)); err != nil {
 		return fmt.Errorf("create table flock, plock: %s", err)
 	}
 
 	var s = setting{Name: "format"}
-	ok, err := m.db.Get(&s)
+	var ok bool
+	err := m.roTxn(func(ses *xorm.Session) (err error) {
+		ok, err = ses.Get(&s)
+		return err
+	})
 	if err != nil {
 		return err
 	}
@@ -312,34 +330,41 @@ func (m *dbMeta) Init(format Format, force bool) error {
 
 func (m *dbMeta) Reset() error {
 	return m.db.DropTables(&setting{}, &counter{},
-		&node{}, &edge{}, &symlink{}, &xattr{},
+		&node{}, &edge{}, &linkParent{}, &symlink{}, &xattr{},
 		&chunk{}, &chunkRef{}, &delslices{},
 		&session{}, &session2{}, &sustained{}, &delfile{},
 		&flock{}, &plock{})
 }
 
-func (m *dbMeta) doLoad() ([]byte, error) {
-	if ok, err := m.db.IsTableExist(&setting{}); err != nil {
-		return nil, err
-	} else if !ok {
-		return nil, nil
-	}
-	s := setting{Name: "format"}
-	_, err := m.db.Get(&s)
-	return []byte(s.Value), err
+func (m *dbMeta) doLoad() (data []byte, err error) {
+	err = m.roTxn(func(ses *xorm.Session) error {
+		if ok, err := ses.IsTableExist(&setting{}); err != nil {
+			return err
+		} else if !ok {
+			return nil
+		}
+		s := setting{Name: "format"}
+		ok, err := ses.Get(&s)
+		if err == nil && ok {
+			data = []byte(s.Value)
+		}
+		return err
+	})
+	return
 }
 
 func (m *dbMeta) doNewSession(sinfo []byte) error {
-	err := m.db.Sync2(new(session2), new(delslices))
+	// add new table
+	err := m.syncTable(new(session2), new(delslices), new(linkParent))
 	if err != nil {
-		return fmt.Errorf("update table session2, delslices: %s", err)
+		return fmt.Errorf("update table session2, delslices, linkParent: %s", err)
 	}
 	// add primary key
-	if err = m.db.Sync2(new(edge), new(chunk), new(xattr), new(sustained)); err != nil {
+	if err = m.syncTable(new(edge), new(chunk), new(xattr), new(sustained)); err != nil {
 		return fmt.Errorf("update table edge, chunk, xattr, sustained: %s", err)
 	}
 	// update the owner from uint64 to int64
-	if err = m.db.Sync2(new(flock), new(plock)); err != nil {
+	if err = m.syncTable(new(flock), new(plock)); err != nil {
 		return fmt.Errorf("update table flock, plock: %s", err)
 	}
 
@@ -393,93 +418,113 @@ func (m *dbMeta) getSession(row interface{}, detail bool) (*Session, error) {
 			frows []flock
 			prows []plock
 		)
-		if err := m.db.Find(&srows, &sustained{Sid: s.Sid}); err != nil {
-			return nil, fmt.Errorf("find sustained %d: %s", s.Sid, err)
-		}
-		s.Sustained = make([]Ino, 0, len(srows))
-		for _, srow := range srows {
-			s.Sustained = append(s.Sustained, srow.Inode)
-		}
+		err := m.roTxn(func(ses *xorm.Session) error {
+			if err := ses.Find(&srows, &sustained{Sid: s.Sid}); err != nil {
+				return fmt.Errorf("find sustained %d: %s", s.Sid, err)
+			}
+			s.Sustained = make([]Ino, 0, len(srows))
+			for _, srow := range srows {
+				s.Sustained = append(s.Sustained, srow.Inode)
+			}
 
-		if err := m.db.Find(&frows, &flock{Sid: s.Sid}); err != nil {
-			return nil, fmt.Errorf("find flock %d: %s", s.Sid, err)
-		}
-		s.Flocks = make([]Flock, 0, len(frows))
-		for _, frow := range frows {
-			s.Flocks = append(s.Flocks, Flock{frow.Inode, uint64(frow.Owner), string(frow.Ltype)})
-		}
+			if err := ses.Find(&frows, &flock{Sid: s.Sid}); err != nil {
+				return fmt.Errorf("find flock %d: %s", s.Sid, err)
+			}
+			s.Flocks = make([]Flock, 0, len(frows))
+			for _, frow := range frows {
+				s.Flocks = append(s.Flocks, Flock{frow.Inode, uint64(frow.Owner), string(frow.Ltype)})
+			}
 
-		if err := m.db.Find(&prows, &plock{Sid: s.Sid}); err != nil {
-			return nil, fmt.Errorf("find plock %d: %s", s.Sid, err)
-		}
-		s.Plocks = make([]Plock, 0, len(prows))
-		for _, prow := range prows {
-			s.Plocks = append(s.Plocks, Plock{prow.Inode, uint64(prow.Owner), prow.Records})
+			if err := ses.Find(&prows, &plock{Sid: s.Sid}); err != nil {
+				return fmt.Errorf("find plock %d: %s", s.Sid, err)
+			}
+			s.Plocks = make([]Plock, 0, len(prows))
+			for _, prow := range prows {
+				s.Plocks = append(s.Plocks, Plock{prow.Inode, uint64(prow.Owner), prow.Records})
+			}
+			return nil
+		})
+		if err != nil {
+			return nil, err
 		}
 	}
 	return &s, nil
 }
 
-func (m *dbMeta) GetSession(sid uint64) (*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)
-	}
-	if ok, err := m.db.IsTableExist(&session{}); err != nil {
-		return nil, err
-	} else if ok {
-		row := session{Sid: sid}
-		if ok, err := m.db.Get(&row); err != nil {
-			return nil, err
+func (m *dbMeta) GetSession(sid uint64, detail bool) (s *Session, err error) {
+	err = m.roTxn(func(ses *xorm.Session) error {
+		row := session2{Sid: sid}
+		if ok, err := ses.Get(&row); err != nil {
+			return err
 		} else if ok {
-			return m.getSession(&row, true)
+			s, err = m.getSession(&row, detail)
+			return err
 		}
-	}
-	return nil, fmt.Errorf("session not found: %d", sid)
+		if ok, err := ses.IsTableExist(&session{}); err != nil {
+			return err
+		} else if ok {
+			row := session{Sid: sid}
+			if ok, err := ses.Get(&row); err != nil {
+				return err
+			} else if ok {
+				s, err = m.getSession(&row, detail)
+				return err
+			}
+		}
+		return fmt.Errorf("session not found: %d", sid)
+	})
+	return
 }
 
 func (m *dbMeta) ListSessions() ([]*Session, error) {
-	var rows []session2
-	err := m.db.Find(&rows)
-	if err != nil {
-		return nil, err
-	}
-	sessions := make([]*Session, 0, len(rows))
-	for i := range rows {
-		s, err := m.getSession(&rows[i], false)
+	var sessions []*Session
+	err := m.roTxn(func(ses *xorm.Session) error {
+		var rows []session2
+		err := ses.Find(&rows)
 		if err != nil {
-			logger.Errorf("get session: %s", err)
-			continue
+			return err
 		}
-		sessions = append(sessions, s)
-	}
-
-	if ok, err := m.db.IsTableExist(&session{}); err != nil {
-		logger.Errorf("Check legacy session table: %s", err)
-	} else if ok {
-		var lrows []session
-		if err = m.db.Find(&lrows); err != nil {
-			logger.Errorf("Scan legacy sessions: %s", err)
-			return sessions, nil
-		}
-		for i := range lrows {
-			s, err := m.getSession(&lrows[i], false)
+		sessions = make([]*Session, 0, len(rows))
+		for i := range rows {
+			s, err := m.getSession(&rows[i], false)
 			if err != nil {
-				logger.Errorf("Get legacy session: %s", err)
+				logger.Errorf("get session: %s", err)
 				continue
 			}
 			sessions = append(sessions, s)
 		}
-	}
-	return sessions, nil
+		if ok, err := ses.IsTableExist(&session{}); err != nil {
+			logger.Errorf("Check legacy session table: %s", err)
+		} else if ok {
+			var lrows []session
+			if err = ses.Find(&lrows); err != nil {
+				logger.Errorf("Scan legacy sessions: %s", err)
+				return nil
+			}
+			for i := range lrows {
+				s, err := m.getSession(&lrows[i], false)
+				if err != nil {
+					logger.Errorf("Get legacy session: %s", err)
+					continue
+				}
+				sessions = append(sessions, s)
+			}
+		}
+		return nil
+	})
+	return sessions, err
 }
 
-func (m *dbMeta) getCounter(name string) (int64, error) {
-	c := counter{Name: name}
-	_, err := m.db.Get(&c)
-	return c.Value, err
+func (m *dbMeta) getCounter(name string) (v int64, err error) {
+	err = m.roTxn(func(s *xorm.Session) error {
+		c := counter{Name: name}
+		_, err := s.Get(&c)
+		if err == nil {
+			v = c.Value
+		}
+		return err
+	})
+	return
 }
 
 func (m *dbMeta) incrCounter(name string, value int64) (int64, error) {
@@ -565,9 +610,12 @@ func (m *dbMeta) shouldRetry(err error) bool {
 		return errors.Is(err, errBusy) || strings.Contains(msg, "database is locked")
 	case "mysql":
 		// MySQL, MariaDB or TiDB
-		return strings.Contains(msg, "try restarting transaction") || strings.Contains(msg, "try again later")
+		return strings.Contains(msg, "try restarting transaction") || strings.Contains(msg, "try again later") ||
+			strings.Contains(msg, "duplicate entry")
 	case "postgres":
-		return strings.Contains(msg, "current transaction is aborted") || strings.Contains(msg, "deadlock detected")
+		return strings.Contains(msg, "current transaction is aborted") || strings.Contains(msg, "deadlock detected") ||
+			strings.Contains(msg, "duplicate key value") || strings.Contains(msg, "could not serialize access") ||
+			strings.Contains(msg, "bad connection") || errors.Is(err, io.EOF) // could not send data to client: No buffer space available
 	default:
 		return false
 	}
@@ -584,15 +632,45 @@ func (m *dbMeta) txn(f func(s *xorm.Session) error) error {
 		_, err = m.db.Transaction(func(s *xorm.Session) (interface{}, error) {
 			return nil, f(s)
 		})
+		if eno, ok := err.(syscall.Errno); ok && eno == 0 {
+			err = nil
+		}
 		if m.shouldRetry(err) {
 			txRestart.Add(1)
 			logger.Debugf("Transaction failed, restart it (tried %d): %s", i+1, err)
 			time.Sleep(time.Millisecond * time.Duration(i*i))
 			continue
 		}
+		return err
+	}
+	logger.Warnf("Already tried 50 times, returning: %s", err)
+	return err
+}
+
+func (m *dbMeta) roTxn(f func(s *xorm.Session) error) error {
+	start := time.Now()
+	defer func() { txDist.Observe(time.Since(start).Seconds()) }()
+	var err error
+	s := m.db.NewSession()
+	defer s.Close()
+	for i := 0; i < 50; i++ {
+		// TODO: read-only
+		if err := s.Begin(); err != nil {
+			logger.Debugf("Start transaction failed, try again (tried %d): %s", i+1, err)
+			time.Sleep(time.Millisecond * time.Duration(i*i))
+			continue
+		}
+		err = f(s)
 		if eno, ok := err.(syscall.Errno); ok && eno == 0 {
 			err = nil
 		}
+		_ = s.Rollback()
+		if m.shouldRetry(err) {
+			txRestart.Add(1)
+			logger.Debugf("Transaction failed, restart it (tried %d): %s", i+1, err)
+			time.Sleep(time.Millisecond * time.Duration(i*i))
+			continue
+		}
 		return err
 	}
 	logger.Warnf("Already tried 50 times, returning: %s", err)
@@ -644,32 +722,40 @@ func (m *dbMeta) flushStats() {
 }
 
 func (m *dbMeta) doLookup(ctx Context, parent Ino, name string, inode *Ino, attr *Attr) syscall.Errno {
-	db := m.db.Table(&edge{})
-	if attr != nil {
-		db = db.Join("INNER", &node{}, "jfs_edge.inode=jfs_node.inode")
-	}
-	nn := namedNode{node: node{Parent: parent}, Name: []byte(name)}
-	exist, err := db.Select("*").Get(&nn)
-	if err != nil {
-		return errno(err)
-	}
-	if !exist {
-		return syscall.ENOENT
-	}
-	*inode = nn.Inode
-	m.parseAttr(&nn.node, attr)
-	return 0
+	return errno(m.roTxn(func(s *xorm.Session) error {
+		s = s.Table(&edge{})
+		nn := namedNode{node: node{Parent: parent}, Name: []byte(name)}
+		var exist bool
+		var err error
+		if attr != nil {
+			s = s.Join("INNER", &node{}, "jfs_edge.inode=jfs_node.inode")
+			exist, err = s.Select("jfs_node.*").Get(&nn)
+		} else {
+			exist, err = s.Select("*").Get(&nn)
+		}
+		if err != nil {
+			return err
+		}
+		if !exist {
+			return syscall.ENOENT
+		}
+		*inode = nn.Inode
+		m.parseAttr(&nn.node, attr)
+		return nil
+	}))
 }
 
 func (m *dbMeta) doGetAttr(ctx Context, inode Ino, attr *Attr) syscall.Errno {
-	var n = node{Inode: inode}
-	ok, err := m.db.Get(&n)
-	if ok {
-		m.parseAttr(&n, attr)
-	} else if err == nil {
-		err = syscall.ENOENT
-	}
-	return errno(err)
+	return errno(m.roTxn(func(s *xorm.Session) error {
+		var n = node{Inode: inode}
+		ok, err := s.Get(&n)
+		if ok {
+			m.parseAttr(&n, attr)
+		} else if err == nil {
+			err = syscall.ENOENT
+		}
+		return err
+	}))
 }
 
 func clearSUGIDSQL(ctx Context, cur *node, set *Attr) {
@@ -783,15 +869,6 @@ func (m *dbMeta) appendSlice(s *xorm.Session, inode Ino, indx uint32, buf []byte
 	return err
 }
 
-func scanChunk(rows *xorm.Rows, c *chunk) error {
-	// FIXME: there is a bug in xorm, which will fill existing slice with partial result
-	// see line 582 in session.go:582 of xorm
-	if len(c.Slices) > 0 {
-		c.Slices = c.Slices[:0]
-	}
-	return rows.Scan(c)
-}
-
 func (m *dbMeta) Truncate(ctx Context, inode Ino, flags uint8, length uint64, attr *Attr) syscall.Errno {
 	defer timeit(time.Now())
 	f := m.of.find(inode)
@@ -821,25 +898,16 @@ func (m *dbMeta) Truncate(ctx Context, inode Ino, flags uint8, length uint64, at
 		if newSpace > 0 && m.checkQuota(newSpace, 0) {
 			return syscall.ENOSPC
 		}
-		var c chunk
-		var zeroChunks []uint32
+		var zeroChunks []chunk
 		var left, right = n.Length, length
 		if left > right {
 			right, left = left, right
 		}
 		if right/ChunkSize-left/ChunkSize > 1 {
-			rows, err := s.Where("inode = ? AND indx > ? AND indx < ?", inode, left/ChunkSize, right/ChunkSize).Cols("indx").ForUpdate().Rows(&c)
+			err := s.Where("inode = ? AND indx > ? AND indx < ?", inode, left/ChunkSize, right/ChunkSize).Cols("indx").ForUpdate().Find(&zeroChunks)
 			if err != nil {
 				return err
 			}
-			for rows.Next() {
-				if err = scanChunk(rows, &c); err != nil {
-					_ = rows.Close()
-					return err
-				}
-				zeroChunks = append(zeroChunks, c.Indx)
-			}
-			_ = rows.Close()
 		}
 
 		l := uint32(right - left)
@@ -850,8 +918,8 @@ func (m *dbMeta) Truncate(ctx Context, inode Ino, flags uint8, length uint64, at
 			return err
 		}
 		buf := marshalSlice(0, 0, 0, 0, ChunkSize)
-		for _, indx := range zeroChunks {
-			if err = m.appendSlice(s, inode, indx, buf); err != nil {
+		for _, c := range zeroChunks {
+			if err = m.appendSlice(s, inode, c.Indx, buf); err != nil {
 				return err
 			}
 		}
@@ -961,10 +1029,16 @@ func (m *dbMeta) Fallocate(ctx Context, inode Ino, mode uint8, off uint64, size
 	return errno(err)
 }
 
-func (m *dbMeta) doReadlink(ctx Context, inode Ino) ([]byte, error) {
-	var l = symlink{Inode: inode}
-	_, err := m.db.Get(&l)
-	return []byte(l.Target), err
+func (m *dbMeta) doReadlink(ctx Context, inode Ino) (target []byte, err error) {
+	err = m.roTxn(func(s *xorm.Session) error {
+		var l = symlink{Inode: inode}
+		ok, err := s.Get(&l)
+		if err == nil && ok {
+			target = []byte(l.Target)
+		}
+		return err
+	})
+	return
 }
 
 func (m *dbMeta) doMknod(ctx Context, parent Ino, name string, _type uint8, mode, cumask uint16, rdev uint32, path string, inode *Ino, attr *Attr) syscall.Errno {
@@ -1149,7 +1223,7 @@ func (m *dbMeta) doUnlink(ctx Context, parent Ino, name string) syscall.Errno {
 				if n.Type == TypeFile && n.Nlink == 0 {
 					opened = m.of.IsOpen(e.Inode)
 				}
-			} else if n.Nlink == 1 {
+			} else if n.Parent > 0 {
 				n.Parent = trash
 			}
 		} else {
@@ -1178,7 +1252,17 @@ func (m *dbMeta) doUnlink(ctx Context, parent Ino, name string) syscall.Errno {
 				return err
 			}
 			if trash > 0 {
-				if err = mustInsert(s, &edge{Parent: trash, Name: []byte(fmt.Sprintf("%d-%d-%s", parent, e.Inode, string(e.Name))), Inode: e.Inode, Type: e.Type}); err != nil {
+				if err = mustInsert(s, &edge{Parent: trash, Name: []byte(m.trashEntry(parent, e.Inode, string(e.Name))), Inode: e.Inode, Type: e.Type}); err != nil {
+					return err
+				}
+				if n.Parent == 0 {
+					if err = mustInsert(s, &linkParent{Inode: e.Inode, Parent: trash}); err != nil {
+						return err
+					}
+				}
+			}
+			if n.Parent == 0 {
+				if _, err = s.Limit(1, 0).Delete(&linkParent{Inode: e.Inode, Parent: parent}); err != nil {
 					return err
 				}
 			}
@@ -1215,6 +1299,11 @@ func (m *dbMeta) doUnlink(ctx Context, parent Ino, name string) syscall.Errno {
 			if _, err := s.Delete(&xattr{Inode: e.Inode}); err != nil {
 				return err
 			}
+			if n.Parent == 0 {
+				if _, err = s.Delete(&linkParent{Inode: e.Inode}); err != nil {
+					return err
+				}
+			}
 		}
 		return err
 	})
@@ -1263,6 +1352,11 @@ func (m *dbMeta) doRmdir(ctx Context, parent Ino, name string) syscall.Errno {
 		if e.Type != TypeDirectory {
 			return syscall.ENOTDIR
 		}
+		var n = node{Inode: e.Inode}
+		ok, err = s.ForUpdate().Get(&n)
+		if err != nil {
+			return err
+		}
 		exist, err := s.ForUpdate().Exist(&edge{Parent: e.Inode})
 		if err != nil {
 			return err
@@ -1270,12 +1364,6 @@ func (m *dbMeta) doRmdir(ctx Context, parent Ino, name string) syscall.Errno {
 		if exist {
 			return syscall.ENOTEMPTY
 		}
-		var n = node{Inode: e.Inode}
-		ok, err = s.ForUpdate().Get(&n)
-		if err != nil {
-			return err
-		}
-
 		now := time.Now().UnixNano() / 1e3
 		if ok {
 			if ctx.Uid() != 0 && pn.Mode&01000 != 0 && ctx.Uid() != pn.Uid && ctx.Uid() != n.Uid {
@@ -1300,7 +1388,7 @@ func (m *dbMeta) doRmdir(ctx Context, parent Ino, name string) syscall.Errno {
 			if _, err = s.Cols("ctime", "parent").Update(&n, &node{Inode: n.Inode}); err != nil {
 				return err
 			}
-			if err = mustInsert(s, &edge{Parent: trash, Name: []byte(fmt.Sprintf("%d-%d-%s", parent, e.Inode, string(e.Name))), Inode: e.Inode, Type: e.Type}); err != nil {
+			if err = mustInsert(s, &edge{Parent: trash, Name: []byte(m.trashEntry(parent, e.Inode, string(e.Name))), Inode: e.Inode, Type: e.Type}); err != nil {
 				return err
 			}
 		} else {
@@ -1333,6 +1421,30 @@ func (m *dbMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentDst
 	var dn node
 	var newSpace, newInode int64
 	err := m.txn(func(s *xorm.Session) error {
+		var t node
+		rows, err := s.ForUpdate().Where("inode IN (?,?)", parentSrc, parentDst).Rows(&t)
+		if err != nil {
+			return err
+		}
+		defer rows.Close()
+		var spn, dpn node
+		for rows.Next() {
+			if err := rows.Scan(&t); err != nil {
+				return err
+			}
+			if t.Inode == parentSrc {
+				spn = t
+			}
+			if t.Inode == parentDst {
+				dpn = t
+			}
+		}
+		if spn.Inode == 0 || dpn.Inode == 0 {
+			return syscall.ENOENT
+		}
+		if spn.Type != TypeDirectory || dpn.Type != TypeDirectory {
+			return syscall.ENOTDIR
+		}
 		var se = edge{Parent: parentSrc, Name: []byte(nameSrc)}
 		ok, err := s.ForUpdate().Get(&se)
 		if err != nil {
@@ -1355,28 +1467,6 @@ func (m *dbMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentDst
 			}
 			return nil
 		}
-		var spn = node{Inode: parentSrc}
-		ok, err = s.ForUpdate().Get(&spn)
-		if err != nil {
-			return err
-		}
-		if !ok {
-			return syscall.ENOENT
-		}
-		if spn.Type != TypeDirectory {
-			return syscall.ENOTDIR
-		}
-		var dpn = node{Inode: parentDst}
-		ok, err = s.ForUpdate().Get(&dpn)
-		if err != nil {
-			return err
-		}
-		if !ok {
-			return syscall.ENOENT
-		}
-		if dpn.Type != TypeDirectory {
-			return syscall.ENOTDIR
-		}
 		var sn = node{Inode: se.Inode}
 		ok, err = s.ForUpdate().Get(&sn)
 		if err != nil {
@@ -1418,11 +1508,15 @@ func (m *dbMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentDst
 			}
 			dn.Ctime = now
 			if exchange {
-				dn.Parent = parentSrc
-				if de.Type == TypeDirectory && parentSrc != parentDst {
-					dpn.Nlink--
-					spn.Nlink++
-					supdate, dupdate = true, true
+				if parentSrc != parentDst {
+					if de.Type == TypeDirectory {
+						dn.Parent = parentSrc
+						dpn.Nlink--
+						spn.Nlink++
+						supdate, dupdate = true, true
+					} else if dn.Parent > 0 {
+						dn.Parent = parentSrc
+					}
 				}
 			} else {
 				if de.Type == TypeDirectory {
@@ -1445,7 +1539,7 @@ func (m *dbMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentDst
 							opened = m.of.IsOpen(dn.Inode)
 						}
 						defer func() { m.of.InvalidateChunk(dino, 0xFFFFFFFE) }()
-					} else if dn.Nlink == 1 {
+					} else if dn.Parent > 0 {
 						dn.Parent = trash
 					}
 				}
@@ -1463,10 +1557,15 @@ func (m *dbMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentDst
 			return syscall.EACCES
 		}
 
-		if se.Type == TypeDirectory && parentSrc != parentDst {
-			spn.Nlink--
-			dpn.Nlink++
-			supdate, dupdate = true, true
+		if parentSrc != parentDst {
+			if se.Type == TypeDirectory {
+				sn.Parent = parentDst
+				spn.Nlink--
+				dpn.Nlink++
+				supdate, dupdate = true, true
+			} else if sn.Parent > 0 {
+				sn.Parent = parentDst
+			}
 		}
 		if supdate || time.Duration(now-spn.Mtime)*1e3 >= minUpdateTime {
 			spn.Mtime = now
@@ -1478,7 +1577,6 @@ func (m *dbMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentDst
 			dpn.Ctime = now
 			dupdate = true
 		}
-		sn.Parent = parentDst
 		sn.Ctime = now
 		if inode != nil {
 			*inode = sn.Inode
@@ -1495,6 +1593,14 @@ func (m *dbMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentDst
 			if _, err := s.Cols("ctime", "parent").Update(dn, &node{Inode: dino}); err != nil {
 				return err
 			}
+			if parentSrc != parentDst && dn.Parent == 0 {
+				if err = mustInsert(s, &linkParent{Inode: dino, Parent: parentSrc}); err != nil {
+					return err
+				}
+				if _, err = s.Limit(1, 0).Delete(&linkParent{Inode: dino, Parent: parentDst}); err != nil {
+					return err
+				}
+			}
 		} else {
 			if n, err := s.Delete(&edge{Parent: parentSrc, Name: se.Name}); err != nil {
 				return err
@@ -1506,14 +1612,27 @@ func (m *dbMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentDst
 					if _, err := s.Cols("ctime", "parent").Update(dn, &node{Inode: dino}); err != nil {
 						return err
 					}
-					name := fmt.Sprintf("%d-%d-%s", parentDst, dino, de.Name)
+					name := m.trashEntry(parentDst, dino, string(de.Name))
 					if err = mustInsert(s, &edge{Parent: trash, Name: []byte(name), Inode: dino, Type: de.Type}); err != nil {
 						return err
 					}
+					if dn.Parent == 0 {
+						if err = mustInsert(s, &linkParent{Inode: dino, Parent: trash}); err != nil {
+							return err
+						}
+						if _, err = s.Limit(1, 0).Delete(&linkParent{Inode: dino, Parent: parentDst}); err != nil {
+							return err
+						}
+					}
 				} else if de.Type != TypeDirectory && dn.Nlink > 0 {
-					if _, err := s.Cols("ctime", "nlink").Update(dn, &node{Inode: dino}); err != nil {
+					if _, err := s.Cols("ctime", "nlink", "parent").Update(dn, &node{Inode: dino}); err != nil {
 						return err
 					}
+					if dn.Parent == 0 {
+						if _, err = s.Limit(1, 0).Delete(&linkParent{Inode: dino, Parent: parentDst}); err != nil {
+							return err
+						}
+					}
 				} else {
 					if de.Type == TypeFile {
 						if opened {
@@ -1546,6 +1665,11 @@ func (m *dbMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentDst
 					if _, err := s.Delete(&xattr{Inode: dino}); err != nil {
 						return err
 					}
+					if dn.Parent == 0 {
+						if _, err = s.Delete(&linkParent{Inode: dino}); err != nil {
+							return err
+						}
+					}
 				}
 				if _, err := s.Delete(&edge{Parent: parentDst, Name: de.Name}); err != nil {
 					return err
@@ -1555,9 +1679,19 @@ func (m *dbMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentDst
 				return err
 			}
 		}
-		if parentDst != parentSrc && !isTrash(parentSrc) && supdate {
-			if _, err := s.Cols("nlink", "mtime", "ctime").Update(&spn, &node{Inode: parentSrc}); err != nil {
-				return err
+		if parentDst != parentSrc {
+			if !isTrash(parentSrc) && supdate {
+				if _, err := s.Cols("nlink", "mtime", "ctime").Update(&spn, &node{Inode: parentSrc}); err != nil {
+					return err
+				}
+			}
+			if sn.Parent == 0 {
+				if err = mustInsert(s, &linkParent{Inode: sn.Inode, Parent: parentDst}); err != nil {
+					return err
+				}
+				if _, err = s.Limit(1, 0).Delete(&linkParent{Inode: sn.Inode, Parent: parentSrc}); err != nil {
+					return err
+				}
 			}
 		}
 		if _, err := s.Cols("ctime", "parent").Update(&sn, &node{Inode: sn.Inode}); err != nil {
@@ -1620,6 +1754,8 @@ func (m *dbMeta) doLink(ctx Context, inode, parent Ino, name string, attr *Attr)
 			pn.Ctime = now
 			updateParent = true
 		}
+		oldParent := n.Parent
+		n.Parent = 0
 		n.Nlink++
 		n.Ctime = now
 
@@ -1631,7 +1767,15 @@ func (m *dbMeta) doLink(ctx Context, inode, parent Ino, name string, attr *Attr)
 				return err
 			}
 		}
-		if _, err := s.Cols("nlink", "ctime").Update(&n, node{Inode: inode}); err != nil {
+		if _, err := s.Cols("nlink", "ctime", "parent").Update(&n, node{Inode: inode}); err != nil {
+			return err
+		}
+		if oldParent > 0 {
+			if err = mustInsert(s, &linkParent{Inode: inode, Parent: oldParent}); err != nil {
+				return err
+			}
+		}
+		if err = mustInsert(s, &linkParent{Inode: inode, Parent: parent}); err != nil {
 			return err
 		}
 		if err == nil {
@@ -1641,117 +1785,130 @@ func (m *dbMeta) doLink(ctx Context, inode, parent Ino, name string, attr *Attr)
 	}))
 }
 
-func (m *dbMeta) doReaddir(ctx Context, inode Ino, plus uint8, entries *[]*Entry) syscall.Errno {
-	dbSession := m.db.Table(&edge{})
-	if plus != 0 {
-		dbSession = dbSession.Join("INNER", &node{}, "jfs_edge.inode=jfs_node.inode")
-	}
-	var nodes []namedNode
-	if err := dbSession.Find(&nodes, &edge{Parent: inode}); err != nil {
-		return errno(err)
-	}
-	for _, n := range nodes {
-		if len(n.Name) == 0 {
-			logger.Errorf("Corrupt entry with empty name: inode %d parent %d", n.Inode, inode)
-			continue
+func (m *dbMeta) doReaddir(ctx Context, inode Ino, plus uint8, entries *[]*Entry, limit int) syscall.Errno {
+	return errno(m.roTxn(func(s *xorm.Session) error {
+		s = s.Table(&edge{})
+		if plus != 0 {
+			s = s.Join("INNER", &node{}, "jfs_edge.inode=jfs_node.inode")
 		}
-		entry := &Entry{
-			Inode: n.Inode,
-			Name:  n.Name,
-			Attr:  &Attr{},
+		if limit > 0 {
+			s = s.Limit(limit, 0)
 		}
-		if plus != 0 {
-			m.parseAttr(&n.node, entry.Attr)
-		} else {
-			entry.Attr.Typ = n.Type
+		var nodes []namedNode
+		if err := s.Find(&nodes, &edge{Parent: inode}); err != nil {
+			return err
 		}
-		*entries = append(*entries, entry)
-	}
-	return 0
+		for _, n := range nodes {
+			if len(n.Name) == 0 {
+				logger.Errorf("Corrupt entry with empty name: inode %d parent %d", n.Inode, inode)
+				continue
+			}
+			entry := &Entry{
+				Inode: n.Inode,
+				Name:  n.Name,
+				Attr:  &Attr{},
+			}
+			if plus != 0 {
+				m.parseAttr(&n.node, entry.Attr)
+			} else {
+				entry.Attr.Typ = n.Type
+			}
+			*entries = append(*entries, entry)
+		}
+		return nil
+	}))
 }
 
 func (m *dbMeta) doCleanStaleSession(sid uint64) error {
 	var fail bool
 	// release locks
-	if _, err := m.db.Delete(flock{Sid: sid}); err != nil {
-		logger.Warnf("Delete flock with sid %d: %d", sid, err)
-		fail = true
-	}
-	if _, err := m.db.Delete(plock{Sid: sid}); err != nil {
-		logger.Warnf("Delete plock with sid %d: %d", sid, err)
+	err := m.txn(func(s *xorm.Session) error {
+		if _, err := s.Delete(flock{Sid: sid}); err != nil {
+			return err
+		}
+		if _, err := s.Delete(plock{Sid: sid}); err != nil {
+			return err
+		}
+		return nil
+	})
+	if err != nil {
+		logger.Warnf("Delete flock/plock with sid %d: %d", sid, err)
 		fail = true
 	}
 
-	var s = sustained{Sid: sid}
-	if rows, err := m.db.Rows(&s); err == nil {
-		var inodes []Ino
-		for rows.Next() {
-			if rows.Scan(&s) == nil {
-				inodes = append(inodes, s.Inode)
-			}
-		}
-		_ = rows.Close()
-		for _, inode := range inodes {
-			if err = m.doDeleteSustainedInode(sid, inode); err != nil {
-				logger.Warnf("Delete sustained inode %d of sid %d: %s", inode, sid, err)
+	var sus []sustained
+	err = m.roTxn(func(ses *xorm.Session) error {
+		sus = nil
+		return ses.Find(&sus, &sustained{Sid: sid})
+	})
+	if err != nil {
+		logger.Warnf("Scan sustained with sid %d: %s", sid, err)
+		fail = true
+	} else {
+		for _, su := range sus {
+			if err = m.doDeleteSustainedInode(sid, su.Inode); err != nil {
+				logger.Warnf("Delete sustained inode %d of sid %d: %s", su.Inode, sid, err)
 				fail = true
 			}
 		}
-	} else {
-		logger.Warnf("Scan sustained with sid %d: %s", sid, err)
-		fail = true
 	}
 
 	if fail {
 		return fmt.Errorf("failed to clean up sid %d", sid)
 	} else {
-		if n, err := m.db.Delete(&session2{Sid: sid}); err != nil {
+		return m.txn(func(s *xorm.Session) error {
+			if n, err := s.Delete(&session2{Sid: sid}); err != nil {
+				return err
+			} else if n == 1 {
+				return nil
+			}
+			ok, err := s.IsTableExist(&session{})
+			if err == nil && ok {
+				_, err = s.Delete(&session{Sid: sid})
+			}
 			return err
-		} else if n == 1 {
-			return nil
-		}
-		ok, err := m.db.IsTableExist(&session{})
-		if err == nil && ok {
-			_, err = m.db.Delete(&session{Sid: sid})
-		}
-		return err
+		})
 	}
 }
 
 func (m *dbMeta) doFindStaleSessions(limit int) ([]uint64, error) {
-	var s session2
-	rows, err := m.db.Where("Expire < ?", time.Now().Unix()).Limit(limit, 0).Rows(&s)
-	if err != nil {
-		return nil, err
-	}
 	var sids []uint64
-	for rows.Next() {
-		if rows.Scan(&s) == nil {
+	_ = m.roTxn(func(ses *xorm.Session) error {
+		var ss []session2
+		err := ses.Where("Expire < ?", time.Now().Unix()).Limit(limit, 0).Find(&ss)
+		if err != nil {
+			return err
+		}
+		for _, s := range ss {
 			sids = append(sids, s.Sid)
 		}
-	}
-	_ = rows.Close()
+		return nil
+	})
+
 	limit -= len(sids)
 	if limit <= 0 {
 		return sids, nil
 	}
 
-	if ok, err := m.db.IsTableExist(&session{}); err != nil {
-		logger.Errorf("Check legacy session table: %s", err)
-	} else if ok {
-		var ls session
-		rows, err = m.db.Where("Heartbeat < ?", time.Now().Add(time.Minute*-5).Unix()).Limit(limit, 0).Rows(&ls)
-		if err != nil {
-			logger.Errorf("Scan stale legacy sessions: %s", err)
-			return sids, nil
-		}
-		for rows.Next() {
-			if rows.Scan(&ls) == nil {
-				sids = append(sids, ls.Sid)
+	err := m.roTxn(func(ses *xorm.Session) error {
+		if ok, err := ses.IsTableExist(&session{}); err != nil {
+			return err
+		} else if ok {
+			var ls []session
+			err := ses.Where("Heartbeat < ?", time.Now().Add(time.Minute*-5).Unix()).Limit(limit, 0).Find(&ls)
+			if err != nil {
+				return err
+			}
+			for _, l := range ls {
+				sids = append(sids, l.Sid)
 			}
 		}
-		_ = rows.Close()
+		return nil
+	})
+	if err != nil {
+		logger.Errorf("Check legacy session table: %s", err)
 	}
+
 	return sids, nil
 }
 
@@ -1792,7 +1949,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
 }
@@ -1809,7 +1966,10 @@ func (m *dbMeta) Read(ctx Context, inode Ino, indx uint32, chunks *[]Slice) sysc
 	}
 	defer timeit(time.Now())
 	var c chunk
-	_, err := m.db.Where("inode=? and indx=?", inode, indx).Get(&c)
+	err := m.roTxn(func(s *xorm.Session) error {
+		_, err := s.Where("inode=? and indx=?", inode, indx).Get(&c)
+		return err
+	})
 	if err != nil {
 		return errno(err)
 	}
@@ -1902,16 +2062,22 @@ func (m *dbMeta) CopyFileRange(ctx Context, fin Ino, offIn uint64, fout Ino, off
 	var newSpace int64
 	defer func() { m.of.InvalidateChunk(fout, 0xFFFFFFFF) }()
 	err := m.txn(func(s *xorm.Session) error {
-		var nin, nout = node{Inode: fin}, node{Inode: fout}
-		ok, err := s.ForUpdate().Get(&nin)
+		var ts []node
+		err := s.ForUpdate().Where("inode IN (?,?)", fin, fout).Find(&ts)
 		if err != nil {
 			return err
 		}
-		ok2, err2 := s.ForUpdate().Get(&nout)
-		if err2 != nil {
-			return err2
+		var nin, nout node
+		for _, t := range ts {
+			if t.Inode == fin {
+				nin = t
+			}
+			if t.Inode == fout {
+				nout = t
+			}
 		}
-		if !ok || !ok2 {
+
+		if nin.Inode == 0 || nout.Inode == 0 {
 			return syscall.ENOENT
 		}
 		if nin.Type != TypeFile {
@@ -1940,21 +2106,15 @@ func (m *dbMeta) CopyFileRange(ctx Context, fin Ino, offIn uint64, fout Ino, off
 		nout.Mtime = now
 		nout.Ctime = now
 
-		var c chunk
-		rows, err := s.Where("inode = ? AND indx >= ? AND indx <= ?", fin, offIn/ChunkSize, (offIn+size)/ChunkSize).ForUpdate().Rows(&c)
+		var cs []chunk
+		err = s.Where("inode = ? AND indx >= ? AND indx <= ?", fin, offIn/ChunkSize, (offIn+size)/ChunkSize).ForUpdate().Find(&cs)
 		if err != nil {
 			return err
 		}
 		chunks := make(map[uint32][]*slice)
-		for rows.Next() {
-			err = scanChunk(rows, &c)
-			if err != nil {
-				_ = rows.Close()
-				return err
-			}
+		for _, c := range cs {
 			chunks[c.Indx] = readSliceBuf(c.Slices)
 		}
-		_ = rows.Close()
 
 		ses := s
 		updateSlices := func(indx uint32, buf []byte, chunkid uint64, size uint32) error {
@@ -2021,35 +2181,44 @@ func (m *dbMeta) CopyFileRange(ctx Context, fin Ino, offIn uint64, fout Ino, off
 	return errno(err)
 }
 
-func (m *dbMeta) doFindDeletedFiles(ts int64, limit int) (map[Ino]uint64, error) {
-	var d delfile
-	rows, err := m.db.Where("expire < ?", ts).Limit(limit, 0).Rows(&d)
-	if err != nil {
-		return nil, err
+func (m *dbMeta) doGetParents(ctx Context, inode Ino) map[Ino]int {
+	var rows []linkParent
+	if err := m.roTxn(func(s *xorm.Session) error {
+		rows = nil
+		return s.Find(&rows, &linkParent{Inode: inode})
+	}); err != nil {
+		logger.Warnf("Scan parent key of inode %d: %s", inode, err)
+		return nil
+	}
+	ps := make(map[Ino]int)
+	for _, row := range rows {
+		ps[row.Parent]++
 	}
+	return ps
+}
+
+func (m *dbMeta) doFindDeletedFiles(ts int64, limit int) (map[Ino]uint64, error) {
 	files := make(map[Ino]uint64)
-	for rows.Next() {
-		if rows.Scan(&d) == nil {
+	err := m.roTxn(func(s *xorm.Session) error {
+		var ds []delfile
+		err := s.Where("expire < ?", ts).Limit(limit, 0).Find(&ds)
+		if err != nil {
+			return err
+		}
+		for _, d := range ds {
 			files[d.Inode] = d.Length
 		}
-	}
-	_ = rows.Close()
-	return files, nil
+		return nil
+	})
+	return files, err
 }
 
 func (m *dbMeta) doCleanupSlices() {
-	var ck chunkRef
-	rows, err := m.db.Where("refs <= 0").Rows(&ck)
-	if err != nil {
-		return
-	}
 	var cks []chunkRef
-	for rows.Next() {
-		if rows.Scan(&ck) == nil {
-			cks = append(cks, ck)
-		}
-	}
-	_ = rows.Close()
+	_ = m.roTxn(func(s *xorm.Session) error {
+		cks = nil
+		return s.Where("refs <= 0").Find(&cks)
+	})
 	for _, ck := range cks {
 		m.deleteSlice(ck.Chunkid, ck.Size)
 	}
@@ -2068,6 +2237,9 @@ func (m *dbMeta) deleteChunk(inode Ino, indx uint32) error {
 		}
 		ss = readSliceBuf(c.Slices)
 		for _, sc := range ss {
+			if sc.chunkid == 0 {
+				continue
+			}
 			_, err = s.Exec("update jfs_chunk_ref set refs=refs-1 where chunkid=? AND size=?", sc.chunkid, sc.size)
 			if err != nil {
 				return err
@@ -2084,9 +2256,18 @@ func (m *dbMeta) deleteChunk(inode Ino, indx uint32) error {
 		return fmt.Errorf("delete slice from chunk %s fail: %s, retry later", inode, err)
 	}
 	for _, s := range ss {
+		if s.chunkid == 0 {
+			continue
+		}
 		var ref = chunkRef{Chunkid: s.chunkid}
-		ok, err := m.db.Get(&ref)
-		if err == nil && ok && ref.Refs <= 0 {
+		err := m.roTxn(func(s *xorm.Session) error {
+			ok, err := s.Get(&ref)
+			if err == nil && !ok {
+				err = errors.New("not found")
+			}
+			return err
+		})
+		if err == nil && ref.Refs <= 0 {
 			m.deleteSlice(s.chunkid, s.size)
 		}
 	}
@@ -2094,47 +2275,35 @@ func (m *dbMeta) deleteChunk(inode Ino, indx uint32) error {
 }
 
 func (m *dbMeta) doDeleteFileData(inode Ino, length uint64) {
-	var c = chunk{Inode: inode}
-	rows, err := m.db.Rows(&c)
-	if err != nil {
-		return
-	}
-	var indexes []uint32
-	for rows.Next() {
-		if scanChunk(rows, &c) == nil {
-			indexes = append(indexes, c.Indx)
-		}
-	}
-	_ = rows.Close()
-	for _, indx := range indexes {
-		err = m.deleteChunk(inode, indx)
+	var indexes []chunk
+	_ = m.roTxn(func(s *xorm.Session) error {
+		indexes = nil
+		return s.Cols("indx").Find(&indexes, &chunk{Inode: inode})
+	})
+	for _, c := range indexes {
+		err := m.deleteChunk(inode, c.Indx)
 		if err != nil {
-			logger.Warnf("deleteChunk inode %d index %d error: %s", inode, indx, err)
+			logger.Warnf("deleteChunk inode %d index %d error: %s", inode, c.Indx, err)
 			return
 		}
 	}
-	_, _ = m.db.Delete(delfile{Inode: inode})
+	_ = m.txn(func(s *xorm.Session) error {
+		_, err := s.Delete(delfile{Inode: inode})
+		return err
+	})
 }
 
 func (m *dbMeta) doCleanupDelayedSlices(edge int64, limit int) (int, error) {
-	var ds delslices
-	rows, err := m.db.Where("deleted < ?", edge).Limit(limit, 0).Rows(&ds)
-	if err != nil {
-		return 0, err
-	}
 	var result []delslices
-	for rows.Next() {
-		ds.Slices = ds.Slices[:0]
-		if rows.Scan(&ds) == nil {
-			result = append(result, ds)
-		}
-	}
-	_ = rows.Close()
+	_ = m.roTxn(func(s *xorm.Session) error {
+		result = nil
+		return s.Where("deleted < ?", edge).Limit(limit, 0).Find(&result)
+	})
 
 	var count int
 	var ss []Slice
 	for _, ds := range result {
-		if err = m.txn(func(ses *xorm.Session) error {
+		if err := m.txn(func(ses *xorm.Session) error {
 			ds := delslices{Chunkid: ds.Chunkid}
 			if ok, e := ses.ForUpdate().Get(&ds); e != nil {
 				return e
@@ -2159,8 +2328,14 @@ func (m *dbMeta) doCleanupDelayedSlices(edge int64, limit int) (int, error) {
 		}
 		for _, s := range ss {
 			var ref = chunkRef{Chunkid: s.Chunkid}
-			ok, e := m.db.Get(&ref)
-			if e == nil && ok && ref.Refs <= 0 {
+			err := m.roTxn(func(s *xorm.Session) error {
+				ok, err := s.Get(&ref)
+				if err == nil && !ok {
+					err = errors.New("not found")
+				}
+				return err
+			})
+			if err == nil && ref.Refs <= 0 {
 				m.deleteSlice(s.Chunkid, s.Size)
 				count++
 			}
@@ -2191,7 +2366,10 @@ func (m *dbMeta) compactChunk(inode Ino, indx uint32, force bool) {
 	}
 
 	var c chunk
-	_, err := m.db.Where("inode=? and indx=?", inode, indx).Get(&c)
+	err := m.roTxn(func(s *xorm.Session) error {
+		_, err := s.Where("inode=? and indx=?", inode, indx).Get(&c)
+		return err
+	})
 	if err != nil {
 		return
 	}
@@ -2249,6 +2427,9 @@ func (m *dbMeta) compactChunk(inode Ino, indx uint32, force bool) {
 			}
 		} else {
 			for _, s_ := range ss {
+				if s_.chunkid == 0 {
+					continue
+				}
 				if _, err := s.Exec("update jfs_chunk_ref set refs=refs-1 where chunkid=? and size=?", s_.chunkid, s_.size); err != nil {
 					return err
 				}
@@ -2259,7 +2440,12 @@ func (m *dbMeta) compactChunk(inode Ino, indx uint32, force bool) {
 	// there could be false-negative that the compaction is successful, double-check
 	if err != nil {
 		var c = chunkRef{Chunkid: chunkid}
-		ok, e := m.db.Get(&c)
+		var ok bool
+		e := m.roTxn(func(s *xorm.Session) error {
+			var e error
+			ok, e = s.Get(&c)
+			return e
+		})
 		if e == nil {
 			if ok {
 				err = nil
@@ -2277,8 +2463,16 @@ func (m *dbMeta) compactChunk(inode Ino, indx uint32, force bool) {
 		m.of.InvalidateChunk(inode, indx)
 		if !trash {
 			for _, s := range ss {
+				if s.chunkid == 0 {
+					continue
+				}
 				var ref = chunkRef{Chunkid: s.chunkid}
-				ok, err := m.db.Get(&ref)
+				var ok bool
+				err := m.roTxn(func(s *xorm.Session) error {
+					var e error
+					ok, e = s.Get(&ref)
+					return e
+				})
 				if err == nil && ok && ref.Refs <= 0 {
 					m.deleteSlice(s.chunkid, s.size)
 				}
@@ -2306,19 +2500,14 @@ func dup(b []byte) []byte {
 }
 
 func (m *dbMeta) CompactAll(ctx Context, bar *utils.Bar) syscall.Errno {
-	var c chunk
-	rows, err := m.db.Where("length(slices) >= ?", sliceBytes*2).Cols("inode", "indx").Rows(&c)
+	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)
+	})
 	if err != nil {
 		return errno(err)
 	}
-	var cs []chunk
-	for rows.Next() {
-		if scanChunk(rows, &c) == nil {
-			c.Slices = dup(c.Slices)
-			cs = append(cs, c)
-		}
-	}
-	_ = rows.Close()
 
 	bar.IncrTotal(int64(len(cs)))
 	for _, c := range cs {
@@ -2333,45 +2522,45 @@ func (m *dbMeta) ListSlices(ctx Context, slices map[Ino][]Slice, delete bool, sh
 	if delete {
 		m.doCleanupSlices()
 	}
-	var c chunk
-	rows, err := m.db.Rows(&c)
-	if err != nil {
-		return errno(err)
-	}
-	defer rows.Close()
-	for rows.Next() {
-		err = scanChunk(rows, &c)
+	err := m.roTxn(func(s *xorm.Session) error {
+		var cs []chunk
+		err := s.Find(&cs)
 		if err != nil {
-			return errno(err)
+			return err
 		}
-		ss := readSliceBuf(c.Slices)
-		for _, s := range ss {
-			if s.chunkid > 0 {
-				slices[c.Inode] = append(slices[c.Inode], Slice{Chunkid: s.chunkid, Size: s.size})
-				if showProgress != nil {
-					showProgress()
+		for _, c := range cs {
+			ss := readSliceBuf(c.Slices)
+			for _, s := range ss {
+				if s.chunkid > 0 {
+					slices[c.Inode] = append(slices[c.Inode], Slice{Chunkid: s.chunkid, Size: s.size})
+					if showProgress != nil {
+						showProgress()
+					}
 				}
 			}
 		}
+		return nil
+	})
+	if err != nil {
+		return errno(err)
 	}
 	if m.fmt.TrashDays == 0 {
 		return 0
 	}
 
-	if ok, err := m.db.IsTableExist(&delslices{}); err != nil {
-		return errno(err)
-	} else if !ok {
-		return 0
-	}
-	var ds delslices
-	rows2, err := m.db.Rows(&ds)
-	if err != nil {
-		return errno(err)
-	}
-	var ss []Slice
-	for rows2.Next() {
-		ds.Slices = ds.Slices[:0]
-		if rows2.Scan(&ds) == nil {
+	err = m.roTxn(func(s *xorm.Session) error {
+		if ok, err := s.IsTableExist(&delslices{}); err != nil {
+			return err
+		} else if !ok {
+			return nil
+		}
+		var dss []delslices
+		err := s.Find(&dss)
+		if err != nil {
+			return err
+		}
+		var ss []Slice
+		for _, ds := range dss {
 			ss = ss[:0]
 			m.decodeDelayedSlices(ds.Slices, &ss)
 			if showProgress != nil {
@@ -2385,79 +2574,71 @@ func (m *dbMeta) ListSlices(ctx Context, slices map[Ino][]Slice, delete bool, sh
 				}
 			}
 		}
-	}
-	_ = rows2.Close()
-	return 0
+		return nil
+	})
+	return errno(err)
 }
 
 func (m *dbMeta) GetXattr(ctx Context, inode Ino, name string, vbuff *[]byte) syscall.Errno {
 	defer timeit(time.Now())
 	inode = m.checkRoot(inode)
-	var x = xattr{Inode: inode, Name: name}
-	ok, err := m.db.Get(&x)
-	if err != nil {
-		return errno(err)
-	}
-	if !ok {
-		return ENOATTR
-	}
-	*vbuff = x.Value
-	return 0
+	return errno(m.roTxn(func(s *xorm.Session) error {
+		var x = xattr{Inode: inode, Name: name}
+		ok, err := s.Get(&x)
+		if err != nil {
+			return err
+		}
+		if !ok {
+			return ENOATTR
+		}
+		*vbuff = x.Value
+		return nil
+	}))
 }
 
 func (m *dbMeta) ListXattr(ctx Context, inode Ino, names *[]byte) syscall.Errno {
 	defer timeit(time.Now())
 	inode = m.checkRoot(inode)
-	var x = xattr{Inode: inode}
-	rows, err := m.db.Where("inode = ?", inode).Rows(&x)
-	if err != nil {
-		return errno(err)
-	}
-	defer rows.Close()
-	*names = nil
-	for rows.Next() {
-		err = rows.Scan(&x)
+	return errno(m.roTxn(func(s *xorm.Session) error {
+		var xs []xattr
+		err := s.Where("inode = ?", inode).Find(&xs, &xattr{Inode: inode})
 		if err != nil {
-			return errno(err)
+			return err
 		}
-		*names = append(*names, []byte(x.Name)...)
-		*names = append(*names, 0)
-	}
-	return 0
+		*names = nil
+		for _, x := range xs {
+			*names = append(*names, []byte(x.Name)...)
+			*names = append(*names, 0)
+		}
+		return nil
+	}))
 }
 
 func (m *dbMeta) doSetXattr(ctx Context, inode Ino, name string, value []byte, flags uint32) syscall.Errno {
 	return errno(m.txn(func(s *xorm.Session) error {
+		var k = &xattr{Inode: inode, Name: name}
 		var x = xattr{Inode: inode, Name: name, Value: value}
-		var err error
-		var n int64
+		ok, err := s.ForUpdate().Get(k)
+		if err != nil {
+			return err
+		}
+		k.Value = nil
 		switch flags {
 		case XattrCreate:
-			n, err = s.Insert(&x)
-			if err != nil || n == 0 {
-				err = syscall.EEXIST
+			if ok {
+				return syscall.EEXIST
 			}
+			err = mustInsert(s, &x)
 		case XattrReplace:
-			n, err = s.Update(&x, &xattr{Inode: inode, Name: name})
-			if err == nil && n == 0 {
-				err = ENOATTR
+			if !ok {
+				return ENOATTR
 			}
+			_, err = s.Update(&x, k)
 		default:
-			if m.db.DriverName() == "postgres" {
-				var r sql.Result
-				r, err = s.Exec("INSERT INTO jfs_xattr(inode, name, value) VALUES(?, ?, ?) "+
-					"ON CONFLICT (inode, name) DO UPDATE SET value=? ", inode, name, value, value)
-				if err == nil {
-					n, err = r.RowsAffected()
-				}
+			if ok {
+				_, err = s.Update(&x, k)
 			} else {
-				n, err = s.Insert(&x)
-				if err != nil || n == 0 {
-					n, err = s.Update(&x, &xattr{Inode: inode, Name: name})
-				}
-			}
-			if err == nil && n == 0 {
-				err = errors.New("not inserted or updated")
+				err = mustInsert(s, &x)
 			}
 		}
 		return err
@@ -2479,9 +2660,9 @@ func (m *dbMeta) doRemoveXattr(ctx Context, inode Ino, name string) syscall.Errn
 
 func (m *dbMeta) dumpEntry(inode Ino, typ uint8) (*DumpedEntry, error) {
 	e := &DumpedEntry{}
-	return e, m.txn(func(db *xorm.Session) error {
+	return e, m.roTxn(func(s *xorm.Session) error {
 		n := &node{Inode: inode}
-		ok, err := db.Get(n)
+		ok, err := s.Get(n)
 		if err != nil {
 			return err
 		}
@@ -2495,7 +2676,7 @@ func (m *dbMeta) dumpEntry(inode Ino, typ uint8) (*DumpedEntry, error) {
 		e.Attr.Inode = inode
 
 		var rows []xattr
-		if err = db.Find(&rows, &xattr{Inode: inode}); err != nil {
+		if err = s.Find(&rows, &xattr{Inode: inode}); err != nil {
 			return err
 		}
 		if len(rows) > 0 {
@@ -2510,7 +2691,7 @@ func (m *dbMeta) dumpEntry(inode Ino, typ uint8) (*DumpedEntry, error) {
 		if attr.Typ == TypeFile {
 			for indx := uint32(0); uint64(indx)*ChunkSize < attr.Length; indx++ {
 				c := &chunk{Inode: inode, Indx: indx}
-				if ok, err = db.Get(c); err != nil {
+				if ok, err = s.Get(c); err != nil {
 					return err
 				}
 				if !ok {
@@ -2526,7 +2707,7 @@ func (m *dbMeta) dumpEntry(inode Ino, typ uint8) (*DumpedEntry, error) {
 			}
 		} else if attr.Typ == TypeSymlink {
 			l := &symlink{Inode: inode}
-			ok, err = db.Get(l)
+			ok, err = s.Get(l)
 			if err != nil {
 				return err
 			}
@@ -2538,6 +2719,7 @@ func (m *dbMeta) dumpEntry(inode Ino, typ uint8) (*DumpedEntry, error) {
 		return nil
 	})
 }
+
 func (m *dbMeta) dumpEntryFast(inode Ino, typ uint8) *DumpedEntry {
 	e := &DumpedEntry{}
 	n, ok := m.snap.node[inode]
@@ -2598,7 +2780,11 @@ func (m *dbMeta) dumpDir(inode Ino, tree *DumpedEntry, bw *bufio.Writer, depth i
 	if m.snap != nil {
 		edges = m.snap.edges[inode]
 	} else {
-		if err := m.db.Find(&edges, &edge{Parent: inode}); err != nil {
+		err := m.roTxn(func(s *xorm.Session) error {
+			edges = nil
+			return s.Find(&edges, &edge{Parent: inode})
+		})
+		if err != nil {
 			return err
 		}
 	}
@@ -2648,67 +2834,70 @@ func (m *dbMeta) dumpDir(inode Ino, tree *DumpedEntry, bw *bufio.Writer, depth i
 }
 
 func (m *dbMeta) makeSnap(bar *utils.Bar) error {
-	m.snap = &dbSnap{
-		node:    make(map[Ino]*node),
-		symlink: make(map[Ino]*symlink),
-		xattr:   make(map[Ino][]*xattr),
-		edges:   make(map[Ino][]*edge),
-		chunk:   make(map[string]*chunk),
-	}
+	return m.roTxn(func(ses *xorm.Session) error {
+		snap := &dbSnap{
+			node:    make(map[Ino]*node),
+			symlink: make(map[Ino]*symlink),
+			xattr:   make(map[Ino][]*xattr),
+			edges:   make(map[Ino][]*edge),
+			chunk:   make(map[string]*chunk),
+		}
+
+		for _, s := range []interface{}{new(node), new(symlink), new(edge), new(xattr), new(chunk)} {
+			if count, err := ses.Count(s); err == nil {
+				bar.IncrTotal(count)
+			} else {
+				return err
+			}
+		}
 
-	for _, s := range []interface{}{new(node), new(symlink), new(edge), new(xattr), new(chunk)} {
-		if count, err := m.db.Count(s); err == nil {
-			bar.IncrTotal(count)
-		} else {
+		bufferSize := 10000
+		if err := ses.Table(&node{}).BufferSize(bufferSize).Iterate(new(node), func(idx int, bean interface{}) error {
+			n := bean.(*node)
+			snap.node[n.Inode] = n
+			bar.Increment()
+			return nil
+		}); err != nil {
 			return err
 		}
-	}
 
-	bufferSize := 10000
-	if err := m.db.BufferSize(bufferSize).Iterate(new(node), func(idx int, bean interface{}) error {
-		n := bean.(*node)
-		m.snap.node[n.Inode] = n
-		bar.Increment()
-		return nil
-	}); err != nil {
-		return err
-	}
-
-	if err := m.db.BufferSize(bufferSize).Iterate(new(symlink), func(idx int, bean interface{}) error {
-		s := bean.(*symlink)
-		m.snap.symlink[s.Inode] = s
-		bar.Increment()
-		return nil
-	}); err != nil {
-		return err
-	}
-	if err := m.db.BufferSize(bufferSize).Iterate(new(edge), func(idx int, bean interface{}) error {
-		e := bean.(*edge)
-		m.snap.edges[e.Parent] = append(m.snap.edges[e.Parent], e)
-		bar.Increment()
-		return nil
-	}); err != nil {
-		return err
-	}
+		if err := ses.Table(&symlink{}).BufferSize(bufferSize).Iterate(new(symlink), func(idx int, bean interface{}) error {
+			s := bean.(*symlink)
+			snap.symlink[s.Inode] = s
+			bar.Increment()
+			return nil
+		}); err != nil {
+			return err
+		}
+		if err := ses.Table(&edge{}).BufferSize(bufferSize).Iterate(new(edge), func(idx int, bean interface{}) error {
+			e := bean.(*edge)
+			snap.edges[e.Parent] = append(snap.edges[e.Parent], e)
+			bar.Increment()
+			return nil
+		}); err != nil {
+			return err
+		}
 
-	if err := m.db.BufferSize(bufferSize).Iterate(new(xattr), func(idx int, bean interface{}) error {
-		x := bean.(*xattr)
-		m.snap.xattr[x.Inode] = append(m.snap.xattr[x.Inode], x)
-		bar.Increment()
-		return nil
-	}); err != nil {
-		return err
-	}
+		if err := ses.Table(&xattr{}).BufferSize(bufferSize).Iterate(new(xattr), func(idx int, bean interface{}) error {
+			x := bean.(*xattr)
+			snap.xattr[x.Inode] = append(snap.xattr[x.Inode], x)
+			bar.Increment()
+			return nil
+		}); err != nil {
+			return err
+		}
 
-	if err := m.db.BufferSize(bufferSize).Iterate(new(chunk), func(idx int, bean interface{}) error {
-		c := bean.(*chunk)
-		m.snap.chunk[fmt.Sprintf("%d-%d", c.Inode, c.Indx)] = c
-		bar.Increment()
+		if err := ses.Table(&chunk{}).BufferSize(bufferSize).Iterate(new(chunk), func(idx int, bean interface{}) error {
+			c := bean.(*chunk)
+			snap.chunk[fmt.Sprintf("%d-%d", c.Inode, c.Indx)] = c
+			bar.Increment()
+			return nil
+		}); err != nil {
+			return err
+		}
+		m.snap = snap
 		return nil
-	}); err != nil {
-		return err
-	}
-	return nil
+	})
 }
 
 func (m *dbMeta) DumpMeta(w io.Writer, root Ino) (err error) {
@@ -2721,14 +2910,6 @@ func (m *dbMeta) DumpMeta(w io.Writer, root Ino) (err error) {
 			}
 		}
 	}()
-	var drows []delfile
-	if err := m.db.Find(&drows); err != nil {
-		return err
-	}
-	dels := make([]*DumpedDelFile, 0, len(drows))
-	for _, row := range drows {
-		dels = append(dels, &DumpedDelFile{row.Inode, row.Length, row.Expire})
-	}
 
 	progress := utils.NewProgress(false, false)
 	var tree, trash *DumpedEntry
@@ -2752,10 +2933,28 @@ func (m *dbMeta) DumpMeta(w io.Writer, root Ino) (err error) {
 	}
 	tree.Name = "FSTree"
 
+	var drows []delfile
 	var crows []counter
-	if err = m.db.Find(&crows); err != nil {
+	var srows []sustained
+	err = m.roTxn(func(s *xorm.Session) error {
+		drows = nil
+		if err := s.Find(&drows); err != nil {
+			return err
+		}
+		crows = nil
+		if err = s.Find(&crows); err != nil {
+			return err
+		}
+		srows = nil
+		return s.Find(&srows)
+	})
+	if err != nil {
 		return err
 	}
+	dels := make([]*DumpedDelFile, 0, len(drows))
+	for _, row := range drows {
+		dels = append(dels, &DumpedDelFile{row.Inode, row.Length, row.Expire})
+	}
 	counters := &DumpedCounters{}
 	for _, row := range crows {
 		switch row.Name {
@@ -2774,10 +2973,6 @@ func (m *dbMeta) DumpMeta(w io.Writer, root Ino) (err error) {
 		}
 	}
 
-	var srows []sustained
-	if err = m.db.Find(&srows); err != nil {
-		return err
-	}
 	ss := make(map[uint64][]Ino)
 	for _, row := range srows {
 		ss[row.Sid] = append(ss[row.Sid], row.Inode)
@@ -2837,18 +3032,25 @@ func (m *dbMeta) loadEntry(e *DumpedEntry, cs *DumpedCounters, refs map[uint64]*
 	logger.Debugf("Loading entry inode %d name %s", inode, unescape(e.Name))
 	attr := e.Attr
 	n := &node{
-		Inode:  inode,
-		Type:   typeFromString(attr.Type),
-		Mode:   attr.Mode,
-		Uid:    attr.Uid,
-		Gid:    attr.Gid,
-		Atime:  attr.Atime*1e6 + int64(attr.Atimensec)/1e3,
-		Mtime:  attr.Mtime*1e6 + int64(attr.Atimensec)/1e3,
-		Ctime:  attr.Ctime*1e6 + int64(attr.Atimensec)/1e3,
-		Nlink:  attr.Nlink,
-		Rdev:   attr.Rdev,
-		Parent: e.Parent,
-	} // Length not set
+		Inode: inode,
+		Type:  typeFromString(attr.Type),
+		Mode:  attr.Mode,
+		Uid:   attr.Uid,
+		Gid:   attr.Gid,
+		Atime: attr.Atime*1e6 + int64(attr.Atimensec)/1e3,
+		Mtime: attr.Mtime*1e6 + int64(attr.Atimensec)/1e3,
+		Ctime: attr.Ctime*1e6 + int64(attr.Atimensec)/1e3,
+		Nlink: attr.Nlink,
+		Rdev:  attr.Rdev,
+	} // Length and Parent not set
+	numParents := len(e.Parents)
+	if numParents == 1 {
+		n.Parent = e.Parents[0]
+	} else if numParents == 0 {
+		logger.Fatalf("No parent for inode: %d", inode)
+	} else if n.Type == TypeDirectory {
+		logger.Fatalf("Too many parents for directory inode %d: %v", inode, e.Parents)
+	}
 	if n.Type == TypeFile {
 		n.Length = attr.Length
 		bar.IncrTotal(int64(len(e.Chunks)))
@@ -2909,6 +3111,12 @@ func (m *dbMeta) loadEntry(e *DumpedEntry, cs *DumpedCounters, refs map[uint64]*
 			beansCh <- &xattr{Inode: inode, Name: x.Name, Value: unescape(x.Value)}
 		}
 	}
+	if numParents > 1 {
+		bar.IncrTotal(int64(len(e.Parents)))
+		for _, parent := range e.Parents {
+			beansCh <- &linkParent{Inode: inode, Parent: parent}
+		}
+	}
 	beansCh <- n
 }
 
@@ -2920,19 +3128,19 @@ func (m *dbMeta) LoadMeta(r io.Reader) error {
 	if len(tables) > 0 {
 		return fmt.Errorf("Database %s is not empty", m.Name())
 	}
-	if err = m.db.Sync2(new(setting), new(counter)); err != nil {
+	if err = m.syncTable(new(setting), new(counter)); err != nil {
 		return fmt.Errorf("create table setting, counter: %s", err)
 	}
-	if err = m.db.Sync2(new(node), new(edge), new(symlink), new(xattr)); err != nil {
-		return fmt.Errorf("create table node, edge, symlink, xattr: %s", err)
+	if err = m.syncTable(new(node), new(edge), new(linkParent), new(symlink), new(xattr)); err != nil {
+		return fmt.Errorf("create table node, edge, linkParent, symlink, xattr: %s", err)
 	}
-	if err = m.db.Sync2(new(chunk), new(chunkRef), new(delslices)); err != nil {
+	if err = m.syncTable(new(chunk), new(chunkRef), new(delslices)); err != nil {
 		return fmt.Errorf("create table chunk, chunk_ref, delslices: %s", err)
 	}
-	if err = m.db.Sync2(new(session2), new(sustained), new(delfile)); err != nil {
+	if err = m.syncTable(new(session2), new(sustained), new(delfile)); err != nil {
 		return fmt.Errorf("create table session2, sustaind, delfile: %s", err)
 	}
-	if err = m.db.Sync2(new(flock), new(plock)); err != nil {
+	if err = m.syncTable(new(flock), new(plock)); err != nil {
 		return fmt.Errorf("create table flock, plock: %s", err)
 	}
 
diff --git a/pkg/meta/sql_lock.go b/pkg/meta/sql_lock.go
index 4195f73e..c944c683 100644
--- a/pkg/meta/sql_lock.go
+++ b/pkg/meta/sql_lock.go
@@ -44,7 +44,8 @@ func (m *dbMeta) Flock(ctx Context, inode Ino, owner_ uint64, ltype uint32, bloc
 				}
 				return err
 			}
-			rows, err := s.ForUpdate().Rows(&flock{Inode: inode})
+			var fs []flock
+			err := s.ForUpdate().Find(&fs, &flock{Inode: inode})
 			if err != nil {
 				return err
 			}
@@ -53,13 +54,9 @@ func (m *dbMeta) Flock(ctx Context, inode Ino, owner_ uint64, ltype uint32, bloc
 				o   int64
 			}
 			var locks = make(map[key]flock)
-			var l flock
-			for rows.Next() {
-				if rows.Scan(&l) == nil {
-					locks[key{l.Sid, l.Owner}] = l
-				}
+			for _, l := range fs {
+				locks[key{l.Sid, l.Owner}] = l
 			}
-			_ = rows.Close()
 
 			if ltype == F_RDLCK {
 				for _, l := range locks {
@@ -186,7 +183,8 @@ func (m *dbMeta) Setlk(ctx Context, inode Ino, owner_ uint64, block bool, ltype
 				}
 				return err
 			}
-			rows, err := s.ForUpdate().Rows(&plock{Inode: inode})
+			var ps []plock
+			err := s.ForUpdate().Find(&ps, &plock{Inode: inode})
 			if err != nil {
 				return err
 			}
@@ -195,14 +193,9 @@ func (m *dbMeta) Setlk(ctx Context, inode Ino, owner_ uint64, block bool, ltype
 				owner int64
 			}
 			var locks = make(map[key][]byte)
-			var l plock
-			for rows.Next() {
-				l.Records = nil
-				if rows.Scan(&l) == nil {
-					locks[key{l.Sid, l.Owner}] = dup(l.Records)
-				}
+			for _, l := range ps {
+				locks[key{l.Sid, l.Owner}] = l.Records
 			}
-			_ = rows.Close()
 			lkey := key{m.sid, owner}
 			for k, d := range locks {
 				if k == lkey {
diff --git a/pkg/meta/tkv.go b/pkg/meta/tkv.go
index 82870701..9c0e8766 100644
--- a/pkg/meta/tkv.go
+++ b/pkg/meta/tkv.go
@@ -160,6 +160,7 @@ All keys:
   C...               counter
   AiiiiiiiiI         inode attribute
   AiiiiiiiiD...      dentry
+  AiiiiiiiiPiiiiiiii parents // for hard links
   AiiiiiiiiCnnnn     file chunks
   AiiiiiiiiS         symlink target
   AiiiiiiiiX...      extented attribute
@@ -167,7 +168,7 @@ All keys:
   Fiiiiiiii          Flocks
   Piiiiiiii          POSIX locks
   Kccccccccnnnn      slice refs
-  Lcccccccctttttttt  delayed slices
+  Lttttttttcccccccc  delayed slices
   SEssssssss         session expire time
   SHssssssss         session heartbeat // for legacy client
   SIssssssss         session info
@@ -182,6 +183,10 @@ func (m *kvMeta) entryKey(parent Ino, name string) []byte {
 	return m.fmtKey("A", parent, "D", name)
 }
 
+func (m *kvMeta) parentKey(inode, parent Ino) []byte {
+	return m.fmtKey("A", inode, "P", parent)
+}
+
 func (m *kvMeta) chunkKey(inode Ino, indx uint32) []byte {
 	return m.fmtKey("A", inode, "C", indx)
 }
@@ -559,7 +564,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 +577,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
 	}
@@ -1117,13 +1122,14 @@ func (m *kvMeta) doUnlink(ctx Context, parent Ino, name string) syscall.Errno {
 				if _type == TypeFile && attr.Nlink == 0 {
 					opened = m.of.IsOpen(inode)
 				}
-			} else if attr.Nlink == 1 {
+			} else if attr.Parent > 0 {
 				attr.Parent = trash
 			}
 		} else {
 			logger.Warnf("no attribute for inode %d (%d, %s)", inode, parent, name)
 			trash = 0
 		}
+
 		defer func() { m.of.InvalidateChunk(inode, 0xFFFFFFFE) }()
 		var updateParent bool
 		if !isTrash(parent) && now.Sub(time.Unix(pattr.Mtime, int64(pattr.Mtimensec))) >= minUpdateTime {
@@ -1141,7 +1147,13 @@ func (m *kvMeta) doUnlink(ctx Context, parent Ino, name string) syscall.Errno {
 		if attr.Nlink > 0 {
 			tx.set(m.inodeKey(inode), m.marshal(&attr))
 			if trash > 0 {
-				tx.set(m.entryKey(trash, fmt.Sprintf("%d-%d-%s", parent, inode, name)), buf)
+				tx.set(m.entryKey(trash, m.trashEntry(parent, inode, name)), buf)
+				if attr.Parent == 0 {
+					tx.incrBy(m.parentKey(inode, trash), 1)
+				}
+			}
+			if attr.Parent == 0 {
+				tx.incrBy(m.parentKey(inode, parent), -1)
 			}
 		} else {
 			switch _type {
@@ -1162,6 +1174,9 @@ func (m *kvMeta) doUnlink(ctx Context, parent Ino, name string) syscall.Errno {
 				newSpace, newInode = -align4K(0), -1
 			}
 			tx.dels(tx.scanKeys(m.xattrKey(inode, ""))...)
+			if attr.Parent == 0 {
+				tx.dels(tx.scanKeys(m.fmtKey("A", inode, "P"))...)
+			}
 		}
 		return nil
 	})
@@ -1234,7 +1249,7 @@ func (m *kvMeta) doRmdir(ctx Context, parent Ino, name string) syscall.Errno {
 		tx.dels(m.entryKey(parent, name))
 		if trash > 0 {
 			tx.set(m.inodeKey(inode), m.marshal(&attr))
-			tx.set(m.entryKey(trash, fmt.Sprintf("%d-%d-%s", parent, inode, name)), buf)
+			tx.set(m.entryKey(trash, m.trashEntry(parent, inode, name)), buf)
 		} else {
 			tx.dels(m.inodeKey(inode))
 			tx.dels(tx.scanKeys(m.xattrKey(inode, ""))...)
@@ -1316,11 +1331,15 @@ func (m *kvMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentDst
 			tattr.Ctime = now.Unix()
 			tattr.Ctimensec = uint32(now.Nanosecond())
 			if exchange {
-				tattr.Parent = parentSrc
-				if dtyp == TypeDirectory && parentSrc != parentDst {
-					dattr.Nlink--
-					sattr.Nlink++
-					supdate, dupdate = true, true
+				if parentSrc != parentDst {
+					if dtyp == TypeDirectory {
+						tattr.Parent = parentSrc
+						dattr.Nlink--
+						sattr.Nlink++
+						supdate, dupdate = true, true
+					} else if tattr.Parent > 0 {
+						tattr.Parent = parentSrc
+					}
 				}
 			} else {
 				if dtyp == TypeDirectory {
@@ -1339,7 +1358,7 @@ func (m *kvMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentDst
 							opened = m.of.IsOpen(dino)
 						}
 						defer func() { m.of.InvalidateChunk(dino, 0xFFFFFFFE) }()
-					} else if tattr.Nlink == 1 {
+					} else if tattr.Parent > 0 {
 						tattr.Parent = trash
 					}
 				}
@@ -1357,10 +1376,15 @@ func (m *kvMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentDst
 			return syscall.EACCES
 		}
 
-		if typ == TypeDirectory && parentSrc != parentDst {
-			sattr.Nlink--
-			dattr.Nlink++
-			supdate, dupdate = true, true
+		if parentSrc != parentDst {
+			if typ == TypeDirectory {
+				iattr.Parent = parentDst
+				sattr.Nlink--
+				dattr.Nlink++
+				supdate, dupdate = true, true
+			} else if iattr.Parent > 0 {
+				iattr.Parent = parentDst
+			}
 		}
 		if supdate || now.Sub(time.Unix(sattr.Mtime, int64(sattr.Mtimensec))) >= minUpdateTime {
 			sattr.Mtime = now.Unix()
@@ -1376,7 +1400,6 @@ func (m *kvMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentDst
 			dattr.Ctimensec = uint32(now.Nanosecond())
 			dupdate = true
 		}
-		iattr.Parent = parentDst
 		iattr.Ctime = now.Unix()
 		iattr.Ctimensec = uint32(now.Nanosecond())
 		if inode != nil {
@@ -1389,14 +1412,25 @@ func (m *kvMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentDst
 		if exchange { // dino > 0
 			tx.set(m.entryKey(parentSrc, nameSrc), dbuf)
 			tx.set(m.inodeKey(dino), m.marshal(&tattr))
+			if parentSrc != parentDst && tattr.Parent == 0 {
+				tx.incrBy(m.parentKey(dino, parentSrc), 1)
+				tx.incrBy(m.parentKey(dino, parentDst), -1)
+			}
 		} else {
 			tx.dels(m.entryKey(parentSrc, nameSrc))
 			if dino > 0 {
 				if trash > 0 {
 					tx.set(m.inodeKey(dino), m.marshal(&tattr))
-					tx.set(m.entryKey(trash, fmt.Sprintf("%d-%d-%s", parentDst, dino, nameDst)), dbuf)
+					tx.set(m.entryKey(trash, m.trashEntry(parentDst, dino, nameDst)), dbuf)
+					if tattr.Parent == 0 {
+						tx.incrBy(m.parentKey(dino, trash), 1)
+						tx.incrBy(m.parentKey(dino, parentDst), -1)
+					}
 				} else if dtyp != TypeDirectory && tattr.Nlink > 0 {
 					tx.set(m.inodeKey(dino), m.marshal(&tattr))
+					if tattr.Parent == 0 {
+						tx.incrBy(m.parentKey(dino, parentDst), -1)
+					}
 				} else {
 					if dtyp == TypeFile {
 						if opened {
@@ -1415,11 +1449,20 @@ func (m *kvMeta) doRename(ctx Context, parentSrc Ino, nameSrc string, parentDst
 						newSpace, newInode = -align4K(0), -1
 					}
 					tx.dels(tx.scanKeys(m.xattrKey(dino, ""))...)
+					if tattr.Parent == 0 {
+						tx.dels(tx.scanKeys(m.fmtKey("A", dino, "P"))...)
+					}
 				}
 			}
 		}
-		if parentDst != parentSrc && !isTrash(parentSrc) && supdate {
-			tx.set(m.inodeKey(parentSrc), m.marshal(&sattr))
+		if parentDst != parentSrc {
+			if !isTrash(parentSrc) && supdate {
+				tx.set(m.inodeKey(parentSrc), m.marshal(&sattr))
+			}
+			if iattr.Parent == 0 {
+				tx.incrBy(m.parentKey(ino, parentDst), 1)
+				tx.incrBy(m.parentKey(ino, parentSrc), -1)
+			}
 		}
 		tx.set(m.inodeKey(ino), m.marshal(&iattr))
 		tx.set(m.entryKey(parentDst, nameDst), buf)
@@ -1466,6 +1509,8 @@ func (m *kvMeta) doLink(ctx Context, inode, parent Ino, name string, attr *Attr)
 			pattr.Ctimensec = uint32(now.Nanosecond())
 			updateParent = true
 		}
+		oldParent := iattr.Parent
+		iattr.Parent = 0
 		iattr.Ctime = now.Unix()
 		iattr.Ctimensec = uint32(now.Nanosecond())
 		iattr.Nlink++
@@ -1474,6 +1519,10 @@ func (m *kvMeta) doLink(ctx Context, inode, parent Ino, name string, attr *Attr)
 			tx.set(m.inodeKey(parent), m.marshal(&pattr))
 		}
 		tx.set(m.inodeKey(inode), m.marshal(&iattr))
+		if oldParent > 0 {
+			tx.incrBy(m.parentKey(inode, oldParent), 1)
+		}
+		tx.incrBy(m.parentKey(inode, parent), 1)
 		if attr != nil {
 			*attr = iattr
 		}
@@ -1481,9 +1530,9 @@ func (m *kvMeta) doLink(ctx Context, inode, parent Ino, name string, attr *Attr)
 	}))
 }
 
-func (m *kvMeta) doReaddir(ctx Context, inode Ino, plus uint8, entries *[]*Entry) syscall.Errno {
+func (m *kvMeta) doReaddir(ctx Context, inode Ino, plus uint8, entries *[]*Entry, limit int) syscall.Errno {
 	// TODO: handle big directory
-	vals, err := m.scanValues(m.entryKey(inode, ""), -1, nil)
+	vals, err := m.scanValues(m.entryKey(inode, ""), limit, nil)
 	if err != nil {
 		return errno(err)
 	}
@@ -1575,7 +1624,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
 }
@@ -1762,6 +1811,22 @@ func (m *kvMeta) CopyFileRange(ctx Context, fin Ino, offIn uint64, fout Ino, off
 	return errno(err)
 }
 
+func (m *kvMeta) doGetParents(ctx Context, inode Ino) map[Ino]int {
+	vals, err := m.scanValues(m.fmtKey("A", inode, "P"), -1, func(k, v []byte) bool {
+		// parents: AiiiiiiiiPiiiiiiii
+		return len(k) == 1+8+1+8 && parseCounter(v) > 0
+	})
+	if err != nil {
+		logger.Warnf("Scan parent key of inode %d: %s", inode, err)
+		return nil
+	}
+	ps := make(map[Ino]int)
+	for k, v := range vals {
+		ps[m.decodeInode([]byte(k[10:]))] = int(parseCounter(v))
+	}
+	return ps
+}
+
 func (m *kvMeta) doFindDeletedFiles(ts int64, limit int) (map[Ino]uint64, error) {
 	klen := 1 + 8 + 8
 	vals, err := m.scanValues(m.fmtKey("D"), limit, func(k, v []byte) bool {
@@ -1806,8 +1871,7 @@ func (m *kvMeta) deleteChunk(inode Ino, indx uint32) error {
 		slices := readSliceBuf(buf)
 		tx.dels(key)
 		for _, s := range slices {
-			r := tx.incrBy(m.sliceKey(s.chunkid, s.size), -1)
-			if r < 0 {
+			if s.chunkid > 0 && tx.incrBy(m.sliceKey(s.chunkid, s.size), -1) < 0 {
 				todel = append(todel, s)
 			}
 		}
@@ -1851,26 +1915,26 @@ func (m *kvMeta) doDeleteFileData(inode Ino, length uint64) {
 }
 
 func (m *kvMeta) doCleanupDelayedSlices(edge int64, limit int) (int, error) {
-	// delayed slices: Lcccccccctttttttt
-	keys, err := m.scanKeys(m.fmtKey("L"))
-	if err != nil {
+	var keys [][]byte
+	if err := m.client.txn(func(tx kvTxn) error {
+		vals := tx.scanRange(m.delSliceKey(0, 0), m.delSliceKey(edge, 0))
+		for k := range vals {
+			if len(k) != 1+8+8 { // delayed slices: Lttttttttcccccccc
+				continue
+			}
+			keys = append(keys, []byte(k))
+		}
+		return nil
+	}); err != nil {
 		logger.Warnf("Scan delayed slices: %s", err)
 		return 0, err
 	}
 
-	klen := 1 + 8 + 8
 	var count int
 	var ss []Slice
 	var rs []int64
 	for _, key := range keys {
-		if len(key) != klen {
-			continue
-		}
-		if m.parseInt64(key[9:]) >= edge {
-			continue
-		}
-
-		if err = m.txn(func(tx kvTxn) error {
+		if err := m.txn(func(tx kvTxn) error {
 			buf := tx.get(key)
 			if len(buf) == 0 {
 				return nil
@@ -1971,7 +2035,9 @@ func (m *kvMeta) compactChunk(inode Ino, indx uint32, force bool) {
 			tx.set(m.delSliceKey(time.Now().Unix(), chunkid), dsbuf)
 		} else {
 			for _, s := range ss {
-				tx.incrBy(m.sliceKey(s.chunkid, s.size), -1)
+				if s.chunkid > 0 {
+					tx.incrBy(m.sliceKey(s.chunkid, s.size), -1)
+				}
 			}
 		}
 		return nil
@@ -1999,7 +2065,7 @@ func (m *kvMeta) compactChunk(inode Ino, indx uint32, force bool) {
 		if !trash {
 			var refs int64
 			for _, s := range ss {
-				if m.client.txn(func(tx kvTxn) error {
+				if s.chunkid > 0 && m.client.txn(func(tx kvTxn) error {
 					refs = tx.incrBy(m.sliceKey(s.chunkid, s.size), 0)
 					return nil
 				}) == nil && refs < 0 {
@@ -2074,7 +2140,7 @@ func (m *kvMeta) ListSlices(ctx Context, slices map[Ino][]Slice, delete bool, sh
 		return 0
 	}
 
-	// delayed slices: Lcccccccctttttttt
+	// delayed slices: Lttttttttcccccccc
 	klen = 1 + 8 + 8
 	result, err = m.scanValues(m.fmtKey("L"), -1, func(k, v []byte) bool {
 		return len(k) == klen
@@ -2152,14 +2218,15 @@ func (m *kvMeta) doSetXattr(ctx Context, inode Ino, name string, value []byte, f
 }
 
 func (m *kvMeta) doRemoveXattr(ctx Context, inode Ino, name string) syscall.Errno {
-	value, err := m.get(m.xattrKey(inode, name))
-	if err != nil {
-		return errno(err)
-	}
-	if value == nil {
-		return ENOATTR
-	}
-	return errno(m.deleteKeys(m.xattrKey(inode, name)))
+	key := m.xattrKey(inode, name)
+	return errno(m.txn(func(tx kvTxn) error {
+		value := tx.get(key)
+		if value == nil {
+			return ENOATTR
+		}
+		tx.dels(key)
+		return nil
+	}))
 }
 
 func (m *kvMeta) dumpEntry(inode Ino, typ uint8) (*DumpedEntry, error) {
@@ -2447,7 +2514,14 @@ func (m *kvMeta) loadEntry(e *DumpedEntry, cs *DumpedCounters, refs map[string]i
 	inode := e.Attr.Inode
 	logger.Debugf("Loading entry inode %d name %s", inode, unescape(e.Name))
 	attr := loadAttr(e.Attr)
-	attr.Parent = e.Parent
+	numParents := len(e.Parents)
+	if numParents == 1 {
+		attr.Parent = e.Parents[0]
+	} else if numParents == 0 {
+		logger.Fatalf("No parent for inode: %d", inode)
+	} else if attr.Typ == TypeDirectory {
+		logger.Fatalf("Too many parents for directory inode %d: %v", inode, e.Parents)
+	}
 	return m.txn(func(tx kvTxn) error {
 		if attr.Typ == TypeFile {
 			attr.Length = e.Attr.Length
@@ -2496,6 +2570,11 @@ func (m *kvMeta) loadEntry(e *DumpedEntry, cs *DumpedCounters, refs map[string]i
 		for _, x := range e.Xattrs {
 			tx.set(m.xattrKey(inode, x.Name), []byte(unescape(x.Value)))
 		}
+		if numParents > 1 {
+			for _, parent := range e.Parents {
+				tx.incrBy(m.parentKey(inode, parent), 1)
+			}
+		}
 		tx.set(m.inodeKey(inode), m.marshal(attr))
 		return nil
 	})
diff --git a/pkg/meta/utils.go b/pkg/meta/utils.go
index efd059d8..f928a394 100644
--- a/pkg/meta/utils.go
+++ b/pkg/meta/utils.go
@@ -285,7 +285,7 @@ func Remove(r Meta, ctx Context, parent Ino, name string) syscall.Errno {
 	if attr.Typ != TypeDirectory {
 		return r.Unlink(ctx, parent, name)
 	}
-	concurrent := make(chan int, 50)
+	concurrent := make(chan int, 10)
 	return emptyEntry(r, ctx, parent, name, inode, concurrent)
 }
 
diff --git a/pkg/object/encrypt.go b/pkg/object/encrypt.go
index a7f1d1c1..a8d4a84d 100644
--- a/pkg/object/encrypt.go
+++ b/pkg/object/encrypt.go
@@ -215,7 +215,7 @@ func (e *encrypted) Get(key string, off, limit int64) (io.ReadCloser, error) {
 	}
 	l := int64(len(plain))
 	if off > l {
-		return nil, io.EOF
+		off = l
 	}
 	if limit == -1 || off+limit > l {
 		limit = l - off
diff --git a/pkg/object/etcd.go b/pkg/object/etcd.go
index fbc918af..06db7cc1 100644
--- a/pkg/object/etcd.go
+++ b/pkg/object/etcd.go
@@ -54,6 +54,9 @@ func (c *etcdClient) Get(key string, off, limit int64) (io.ReadCloser, error) {
 	}
 	for _, pair := range resp.Kvs {
 		if string(pair.Key) == key {
+			if off > int64(len(pair.Value)) {
+				off = int64(len(pair.Value))
+			}
 			data := pair.Value[off:]
 			if limit > 0 && limit < int64(len(data)) {
 				data = data[:limit]
diff --git a/pkg/object/mem.go b/pkg/object/mem.go
index 16569bf4..925ef166 100644
--- a/pkg/object/mem.go
+++ b/pkg/object/mem.go
@@ -85,6 +85,9 @@ func (m *memStore) Get(key string, off, limit int64) (io.ReadCloser, error) {
 	if !ok {
 		return nil, errors.New("not exists")
 	}
+	if off > int64(len(d.data)) {
+		off = int64(len(d.data))
+	}
 	data := d.data[off:]
 	if limit > 0 && limit < int64(len(data)) {
 		data = data[:limit]
diff --git a/pkg/object/redis.go b/pkg/object/redis.go
index 73cb37f5..1dfa2525 100644
--- a/pkg/object/redis.go
+++ b/pkg/object/redis.go
@@ -53,6 +53,9 @@ func (r *redisStore) Get(key string, off, limit int64) (io.ReadCloser, error) {
 	if err != nil {
 		return nil, err
 	}
+	if off > int64(len(data)) {
+		off = int64(len(data))
+	}
 	data = data[off:]
 	if limit > 0 && limit < int64(len(data)) {
 		data = data[:limit]
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/object/sql.go b/pkg/object/sql.go
index ce058831..a86d1cfe 100644
--- a/pkg/object/sql.go
+++ b/pkg/object/sql.go
@@ -63,6 +63,9 @@ func (s *sqlStore) Get(key string, off, limit int64) (io.ReadCloser, error) {
 	if !ok {
 		return nil, errors.New("not found")
 	}
+	if off > int64(len(b.Data)) {
+		off = int64(len(b.Data))
+	}
 	data := b.Data[off:]
 	if limit > 0 && limit < int64(len(data)) {
 		data = data[:limit]
@@ -122,14 +125,14 @@ func (s *sqlStore) List(prefix, marker string, limit int64) ([]Object, error) {
 	if marker == "" {
 		marker = prefix
 	}
-	var b blob
-	rows, err := s.db.Where("key >= ?", marker).Limit(int(limit)).Cols("key", "size", "modified").Rows(&b)
+	var bs []blob
+	err := s.db.Where("key >= ?", marker).Limit(int(limit)).Cols("key", "size", "modified").OrderBy("key").Find(&bs)
 	if err != nil {
 		return nil, err
 	}
 	var objs []Object
-	for rows.Next() {
-		if rows.Scan(&b) == nil && strings.HasPrefix(b.Key, prefix) {
+	for _, b := range bs {
+		if strings.HasPrefix(b.Key, prefix) {
 			objs = append(objs, &obj{
 				key:   b.Key,
 				size:  b.Size,
diff --git a/pkg/object/tikv.go b/pkg/object/tikv.go
index 699a5f21..72958f84 100644
--- a/pkg/object/tikv.go
+++ b/pkg/object/tikv.go
@@ -54,6 +54,9 @@ func (t *tikv) Get(key string, off, limit int64) (io.ReadCloser, error) {
 	if err != nil {
 		return nil, err
 	}
+	if off > int64(len(d)) {
+		off = int64(len(d))
+	}
 	data := d[off:]
 	if limit > 0 && limit < int64(len(data)) {
 		data = data[:limit]
diff --git a/pkg/sync/sync.go b/pkg/sync/sync.go
index b4f2d389..64b1d77d 100644
--- a/pkg/sync/sync.go
+++ b/pkg/sync/sync.go
@@ -550,7 +550,7 @@ func worker(tasks <-chan object.Object, src, dst object.ObjectStorage, config *C
 			}
 			if err == nil {
 				if mc, ok := dst.(object.MtimeChanger); ok {
-					if err = mc.Chtimes(obj.Key(), obj.Mtime()); err != nil {
+					if err = mc.Chtimes(obj.Key(), obj.Mtime()); err != nil && !errors.Is(err, utils.ENOTSUP) {
 						logger.Warnf("Update mtime of %s: %s", key, err)
 					}
 				}
diff --git a/pkg/utils/progress.go b/pkg/utils/progress.go
index 71003444..68384861 100644
--- a/pkg/utils/progress.go
+++ b/pkg/utils/progress.go
@@ -139,6 +139,25 @@ func (p *Progress) AddByteSpinner(name string) *Bar {
 	return &Bar{Bar: b}
 }
 
+func (p *Progress) AddIoSpeedBar(name string, total int64) *Bar {
+	b := p.Progress.Add(0,
+		mpb.NewBarFiller(mpb.BarStyle()),
+		mpb.PrependDecorators(
+			decor.Name(name+": ", decor.WCSyncWidth),
+			decor.CountersKibiByte("% .1f / % .1f"),
+		),
+		mpb.AppendDecorators(
+			decor.OnComplete(decor.Percentage(decor.WC{W: 5}), "done"),
+			decor.OnComplete(
+				decor.AverageETA(decor.ET_STYLE_GO, decor.WC{W: 6}), "",
+			),
+		),
+	)
+	b.SetTotal(total, false)
+	p.bars = append(p.bars, b)
+	return &Bar{Bar: b}
+}
+
 func (p *Progress) AddDoubleSpinner(name string) *DoubleSpinner {
 	return &DoubleSpinner{
 		p.AddCountSpinner(name).Bar,
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())
 		}
diff --git a/pkg/vfs/fill.go b/pkg/vfs/fill.go
index 3f990a81..ce59ca4d 100644
--- a/pkg/vfs/fill.go
+++ b/pkg/vfs/fill.go
@@ -19,6 +19,7 @@ package vfs
 import (
 	"fmt"
 	"path"
+	"strconv"
 	"strings"
 	"sync"
 	"syscall"
@@ -74,8 +75,16 @@ func (v *VFS) fillCache(paths []string, concurrent int) {
 }
 
 func (v *VFS) resolve(p string, inode *Ino, attr *Attr) syscall.Errno {
-	p = strings.Trim(p, "/")
 	ctx := meta.Background
+	var inodePrefix = "inode:"
+	if strings.HasPrefix(p, inodePrefix) {
+		i, err := strconv.ParseUint(p[len(inodePrefix):], 10, 64)
+		if err == nil {
+			*inode = meta.Ino(i)
+			return v.Meta.GetAttr(ctx, meta.Ino(i), attr)
+		}
+	}
+	p = strings.Trim(p, "/")
 	err := v.Meta.Resolve(ctx, 1, p, inode, attr)
 	if err != syscall.ENOTSUP {
 		return err
diff --git a/pkg/vfs/vfs.go b/pkg/vfs/vfs.go
index 9086da6d..6c900bc2 100644
--- a/pkg/vfs/vfs.go
+++ b/pkg/vfs/vfs.go
@@ -35,7 +35,7 @@ type Context = LogContext
 
 const (
 	rootID      = 1
-	maxName     = 255
+	maxName     = meta.MaxName
 	maxSymlink  = 4096
 	maxFileSize = meta.ChunkSize << 31
 )
@@ -70,7 +70,7 @@ var (
 func (v *VFS) Lookup(ctx Context, parent Ino, name string) (entry *meta.Entry, err syscall.Errno) {
 	var inode Ino
 	var attr = &Attr{}
-	if parent == rootID {
+	if parent == rootID || name == ".control" {
 		n := getInternalNodeByName(name)
 		if n != nil {
 			entry = &meta.Entry{Inode: n.inode, Attr: n.attr}


Test patch:
diff --git a/cmd/bench_test.go b/cmd/bench_test.go
index f5f6e42b..bd90a691 100644
--- a/cmd/bench_test.go
+++ b/cmd/bench_test.go
@@ -31,3 +31,9 @@ func TestBench(t *testing.T) {
 		t.Fatalf("test bench failed: %s", err)
 	}
 }
+
+func TestBenchForObject(t *testing.T) {
+	if err := Main([]string{"", "objbench", testMountPoint + "/", "-p", "4"}); err != nil {
+		t.Fatalf("test bench failed: %s", err)
+	}
+}
diff --git a/pkg/meta/redis_test.go b/pkg/meta/redis_test.go
index ad9ddc97..5f081316 100644
--- a/pkg/meta/redis_test.go
+++ b/pkg/meta/redis_test.go
@@ -21,8 +21,10 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"reflect"
 	"runtime"
 	"strconv"
+	"strings"
 	"sync"
 	"syscall"
 	"testing"
@@ -65,6 +67,7 @@ func testMeta(t *testing.T, m Meta) {
 	testMetaClient(t, m)
 	testTruncateAndDelete(t, m)
 	testTrash(t, m)
+	testParents(t, m)
 	testRemove(t, m)
 	testStickyBit(t, m)
 	testLocks(t, m)
@@ -463,6 +466,9 @@ func testMetaClient(t *testing.T, m Meta) {
 	if st := m.SetXattr(ctx, inode, "a", []byte("v3"), XattrCreate); st != syscall.EEXIST {
 		t.Fatalf("setxattr: %s", st)
 	}
+	if st := m.SetXattr(ctx, inode, "a", []byte("v3"), XattrReplace); st != 0 {
+		t.Fatalf("setxattr: %s", st)
+	}
 	if st := m.SetXattr(ctx, inode, "a", []byte("v4"), XattrReplace); st != 0 {
 		t.Fatalf("setxattr: %s", st)
 	}
@@ -1109,7 +1115,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 +1126,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) {
@@ -1198,6 +1204,17 @@ func testTrash(t *testing.T, m Meta) {
 	if st := m.Unlink(ctx, 1, "d"); st != 0 {
 		t.Fatalf("unlink d: %s", st)
 	}
+	lname := strings.Repeat("f", MaxName)
+	if st := m.Create(ctx, 1, lname, 0644, 022, 0, &inode, attr); st != 0 {
+		t.Fatalf("create %s: %s", lname, st)
+	}
+	if st := m.Unlink(ctx, 1, lname); st != 0 {
+		t.Fatalf("unlink %s: %s", lname, st)
+	}
+	tname := fmt.Sprintf("1-%d-%s", inode, lname)[:MaxName]
+	if st := m.Lookup(ctx, TrashInode+1, tname, &inode, attr); st != 0 || attr.Parent != TrashInode+1 {
+		t.Fatalf("lookup subTrash/%s: %s, attr %+v", tname, st, attr)
+	}
 	var entries []*Entry
 	if st := m.Readdir(ctx, 1, 0, &entries); st != 0 {
 		t.Fatalf("readdir: %s", st)
@@ -1209,7 +1226,7 @@ func testTrash(t *testing.T, m Meta) {
 	if st := m.Readdir(ctx, TrashInode+1, 0, &entries); st != 0 {
 		t.Fatalf("readdir: %s", st)
 	}
-	if len(entries) != 7 {
+	if len(entries) != 8 {
 		t.Fatalf("entries: %d", len(entries))
 	}
 	ctx2 := NewContext(1000, 1, []uint32{1})
@@ -1235,6 +1252,79 @@ func testTrash(t *testing.T, m Meta) {
 	}
 }
 
+func testParents(t *testing.T, m Meta) {
+	if err := m.Init(Format{Name: "test"}, false); err != nil {
+		t.Fatalf("init: %s", err)
+	}
+	ctx := Background
+	var inode, parent Ino
+	var attr = &Attr{}
+	if st := m.Create(ctx, 1, "f", 0644, 022, 0, &inode, attr); st != 0 {
+		t.Fatalf("create f: %s", st)
+	}
+	if attr.Parent != 1 {
+		t.Fatalf("expect parent 1, but got %d", attr.Parent)
+	}
+	checkParents := func(inode Ino, expect map[Ino]int) {
+		if ps := m.GetParents(ctx, inode); ps == nil {
+			t.Fatalf("get parents of inode %d returns nil", inode)
+		} else if !reflect.DeepEqual(ps, expect) {
+			t.Fatalf("expect parents %v, but got %v", expect, ps)
+		}
+	}
+	checkParents(inode, map[Ino]int{1: 1})
+
+	if st := m.Link(ctx, inode, 1, "l1", attr); st != 0 {
+		t.Fatalf("link l1 -> f: %s", st)
+	}
+	if attr.Parent != 0 {
+		t.Fatalf("expect parent 0, but got %d", attr.Parent)
+	}
+	checkParents(inode, map[Ino]int{1: 2})
+
+	if st := m.Mkdir(ctx, 1, "d", 0755, 022, 0, &parent, attr); st != 0 {
+		t.Fatalf("mkdir d: %s", st)
+	}
+	if st := m.Link(ctx, inode, parent, "l2", attr); st != 0 {
+		t.Fatalf("link l2 -> f: %s", st)
+	}
+	if st := m.Link(ctx, inode, parent, "l3", attr); st != 0 {
+		t.Fatalf("link l3 -> f: %s", st)
+	}
+	checkParents(inode, map[Ino]int{1: 2, parent: 2})
+
+	if st := m.Unlink(ctx, 1, "f"); st != 0 {
+		t.Fatalf("unlink f: %s", st)
+	}
+	if st := m.Create(ctx, 1, "f2", 0644, 022, 0, &inode, attr); st != 0 {
+		t.Fatalf("create f2: %s", st)
+	}
+	if st := m.Rename(ctx, 1, "f2", 1, "l1", 0, &inode, attr); st != 0 {
+		t.Fatalf("rename f2 -> l1: %s", st)
+	}
+	if st := m.Lookup(ctx, parent, "l2", &inode, attr); st != 0 {
+		t.Fatalf("lookup d/l2: %s", st)
+	}
+	if attr.Parent != 0 {
+		t.Fatalf("expect parent 0, but got %d", attr.Parent)
+	}
+	if st := m.Unlink(ctx, parent, "l2"); st != 0 {
+		t.Fatalf("unlink d/l2: %s", st)
+	}
+	checkParents(inode, map[Ino]int{parent: 1})
+
+	// clean up
+	if st := m.Unlink(ctx, 1, "l1"); st != 0 {
+		t.Fatalf("unlink l1: %s", st)
+	}
+	if st := m.Unlink(ctx, parent, "l3"); st != 0 {
+		t.Fatalf("unlink d/l3: %s", st)
+	}
+	if st := m.Rmdir(ctx, 1, "d"); st != 0 {
+		t.Fatalf("rmdir d: %s", st)
+	}
+}
+
 func testOpenCache(t *testing.T, m Meta) {
 	ctx := Background
 	var inode Ino
@@ -1276,6 +1366,9 @@ func testReadOnly(t *testing.T, m Meta) {
 
 	var inode Ino
 	var attr = &Attr{}
+	if st := m.GetAttr(ctx, 1, attr); st != 0 {
+		t.Fatalf("getattr 1: %s", st)
+	}
 	if st := m.Mkdir(ctx, 1, "d", 0640, 022, 0, &inode, attr); st != syscall.EROFS {
 		t.Fatalf("mkdir d: %s", st)
 	}
diff --git a/pkg/object/object_storage_test.go b/pkg/object/object_storage_test.go
index 6a5b9f41..bebf3651 100644
--- a/pkg/object/object_storage_test.go
+++ b/pkg/object/object_storage_test.go
@@ -90,16 +90,22 @@ func testStorage(t *testing.T, s ObjectStorage) {
 	}
 
 	if d, e := get(s, "test", 0, -1); d != "hello" {
-		t.Fatalf("expect hello, but got %v, error:%s", d, e)
+		t.Fatalf("expect hello, but got %v, error: %s", d, e)
+	}
+	if d, e := get(s, "test", 2, -1); d != "llo" {
+		t.Fatalf("expect llo, but got %v, error: %s", d, e)
 	}
 	if d, e := get(s, "test", 2, 3); d != "llo" {
-		t.Fatalf("expect llo, but got %v, error:%s", d, e)
+		t.Fatalf("expect llo, but got %v, error: %s", d, e)
 	}
 	if d, e := get(s, "test", 2, 2); d != "ll" {
-		t.Fatalf("expect ll, but got %v, error:%s", d, e)
+		t.Fatalf("expect ll, but got %v, error: %s", d, e)
 	}
 	if d, e := get(s, "test", 4, 2); d != "o" {
-		t.Errorf("out-of-range get: 'o', but got %v, error:%s", len(d), e)
+		t.Errorf("out-of-range get: 'o', but got %v, error: %s", len(d), e)
+	}
+	if d, e := get(s, "test", 6, 2); d != "" {
+		t.Errorf("out-of-range get: '', but got %v, error: %s", len(d), e)
 	}
 	switch s.(*withPrefix).os.(type) {
 	case FileSystem: