-
Notifications
You must be signed in to change notification settings - Fork 461
/
chain_index.go
208 lines (170 loc) · 5.25 KB
/
chain_index.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
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
package chain
import (
"context"
"fmt"
"hash/maphash"
"os"
"strconv"
"github.com/puzpuzpuz/xsync/v2"
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/venus/pkg/shardedmutex"
"github.com/filecoin-project/venus/venus-shared/types"
)
// DefaultChainIndexCacheSize no longer sets the maximum size, just the initial size of the map.
var DefaultChainIndexCacheSize = 1 << 15
func init() {
if s := os.Getenv("CHAIN_INDEX_CACHE"); s != "" {
lcic, err := strconv.Atoi(s)
if err != nil {
log.Errorf("failed to parse 'CHAIN_INDEX_CACHE' env var: %s", err)
}
DefaultChainIndexCacheSize = lcic
}
}
// ChainIndex tipset height index, used to getting tipset by height quickly
type ChainIndex struct { //nolint:revive
indexCache *xsync.MapOf[types.TipSetKey, *lbEntry]
fillCacheLock shardedmutex.ShardedMutexFor[types.TipSetKey]
loadTipSet loadTipSetFunc
skipLength abi.ChainEpoch
}
func maphashTSK(s maphash.Seed, tsk types.TipSetKey) uint64 {
return maphash.Bytes(s, tsk.Bytes())
}
// NewChainIndex return a new chain index with arc cache
func NewChainIndex(lts loadTipSetFunc) *ChainIndex {
return &ChainIndex{
indexCache: xsync.NewTypedMapOfPresized[types.TipSetKey, *lbEntry](maphashTSK, DefaultChainIndexCacheSize),
fillCacheLock: shardedmutex.NewFor(maphashTSK, 32),
loadTipSet: lts,
skipLength: 20,
}
}
type lbEntry struct {
targetHeight abi.ChainEpoch
target types.TipSetKey
}
// GetTipSetByHeight get tipset at specify height from specify tipset
// the tipset within the skiplength is directly obtained by reading the database.
// if the height difference exceeds the skiplength, the tipset is read from caching.
// if the caching fails, the tipset is obtained by reading the database and updating the cache
func (ci *ChainIndex) GetTipSetByHeight(ctx context.Context, from *types.TipSet, to abi.ChainEpoch) (*types.TipSet, error) {
if from.Height()-to <= ci.skipLength {
return ci.walkBack(ctx, from, to)
}
rounded, err := ci.roundDown(ctx, from)
if err != nil {
return nil, fmt.Errorf("failed to round down: %w", err)
}
cur := rounded.Key()
for {
lbe, ok := ci.indexCache.Load(cur) // check the cache
if !ok {
lk := ci.fillCacheLock.GetLock(cur)
lk.Lock() // if entry is missing, take the lock
lbe, ok = ci.indexCache.Load(cur) // check if someone else added it while we waited for lock
if !ok {
fc, err := ci.fillCache(ctx, cur)
if err != nil {
lk.Unlock()
return nil, fmt.Errorf("failed to fill cache: %w", err)
}
lbe = fc
ci.indexCache.Store(cur, lbe)
}
lk.Unlock()
}
if to == lbe.targetHeight {
ts, err := ci.loadTipSet(ctx, lbe.target)
if err != nil {
return nil, fmt.Errorf("failed to load tipset: %w", err)
}
return ts, nil
}
if to > lbe.targetHeight {
ts, err := ci.loadTipSet(ctx, cur)
if err != nil {
return nil, fmt.Errorf("failed to load tipset: %w", err)
}
return ci.walkBack(ctx, ts, to)
}
cur = lbe.target
}
}
// GetTipsetByHeightWithoutCache get the tipset of specific height by reading the database directly
func (ci *ChainIndex) GetTipsetByHeightWithoutCache(ctx context.Context, from *types.TipSet, to abi.ChainEpoch) (*types.TipSet, error) {
return ci.walkBack(ctx, from, to)
}
// Caller must hold indexCacheLk
func (ci *ChainIndex) fillCache(ctx context.Context, tsk types.TipSetKey) (*lbEntry, error) {
ts, err := ci.loadTipSet(ctx, tsk)
if err != nil {
return nil, fmt.Errorf("failed to load tipset: %w", err)
}
if ts.Height() == 0 {
return &lbEntry{
targetHeight: 0,
target: tsk,
}, nil
}
// will either be equal to ts.Height, or at least > ts.Parent.Height()
rheight := ci.roundHeight(ts.Height())
parent, err := ci.loadTipSet(ctx, ts.Parents())
if err != nil {
return nil, err
}
rheight -= ci.skipLength
if rheight < 0 {
rheight = 0
}
var skipTarget *types.TipSet
if parent.Height() < rheight {
skipTarget = parent
} else {
skipTarget, err = ci.walkBack(ctx, parent, rheight)
if err != nil {
return nil, fmt.Errorf("fillCache walkback: %w", err)
}
}
lbe := &lbEntry{
targetHeight: skipTarget.Height(),
target: skipTarget.Key(),
}
return lbe, nil
}
// floors to nearest skipLength multiple
func (ci *ChainIndex) roundHeight(h abi.ChainEpoch) abi.ChainEpoch {
return (h / ci.skipLength) * ci.skipLength
}
func (ci *ChainIndex) roundDown(ctx context.Context, ts *types.TipSet) (*types.TipSet, error) {
target := ci.roundHeight(ts.Height())
rounded, err := ci.walkBack(ctx, ts, target)
if err != nil {
return nil, fmt.Errorf("failed to walk back: %w", err)
}
return rounded, nil
}
func (ci *ChainIndex) walkBack(ctx context.Context, from *types.TipSet, to abi.ChainEpoch) (*types.TipSet, error) {
if to > from.Height() {
return nil, fmt.Errorf("looking for tipset with height greater than start point")
}
if to == from.Height() {
return from, nil
}
ts := from
for {
pts, err := ci.loadTipSet(ctx, ts.Parents())
if err != nil {
return nil, fmt.Errorf("failed to load tipset: %w", err)
}
if to > pts.Height() {
// in case pts is lower than the epoch we're looking for (null blocks)
// return a tipset above that height
return ts, nil
}
if to == pts.Height() {
return pts, nil
}
ts = pts
}
}