localcache

cache.go

package localcache

import "context"

type Cache[K comparable, T any] interface {
	MGet(ctx context.Context, ids []K) ([]T, error)
}

type Storage[K comparable, T any] interface {
	Load(ctx context.Context, ids []K) (map[K][]byte, error)
	KeyFactory(id K) []byte

	// KeyRelease 如果希望使用 sync.Pool 可以在该方法进行归还操作
	KeyRelease([]byte)

	// Unmarshal  bytes 在多次调用中可能会被复用，请避免使用 unsafe 方法
	Unmarshal(bytes []byte) (T, error)
}

fastcache.go

package localcache

import (
	"context"
	"encoding/binary"

	"github.com/VictoriaMetrics/fastcache"
	"github.com/libp2p/go-buffer-pool"
)

const timeMaxLen = 2
const timeFilter = 0xffff

type fastcacheVariant[K comparable, T any] struct {
	local *fastcache.Cache

	loader Storage[K, T]

	Now      func() int64
	lifetime uint16
}

func WithFastCache[K comparable, T any](cap int, lifetime uint64, loader Storage[K, T]) Cache[K, T] {
	return &fastcacheVariant[K, T]{
		local:    fastcache.New(cap),
		lifetime: uint16(lifetime),
		loader:   loader,
	}
}

func (f *fastcacheVariant[K, T]) MGet(ctx context.Context, ids []K) ([]T, error) {
	var items = make([]T, 0, len(ids))
	var (
		missing []K
		err     error
	)
	items, missing, err = f.fromLocal(ctx, items, ids)
	if err != nil {
		return nil, err
	}
	if len(missing) == 0 {
		return items, nil
	}
	items, err = f.loadMore(ctx, items, missing)
	if err != nil {
		return items, err
	}
	return items, nil
}

func (f *fastcacheVariant[K, T]) fromLocal(ctx context.Context, items []T, ids []K) ([]T, []K, error) {
	var missing = make([]K, 0, len(ids))
	now := uint16(f.Now() % timeFilter)
	for _, id := range ids {
		buff := f.local.Get(nil, f.loader.KeyFactory(id))
		if len(buff) == 0 {
			missing = append(missing, id)
			continue
		}
		duration := now - binary.BigEndian.Uint16(buff[:timeMaxLen])
		if duration > f.lifetime {
			missing = append(missing, id)
			continue
		}
		val, err := f.loader.Unmarshal(buff[timeMaxLen:])
		if err != nil {
			return nil, nil, err
		}
		items = append(items, val)
	}
	return items, missing, nil
}

func (f *fastcacheVariant[K, T]) loadMore(ctx context.Context, items []T, missing []K) ([]T, error) {

	fromLoader, err := f.loader.Load(ctx, missing)
	if err != nil {
		return nil, err
	}
	now := uint64(f.Now())

	timeBuff := f.getBuff(timeMaxLen)
	binary.BigEndian.AppendUint16(timeBuff[:0], uint16(now%timeFilter))
	for _, id := range missing {
		content, ok := fromLoader[id]
		if !ok {
			//TODO:: not found?
			continue
		}
		buff := f.getBuff(len(content) + timeMaxLen)

		copy(buff, timeBuff)             //cache time
		copy(buff[timeMaxLen:], content) //cache content

		f.local.Set(f.loader.KeyFactory(id), buff)

		item, err := f.loader.Unmarshal(content)
		if err != nil {
			//TODO:: return?
			return nil, err
		}
		items = append(items, item)
		f.releaseBuff(buff)
	}
	return items, nil
}

func (f *fastcacheVariant[K, T]) getBuff(l int) []byte {
	return pool.Get(l)
}

func (f *fastcacheVariant[K, T]) releaseBuff(slice []byte) {
	pool.Put(slice)
}