mirror-seaweedfs/weed/util/chunk_cache/chunk_cache.go

package chunk_cache

import (
	"encoding/binary"
	"errors"
	"sync"

	"github.com/seaweedfs/seaweedfs/weed/glog"
	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
	"github.com/seaweedfs/seaweedfs/weed/storage/types"
)

var ErrorOutOfBounds = errors.New("attempt to read out of bounds")

const cacheHeaderSize = 8 // 4 bytes volumeId + 4 bytes cookie

// parseCacheHeader extracts volume ID and cookie from the 8-byte cache header
func parseCacheHeader(header []byte) (needle.VolumeId, types.Cookie) {
	volumeId := needle.VolumeId(binary.BigEndian.Uint32(header[0:4]))
	cookie := types.BytesToCookie(header[4:8])
	return volumeId, cookie
}

type ChunkCache interface {
	ReadChunkAt(data []byte, fileId string, offset uint64) (n int, err error)
	SetChunk(fileId string, data []byte)
	IsInCache(fileId string, lockNeeded bool) (answer bool)
	GetMaxFilePartSizeInCache() (answer uint64)
}

// a global cache for recently accessed file chunks
type TieredChunkCache struct {
	memCache   *ChunkCacheInMemory
	diskCaches []*OnDiskCacheLayer
	sync.RWMutex
	onDiskCacheSizeLimit0  uint64
	onDiskCacheSizeLimit1  uint64
	onDiskCacheSizeLimit2  uint64
	maxFilePartSizeInCache uint64
}

var _ ChunkCache = &TieredChunkCache{}

func NewTieredChunkCache(maxEntries int64, dir string, diskSizeInUnit int64, unitSize int64) *TieredChunkCache {

	c := &TieredChunkCache{
		memCache: NewChunkCacheInMemory(maxEntries),
	}
	c.diskCaches = make([]*OnDiskCacheLayer, 3)
	c.onDiskCacheSizeLimit0 = uint64(unitSize)
	c.onDiskCacheSizeLimit1 = 4 * c.onDiskCacheSizeLimit0
	c.onDiskCacheSizeLimit2 = 2 * c.onDiskCacheSizeLimit1
	c.diskCaches[0] = NewOnDiskCacheLayer(dir, "c0_2", diskSizeInUnit*unitSize/8, 2)
	c.diskCaches[1] = NewOnDiskCacheLayer(dir, "c1_3", diskSizeInUnit*unitSize/4+diskSizeInUnit*unitSize/8, 3)
	c.diskCaches[2] = NewOnDiskCacheLayer(dir, "c2_2", diskSizeInUnit*unitSize/2, 2)
	c.maxFilePartSizeInCache = uint64(unitSize*diskSizeInUnit) / 4

	return c
}

func (c *TieredChunkCache) GetMaxFilePartSizeInCache() (answer uint64) {
	if c == nil {
		return 0
	}
	return c.maxFilePartSizeInCache
}

func (c *TieredChunkCache) IsInCache(fileId string, lockNeeded bool) (answer bool) {
	if c == nil {
		return false
	}

	if lockNeeded {
		c.RLock()
		defer c.RUnlock()
	}

	item := c.memCache.cache.Get(fileId)
	if item != nil {
		glog.V(4).Infof("fileId %s is in memcache", fileId)
		return true
	}

	fid, err := needle.ParseFileIdFromString(fileId)
	if err != nil {
		glog.V(4).Infof("failed to parse file id %s", fileId)
		return false
	}

	// Check disk cache with volume ID and cookie validation
	for i, diskCacheLayer := range c.diskCaches {
		for k, v := range diskCacheLayer.diskCaches {
			if nv, ok := v.nm.Get(fid.Key); ok {
				// Read cache header to check volume ID and cookie
				headerBytes := make([]byte, cacheHeaderSize)
				if readN, readErr := v.DataBackend.ReadAt(headerBytes, nv.Offset.ToActualOffset()); readErr == nil && readN == cacheHeaderSize {
					// Parse volume ID and cookie from header
					storedVolumeId, storedCookie := parseCacheHeader(headerBytes)

					if storedVolumeId == fid.VolumeId && storedCookie == fid.Cookie {
						glog.V(4).Infof("fileId %s is in diskCaches[%d].volume[%d]", fileId, i, k)
						return true
					}
					glog.V(4).Infof("fileId %s header mismatch in diskCaches[%d].volume[%d]: stored volume %d cookie %x, expected volume %d cookie %x",
						fileId, i, k, storedVolumeId, storedCookie, fid.VolumeId, fid.Cookie)
				}
			}
		}
	}
	return false
}

func (c *TieredChunkCache) ReadChunkAt(data []byte, fileId string, offset uint64) (n int, err error) {
	if c == nil {
		return 0, nil
	}

	c.RLock()
	defer c.RUnlock()

	minSize := offset + uint64(len(data))
	if minSize <= c.onDiskCacheSizeLimit0 {
		n, err = c.memCache.readChunkAt(data, fileId, offset)
		if err != nil {
			glog.Errorf("failed to read from memcache: %s", err)
		}
		if n == int(len(data)) {
			return n, nil
		}
	}

	fid, err := needle.ParseFileIdFromString(fileId)
	if err != nil {
		glog.Errorf("failed to parse file id %s", fileId)
		return 0, nil
	}

	// Try disk caches with volume ID and cookie validation
	if minSize <= c.onDiskCacheSizeLimit0 {
		n, err = c.readChunkAtWithHeaderValidation(data, fid, offset, 0)
		if n == int(len(data)) {
			return
		}
	}
	if minSize <= c.onDiskCacheSizeLimit1 {
		n, err = c.readChunkAtWithHeaderValidation(data, fid, offset, 1)
		if n == int(len(data)) {
			return
		}
	}
	{
		n, err = c.readChunkAtWithHeaderValidation(data, fid, offset, 2)
		if n == int(len(data)) {
			return
		}
	}

	return 0, nil

}

func (c *TieredChunkCache) SetChunk(fileId string, data []byte) {
	if c == nil {
		return
	}
	c.Lock()
	defer c.Unlock()

	glog.V(4).Infof("SetChunk %s size %d\n", fileId, len(data))
	if c.IsInCache(fileId, false) {
		glog.V(4).Infof("fileId %s is already in cache", fileId)
		return
	}

	c.doSetChunk(fileId, data)
}

func (c *TieredChunkCache) doSetChunk(fileId string, data []byte) {
	// Disk cache format: [4-byte volumeId][4-byte cookie][chunk data]
	// Memory cache format: full fileId as key -> raw data (unchanged)

	// Memory cache unchanged - uses full fileId
	if len(data) <= int(c.onDiskCacheSizeLimit0) {
		c.memCache.SetChunk(fileId, data)
	}

	fid, err := needle.ParseFileIdFromString(fileId)
	if err != nil {
		glog.Errorf("failed to parse file id %s", fileId)
		return
	}

	// Prepend volume ID and cookie to data for disk cache
	// Format: [4-byte volumeId][4-byte cookie][chunk data]
	headerBytes := make([]byte, cacheHeaderSize)
	// Store volume ID in first 4 bytes using big-endian
	binary.BigEndian.PutUint32(headerBytes[0:4], uint32(fid.VolumeId))
	// Store cookie in next 4 bytes
	types.CookieToBytes(headerBytes[4:8], fid.Cookie)
	dataWithHeader := append(headerBytes, data...)

	// Store with volume ID and cookie header in disk cache
	if len(data) <= int(c.onDiskCacheSizeLimit0) {
		c.diskCaches[0].setChunk(fid.Key, dataWithHeader)
	} else if len(data) <= int(c.onDiskCacheSizeLimit1) {
		c.diskCaches[1].setChunk(fid.Key, dataWithHeader)
	} else {
		c.diskCaches[2].setChunk(fid.Key, dataWithHeader)
	}

}

func (c *TieredChunkCache) Shutdown() {
	if c == nil {
		return
	}
	c.Lock()
	defer c.Unlock()
	for _, diskCache := range c.diskCaches {
		diskCache.shutdown()
	}
}

// readChunkAtWithHeaderValidation reads from disk cache with volume ID and cookie validation
func (c *TieredChunkCache) readChunkAtWithHeaderValidation(data []byte, fid *needle.FileId, offset uint64, cacheLevel int) (n int, err error) {
	// Step 1: Read and validate header (volume ID + cookie)
	headerBuffer := make([]byte, cacheHeaderSize)
	headerRead, err := c.diskCaches[cacheLevel].readChunkAt(headerBuffer, fid.Key, 0)

	if err != nil {
		glog.V(4).Infof("failed to read header for %s from cache level %d: %v",
			fid.String(), cacheLevel, err)
		return 0, err
	}

	if headerRead < cacheHeaderSize {
		glog.V(4).Infof("insufficient data for header validation for %s from cache level %d: read %d bytes",
			fid.String(), cacheLevel, headerRead)
		return 0, nil // Not enough data for header - likely old format, treat as cache miss
	}

	// Parse volume ID and cookie from header
	storedVolumeId, storedCookie := parseCacheHeader(headerBuffer)

	// Validate both volume ID and cookie
	if storedVolumeId != fid.VolumeId || storedCookie != fid.Cookie {
		glog.V(4).Infof("header mismatch for %s in cache level %d: stored volume %d cookie %x, expected volume %d cookie %x (possible old format)",
			fid.String(), cacheLevel, storedVolumeId, storedCookie, fid.VolumeId, fid.Cookie)
		return 0, nil // Treat as cache miss - could be old format or actual mismatch
	}

	// Step 2: Read actual data from the offset position (after header)
	// The disk cache has format: [4-byte volumeId][4-byte cookie][actual chunk data]
	// We want to read from position: cacheHeaderSize + offset
	dataOffset := cacheHeaderSize + offset
	n, err = c.diskCaches[cacheLevel].readChunkAt(data, fid.Key, dataOffset)

	if err != nil {
		glog.V(4).Infof("failed to read data at offset %d for %s from cache level %d: %v",
			offset, fid.String(), cacheLevel, err)
		return 0, err
	}

	return n, nil
}

func min(x, y int) int {
	if x < y {
		return x
	}
	return y
}