-
Notifications
You must be signed in to change notification settings - Fork 1.2k
/
ipldstore.go
92 lines (73 loc) · 1.99 KB
/
ipldstore.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
package ipldstore
import (
"bytes"
"context"
"fmt"
"github.com/filecoin-project/lotus/tools/stats/metrics"
lru "github.com/hashicorp/golang-lru"
"github.com/ipfs/go-cid"
cbg "github.com/whyrusleeping/cbor-gen"
"go.opencensus.io/stats"
)
type ApiIpldStore struct {
ctx context.Context
api apiIpldStoreApi
cache *lru.TwoQueueCache
cacheSize int
}
type apiIpldStoreApi interface {
ChainReadObj(context.Context, cid.Cid) ([]byte, error)
}
func NewApiIpldStore(ctx context.Context, api apiIpldStoreApi, cacheSize int) (*ApiIpldStore, error) {
store := &ApiIpldStore{
ctx: ctx,
api: api,
cacheSize: cacheSize,
}
cache, err := lru.New2Q(store.cacheSize)
if err != nil {
return nil, err
}
store.cache = cache
return store, nil
}
func (ht *ApiIpldStore) Context() context.Context {
return ht.ctx
}
func (ht *ApiIpldStore) read(ctx context.Context, c cid.Cid) ([]byte, error) {
stats.Record(ctx, metrics.IpldStoreCacheMiss.M(1))
done := metrics.Timer(ctx, metrics.IpldStoreReadDuration)
defer done()
return ht.api.ChainReadObj(ctx, c)
}
func (ht *ApiIpldStore) Get(ctx context.Context, c cid.Cid, out interface{}) error {
done := metrics.Timer(ctx, metrics.IpldStoreGetDuration)
defer done()
defer func() {
stats.Record(ctx, metrics.IpldStoreCacheSize.M(int64(ht.cacheSize)))
stats.Record(ctx, metrics.IpldStoreCacheLength.M(int64(ht.cache.Len())))
}()
var raw []byte
if a, ok := ht.cache.Get(c); ok {
stats.Record(ctx, metrics.IpldStoreCacheHit.M(1))
raw = a.([]byte)
} else {
bs, err := ht.read(ctx, c)
if err != nil {
return err
}
raw = bs
}
cu, ok := out.(cbg.CBORUnmarshaler)
if ok {
if err := cu.UnmarshalCBOR(bytes.NewReader(raw)); err != nil {
return err
}
ht.cache.Add(c, raw)
return nil
}
return fmt.Errorf("Object does not implement CBORUnmarshaler")
}
func (ht *ApiIpldStore) Put(ctx context.Context, v interface{}) (cid.Cid, error) {
return cid.Undef, fmt.Errorf("Put is not implemented on ApiIpldStore")
}