lotus/chain/events/tscache.go

218 lines
5.5 KiB
Go
Raw Normal View History

2019-09-05 07:40:50 +00:00
package events
2019-09-03 17:45:55 +00:00
import (
"context"
"sync"
2019-09-03 17:45:55 +00:00
"golang.org/x/xerrors"
2022-06-14 15:00:51 +00:00
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/lotus/chain/types"
2019-09-03 17:45:55 +00:00
)
type tsCacheAPI interface {
2021-08-04 00:10:30 +00:00
ChainGetTipSetAfterHeight(context.Context, abi.ChainEpoch, types.TipSetKey) (*types.TipSet, error)
ChainGetTipSetByHeight(context.Context, abi.ChainEpoch, types.TipSetKey) (*types.TipSet, error)
2021-08-04 00:10:30 +00:00
ChainGetTipSet(context.Context, types.TipSetKey) (*types.TipSet, error)
ChainHead(context.Context) (*types.TipSet, error)
}
2019-09-03 17:45:55 +00:00
// tipSetCache implements a simple ring-buffer cache to keep track of recent
// tipsets
type tipSetCache struct {
mu sync.RWMutex
2021-08-04 00:10:30 +00:00
byKey map[types.TipSetKey]*types.TipSet
byHeight []*types.TipSet
start int // chain head (end)
len int
storage tsCacheAPI
2019-09-03 17:45:55 +00:00
}
2021-08-04 00:10:30 +00:00
func newTSCache(storage tsCacheAPI, cap abi.ChainEpoch) *tipSetCache {
2019-09-03 17:45:55 +00:00
return &tipSetCache{
2021-08-04 00:10:30 +00:00
byKey: make(map[types.TipSetKey]*types.TipSet, cap),
byHeight: make([]*types.TipSet, cap),
start: 0,
len: 0,
storage: storage,
2019-09-03 17:45:55 +00:00
}
}
2021-08-04 00:10:30 +00:00
func (tsc *tipSetCache) ChainGetTipSet(ctx context.Context, tsk types.TipSetKey) (*types.TipSet, error) {
if ts, ok := tsc.byKey[tsk]; ok {
return ts, nil
}
return tsc.storage.ChainGetTipSet(ctx, tsk)
}
func (tsc *tipSetCache) ChainGetTipSetByHeight(ctx context.Context, height abi.ChainEpoch, tsk types.TipSetKey) (*types.TipSet, error) {
return tsc.get(ctx, height, tsk, true)
}
func (tsc *tipSetCache) ChainGetTipSetAfterHeight(ctx context.Context, height abi.ChainEpoch, tsk types.TipSetKey) (*types.TipSet, error) {
return tsc.get(ctx, height, tsk, false)
}
func (tsc *tipSetCache) get(ctx context.Context, height abi.ChainEpoch, tsk types.TipSetKey, prev bool) (*types.TipSet, error) {
fallback := tsc.storage.ChainGetTipSetAfterHeight
if prev {
fallback = tsc.storage.ChainGetTipSetByHeight
}
tsc.mu.RLock()
// Nothing in the cache?
if tsc.len == 0 {
tsc.mu.RUnlock()
log.Warnf("tipSetCache.get: cache is empty, requesting from storage (h=%d)", height)
return fallback(ctx, height, tsk)
}
// Resolve the head.
head := tsc.byHeight[tsc.start]
if !tsk.IsEmpty() {
// Not on this chain?
var ok bool
head, ok = tsc.byKey[tsk]
if !ok {
tsc.mu.RUnlock()
return fallback(ctx, height, tsk)
}
}
headH := head.Height()
tailH := headH - abi.ChainEpoch(tsc.len)
if headH == height {
tsc.mu.RUnlock()
return head, nil
} else if headH < height {
tsc.mu.RUnlock()
// If the user doesn't pass a tsk, we assume "head" is the last tipset we processed.
return nil, xerrors.Errorf("requested epoch is in the future")
} else if height < tailH {
log.Warnf("tipSetCache.get: requested tipset not in cache, requesting from storage (h=%d; tail=%d)", height, tailH)
tsc.mu.RUnlock()
return fallback(ctx, height, head.Key())
}
direction := 1
if prev {
direction = -1
}
var ts *types.TipSet
for i := 0; i < tsc.len && ts == nil; i += direction {
ts = tsc.byHeight[normalModulo(tsc.start-int(headH-height)+i, len(tsc.byHeight))]
}
tsc.mu.RUnlock()
return ts, nil
}
func (tsc *tipSetCache) ChainHead(ctx context.Context) (*types.TipSet, error) {
tsc.mu.RLock()
best := tsc.byHeight[tsc.start]
tsc.mu.RUnlock()
if best == nil {
return tsc.storage.ChainHead(ctx)
}
return best, nil
}
2019-09-03 17:45:55 +00:00
2021-08-04 00:10:30 +00:00
func (tsc *tipSetCache) add(to *types.TipSet) error {
tsc.mu.Lock()
defer tsc.mu.Unlock()
2019-09-03 17:45:55 +00:00
if tsc.len > 0 {
2021-08-04 00:10:30 +00:00
best := tsc.byHeight[tsc.start]
if best.Height() >= to.Height() {
return xerrors.Errorf("tipSetCache.add: expected new tipset height to be at least %d, was %d", tsc.byHeight[tsc.start].Height()+1, to.Height())
2019-09-03 17:45:55 +00:00
}
2021-08-04 00:10:30 +00:00
if best.Key() != to.Parents() {
return xerrors.Errorf(
"tipSetCache.add: expected new tipset %s (%d) to follow %s (%d), its parents are %s",
2021-08-04 00:10:30 +00:00
to.Key(), to.Height(), best.Key(), best.Height(), best.Parents(),
)
}
2019-09-03 17:45:55 +00:00
}
2021-08-04 00:10:30 +00:00
nextH := to.Height()
2019-10-04 22:43:04 +00:00
if tsc.len > 0 {
2021-08-04 00:10:30 +00:00
nextH = tsc.byHeight[tsc.start].Height() + 1
2019-10-04 22:43:04 +00:00
}
// fill null blocks
2021-08-04 00:10:30 +00:00
for nextH != to.Height() {
tsc.start = normalModulo(tsc.start+1, len(tsc.byHeight))
was := tsc.byHeight[tsc.start]
if was != nil {
tsc.byHeight[tsc.start] = nil
delete(tsc.byKey, was.Key())
}
if tsc.len < len(tsc.byHeight) {
2019-10-04 22:43:04 +00:00
tsc.len++
}
nextH++
}
2021-08-04 00:10:30 +00:00
tsc.start = normalModulo(tsc.start+1, len(tsc.byHeight))
was := tsc.byHeight[tsc.start]
if was != nil {
delete(tsc.byKey, was.Key())
}
tsc.byHeight[tsc.start] = to
if tsc.len < len(tsc.byHeight) {
2019-09-03 17:45:55 +00:00
tsc.len++
}
2021-08-04 00:10:30 +00:00
tsc.byKey[to.Key()] = to
2019-09-03 17:45:55 +00:00
return nil
}
2021-08-04 00:10:30 +00:00
func (tsc *tipSetCache) revert(from *types.TipSet) error {
tsc.mu.Lock()
defer tsc.mu.Unlock()
2021-08-04 00:10:30 +00:00
return tsc.revertUnlocked(from)
}
func (tsc *tipSetCache) revertUnlocked(ts *types.TipSet) error {
2019-09-03 17:45:55 +00:00
if tsc.len == 0 {
2019-12-03 17:38:37 +00:00
return nil // this can happen, and it's fine
2019-09-03 17:45:55 +00:00
}
2021-08-04 00:10:30 +00:00
was := tsc.byHeight[tsc.start]
if !was.Equals(ts) {
2019-09-03 17:45:55 +00:00
return xerrors.New("tipSetCache.revert: revert tipset didn't match cache head")
}
2021-08-04 00:10:30 +00:00
delete(tsc.byKey, was.Key())
2019-09-03 17:45:55 +00:00
2021-08-04 00:10:30 +00:00
tsc.byHeight[tsc.start] = nil
tsc.start = normalModulo(tsc.start-1, len(tsc.byHeight))
2019-09-03 17:45:55 +00:00
tsc.len--
2019-10-04 22:43:04 +00:00
_ = tsc.revertUnlocked(nil) // revert null block gap
2019-09-03 17:45:55 +00:00
return nil
}
2021-08-04 00:10:30 +00:00
func (tsc *tipSetCache) observer() TipSetObserver {
return (*tipSetCacheObserver)(tsc)
}
2021-08-04 00:10:30 +00:00
type tipSetCacheObserver tipSetCache
2019-09-03 17:45:55 +00:00
2021-08-04 00:10:30 +00:00
var _ TipSetObserver = new(tipSetCacheObserver)
2019-09-03 17:45:55 +00:00
2021-08-04 00:10:30 +00:00
func (tsc *tipSetCacheObserver) Apply(_ context.Context, _, to *types.TipSet) error {
return (*tipSetCache)(tsc).add(to)
2019-09-03 17:45:55 +00:00
}
2021-08-04 00:10:30 +00:00
func (tsc *tipSetCacheObserver) Revert(ctx context.Context, from, _ *types.TipSet) error {
return (*tipSetCache)(tsc).revert(from)
2019-09-03 17:45:55 +00:00
}
func normalModulo(n, m int) int {
2019-09-30 22:37:01 +00:00
return ((n % m) + m) % m
}