Repository: juicedata/juicefs
Pull request #2047 (2022-05-18 11:54:58)

test upload small and big files with multiple threads to gateway.


Patch:
diff --git a/.github/workflows/gateway.yml b/.github/workflows/gateway.yml
index c9d841eb..034cd94e 100644
--- a/.github/workflows/gateway.yml
+++ b/.github/workflows/gateway.yml
@@ -19,7 +19,7 @@ on:
   workflow_dispatch:
 
 jobs:
-  gateway-test:
+  upload-small-files:
     runs-on: ubuntu-latest
     steps:
       - name: Checkout
@@ -31,12 +31,112 @@ jobs:
         run: |
           make juicefs
 
-      - name: Init 
+      - 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
 
-      - name: Test sync to gateway with multiple process
+          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
@@ -46,16 +146,9 @@ jobs:
           export MINIO_ROOT_PASSWORD=minioadmin
           ./juicefs format $meta_url $volume
           ./juicefs gateway $meta_url localhost:9000 &
-          ./juicefs sync ./  s3://minioadmin:minioadmin@localhost:9000/$volume/source_dir/ --no-https -p 20
+          ./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: 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() }}
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/pkg/meta/base.go b/pkg/meta/base.go
index a742407d..5031ef1e 100644
--- a/pkg/meta/base.go
+++ b/pkg/meta/base.go
@@ -70,6 +70,8 @@ type engine interface {
 	doRename(ctx Context, parentSrc Ino, nameSrc string, parentDst Ino, nameDst string, flags uint32, inode *Ino, attr *Attr) syscall.Errno
 	doSetXattr(ctx Context, inode Ino, name string, value []byte, flags uint32) syscall.Errno
 	doRemoveXattr(ctx Context, inode Ino, name string) syscall.Errno
+
+	GetSession(sid uint64, detail bool) (*Session, error)
 }
 
 type baseMeta struct {
@@ -83,6 +85,7 @@ type baseMeta struct {
 	of           *openfiles
 	removedFiles map[Ino]bool
 	compacting   map[uint64]bool
+	maxDeleting  chan struct{}
 	symlinks     *sync.Map
 	msgCallbacks *msgCallbacks
 	newSpace     int64
@@ -111,6 +114,7 @@ func newBaseMeta(conf *Config) baseMeta {
 		of:           newOpenFiles(conf.OpenCache),
 		removedFiles: make(map[Ino]bool),
 		compacting:   make(map[uint64]bool),
+		maxDeleting:  make(chan struct{}, 100),
 		symlinks:     &sync.Map{},
 		msgCallbacks: &msgCallbacks{
 			callbacks: make(map[uint32]MsgCallback),
@@ -231,7 +235,12 @@ func (m *baseMeta) CleanStaleSessions() {
 		return
 	}
 	for _, sid := range sids {
-		logger.Infof("clean up stale session %d: %s", sid, m.en.doCleanStaleSession(sid))
+		s, err := m.en.GetSession(sid, false)
+		if err != nil {
+			logger.Warnf("Get session info %d: %s", sid, err)
+			s = &Session{Sid: sid}
+		}
+		logger.Infof("clean up stale session %d %+v: %s", sid, s.SessionInfo, m.en.doCleanStaleSession(sid))
 	}
 }
 
@@ -782,9 +791,7 @@ func (m *baseMeta) Close(ctx Context, inode Ino) syscall.Errno {
 		defer m.Unlock()
 		if m.removedFiles[inode] {
 			delete(m.removedFiles, inode)
-			go func() {
-				_ = m.en.doDeleteSustainedInode(m.sid, inode)
-			}()
+			_ = m.en.doDeleteSustainedInode(m.sid, inode)
 		}
 	}
 	return 0
@@ -845,7 +852,19 @@ func (m *baseMeta) fileDeleted(opened bool, inode Ino, length uint64) {
 		m.removedFiles[inode] = true
 		m.Unlock()
 	} else {
-		go m.en.doDeleteFileData(inode, length)
+		m.tryDeleteFileData(inode, length)
+	}
+}
+
+func (m *baseMeta) tryDeleteFileData(inode Ino, length uint64) {
+	select {
+	case m.maxDeleting <- struct{}{}:
+		go func() {
+			m.en.doDeleteFileData(inode, length)
+			<-m.maxDeleting
+		}()
+	default:
+		// will be cleanup later
 	}
 }
 
diff --git a/pkg/meta/interface.go b/pkg/meta/interface.go
index 7664e50a..361851ea 100644
--- a/pkg/meta/interface.go
+++ b/pkg/meta/interface.go
@@ -249,7 +249,7 @@ type Meta interface {
 	// CloseSession does cleanup and close the session.
 	CloseSession() error
 	// GetSession retrieves information of session with sid
-	GetSession(sid uint64) (*Session, error)
+	GetSession(sid uint64, detail bool) (*Session, error)
 	// ListSessions returns all client sessions.
 	ListSessions() ([]*Session, error)
 	// CleanStaleSessions cleans up sessions not active for more than 5 minutes
diff --git a/pkg/meta/redis.go b/pkg/meta/redis.go
index befe405f..a1eeb6e8 100644
--- a/pkg/meta/redis.go
+++ b/pkg/meta/redis.go
@@ -408,7 +408,7 @@ func (m *redisMeta) getSession(sid string, detail bool) (*Session, error) {
 	return &s, nil
 }
 
-func (m *redisMeta) GetSession(sid uint64) (*Session, error) {
+func (m *redisMeta) GetSession(sid uint64, detail bool) (*Session, error) {
 	var legacy bool
 	key := strconv.FormatUint(sid, 10)
 	score, err := m.rdb.ZScore(Background, m.allSessions(), key).Result()
@@ -422,7 +422,7 @@ func (m *redisMeta) GetSession(sid uint64) (*Session, error) {
 	if err != nil {
 		return nil, err
 	}
-	s, err := m.getSession(key, true)
+	s, err := m.getSession(key, detail)
 	if err != nil {
 		return nil, err
 	}
@@ -1874,7 +1874,7 @@ func (m *redisMeta) doDeleteSustainedInode(sid uint64, inode Ino) error {
 	})
 	if err == nil {
 		m.updateStats(newSpace, -1)
-		go m.doDeleteFileData(inode, attr.Length)
+		m.tryDeleteFileData(inode, attr.Length)
 	}
 	return err
 }
diff --git a/pkg/meta/sql.go b/pkg/meta/sql.go
index c8e2d013..2fd41b97 100644
--- a/pkg/meta/sql.go
+++ b/pkg/meta/sql.go
@@ -420,12 +420,12 @@ func (m *dbMeta) getSession(row interface{}, detail bool) (*Session, error) {
 	return &s, nil
 }
 
-func (m *dbMeta) GetSession(sid uint64) (*Session, error) {
+func (m *dbMeta) GetSession(sid uint64, detail bool) (*Session, error) {
 	row := session2{Sid: sid}
 	if ok, err := m.db.Get(&row); err != nil {
 		return nil, err
 	} else if ok {
-		return m.getSession(&row, true)
+		return m.getSession(&row, detail)
 	}
 	if ok, err := m.db.IsTableExist(&session{}); err != nil {
 		return nil, err
@@ -434,7 +434,7 @@ func (m *dbMeta) GetSession(sid uint64) (*Session, error) {
 		if ok, err := m.db.Get(&row); err != nil {
 			return nil, err
 		} else if ok {
-			return m.getSession(&row, true)
+			return m.getSession(&row, detail)
 		}
 	}
 	return nil, fmt.Errorf("session not found: %d", sid)
@@ -1792,7 +1792,7 @@ func (m *dbMeta) doDeleteSustainedInode(sid uint64, inode Ino) error {
 	})
 	if err == nil {
 		m.updateStats(newSpace, -1)
-		go m.doDeleteFileData(inode, n.Length)
+		m.tryDeleteFileData(inode, n.Length)
 	}
 	return err
 }
diff --git a/pkg/meta/tkv.go b/pkg/meta/tkv.go
index 82870701..c21e754a 100644
--- a/pkg/meta/tkv.go
+++ b/pkg/meta/tkv.go
@@ -559,7 +559,7 @@ func (m *kvMeta) getSession(sid uint64, detail bool) (*Session, error) {
 	return &s, nil
 }
 
-func (m *kvMeta) GetSession(sid uint64) (*Session, error) {
+func (m *kvMeta) GetSession(sid uint64, detail bool) (*Session, error) {
 	var legacy bool
 	value, err := m.get(m.sessionKey(sid))
 	if err == nil && value == nil {
@@ -572,7 +572,7 @@ func (m *kvMeta) GetSession(sid uint64) (*Session, error) {
 	if value == nil {
 		return nil, fmt.Errorf("session not found: %d", sid)
 	}
-	s, err := m.getSession(sid, true)
+	s, err := m.getSession(sid, detail)
 	if err != nil {
 		return nil, err
 	}
@@ -1575,7 +1575,7 @@ func (m *kvMeta) doDeleteSustainedInode(sid uint64, inode Ino) error {
 	})
 	if err == nil {
 		m.updateStats(newSpace, -1)
-		go m.doDeleteFileData(inode, attr.Length)
+		m.tryDeleteFileData(inode, attr.Length)
 	}
 	return err
 }


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