Golang 高性能 LocalCache：BigCache 设计与分析

type cacheShard struct {
	hashmap     map[uint64]uint32	// 上面的 map
	entries     queue.BytesQueue	// 数据存储
	lock        sync.RWMutex
	entryBuffer []byte
	onRemove    onRemoveCallback

	isVerbose    bool
	statsEnabled bool
	logger       Logger
	clock        clock
	lifeWindow   uint64

	hashmapStats map[uint64]uint32
	stats        Stats
}

const (
    timestampSizeInBytes = 8  // 存放时间戳
    hashSizeInBytes      = 8  // 存放 hash 值
    keySizeInBytes       = 2  // 存放 key 的长度
    headersSizeInBytes   = timestampSizeInBytes + hashSizeInBytes + keySizeInBytes // header 的长度
)

func wrapEntry(timestamp uint64, hash uint64, key string, entry []byte, buffer *[]byte) []byte {
    keyLength := len(key)  // key 的长度
    blobLength := len(entry) + headersSizeInBytes + keyLength

    if blobLength > len(*buffer) {
        *buffer = make([]byte, blobLength)
    }
    blob := *buffer
  	// 数据存储采用小端序 LittleEndian
    binary.LittleEndian.PutUint64(blob, timestamp)
    binary.LittleEndian.PutUint64(blob[timestampSizeInBytes:], hash)
    binary.LittleEndian.PutUint16(blob[timestampSizeInBytes+hashSizeInBytes:], uint16(keyLength))
    copy(blob[headersSizeInBytes:], key)
    copy(blob[headersSizeInBytes+keyLength:], entry)
    return blob[:blobLength]
}


func readEntry(data []byte) []byte {
	length := binary.LittleEndian.Uint16(data[timestampSizeInBytes+hashSizeInBytes:])
	return data[headersSizeInBytes+length:]
}

func readTimestampFromEntry(data []byte) uint64 {
	return binary.LittleEndian.Uint64(data)
}

func readKeyFromEntry(data []byte) string {
	length := binary.LittleEndian.Uint16(data[timestampSizeInBytes+hashSizeInBytes:])
	return string(data[headersSizeInBytes : headersSizeInBytes+length])
}

func readHashFromEntry(data []byte) uint64 {
	return binary.LittleEndian.Uint64(data[timestampSizeInBytes:])
}

func resetKeyFromEntry(data []byte) {
	binary.LittleEndian.PutUint64(data[timestampSizeInBytes:], 0)
}

func (q *BytesQueue) push(data []byte, len int) {
    // 得到 binary 的长度（uint64(len)）
    // 其中 len 为 pack 后的数据长度
    headerEntrySize := binary.PutUvarint(q.headerBuffer, uint64(len))
    // 复制长度
	q.copy(q.headerBuffer, headerEntrySize)
    // 复制 pack 后的数据
	q.copy(data, len)

	if q.tail > q.head {
		q.rightMargin = q.tail
	}
	if q.tail == q.head {
		q.full = true
	}

	q.count++
}

func (q *BytesQueue) copy(data []byte, len int) {
	q.tail += copy(q.array[q.tail:], data[:len])
}

type cacheShard struct {
	//...
	hashmap     map[uint64]uint32
	entries     queue.BytesQueue
	lock        sync.RWMutex
	entryBuffer []byte
	//...
}

// BytesQueue is a non-thread safe queue type of fifo based on bytes array.
// For every push operation index of entry is returned. It can be used to read the entry later
type BytesQueue struct {
	full            bool
	array           []byte      // 真正存储数据的地方
	capacity        int			// array 的容量
	maxCapacity     int			// array 可申请的最大容量
	head            int
	tail            int 		// 下次可以插入 item 的位置
	count           int			// 当前插入的 item 数量
	rightMargin     int
	headerBuffer    []byte		// 插入前做临时 buffer 所用（slice-copy）
	verbose         bool        // 打印 log 开关
	initialCapacity int			// BytesQueue 创建时，array 的初始容量
}

func (q *BytesQueue) copy(data []byte, len int) {
	q.tail += copy(q.array[q.tail:], data[:len])
}

//Pop reads the oldest entry from queue and moves head pointer to the next one
func (q *BytesQueue) Pop() ([]byte, error) {
	data, headerEntrySize, err := q.peek(q.head)
	if err != nil {
		return nil, err
	}
	size := len(data)

	q.head += headerEntrySize + size
	q.count--

	if q.head == q.rightMargin {
		q.head = leftMarginIndex
		if q.tail == q.rightMargin {
			q.tail = leftMarginIndex
		}
		q.rightMargin = q.tail
	}

	q.full = false

	return data, nil
}

type BigCache struct {
    shards       []*cacheShard	// 分片结构，每一个 cacheShard 中都有一个读写锁
    lifeWindow   uint64 // 数据过期时间
    clock        clock  // 获取数据存储时的时间戳
    hash         Hasher // 计算请求数据 key 的 hash 值（通用方法）
    config       Config
    shardMask    uint64 //len(shards) - 1，必须为 2 的次幂
    maxShardSize uint32 // 每个 shard 中 byte 数组的最大容量
    close        chan struct{}
}

type cacheShard struct {
	hashmap     map[uint64]uint32
	// 存储数据在 entries 中具体位置, key 为数据 key 的 hash，value 为位置（key -> value index）
	entries     queue.BytesQueue		// 数据最终存储的位置
	lock        sync.RWMutex
	entryBuffer []byte
	onRemove    onRemoveCallback	 // 删除数据时提供的回调函数

	isVerbose    bool
	statsEnabled bool
	logger       Logger
	clock        clock
	lifeWindow   uint64

	hashmapStats map[uint64]uint32
	stats        Stats
}

func (c *BigCache) Get(key string) ([]byte, error) {
	hashedKey := c.hash.Sum64(key) // string -> uint64
	shard := c.getShard(hashedKey) // 获取 key 所在的 shard 位置
	return shard.get(key, hashedKey)
}

// Set saves entry under the key
func (c *BigCache) Set(key string, entry []byte) error {
	hashedKey := c.hash.Sum64(key)
	shard := c.getShard(hashedKey)
	return shard.set(key, hashedKey, entry)
}

// Delete removes the key
func (c *BigCache) Delete(key string) error {
	hashedKey := c.hash.Sum64(key)
	shard := c.getShard(hashedKey)
	return shard.del(key, hashedKey)
}

//getShard 方法：通过 `hashedKey` 获取数据存储到哪一个 `shard 中
func (c *BigCache) getShard(hashedKey uint64) (shard *cacheShard) {
	return c.shards[hashedKey&c.shardMask]
}

// Set saves entry under the key
func (c *BigCache) Set(key string, entry []byte) error {
    hashedKey := c.hash.Sum64(key)
    shard := c.getShard(hashedKey)
    return shard.set(key, hashedKey, entry)
}

func (s *cacheShard) set(key string, hashedKey uint64, entry []byte) error {
    currentTimestamp := uint64(s.clock.epoch())
    s.lock.Lock()
    // 冲突检测
    if previousIndex := s.hashmap[hashedKey]; previousIndex != 0 {
        if previousEntry, err := s.entries.Get(int(previousIndex)); err == nil {
            resetKeyFromEntry(previousEntry)
        }
    }
    // 每次插入时，都会检查
    // 删除队列头部的过期数据
    if oldestEntry, err := s.entries.Peek(); err == nil {
        s.onEvict(oldestEntry, currentTimestamp, s.removeOldestEntry)
    }

    // pack key 和 value
    w := wrapEntry(currentTimestamp, hashedKey, key, entry, &s.entryBuffer)

    // 这里是个 for 循环
    for {
        if index, err := s.entries.Push(w); err == nil {
            s.hashmap[hashedKey] = uint32(index)    // 插入成功！最终 hashmap 中保存的是 index
            // 即插入前 q.tail 的位置
            s.lock.Unlock()
            return nil
        }
        // 在插入数据时，如果插入失败了，bigCache 会将 BytesQueue 的头部数据删除掉，尝试通过删除已插入的数据来解决因 BytesQueue 存储不足而插入失败的情况，然后重新尝试 push 操作。如果再次失败了，bigCache 会继续尝试，直到 bigCache 中已无可删除的数据了为止。
        if s.removeOldestEntry(NoSpace) != nil {
            s.lock.Unlock()
            return fmt.Errorf("entry is bigger than max shard size")
        }
    }
}

// Push copies entry at the end of queue and moves tail pointer. Allocates more space if needed.
// Returns index for pushed data or error if maximum size queue limit is reached.
func (q *BytesQueue) Push(data []byte) (int, error) {
	dataLen := len(data)
	headerEntrySize := getUvarintSize(uint32(dataLen))

	if !q.canInsertAfterTail(dataLen + headerEntrySize) {
		if q.canInsertBeforeHead(dataLen + headerEntrySize) {
			q.tail = leftMarginIndex
		} else if q.capacity+headerEntrySize+dataLen >= q.maxCapacity && q.maxCapacity > 0 {
			return -1, &queueError{"Full queue. Maximum size limit reached."}
		} else {
            // 尝试扩容，这里性能可能会受影响
			q.allocateAdditionalMemory(dataLen + headerEntrySize)
		}
	}

    // 插入的位置（保存插入前的位置）
	index := q.tail

	q.push(data, dataLen)

	// 返回 index，hashmap 保存的正是此位置
	return index, nil
}

// canInsertAfterTail returns true if it's possible to insert an entry of size of need after the tail of the queue
func (q *BytesQueue) canInsertAfterTail(need int) bool {
	if q.full {
		return false
	}
	if q.tail >= q.head {
		return q.capacity-q.tail >= need
    }
    // 当 q.tail 小于 q.head 时，分为下面几种情况
	// 1. there is exactly need bytes between head and tail, so we do not need
	// to reserve extra space for a potential empty entry when realloc this queue
	// 2. still have unused space between tail and head, then we must reserve
	// at least headerEntrySize bytes so we can put an empty entry
	return q.head-q.tail == need || q.head-q.tail >= need+minimumHeaderSize
}


// canInsertBeforeHead returns true if it's possible to insert an entry of size of need before the head of the queue
func (q *BytesQueue) canInsertBeforeHead(need int) bool {
	if q.full {
		return false
	}
	if q.tail >= q.head {
		return q.head-leftMarginIndex == need || q.head-leftMarginIndex >= need+minimumHeaderSize
	}
	return q.head-q.tail == need || q.head-q.tail >= need+minimumHeaderSize
}

func (s *cacheShard) removeOldestEntry(reason RemoveReason) error {
	oldest, err := s.entries.Pop()
	if err == nil {
		hash := readHashFromEntry(oldest)
		if hash == 0 {
			// entry has been explicitly deleted with resetKeyFromEntry, ignore
			return nil
		}
		delete(s.hashmap, hash)
		s.onRemove(oldest, reason)
		if s.statsEnabled {
			delete(s.hashmapStats, hash)
		}
		return nil
	}
	return err
}

// Delete removes the key
func (c *BigCache) Delete(key string) error {
 	hashedKey := c.hash.Sum64(key)
 	shard := c.getShard(hashedKey)
 	return shard.del(hashedKey)
}

func (s *cacheShard) del(key string, hashedKey uint64) error {
    s.lock.RLock()
    itemIndex := s.hashmap[hashedKey]
    if itemIndex == 0 {
        s.lock.RUnlock()
        s.delmiss()
        return notFound(key)
    }
    wrappedEntry, err := s.entries.Get(int(itemIndex))
    if err != nil {
        s.lock.RUnlock()
        s.delmiss()
        return err
    }
    s.lock.RUnlock()

    s.lock.Lock()
    {
        delete(s.hashmap, hashedKey)
        s.onRemove(wrappedEntry, Deleted)
        resetKeyFromEntry(wrappedEntry)
    }
    s.lock.Unlock()
    return nil
}

// Get reads entry for the key.
// It returns an ErrEntryNotFound when
// no entry exists for the given key.
func (c *BigCache) Get(key string) ([]byte, error) {
	hashedKey := c.hash.Sum64(key)
	shard := c.getShard(hashedKey)
	return shard.get(key, hashedKey)
}

//根据uint64的key值获取shard分片index
func (c *BigCache) getShard(hashedKey uint64) (shard *cacheShard) {
	return c.shards[hashedKey&c.shardMask]
}

func (s *cacheShard) get(key string, hashedKey uint64) ([]byte, error) {
	//shard加锁
    s.lock.RLock()

    //getWrappedEntry 先根据s.hashmap查询到itemIndex，itemIndex即是key保存在byteQueue的array中的位置
	wrappedEntry, err := s.getWrappedEntry(hashedKey)
	if err != nil {
		s.lock.RUnlock()
		return nil, err
	}

	//解码原始数据，先严格检查key是否与value中保存的是否一致
	if entryKey := readKeyFromEntry(wrappedEntry); key != entryKey {
		// hash发生了冲突，有概率发生
		s.lock.RUnlock()
		s.collision()
		if s.isVerbose {
			s.logger.Printf("Collision detected. Both %q and %q have the same hash %x", key, entryKey, hashedKey)
		}
		return nil, ErrEntryNotFound
	}
	//解码原始数据
	entry := readEntry(wrappedEntry)
	s.lock.RUnlock()

	//查询命中计数
	s.hit(hashedKey)

	return entry, nil
}

func (s *cacheShard) getWrappedEntry(hashedKey uint64) ([]byte, error) {
	//查询key在byteQueue中的位置
	itemIndex := s.hashmap[hashedKey]

    // 未找到
	if itemIndex == 0 {
		s.miss()
		return nil, ErrEntryNotFound
	}

    //itemIndex 是索引（注意itemIndex是uint32类型）
	wrappedEntry, err := s.entries.Get(int(itemIndex))
	if err != nil {
		s.miss()
		return nil, err
	}

	return wrappedEntry, err
}

// Get reads entry from index
func (q *BytesQueue) Get(index int) ([]byte, error) {
    data, _, err := q.peek(index)
    return data, err
}

func (q *BytesQueue) peek(index int) ([]byte, int, error) {
    if q.count == 0 {
        return nil, 0, &queueError{"Empty queue"}
    }

    if index <= 0 {
        return nil, 0, &queueError{"Index must be grater than zero. Invalid index."}
    }

    if index+headerEntrySize >= len(q.array) {
        return nil, 0, &queueError{"Index out of range"}
    }

    blockSize := int(binary.LittleEndian.Uint32(q.array[index : index+headerEntrySize]))
    return q.array[index+headerEntrySize : index+headerEntrySize+blockSize], blockSize, nil
}

...
if config.CleanWindow > 0 {
    go func() {
        ticker := time.NewTicker(config.CleanWindow)
        defer ticker.Stop()
        for {
            select {
            case t := <-ticker.C:
                cache.cleanUp(uint64(t.Unix()))
            case <-cache.close:
                return
            }
        }
    }()
}
...

func (c *BigCache) cleanUp(currentTimestamp uint64) {
	for _, shard := range c.shards {
		shard.cleanUp(currentTimestamp)
	}
}

func (s *cacheShard) cleanUp(currentTimestamp uint64) {
	s.lock.Lock()
	for {
		if oldestEntry, err := s.entries.Peek(); err != nil {
			break
		} else if evicted := s.onEvict(oldestEntry, currentTimestamp, s.removeOldestEntry); !evicted {
			break
		}
	}
	s.lock.Unlock()
}

// Peek reads the oldest entry from list without moving head pointer
func (q *BytesQueue) Peek() ([]byte, error) {
	data, _, err := q.peek(q.head)
	return data, err
}

func (q *BytesQueue) allocateAdditionalMemory(minimum int) {
    start := time.Now()
    if q.capacity < minimum {
        q.capacity += minimum
    }
    q.capacity = q.capacity * 2
    if q.capacity > q.maxCapacity && q.maxCapacity > 0 {
        q.capacity = q.maxCapacity
    }
    oldArray := q.array
    q.array = make([]byte, q.capacity)
    if leftMarginIndex != q.rightMargin {
        copy(q.array, oldArray[:q.rightMargin])
        if q.tail < q.head {
            emptyBlobLen := q.head - q.tail - headerEntrySize
            q.push(make([]byte, emptyBlobLen), emptyBlobLen)
            q.head = leftMarginIndex // head 指向了后插入的数据
            q.tail = q.rightMargin
        }
    }
}

type fnv64a struct{}

const (
	// offset64 FNVa offset basis. See https://en.wikipedia.org/wiki/Fowler–Noll–Vo_hash_function#FNV-1a_hash
	offset64 = 14695981039346656037
	// prime64 FNVa prime value. See https://en.wikipedia.org/wiki/Fowler–Noll–Vo_hash_function#FNV-1a_hash
	prime64 = 1099511628211
)

// Sum64 gets the string and returns its uint64 hash value.
func (f fnv64a) Sum64(key string) uint64 {
	var hash uint64 = offset64
	for i := 0; i <len(key); i++ {
		hash ^= uint64(key[i])
		hash *= prime64
	}

	return hash
}

//...
// 从序列化的存储中抽取 key 并比较
if entryKey := readKeyFromEntry(wrappedEntry); key != entryKey {
	if s.isVerbose {
		s.logger.Printf("Collision detected. Both %q and %q have the same hash %x", key, entryKey, hashedKey)
	}
	s.lock.RUnlock()
	s.collision()
	return nil, ErrEntryNotFound
}
//...