Repository: matrixorigin/matrixone
Pull request #12808 (2023-11-17 03:47:19)

fileservice: various fixes


## What type of PR is this?

- [ ] API-change
- [ ] BUG
- [x] Improvement
- [ ] Documentation
- [ ] Feature
- [ ] Test and CI
- [ ] Code Refactoring

## Which issue(s) this PR fixes:

issue #12779 

## What this PR does / why we need it:

fileservice: disk cache logging tunings

blockio: use PrefetchFile

fileservice: optimize DiskCache.writeFile

fileservice: refine S3FS.PrefetchFile tests

fileservice: add slow io lock wait warnings


Patch:
diff --git a/pkg/fileservice/disk_cache.go b/pkg/fileservice/disk_cache.go
index afb4fc5c2..d59de9771 100644
--- a/pkg/fileservice/disk_cache.go
+++ b/pkg/fileservice/disk_cache.go
@@ -171,7 +171,7 @@ func (d *DiskCache) Read(
 		if err := entry.ReadFromOSFile(file); err != nil {
 			// ignore error
 			numError++
-			logutil.Info("read disk cache error", zap.Any("error", err))
+			logutil.Warn("read disk cache error", zap.Any("error", err))
 			continue
 		}
 
@@ -279,39 +279,42 @@ func (d *DiskCache) writeFile(
 	err = os.MkdirAll(dir, 0755)
 	if err != nil {
 		numError++
-		logutil.Info("write disk cache error", zap.Any("error", err))
+		logutil.Warn("write disk cache error", zap.Any("error", err))
 		return false, nil // ignore error
 	}
 	f, err := os.CreateTemp(dir, "*")
 	if err != nil {
 		numError++
-		logutil.Info("write disk cache error", zap.Any("error", err))
+		logutil.Warn("write disk cache error", zap.Any("error", err))
 		return false, nil // ignore error
 	}
 	numCreate++
 	from, err := openReader(ctx)
 	if err != nil {
 		numError++
-		logutil.Info("write disk cache error", zap.Any("error", err))
+		logutil.Warn("write disk cache error", zap.Any("error", err))
 		return false, nil // ignore error
 	}
 	defer from.Close()
-	n, err := io.Copy(f, from)
+	var buf []byte
+	put := ioBufferPool.Get(&buf)
+	defer put.Put()
+	n, err := io.CopyBuffer(f, from, buf)
 	if err != nil {
 		f.Close()
 		os.Remove(f.Name())
 		numError++
-		logutil.Info("write disk cache error", zap.Any("error", err))
+		logutil.Warn("write disk cache error", zap.Any("error", err))
 		return false, nil // ignore error
 	}
 	if err := f.Close(); err != nil {
 		numError++
-		logutil.Info("write disk cache error", zap.Any("error", err))
+		logutil.Warn("write disk cache error", zap.Any("error", err))
 		return false, nil // ignore error
 	}
 	if err := os.Rename(f.Name(), diskPath); err != nil {
 		numError++
-		logutil.Info("write disk cache error", zap.Any("error", err))
+		logutil.Warn("write disk cache error", zap.Any("error", err))
 		return false, nil // ignore error
 	}
 
@@ -343,7 +346,7 @@ func (d *DiskCache) triggerEvict(ctx context.Context, bytesWritten int64) {
 		if newlyWrittenThreshold > 0 && d.evictState.newlyWritten >= newlyWrittenThreshold {
 			if d.evictState.timer.Stop() {
 				// evict immediately
-				logutil.Debug("disk cache: newly written bytes may excceeds eviction target, start immediately",
+				logutil.Info("disk cache: newly written bytes may excceeds eviction target, start immediately",
 					zap.Any("newly-written", d.evictState.newlyWritten),
 					zap.Any("newly-written-threshold", newlyWrittenThreshold),
 				)
@@ -420,8 +423,8 @@ func (d *DiskCache) evict(ctx context.Context) {
 
 	target := int64(float64(d.capacity) * d.evictTarget)
 
-	logutil.Info("disk cache eviction",
-		zap.Any("bytes", sumSize),
+	logutil.Info("disk cache eviction check",
+		zap.Any("used bytes", sumSize),
 		zap.Any("target", target),
 	)
 
@@ -432,9 +435,9 @@ func (d *DiskCache) evict(ctx context.Context) {
 			perfcounter.Update(ctx, func(set *perfcounter.CounterSet) {
 				set.FileService.Cache.Disk.Evict.Add(numDeleted)
 			}, d.perfCounterSets...)
-			logutil.Debug("disk cache: eviction finished",
-				zap.Any("files", numDeleted),
-				zap.Any("bytes", bytesDeleted),
+			logutil.Info("disk cache eviction finished",
+				zap.Any("deleted files", numDeleted),
+				zap.Any("deleted bytes", bytesDeleted),
 			)
 		}
 	}()
diff --git a/pkg/fileservice/io_lock.go b/pkg/fileservice/io_lock.go
index 702dbdd72..927b484cd 100644
--- a/pkg/fileservice/io_lock.go
+++ b/pkg/fileservice/io_lock.go
@@ -14,7 +14,13 @@
 
 package fileservice
 
-import "sync"
+import (
+	"sync"
+	"time"
+
+	"github.com/matrixorigin/matrixone/pkg/logutil"
+	"go.uber.org/zap"
+)
 
 type IOLockKey struct {
 	File string
@@ -24,20 +30,38 @@ type IOLocks struct {
 	locks sync.Map
 }
 
+var slowIOWaitDuration = time.Second * 10
+
 func (i *IOLocks) Lock(key IOLockKey) (unlock func(), wait func()) {
 	ch := make(chan struct{})
 	v, loaded := i.locks.LoadOrStore(key, ch)
+
 	if loaded {
 		// not locked
 		wait = func() {
-			<-v.(chan struct{})
+			t0 := time.Now()
+			for {
+				timer := time.NewTimer(slowIOWaitDuration)
+				select {
+				case <-v.(chan struct{}):
+					timer.Stop()
+					return
+				case <-timer.C:
+					logutil.Warn("wait io lock for too long",
+						zap.Any("wait", time.Since(t0)),
+						zap.Any("key", key),
+					)
+				}
+			}
 		}
 		return
 	}
+
 	// locked
 	unlock = func() {
 		i.locks.Delete(key)
 		close(ch)
 	}
+
 	return
 }
diff --git a/pkg/fileservice/s3_fs.go b/pkg/fileservice/s3_fs.go
index ebf3996e5..2f81401ea 100644
--- a/pkg/fileservice/s3_fs.go
+++ b/pkg/fileservice/s3_fs.go
@@ -251,6 +251,15 @@ func (s *S3FS) PrefetchFile(ctx context.Context, filePath string) error {
 		return err
 	}
 
+	unlock, wait := s.ioLocks.Lock(IOLockKey{
+		File: filePath,
+	})
+	if unlock != nil {
+		defer unlock()
+	} else {
+		wait()
+	}
+
 	// load to disk cache
 	if s.diskCache != nil {
 		if err := s.diskCache.SetFile(
@@ -271,33 +280,13 @@ func (s *S3FS) newReadCloser(ctx context.Context, filePath string) (io.ReadClose
 		return nil, err
 	}
 
-	unlock, wait := s.ioLocks.Lock(IOLockKey{
-		File: filePath,
-	})
-	if unlock == nil {
-		wait()
-	}
-
 	key := s.pathToKey(filePath)
 	r, err := s.storage.Read(ctx, key, ptrTo[int64](0), (*int64)(nil))
 	if err != nil {
-		if unlock != nil {
-			unlock()
-		}
 		return nil, err
 	}
 
-	return &readCloser{
-		closeFunc: func() error {
-			// unlock
-			if unlock != nil {
-				unlock()
-			}
-			// close
-			return r.Close()
-		},
-		r: r,
-	}, nil
+	return r, nil
 }
 
 func (s *S3FS) Write(ctx context.Context, vector IOVector) error {


Test patch:
diff --git a/pkg/fileservice/s3_fs_test.go b/pkg/fileservice/s3_fs_test.go
index ff7500c49..fd1d21c10 100644
--- a/pkg/fileservice/s3_fs_test.go
+++ b/pkg/fileservice/s3_fs_test.go
@@ -15,6 +15,7 @@
 package fileservice
 
 import (
+	"bytes"
 	"context"
 	"encoding/csv"
 	"encoding/json"
@@ -910,13 +911,15 @@ func TestS3PrefetchFile(t *testing.T) {
 	)
 	assert.Nil(t, err)
 
+	data := bytes.Repeat([]byte("abcd"), 2<<20)
+
 	// write file
 	err = fs.Write(ctx, IOVector{
 		FilePath: "foo/bar",
 		Entries: []IOEntry{
 			{
-				Size: 3,
-				Data: []byte("foo"),
+				Size: int64(len(data)),
+				Data: data,
 			},
 		},
 		Policy: SkipDiskCache | SkipMemoryCache,
@@ -928,19 +931,26 @@ func TestS3PrefetchFile(t *testing.T) {
 	err = fs.PrefetchFile(ctx, "foo/bar")
 	assert.Nil(t, err)
 	assert.Equal(t, int64(1), pcSet.FileService.Cache.Disk.WriteFile.Load())
+	err = fs.PrefetchFile(ctx, "foo/bar")
+	assert.Nil(t, err)
+	assert.Equal(t, int64(1), pcSet.FileService.Cache.Disk.WriteFile.Load())
 
 	// read
-	vec := &IOVector{
-		FilePath: "foo/bar",
-		Entries: []IOEntry{
-			{
-				Size: 3,
+	lastHit := int64(0)
+	for i := 1; i < len(data); i += len(data) / 1000 {
+		vec := &IOVector{
+			FilePath: "foo/bar",
+			Entries: []IOEntry{
+				{
+					Size: int64(i),
+				},
 			},
-		},
+		}
+		err = fs.Read(ctx, vec)
+		assert.Nil(t, err)
+		assert.Equal(t, data[:i], vec.Entries[0].Data)
+		assert.Equal(t, lastHit+1, pcSet.FileService.Cache.Disk.Hit.Load())
+		lastHit++
 	}
-	err = fs.Read(ctx, vec)
-	assert.Nil(t, err)
-	assert.Equal(t, []byte("foo"), vec.Entries[0].Data)
-	assert.Equal(t, int64(1), pcSet.FileService.Cache.Disk.Hit.Load())
 
 }