bug fix: s3fifo with ghost

cmd/mo-service/config.go

@@ -301,7 +301,7 @@ func (c *Config) setDefaultValue() error {
 
 func (c *Config) initMetaCache() {
 	if c.MetaCache.MemoryCapacity > 0 {
-		objectio.InitMetaCache(int64(c.MetaCache.MemoryCapacity))
+		objectio.InitMetaCache(int64(c.MetaCache.MemoryCapacity), c.MetaCache.DisableS3Fifo)
 	}
 }
 

pkg/embed/config.go

@@ -279,7 +279,7 @@ func (c *ServiceConfig) setDefaultValue() error {
 
 func (c *ServiceConfig) initMetaCache() {
 	if c.MetaCache.MemoryCapacity > 0 {
-		objectio.InitMetaCache(int64(c.MetaCache.MemoryCapacity))
+		objectio.InitMetaCache(int64(c.MetaCache.MemoryCapacity), c.MetaCache.DisableS3Fifo)
 	}
 }
 

pkg/fileservice/bytes.go

@@ -16,53 +16,75 @@ package fileservice
 
 import (
 	"context"
-	"sync/atomic"
+	"sync"
 
 	"github.com/matrixorigin/matrixone/pkg/common/malloc"
 	"github.com/matrixorigin/matrixone/pkg/fileservice/fscache"
 )
 
 type Bytes struct {
+	mu          sync.Mutex
 	bytes       []byte
 	deallocator malloc.Deallocator
-	deallocated uint32
-	_refs       atomic.Int32
-	refs        *atomic.Int32
+	refs        int32
 }
 
 func (b *Bytes) Size() int64 {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	if b.bytes == nil {
+		panic("fileservice.Bytes.Size() buffer already deallocated")
+	}
 	return int64(len(b.bytes))
 }
 
 func (b *Bytes) Bytes() []byte {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	if b.bytes == nil {
+		panic("fileservice.Bytes.Bytes() buffer already deallocated")
+	}
 	return b.bytes
 }
 
 func (b *Bytes) Slice(length int) fscache.Data {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	if b.bytes == nil {
+		panic("fileservice.Bytes.Slice() buffer already deallocated")
+	}
 	b.bytes = b.bytes[:length]
 	return b
 }
 
 func (b *Bytes) Retain() {
-	if b.refs != nil {
-		b.refs.Add(1)
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	if b.bytes == nil {
+		panic("fileservice.Bytes.Retain() buffer already deallocated")
 	}
+
+	b.refs += 1
 }
 
-func (b *Bytes) Release() {
-	if b.refs != nil {
-		if n := b.refs.Add(-1); n == 0 {
-			if b.deallocator != nil &&
-				atomic.CompareAndSwapUint32(&b.deallocated, 0, 1) {
-				b.deallocator.Deallocate(malloc.NoHints)
-			}
-		}
-	} else {
-		if b.deallocator != nil &&
-			atomic.CompareAndSwapUint32(&b.deallocated, 0, 1) {
+func (b *Bytes) Release() bool {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	if b.bytes == nil {
+		panic("fileservice.Bytes.Release() double free")
+	}
+
+	b.refs -= 1
+	if b.refs == 0 {
+		if b.deallocator != nil {
 			b.deallocator.Deallocate(malloc.NoHints)
+			b.bytes = nil
+			return true
 		}
 	}
+	return false
 }
 
 type bytesAllocator struct {
@@ -79,9 +101,8 @@ func (b *bytesAllocator) allocateCacheData(size int, hints malloc.Hints) fscache
 	bytes := &Bytes{
 		bytes:       slice,
 		deallocator: dec,
+		refs:        1,
 	}
-	bytes._refs.Store(1)
-	bytes.refs = &bytes._refs
 	return bytes
 }
 

pkg/fileservice/bytes_test.go

@@ -28,7 +28,28 @@ func TestBytes(t *testing.T) {
 		bs := &Bytes{
 			bytes:       bytes,
 			deallocator: deallocator,
+			refs:        1,
 		}
 		bs.Release()
 	})
 }
+
+func TestBytesPanic(t *testing.T) {
+	bytes, deallocator, err := ioAllocator().Allocate(42, malloc.NoHints)
+	assert.Nil(t, err)
+	bs := &Bytes{
+		bytes:       bytes,
+		deallocator: deallocator,
+		refs:        1,
+	}
+
+	released := bs.Release()
+	assert.Equal(t, released, true)
+
+	assert.Panics(t, func() { bs.Release() }, "Bytes.Release panic()")
+	assert.Panics(t, func() { bs.Size() }, "Bytes.Size panic()")
+	assert.Panics(t, func() { bs.Bytes() }, "Bytes.Bytes panic()")
+	assert.Panics(t, func() { bs.Slice(0) }, "Bytes.Slice panic()")
+	assert.Panics(t, func() { bs.Retain() }, "Bytes.Retain panic()")
+
+}

pkg/fileservice/cache.go

@@ -43,6 +43,7 @@ type CacheConfig struct {
 	RemoteCacheEnabled   bool           `toml:"remote-cache-enabled"`
 	RPC                  morpc.Config   `toml:"rpc"`
 	CheckOverlaps        bool           `toml:"check-overlaps"`
+	DisableS3Fifo        bool           `toml:"disable-s3fifo"`
 
 	QueryClient      client.QueryClient            `json:"-"`
 	KeyRouterFactory KeyRouterFactory[pb.CacheKey] `json:"-"`

pkg/fileservice/cache_test.go

@@ -32,7 +32,7 @@ func Test_readCache(t *testing.T) {
 	slowCacheReadThreshold = time.Second
 
 	size := int64(128)
-	m := NewMemCache(fscache.ConstCapacity(size), nil, nil, "")
+	m := NewMemCache(fscache.ConstCapacity(size), nil, nil, "", false)
 	defer m.Close(ctx)
 
 	ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*3)

pkg/fileservice/disk_cache.go

@@ -60,6 +60,7 @@ func NewDiskCache(
 	asyncLoad bool,
 	cacheDataAllocator CacheDataAllocator,
 	name string,
+	disable_s3fifo bool,
 ) (ret *DiskCache, err error) {
 
 	err = os.MkdirAll(path, 0755)
@@ -119,6 +120,7 @@ func NewDiskCache(
 					)
 				}
 			},
+			disable_s3fifo,
 		),
 	}
 	ret.updatingPaths.Cond = sync.NewCond(new(sync.Mutex))

pkg/fileservice/disk_cache_test.go

@@ -58,7 +58,7 @@ func TestDiskCache(t *testing.T) {
 	}
 
 	// new
-	cache, err := NewDiskCache(ctx, dir, fscache.ConstCapacity(1<<20), nil, false, nil, "")
+	cache, err := NewDiskCache(ctx, dir, fscache.ConstCapacity(1<<20), nil, false, nil, "", false)
 	assert.Nil(t, err)
 	defer cache.Close(ctx)
 
@@ -143,7 +143,7 @@ func TestDiskCache(t *testing.T) {
 	testRead(cache)
 
 	// new cache instance and read
-	cache, err = NewDiskCache(ctx, dir, fscache.ConstCapacity(1<<20), nil, false, nil, "")
+	cache, err = NewDiskCache(ctx, dir, fscache.ConstCapacity(1<<20), nil, false, nil, "", false)
 	assert.Nil(t, err)
 	defer cache.Close(ctx)
 
@@ -152,7 +152,7 @@ func TestDiskCache(t *testing.T) {
 	assert.Equal(t, 1, numWritten)
 
 	// new cache instance and update
-	cache, err = NewDiskCache(ctx, dir, fscache.ConstCapacity(1<<20), nil, false, nil, "")
+	cache, err = NewDiskCache(ctx, dir, fscache.ConstCapacity(1<<20), nil, false, nil, "", false)
 	assert.Nil(t, err)
 	defer cache.Close(ctx)
 
@@ -178,7 +178,7 @@ func TestDiskCacheWriteAgain(t *testing.T) {
 	})
 	ctx = OnDiskCacheEvict(ctx, nil) // for coverage
 
-	cache, err := NewDiskCache(ctx, dir, fscache.ConstCapacity(4096), nil, false, nil, "")
+	cache, err := NewDiskCache(ctx, dir, fscache.ConstCapacity(4096), nil, false, nil, "", false)
 	assert.Nil(t, err)
 	defer cache.Close(ctx)
 
@@ -243,7 +243,7 @@ func TestDiskCacheWriteAgain(t *testing.T) {
 func TestDiskCacheFileCache(t *testing.T) {
 	dir := t.TempDir()
 	ctx := context.Background()
-	cache, err := NewDiskCache(ctx, dir, fscache.ConstCapacity(1<<20), nil, false, nil, "")
+	cache, err := NewDiskCache(ctx, dir, fscache.ConstCapacity(1<<20), nil, false, nil, "", false)
 	assert.Nil(t, err)
 	defer cache.Close(ctx)
 
@@ -303,7 +303,7 @@ func TestDiskCacheDirSize(t *testing.T) {
 
 	dir := t.TempDir()
 	capacity := 1 << 20
-	cache, err := NewDiskCache(ctx, dir, fscache.ConstCapacity(int64(capacity)), nil, false, nil, "")
+	cache, err := NewDiskCache(ctx, dir, fscache.ConstCapacity(int64(capacity)), nil, false, nil, "", false)
 	assert.Nil(t, err)
 	defer cache.Close(ctx)
 
@@ -366,6 +366,7 @@ func benchmarkDiskCacheWriteThenRead(
 		false,
 		nil,
 		"",
+		false,
 	)
 	if err != nil {
 		b.Fatal(err)
@@ -461,6 +462,7 @@ func benchmarkDiskCacheReadRandomOffsetAtLargeFile(
 		false,
 		nil,
 		"",
+		false,
 	)
 	if err != nil {
 		b.Fatal(err)
@@ -532,6 +534,7 @@ func BenchmarkDiskCacheMultipleIOEntries(b *testing.B) {
 		false,
 		nil,
 		"",
+		false,
 	)
 	if err != nil {
 		b.Fatal(err)
@@ -603,7 +606,7 @@ func TestDiskCacheClearFiles(t *testing.T) {
 	assert.Nil(t, err)
 	numFiles := len(files)
 
-	_, err = NewDiskCache(ctx, dir, fscache.ConstCapacity(1<<20), nil, false, nil, "")
+	_, err = NewDiskCache(ctx, dir, fscache.ConstCapacity(1<<20), nil, false, nil, "", false)
 	assert.Nil(t, err)
 
 	files, err = filepath.Glob(filepath.Join(dir, "*"))
@@ -617,7 +620,7 @@ func TestDiskCacheClearFiles(t *testing.T) {
 func TestDiskCacheBadWrite(t *testing.T) {
 	dir := t.TempDir()
 	ctx := context.Background()
-	cache, err := NewDiskCache(ctx, dir, fscache.ConstCapacity(1<<20), nil, false, nil, "")
+	cache, err := NewDiskCache(ctx, dir, fscache.ConstCapacity(1<<20), nil, false, nil, "", false)
 	assert.Nil(t, err)
 
 	written, err := cache.writeFile(
@@ -652,6 +655,7 @@ func TestDiskCacheGlobalSizeHint(t *testing.T) {
 		false,
 		nil,
 		"test",
+		false,
 	)
 	assert.Nil(t, err)
 	defer cache.Close(ctx)
@@ -684,7 +688,7 @@ func TestDiskCacheGlobalSizeHint(t *testing.T) {
 
 func TestDiskCacheSetFromFile(t *testing.T) {
 	ctx := context.Background()
-	cache, err := NewDiskCache(ctx, t.TempDir(), fscache.ConstCapacity(1<<30), nil, false, nil, "")
+	cache, err := NewDiskCache(ctx, t.TempDir(), fscache.ConstCapacity(1<<30), nil, false, nil, "", false)
 	require.Nil(t, err)
 	defer cache.Close(ctx)
 
@@ -709,7 +713,7 @@ func TestDiskCacheSetFromFile(t *testing.T) {
 
 func TestDiskCacheQuotaExceeded(t *testing.T) {
 	ctx := context.Background()
-	cache, err := NewDiskCache(ctx, t.TempDir(), fscache.ConstCapacity(3), nil, false, nil, "")
+	cache, err := NewDiskCache(ctx, t.TempDir(), fscache.ConstCapacity(3), nil, false, nil, "", false)
 	require.Nil(t, err)
 	defer cache.Close(ctx)