modelRT/real-time-data/cache.go

// Package realtimedata define real time data operation functions
package realtimedata

import (
	"container/heap"
	"fmt"
	"sync"
	"time"
)

// DataItem define structure for storing data, insertion time, and last access time
type DataItem struct {
	Data       any
	InsertTime time.Time
	LastAccess time.Time
	Index      int
}

// priorityQueueItem implement heap.Interface
type priorityQueueItem struct {
	item *DataItem
	prio int64
}

// priorityQueue implement heap.Interface
type priorityQueue []*priorityQueueItem

func (pq priorityQueue) Len() int { return len(pq) }

func (pq priorityQueue) Less(i, j int) bool {
	// Note: use the UnixNano value of LastAccess time, with smaller values indicating earlier. Therefore, we need to perform reverse comparison to implement LRU like behavior
	return pq[i].prio < pq[j].prio
}

func (pq priorityQueue) Swap(i, j int) {
	pq[i], pq[j] = pq[j], pq[i]
	pq[i].item.Index = i
	pq[j].item.Index = j
}

func (pq *priorityQueue) Push(x any) {
	n := len(*pq)
	queueItem := x.(*priorityQueueItem)
	queueItem.item.Index = n
	*pq = append(*pq, queueItem)
}

func (pq *priorityQueue) Pop() any {
	old := *pq
	n := len(old)
	queueItem := old[n-1]
	queueItem.item.Index = -1
	*pq = old[0 : n-1]
	return queueItem
}

// update updates the priority of an item in the priority queue.
func (pq *priorityQueue) update(item *DataItem, newPrio int64) {
	item.LastAccess = time.Unix(0, newPrio)
	pqItem := (*pq)[item.Index]
	pqItem.prio = newPrio
	heap.Fix(pq, item.Index)
}

type TimeCache struct {
	mu       sync.Mutex
	capacity int
	items    map[any]*DataItem
	pq       priorityQueue
}

// NewTimeCache 创建一个新的 TimeCache 实例
func NewTimeCache(capacity int) *TimeCache {
	return &TimeCache{
		capacity: capacity,
		items:    make(map[any]*DataItem),
		pq:       make(priorityQueue, 0, capacity),
	}
}

// Add 添加一个新项到缓存中
func (tc *TimeCache) Add(data any) {
	tc.mu.Lock()
	defer tc.mu.Unlock()

	now := time.Now()

	// 如果数据已经存在，则更新它的最后访问时间
	if existingItem, ok := tc.items[data]; ok {
		prio := existingItem.LastAccess.UnixNano()
		tc.pq.update(existingItem, prio+1) // 实际上我们不需要 +1，但这里为了演示如何更新优先级
		// 注意：上面的更新操作是多余的，因为我们总是可以重新插入一个新项并删除旧项
		// 因此，下面的代码将替换上面的更新操作
		tc.pq.Remove(tc.pq.search(existingItem))
		delete(tc.items, data)
	}

	newItem := &DataItem{
		Data:       data,
		InsertTime: now,
		LastAccess: now,
		Index:      -1, // 初始时索引无效
	}
	pqItem := &priorityQueueItem{
		item: newItem,
		prio: newItem.LastAccess.UnixNano(),
	}
	heap.Push(&tc.pq, pqItem)
	newItem.Index = pqItem.item.Index
	tc.items[data] = newItem

	// 如果缓存超过了容量，移除最后访问时间最早的项
	if tc.pq.Len() > tc.capacity {
		oldestItem := heap.Pop(&tc.pq).(*priorityQueueItem).item
		delete(tc.items, oldestItem.Data)
	}
}

// search 辅助函数，用于在优先级队列中搜索一个项
func (pq *priorityQueue) search(item *DataItem) int {
	for i, pqItem := range *pq {
		if pqItem.item == item {
			return i
		}
	}
	return -1 // 未找到
}

// Remove 辅助函数，用于从优先级队列中移除一个项（注意：这不是 heap.Interface 的一部分）
func (pq *priorityQueue) Remove(index int) {
	pqItem := (*pq)[index]
	pqItem.item.Index = -1 // 将索引设为无效值
	*pq = append((*pq)[:index], (*pq)[index+1:]...)
	heap.Fix(pq, index)
}

func (tc *TimeCache) PrintCache() {
	tc.mu.Lock()
	defer tc.mu.Unlock()
	for _, item := range tc.items {
		fmt.Printf("Data: %v, InsertTime: %s, LastAccess: %s\n", item.Data, item.InsertTime, item.LastAccess)
	}
}