-
Notifications
You must be signed in to change notification settings - Fork 567
/
cache.go
103 lines (94 loc) · 2.35 KB
/
cache.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
package index
import (
"bytes"
"context"
"io"
"sort"
"sync"
"github.com/hashicorp/golang-lru/simplelru"
"github.com/pachyderm/pachyderm/v2/src/internal/errors"
"github.com/pachyderm/pachyderm/v2/src/internal/pbutil"
"github.com/pachyderm/pachyderm/v2/src/internal/storage/chunk"
)
type Cache struct {
storage *chunk.Storage
cache *simplelru.LRU
mu sync.Mutex
}
func NewCache(storage *chunk.Storage, size int) *Cache {
lruCache, err := simplelru.NewLRU(size, nil)
if err != nil {
panic(err)
}
return &Cache{
storage: storage,
cache: lruCache,
}
}
func (c *Cache) Get(ctx context.Context, chunkRef *chunk.DataRef, filter *pathFilter, w io.Writer) error {
c.mu.Lock()
v, ok := c.cache.Get(string(chunkRef.Ref.Id))
c.mu.Unlock()
if ok {
return get(v.(*cachedChunk), filter, w)
}
cr := c.storage.NewReader(ctx, []*chunk.DataRef{chunkRef})
buf := &bytes.Buffer{}
if err := cr.Get(buf); err != nil {
return err
}
cachedChunk, err := c.computeCachedChunk(buf.Bytes())
if err != nil {
return err
}
c.mu.Lock()
c.cache.Add(string(chunkRef.Ref.Id), cachedChunk)
c.mu.Unlock()
return get(cachedChunk, filter, w)
}
type cachedChunk struct {
data []byte
pathOffsets []*pathOffset
}
type pathOffset struct {
upper string
offset int
}
func (c *Cache) computeCachedChunk(data []byte) (*cachedChunk, error) {
br := bytes.NewReader(data)
pbr := pbutil.NewReader(br)
var pathOffsets []*pathOffset
for {
pathOffset := &pathOffset{
offset: len(data) - br.Len(),
}
idx := &Index{}
if err := pbr.Read(idx); err != nil {
break
}
pathOffset.upper = idx.Path
if idx.Range != nil {
pathOffset.upper = idx.Range.LastPath
}
pathOffsets = append(pathOffsets, pathOffset)
}
return &cachedChunk{
data: data,
pathOffsets: pathOffsets,
}, nil
}
func get(cachedChunk *cachedChunk, filter *pathFilter, w io.Writer) error {
if len(cachedChunk.pathOffsets) == 0 {
_, err := w.Write(cachedChunk.data)
return errors.EnsureStack(err)
}
i := sort.Search(len(cachedChunk.pathOffsets), func(i int) bool {
return atStart(cachedChunk.pathOffsets[i].upper, filter)
})
if i >= len(cachedChunk.pathOffsets) {
_, err := w.Write(cachedChunk.data[cachedChunk.pathOffsets[i-1].offset:])
return errors.EnsureStack(err)
}
_, err := w.Write(cachedChunk.data[cachedChunk.pathOffsets[i].offset:])
return errors.EnsureStack(err)
}