git-bug/cache/subcache.go

561 lines
13 KiB
Go
Raw Normal View History

2022-11-19 13:33:12 +03:00
package cache
import (
"bytes"
"encoding/gob"
"fmt"
2022-12-23 01:19:31 +03:00
"path/filepath"
2022-11-19 13:33:12 +03:00
"sync"
"github.com/pkg/errors"
"github.com/MichaelMure/git-bug/entities/identity"
2022-11-19 13:33:12 +03:00
"github.com/MichaelMure/git-bug/entity"
"github.com/MichaelMure/git-bug/repository"
)
type Excerpt interface {
Id() entity.Id
setId(id entity.Id)
2022-11-19 13:33:12 +03:00
}
type CacheEntity interface {
2022-12-19 20:09:59 +03:00
Id() entity.Id
2022-11-19 13:33:12 +03:00
NeedCommit() bool
2022-12-23 01:19:31 +03:00
Lock()
2022-11-19 13:33:12 +03:00
}
2022-11-29 15:01:53 +03:00
type getUserIdentityFunc func() (*IdentityCache, error)
2022-11-19 13:33:12 +03:00
// Actions expose a number of action functions on Entities, to give upper layers (cache) a way to normalize interactions.
// Note: ideally this wouldn't exist, the cache layer would assume that everything is an entity/dag, and directly use the
// functions from this package, but right now identities are not using that framework.
type Actions[EntityT entity.Interface] struct {
ReadWithResolver func(repo repository.ClockedRepo, resolvers entity.Resolvers, id entity.Id) (EntityT, error)
ReadAllWithResolver func(repo repository.ClockedRepo, resolvers entity.Resolvers) <-chan entity.StreamedEntity[EntityT]
Remove func(repo repository.ClockedRepo, id entity.Id) error
MergeAll func(repo repository.ClockedRepo, resolvers entity.Resolvers, remote string, mergeAuthor identity.Interface) <-chan entity.MergeResult
}
var _ cacheMgmt = &SubCache[entity.Interface, Excerpt, CacheEntity]{}
2022-12-19 20:09:59 +03:00
type SubCache[EntityT entity.Interface, ExcerptT Excerpt, CacheT CacheEntity] struct {
2022-11-19 13:33:12 +03:00
repo repository.ClockedRepo
2022-11-29 15:01:53 +03:00
resolvers func() entity.Resolvers
2022-11-19 13:33:12 +03:00
getUserIdentity getUserIdentityFunc
makeCached func(entity EntityT, entityUpdated func(id entity.Id) error) CacheT
makeExcerpt func(CacheT) ExcerptT
makeIndexData func(CacheT) []string
actions Actions[EntityT]
2022-11-19 13:33:12 +03:00
typename string
namespace string
version uint
maxLoaded int
mu sync.RWMutex
excerpts map[entity.Id]ExcerptT
cached map[entity.Id]CacheT
lru *lruIdCache
}
2022-12-19 20:09:59 +03:00
func NewSubCache[EntityT entity.Interface, ExcerptT Excerpt, CacheT CacheEntity](
2022-11-19 13:33:12 +03:00
repo repository.ClockedRepo,
2022-12-19 20:09:59 +03:00
resolvers func() entity.Resolvers, getUserIdentity getUserIdentityFunc,
makeCached func(entity EntityT, entityUpdated func(id entity.Id) error) CacheT,
makeExcerpt func(CacheT) ExcerptT,
makeIndexData func(CacheT) []string,
actions Actions[EntityT],
2022-11-19 13:33:12 +03:00
typename, namespace string,
2022-12-19 20:09:59 +03:00
version uint, maxLoaded int) *SubCache[EntityT, ExcerptT, CacheT] {
return &SubCache[EntityT, ExcerptT, CacheT]{
2022-11-19 13:33:12 +03:00
repo: repo,
resolvers: resolvers,
getUserIdentity: getUserIdentity,
makeCached: makeCached,
makeExcerpt: makeExcerpt,
makeIndexData: makeIndexData,
actions: actions,
2022-11-19 13:33:12 +03:00
typename: typename,
namespace: namespace,
version: version,
maxLoaded: maxLoaded,
excerpts: make(map[entity.Id]ExcerptT),
cached: make(map[entity.Id]CacheT),
lru: newLRUIdCache(),
}
}
2022-12-19 20:09:59 +03:00
func (sc *SubCache[EntityT, ExcerptT, CacheT]) Typename() string {
2022-12-08 14:30:08 +03:00
return sc.typename
}
2022-11-19 13:33:12 +03:00
// Load will try to read from the disk the entity cache file
2022-12-19 20:09:59 +03:00
func (sc *SubCache[EntityT, ExcerptT, CacheT]) Load() error {
2022-11-19 13:33:12 +03:00
sc.mu.Lock()
defer sc.mu.Unlock()
2022-12-23 01:19:31 +03:00
f, err := sc.repo.LocalStorage().Open(filepath.Join("cache", sc.namespace))
2022-11-19 13:33:12 +03:00
if err != nil {
return err
}
aux := struct {
Version uint
Excerpts map[entity.Id]ExcerptT
}{}
decoder := gob.NewDecoder(f)
2022-11-19 13:33:12 +03:00
err = decoder.Decode(&aux)
if err != nil {
_ = f.Close()
return err
}
err = f.Close()
2022-11-19 13:33:12 +03:00
if err != nil {
return err
}
if aux.Version != sc.version {
return fmt.Errorf("unknown %s cache format version %v", sc.namespace, aux.Version)
}
// the id is not serialized in the excerpt itself (non-exported field in go, long story ...),
// so we fix it here, which doubles as enforcing coherency.
for id, excerpt := range aux.Excerpts {
excerpt.setId(id)
}
2022-11-19 13:33:12 +03:00
sc.excerpts = aux.Excerpts
2022-12-23 01:19:31 +03:00
index, err := sc.repo.GetIndex(sc.namespace)
2022-11-19 13:33:12 +03:00
if err != nil {
return err
}
// simple heuristic to detect a mismatch between the index and the entities
count, err := index.DocCount()
if err != nil {
return err
}
if count != uint64(len(sc.excerpts)) {
return fmt.Errorf("count mismatch between bleve and %s excerpts", sc.namespace)
}
// TODO: find a way to check lamport clocks
2022-11-19 13:33:12 +03:00
return nil
}
// Write will serialize on disk the entity cache file
func (sc *SubCache[EntityT, ExcerptT, CacheT]) write() error {
2022-11-19 13:33:12 +03:00
sc.mu.RLock()
defer sc.mu.RUnlock()
var data bytes.Buffer
aux := struct {
Version uint
Excerpts map[entity.Id]ExcerptT
}{
Version: sc.version,
Excerpts: sc.excerpts,
}
encoder := gob.NewEncoder(&data)
err := encoder.Encode(aux)
if err != nil {
return err
}
2022-12-23 01:19:31 +03:00
f, err := sc.repo.LocalStorage().Create(filepath.Join("cache", sc.namespace))
2022-11-19 13:33:12 +03:00
if err != nil {
return err
}
_, err = f.Write(data.Bytes())
if err != nil {
_ = f.Close()
2022-11-19 13:33:12 +03:00
return err
}
return f.Close()
}
func (sc *SubCache[EntityT, ExcerptT, CacheT]) Build() <-chan BuildEvent {
out := make(chan BuildEvent)
2022-12-08 14:30:08 +03:00
go func() {
defer close(out)
2022-12-08 14:30:08 +03:00
out <- BuildEvent{
Typename: sc.typename,
Event: BuildEventStarted,
}
2022-12-19 20:09:59 +03:00
sc.excerpts = make(map[entity.Id]ExcerptT)
2022-12-08 14:30:08 +03:00
allEntities := sc.actions.ReadAllWithResolver(sc.repo, sc.resolvers())
2022-12-19 20:09:59 +03:00
index, err := sc.repo.GetIndex(sc.namespace)
if err != nil {
out <- BuildEvent{
Typename: sc.typename,
Err: err,
}
return
2022-12-08 14:30:08 +03:00
}
// wipe the index just to be sure
err = index.Clear()
if err != nil {
out <- BuildEvent{
Typename: sc.typename,
Err: err,
}
return
}
2022-12-08 14:30:08 +03:00
indexer, indexEnd := index.IndexBatch()
for e := range allEntities {
if e.Err != nil {
out <- BuildEvent{
Typename: sc.typename,
Err: e.Err,
}
return
}
cached := sc.makeCached(e.Entity, sc.entityUpdated)
sc.excerpts[e.Entity.Id()] = sc.makeExcerpt(cached)
// might as well keep them in memory
sc.cached[e.Entity.Id()] = cached
indexData := sc.makeIndexData(cached)
if err := indexer(e.Entity.Id().String(), indexData); err != nil {
out <- BuildEvent{
Typename: sc.typename,
Err: err,
}
return
}
out <- BuildEvent{
Typename: sc.typename,
Event: BuildEventProgress,
Progress: e.CurrentEntity,
Total: e.TotalEntities,
}
2022-12-08 14:30:08 +03:00
}
2022-11-29 15:01:53 +03:00
err = indexEnd()
if err != nil {
out <- BuildEvent{
Typename: sc.typename,
Err: err,
}
return
}
2022-12-19 20:09:59 +03:00
err = sc.write()
if err != nil {
out <- BuildEvent{
Typename: sc.typename,
Err: err,
}
return
}
out <- BuildEvent{
Typename: sc.typename,
Event: BuildEventFinished,
}
}()
return out
2022-11-29 15:01:53 +03:00
}
2022-11-19 13:33:12 +03:00
func (sc *SubCache[EntityT, ExcerptT, CacheT]) SetCacheSize(size int) {
sc.maxLoaded = size
sc.evictIfNeeded()
}
2022-12-19 20:09:59 +03:00
func (sc *SubCache[EntityT, ExcerptT, CacheT]) Close() error {
2022-11-29 15:01:53 +03:00
sc.mu.Lock()
defer sc.mu.Unlock()
sc.excerpts = nil
sc.cached = make(map[entity.Id]CacheT)
return nil
2022-11-19 13:33:12 +03:00
}
// AllIds return all known bug ids
2022-12-19 20:09:59 +03:00
func (sc *SubCache[EntityT, ExcerptT, CacheT]) AllIds() []entity.Id {
2022-11-19 13:33:12 +03:00
sc.mu.RLock()
defer sc.mu.RUnlock()
result := make([]entity.Id, len(sc.excerpts))
i := 0
for _, excerpt := range sc.excerpts {
result[i] = excerpt.Id()
i++
}
return result
}
// Resolve retrieve an entity matching the exact given id
2022-12-19 20:09:59 +03:00
func (sc *SubCache[EntityT, ExcerptT, CacheT]) Resolve(id entity.Id) (CacheT, error) {
2022-11-19 13:33:12 +03:00
sc.mu.RLock()
cached, ok := sc.cached[id]
if ok {
sc.lru.Get(id)
sc.mu.RUnlock()
return cached, nil
}
sc.mu.RUnlock()
e, err := sc.actions.ReadWithResolver(sc.repo, sc.resolvers(), id)
2022-11-19 13:33:12 +03:00
if err != nil {
2022-12-08 14:30:08 +03:00
return *new(CacheT), err
2022-11-19 13:33:12 +03:00
}
2022-12-19 20:09:59 +03:00
cached = sc.makeCached(e, sc.entityUpdated)
2022-11-19 13:33:12 +03:00
sc.mu.Lock()
sc.cached[id] = cached
sc.lru.Add(id)
sc.mu.Unlock()
sc.evictIfNeeded()
return cached, nil
}
// ResolvePrefix retrieve an entity matching an id prefix. It fails if multiple
// entity match.
2022-12-19 20:09:59 +03:00
func (sc *SubCache[EntityT, ExcerptT, CacheT]) ResolvePrefix(prefix string) (CacheT, error) {
2022-11-19 13:33:12 +03:00
return sc.ResolveMatcher(func(excerpt ExcerptT) bool {
return excerpt.Id().HasPrefix(prefix)
})
}
2022-12-19 20:09:59 +03:00
func (sc *SubCache[EntityT, ExcerptT, CacheT]) ResolveMatcher(f func(ExcerptT) bool) (CacheT, error) {
2022-11-19 13:33:12 +03:00
id, err := sc.resolveMatcher(f)
if err != nil {
2022-12-08 14:30:08 +03:00
return *new(CacheT), err
2022-11-19 13:33:12 +03:00
}
return sc.Resolve(id)
}
// ResolveExcerpt retrieve an Excerpt matching the exact given id
2022-12-19 20:09:59 +03:00
func (sc *SubCache[EntityT, ExcerptT, CacheT]) ResolveExcerpt(id entity.Id) (ExcerptT, error) {
2022-11-19 13:33:12 +03:00
sc.mu.RLock()
defer sc.mu.RUnlock()
excerpt, ok := sc.excerpts[id]
if !ok {
2022-12-08 14:30:08 +03:00
return *new(ExcerptT), entity.NewErrNotFound(sc.typename)
2022-11-19 13:33:12 +03:00
}
return excerpt, nil
}
// ResolveExcerptPrefix retrieve an Excerpt matching an id prefix. It fails if multiple
// entity match.
2022-12-19 20:09:59 +03:00
func (sc *SubCache[EntityT, ExcerptT, CacheT]) ResolveExcerptPrefix(prefix string) (ExcerptT, error) {
2022-11-19 13:33:12 +03:00
return sc.ResolveExcerptMatcher(func(excerpt ExcerptT) bool {
return excerpt.Id().HasPrefix(prefix)
})
}
2022-12-19 20:09:59 +03:00
func (sc *SubCache[EntityT, ExcerptT, CacheT]) ResolveExcerptMatcher(f func(ExcerptT) bool) (ExcerptT, error) {
2022-11-19 13:33:12 +03:00
id, err := sc.resolveMatcher(f)
if err != nil {
2022-12-08 14:30:08 +03:00
return *new(ExcerptT), err
2022-11-19 13:33:12 +03:00
}
return sc.ResolveExcerpt(id)
}
2022-12-19 20:09:59 +03:00
func (sc *SubCache[EntityT, ExcerptT, CacheT]) resolveMatcher(f func(ExcerptT) bool) (entity.Id, error) {
2022-11-19 13:33:12 +03:00
sc.mu.RLock()
defer sc.mu.RUnlock()
// preallocate but empty
matching := make([]entity.Id, 0, 5)
for _, excerpt := range sc.excerpts {
if f(excerpt) {
matching = append(matching, excerpt.Id())
}
}
if len(matching) > 1 {
return entity.UnsetId, entity.NewErrMultipleMatch(sc.typename, matching)
}
if len(matching) == 0 {
return entity.UnsetId, entity.NewErrNotFound(sc.typename)
}
return matching[0], nil
}
2022-12-19 20:09:59 +03:00
func (sc *SubCache[EntityT, ExcerptT, CacheT]) add(e EntityT) (CacheT, error) {
2022-11-19 13:33:12 +03:00
sc.mu.Lock()
if _, has := sc.cached[e.Id()]; has {
sc.mu.Unlock()
2022-12-08 14:30:08 +03:00
return *new(CacheT), fmt.Errorf("entity %s already exist in the cache", e.Id())
2022-11-19 13:33:12 +03:00
}
2022-12-19 20:09:59 +03:00
cached := sc.makeCached(e, sc.entityUpdated)
2022-11-19 13:33:12 +03:00
sc.cached[e.Id()] = cached
sc.lru.Add(e.Id())
sc.mu.Unlock()
sc.evictIfNeeded()
// force the write of the excerpt
err := sc.entityUpdated(e.Id())
if err != nil {
2022-12-08 14:30:08 +03:00
return *new(CacheT), err
2022-11-19 13:33:12 +03:00
}
return cached, nil
}
2022-12-19 20:09:59 +03:00
func (sc *SubCache[EntityT, ExcerptT, CacheT]) Remove(prefix string) error {
2022-11-19 13:33:12 +03:00
e, err := sc.ResolvePrefix(prefix)
if err != nil {
return err
}
sc.mu.Lock()
err = sc.actions.Remove(sc.repo, e.Id())
2022-11-19 13:33:12 +03:00
if err != nil {
sc.mu.Unlock()
2022-11-19 13:33:12 +03:00
return err
}
delete(sc.cached, e.Id())
delete(sc.excerpts, e.Id())
sc.lru.Remove(e.Id())
sc.mu.Unlock()
return sc.write()
}
func (sc *SubCache[EntityT, ExcerptT, CacheT]) MergeAll(remote string) <-chan entity.MergeResult {
out := make(chan entity.MergeResult)
2022-11-19 13:33:12 +03:00
// Intercept merge results to update the cache properly
go func() {
defer close(out)
author, err := sc.getUserIdentity()
if err != nil {
out <- entity.NewMergeError(err, "")
return
}
results := sc.actions.MergeAll(sc.repo, sc.resolvers(), remote, author)
for result := range results {
out <- result
if result.Err != nil {
continue
}
switch result.Status {
case entity.MergeStatusNew, entity.MergeStatusUpdated:
e := result.Entity.(EntityT)
cached := sc.makeCached(e, sc.entityUpdated)
sc.mu.Lock()
sc.excerpts[result.Id] = sc.makeExcerpt(cached)
// might as well keep them in memory
sc.cached[result.Id] = cached
sc.mu.Unlock()
}
}
err = sc.write()
if err != nil {
out <- entity.NewMergeError(err, "")
return
}
}()
return out
2022-11-19 13:33:12 +03:00
}
func (sc *SubCache[EntityT, ExcerptT, CacheT]) GetNamespace() string {
return sc.namespace
2022-11-19 13:33:12 +03:00
}
// entityUpdated is a callback to trigger when the excerpt of an entity changed
2022-12-19 20:09:59 +03:00
func (sc *SubCache[EntityT, ExcerptT, CacheT]) entityUpdated(id entity.Id) error {
2022-11-19 13:33:12 +03:00
sc.mu.Lock()
e, ok := sc.cached[id]
2022-11-19 13:33:12 +03:00
if !ok {
sc.mu.Unlock()
// if the bug is not loaded at this point, it means it was loaded before
// but got evicted. Which means we potentially have multiple copies in
// memory and thus concurrent write.
// Failing immediately here is the simple and safe solution to avoid
// complicated data loss.
return errors.New("entity missing from cache")
2022-11-19 13:33:12 +03:00
}
sc.lru.Get(id)
// sc.excerpts[id] = bug2.NewBugExcerpt(b.bug, b.Snapshot())
sc.excerpts[id] = sc.makeExcerpt(e)
2022-11-19 13:33:12 +03:00
sc.mu.Unlock()
2022-12-23 01:19:31 +03:00
index, err := sc.repo.GetIndex(sc.namespace)
if err != nil {
return err
}
err = index.IndexOne(e.Id().String(), sc.makeIndexData(e))
if err != nil {
2022-11-19 13:33:12 +03:00
return err
}
return sc.write()
2022-11-19 13:33:12 +03:00
}
// evictIfNeeded will evict an entity from the cache if needed
2022-12-19 20:09:59 +03:00
func (sc *SubCache[EntityT, ExcerptT, CacheT]) evictIfNeeded() {
2022-11-19 13:33:12 +03:00
sc.mu.Lock()
defer sc.mu.Unlock()
if sc.lru.Len() <= sc.maxLoaded {
return
}
for _, id := range sc.lru.GetOldestToNewest() {
b := sc.cached[id]
if b.NeedCommit() {
continue
}
2022-12-23 01:19:31 +03:00
// as a form of assurance that evicted entities don't get manipulated, we lock them here.
// if something try to do it anyway, it will lock the program and make it obvious.
b.Lock()
2022-11-19 13:33:12 +03:00
sc.lru.Remove(id)
delete(sc.cached, id)
if sc.lru.Len() <= sc.maxLoaded {
return
}
}
}