(skywalking-banyandb) branch main updated: Remove redundant sync operation in fs (#399)

[wu...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

wusheng pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git


The following commit(s) were added to refs/heads/main by this push:
     new 71ad3c68 Remove redundant sync operation in fs (#399)
71ad3c68 is described below

commit 71ad3c681b2c155bfffdb869f137ff2fcfa73997
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Wed Mar 6 09:49:31 2024 +0800

    Remove redundant sync operation in fs (#399)
---
 .github/workflows/flaky-test.yml            |  4 ++--
 pkg/fs/local_file_system.go                 | 23 +++--------------------
 pkg/fs/local_file_system_nix.go             | 10 ++++++++++
 pkg/fs/local_file_system_windows.go         |  4 ++++
 test/stress/cases/istio/istio_suite_test.go | 13 +++++++------
 5 files changed, 26 insertions(+), 28 deletions(-)

diff --git a/.github/workflows/flaky-test.yml b/.github/workflows/flaky-test.yml
index 3420da1d..e789870c 100644
--- a/.github/workflows/flaky-test.yml
+++ b/.github/workflows/flaky-test.yml
@@ -28,5 +28,5 @@ jobs:
     if: github.repository == 'apache/skywalking-banyandb'
     uses: ./.github/workflows/test.yml
     with:
-      options: --repeat 49 --label-filter '(integration&&!slow)||banyand'
-      timeout-minutes: 40
+      options: --repeat 3 --label-filter '(integration&&!slow)||banyand'
+      timeout-minutes: 50
diff --git a/pkg/fs/local_file_system.go b/pkg/fs/local_file_system.go
index 7a8b1904..b943a6f1 100644
--- a/pkg/fs/local_file_system.go
+++ b/pkg/fs/local_file_system.go
@@ -100,7 +100,7 @@ func (fs *localFileSystem) mkdir(path string, permission Mode) {
 		fs.logger.Panic().Str("path", path).Err(err).Msg("failed to create directory")
 	}
 	parentDirPath := filepath.Dir(path)
-	fs.syncPath(parentDirPath)
+	fs.SyncPath(parentDirPath)
 }
 
 func (fs *localFileSystem) pathExist(path string) bool {
@@ -113,20 +113,6 @@ func (fs *localFileSystem) pathExist(path string) bool {
 	return true
 }
 
-func (fs *localFileSystem) syncPath(path string) {
-	d, err := os.Open(path)
-	if err != nil {
-		fs.logger.Panic().Str("path", path).Err(err).Msg("failed to open directory")
-	}
-	if err := d.Sync(); err != nil {
-		_ = d.Close()
-		fs.logger.Panic().Str("path", path).Err(err).Msg("failed to sync directory")
-	}
-	if err := d.Close(); err != nil {
-		fs.logger.Panic().Str("path", path).Err(err).Msg("ailed to sync directory")
-	}
-}
-
 func (fs *localFileSystem) ReadDir(dirname string) []DirEntry {
 	des, err := os.ReadDir(dirname)
 	if err != nil {
@@ -418,11 +404,8 @@ func (file *LocalFile) Path() string {
 
 // Close is used to close File.
 func (file *LocalFile) Close() error {
-	if err := file.file.Sync(); err != nil {
-		return &FileSystemError{
-			Code:    closeError,
-			Message: fmt.Sprintf("Close File error, directory name: %s, error message: %s", file.file.Name(), err),
-		}
+	if err := syncFile(file.file); err != nil {
+		return err
 	}
 
 	if err := file.file.Close(); err != nil {
diff --git a/pkg/fs/local_file_system_nix.go b/pkg/fs/local_file_system_nix.go
index 75db2649..4f2ebf0a 100644
--- a/pkg/fs/local_file_system_nix.go
+++ b/pkg/fs/local_file_system_nix.go
@@ -72,3 +72,13 @@ func (fs *localFileSystem) SyncPath(name string) {
 		fs.logger.Panic().Str("name", name).Err(err).Msg("failed to close file")
 	}
 }
+
+func syncFile(file *os.File) error {
+	if err := file.Sync(); err != nil {
+		return &FileSystemError{
+			Code:    flushError,
+			Message: fmt.Sprintf("Flush File error, directory name: %s, error message: %s", file.Name(), err),
+		}
+	}
+	return nil
+}
diff --git a/pkg/fs/local_file_system_windows.go b/pkg/fs/local_file_system_windows.go
index ab6d5c91..969babc6 100644
--- a/pkg/fs/local_file_system_windows.go
+++ b/pkg/fs/local_file_system_windows.go
@@ -59,3 +59,7 @@ func (*localFileSystem) CreateLockFile(name string, permission Mode) (File, erro
 }
 
 func (fs *localFileSystem) SyncPath(_ string) {}
+
+func syncFile(_ *os.File) error {
+	return nil
+}
diff --git a/test/stress/cases/istio/istio_suite_test.go b/test/stress/cases/istio/istio_suite_test.go
index a191a2fc..5377b72e 100644
--- a/test/stress/cases/istio/istio_suite_test.go
+++ b/test/stress/cases/istio/istio_suite_test.go
@@ -25,6 +25,7 @@ import (
 	"io"
 	"os"
 	"path/filepath"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -103,8 +104,8 @@ var _ = g.Describe("Istio", func() {
 
 		fmt.Printf("written %d items in %s\n", writtenCount, endTime.Sub(startTime).String())
 		fmt.Printf("throughput: %f items/s\n", float64(writtenCount)/endTime.Sub(startTime).Seconds())
-		fmt.Printf("throughput(kb/s) %f\n", float64(bc.bytesSent)/endTime.Sub(startTime).Seconds()/1024)
-		fmt.Printf("latency: %s\n", bc.totalLatency/time.Duration(writtenCount))
+		fmt.Printf("throughput(kb/s) %f\n", float64(bc.bytesSent.Load())/endTime.Sub(startTime).Seconds()/1024)
+		fmt.Printf("latency: %s\n", time.Duration(bc.totalLatency.Load())/time.Duration(writtenCount))
 	})
 })
 
@@ -263,8 +264,8 @@ func printDiskUsage(dir string, maxDepth, curDepth int) {
 }
 
 type clientCounter struct {
-	bytesSent    int
-	totalLatency time.Duration
+	bytesSent    atomic.Int64
+	totalLatency atomic.Int64
 }
 
 func (*clientCounter) HandleConn(context.Context, stats.ConnStats) {}
@@ -276,9 +277,9 @@ func (c *clientCounter) TagRPC(ctx context.Context, _ *stats.RPCTagInfo) context
 func (c *clientCounter) HandleRPC(_ context.Context, s stats.RPCStats) {
 	switch s := s.(type) {
 	case *stats.OutPayload:
-		c.bytesSent += s.WireLength
+		c.bytesSent.Add(int64(s.WireLength))
 	case *stats.End:
-		c.totalLatency += s.EndTime.Sub(s.BeginTime)
+		c.totalLatency.Add(int64(s.EndTime.Sub(s.BeginTime)))
 	}
 }