cerc-io/lotus
11
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 27 Wiki Activity

refactor: update cache to the new generic version (#10463)

Browse Source 
- Adds type safety.
- Reduces allocations.
- Fixes the drand cache (was storing by value, but retrieving by pointer)
This commit is contained in:
Steven Allen
2023-03-13 15:29:09 -07:00
committed by GitHub
parent 97a9921cdd
commit dcb49dc8ee
17 changed files with 95 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files
chain/badtscache.go
+4 -9
View File
@@ -3,14 +3,14 @@ package chain
import (
"fmt"
lru "github.com/hashicorp/golang-lru"
lru "github.com/hashicorp/golang-lru/v2"
"github.com/ipfs/go-cid"
"github.com/filecoin-project/lotus/build"
)
type BadBlockCache struct {
badBlocks *lru.ARCCache
badBlocks *lru.ARCCache[cid.Cid, BadBlockReason]
}
type BadBlockReason struct {
@@ -43,7 +43,7 @@ func (bbr BadBlockReason) String() string {
}
func NewBadBlockCache() *BadBlockCache {
cache, err := lru.NewARC(build.BadBlockCacheSize)
cache, err := lru.NewARC[cid.Cid, BadBlockReason](build.BadBlockCacheSize)
if err != nil {
panic(err) // ok
}
@@ -66,10 +66,5 @@ func (bts *BadBlockCache) Purge() {
}
func (bts *BadBlockCache) Has(c cid.Cid) (BadBlockReason, bool) {
rval, ok := bts.badBlocks.Get(c)
if !ok {
return BadBlockReason{}, false
}
return rval.(BadBlockReason), true
return bts.badBlocks.Get(c)
}
chain/beacon/drand/drand.go
+6 -10
View File
@@ -12,7 +12,7 @@ import (
gclient "github.com/drand/drand/lp2p/client"
"github.com/drand/kyber"
kzap "github.com/go-kit/kit/log/zap"
lru "github.com/hashicorp/golang-lru"
lru "github.com/hashicorp/golang-lru/v2"
logging "github.com/ipfs/go-log/v2"
pubsub "github.com/libp2p/go-libp2p-pubsub"
"go.uber.org/zap/zapcore"
@@ -61,7 +61,7 @@ type DrandBeacon struct {
filGenTime uint64
filRoundTime uint64
localCache *lru.Cache
localCache *lru.Cache[uint64, *types.BeaconEntry]
}
// DrandHTTPClient interface overrides the user agent used by drand
@@ -110,7 +110,7 @@ func NewDrandBeacon(genesisTs, interval uint64, ps *pubsub.PubSub, config dtypes
return nil, xerrors.Errorf("creating drand client: %w", err)
}
lc, err := lru.New(1024)
lc, err := lru.New[uint64, *types.BeaconEntry](1024)
if err != nil {
return nil, err
}
@@ -160,16 +160,12 @@ func (db *DrandBeacon) Entry(ctx context.Context, round uint64) <-chan beacon.Re
return out
}
func (db *DrandBeacon) cacheValue(e types.BeaconEntry) {
db.localCache.Add(e.Round, e)
db.localCache.Add(e.Round, &e)
}
func (db *DrandBeacon) getCachedValue(round uint64) *types.BeaconEntry {
v, ok := db.localCache.Get(round)
if !ok {
return nil
}
e, _ := v.(types.BeaconEntry)
return &e
v, _ := db.localCache.Get(round)
return v
}
func (db *DrandBeacon) VerifyEntry(curr types.BeaconEntry, prev types.BeaconEntry) error {
chain/block_receipt_tracker.go
+5 -7
View File
@@ -5,7 +5,7 @@ import (
"sync"
"time"
lru "github.com/hashicorp/golang-lru"
lru "github.com/hashicorp/golang-lru/v2"
"github.com/libp2p/go-libp2p/core/peer"
"github.com/filecoin-project/lotus/build"
@@ -17,7 +17,7 @@ type blockReceiptTracker struct {
// using an LRU cache because i don't want to handle all the edge cases for
// manual cleanup and maintenance of a fixed size set
cache *lru.Cache
cache *lru.Cache[types.TipSetKey, *peerSet]
}
type peerSet struct {
@@ -25,7 +25,7 @@ type peerSet struct {
}
func newBlockReceiptTracker() *blockReceiptTracker {
c, _ := lru.New(512)
c, _ := lru.New[types.TipSetKey, *peerSet](512)
return &blockReceiptTracker{
cache: c,
}
@@ -46,20 +46,18 @@ func (brt *blockReceiptTracker) Add(p peer.ID, ts *types.TipSet) {
return
}
val.(*peerSet).peers[p] = build.Clock.Now()
val.peers[p] = build.Clock.Now()
}
func (brt *blockReceiptTracker) GetPeers(ts *types.TipSet) []peer.ID {
brt.lk.Lock()
defer brt.lk.Unlock()
val, ok := brt.cache.Get(ts.Key())
ps, ok := brt.cache.Get(ts.Key())
if !ok {
return nil
}
ps := val.(*peerSet)
out := make([]peer.ID, 0, len(ps.peers))
for p := range ps.peers {
out = append(out, p)
chain/events/message_cache.go
+9 -9
View File
@@ -4,7 +4,7 @@ import (
"context"
"sync"
lru "github.com/hashicorp/golang-lru"
lru "github.com/hashicorp/golang-lru/v2"
"github.com/ipfs/go-cid"
"github.com/filecoin-project/lotus/api"
@@ -14,14 +14,14 @@ type messageCache struct {
api EventAPI
blockMsgLk sync.Mutex
blockMsgCache *lru.ARCCache
blockMsgCache *lru.ARCCache[cid.Cid, *api.BlockMessages]
}
func newMessageCache(api EventAPI) *messageCache {
blsMsgCache, _ := lru.NewARC(500)
func newMessageCache(a EventAPI) *messageCache {
blsMsgCache, _ := lru.NewARC[cid.Cid, *api.BlockMessages](500)
return &messageCache{
api: api,
api: a,
blockMsgCache: blsMsgCache,
}
}
@@ -30,14 +30,14 @@ func (c *messageCache) ChainGetBlockMessages(ctx context.Context, blkCid cid.Cid
c.blockMsgLk.Lock()
defer c.blockMsgLk.Unlock()
msgsI, ok := c.blockMsgCache.Get(blkCid)
msgs, ok := c.blockMsgCache.Get(blkCid)
var err error
if !ok {
msgsI, err = c.api.ChainGetBlockMessages(ctx, blkCid)
msgs, err = c.api.ChainGetBlockMessages(ctx, blkCid)
if err != nil {
return nil, err
}
c.blockMsgCache.Add(blkCid, msgsI)
c.blockMsgCache.Add(blkCid, msgs)
}
return msgsI.(*api.BlockMessages), nil
return msgs, nil
}
chain/messagepool/messagepool.go
+9 -14
View File
@@ -12,7 +12,7 @@ import (
"time"
"github.com/hashicorp/go-multierror"
lru "github.com/hashicorp/golang-lru"
lru "github.com/hashicorp/golang-lru/v2"
"github.com/ipfs/go-cid"
"github.com/ipfs/go-datastore"
"github.com/ipfs/go-datastore/namespace"
@@ -159,7 +159,7 @@ type MessagePool struct {
// pruneCooldown is a channel used to allow a cooldown time between prunes
pruneCooldown chan struct{}
blsSigCache *lru.TwoQueueCache
blsSigCache *lru.TwoQueueCache[cid.Cid, crypto.Signature]
changes *lps.PubSub
@@ -167,9 +167,9 @@ type MessagePool struct {
netName dtypes.NetworkName
sigValCache *lru.TwoQueueCache
sigValCache *lru.TwoQueueCache[string, struct{}]
nonceCache *lru.Cache
nonceCache *lru.Cache[nonceCacheKey, uint64]
evtTypes [3]journal.EventType
journal journal.Journal
@@ -369,9 +369,9 @@ func (ms *msgSet) toSlice() []*types.SignedMessage {
}
func New(ctx context.Context, api Provider, ds dtypes.MetadataDS, us stmgr.UpgradeSchedule, netName dtypes.NetworkName, j journal.Journal) (*MessagePool, error) {
cache, _ := lru.New2Q(build.BlsSignatureCacheSize)
verifcache, _ := lru.New2Q(build.VerifSigCacheSize)
noncecache, _ := lru.New(256)
cache, _ := lru.New2Q[cid.Cid, crypto.Signature](build.BlsSignatureCacheSize)
verifcache, _ := lru.New2Q[string, struct{}](build.VerifSigCacheSize)
noncecache, _ := lru.New[nonceCacheKey, uint64](256)
cfg, err := loadConfig(ctx, ds)
if err != nil {
@@ -1053,7 +1053,7 @@ func (mp *MessagePool) getStateNonce(ctx context.Context, addr address.Address,
n, ok := mp.nonceCache.Get(nk)
if ok {
return n.(uint64), nil
return n, nil
}
act, err := mp.api.GetActorAfter(addr, ts)
@@ -1473,15 +1473,10 @@ func (mp *MessagePool) MessagesForBlocks(ctx context.Context, blks []*types.Bloc
}
func (mp *MessagePool) RecoverSig(msg *types.Message) *types.SignedMessage {
val, ok := mp.blsSigCache.Get(msg.Cid())
sig, ok := mp.blsSigCache.Get(msg.Cid())
if !ok {
return nil
}
sig, ok := val.(crypto.Signature)
if !ok {
log.Errorf("value in signature cache was not a signature (got %T)", val)
return nil
}
return &types.SignedMessage{
Message: *msg,
chain/store/messages.go
+2 -4
View File
@@ -207,9 +207,7 @@ type mmCids struct {
}
func (cs *ChainStore) ReadMsgMetaCids(ctx context.Context, mmc cid.Cid) ([]cid.Cid, []cid.Cid, error) {
o, ok := cs.mmCache.Get(mmc)
if ok {
mmcids := o.(*mmCids)
if mmcids, ok := cs.mmCache.Get(mmc); ok {
return mmcids.bls, mmcids.secpk, nil
}
@@ -229,7 +227,7 @@ func (cs *ChainStore) ReadMsgMetaCids(ctx context.Context, mmc cid.Cid) ([]cid.C
return nil, nil, xerrors.Errorf("loading secpk message cids for block: %w", err)
}
cs.mmCache.Add(mmc, &mmCids{
cs.mmCache.Add(mmc, mmCids{
bls: blscids,
secpk: secpkcids,
})
chain/store/store.go
+7 -8
View File
@@ -11,7 +11,7 @@ import (
"sync"
"time"
lru "github.com/hashicorp/golang-lru"
lru "github.com/hashicorp/golang-lru/v2"
"github.com/ipfs/go-cid"
dstore "github.com/ipfs/go-datastore"
"github.com/ipfs/go-datastore/query"
@@ -120,8 +120,8 @@ type ChainStore struct {
reorgCh chan<- reorg
reorgNotifeeCh chan ReorgNotifee
mmCache *lru.ARCCache // msg meta cache (mh.Messages -> secp, bls []cid)
tsCache *lru.ARCCache
mmCache *lru.ARCCache[cid.Cid, mmCids]
tsCache *lru.ARCCache[types.TipSetKey, *types.TipSet]
evtTypes [1]journal.EventType
journal journal.Journal
@@ -133,8 +133,8 @@ type ChainStore struct {
}
func NewChainStore(chainBs bstore.Blockstore, stateBs bstore.Blockstore, ds dstore.Batching, weight WeightFunc, j journal.Journal) *ChainStore {
c, _ := lru.NewARC(DefaultMsgMetaCacheSize)
tsc, _ := lru.NewARC(DefaultTipSetCacheSize)
c, _ := lru.NewARC[cid.Cid, mmCids](DefaultMsgMetaCacheSize)
tsc, _ := lru.NewARC[types.TipSetKey, *types.TipSet](DefaultTipSetCacheSize)
if j == nil {
j = journal.NilJournal()
}
@@ -818,9 +818,8 @@ func (cs *ChainStore) GetBlock(ctx context.Context, c cid.Cid) (*types.BlockHead
}
func (cs *ChainStore) LoadTipSet(ctx context.Context, tsk types.TipSetKey) (*types.TipSet, error) {
v, ok := cs.tsCache.Get(tsk)
if ok {
return v.(*types.TipSet), nil
if ts, ok := cs.tsCache.Get(tsk); ok {
return ts, nil
}
// Fetch tipset block headers from blockstore in parallel
chain/sub/incoming.go
+16 -21
View File
@@ -7,7 +7,7 @@ import (
"sync"
"time"
lru "github.com/hashicorp/golang-lru"
lru "github.com/hashicorp/golang-lru/v2"
bserv "github.com/ipfs/go-blockservice"
"github.com/ipfs/go-cid"
blocks "github.com/ipfs/go-libipfs/blocks"
@@ -217,7 +217,7 @@ func fetchCids(
type BlockValidator struct {
self peer.ID
peers *lru.TwoQueueCache
peers *lru.TwoQueueCache[peer.ID, int]
killThresh int
@@ -231,7 +231,7 @@ type BlockValidator struct {
}
func NewBlockValidator(self peer.ID, chain *store.ChainStore, cns consensus.Consensus, blacklist func(peer.ID)) *BlockValidator {
p, _ := lru.New2Q(4096)
p, _ := lru.New2Q[peer.ID, int](4096)
return &BlockValidator{
self: self,
peers: p,
@@ -244,21 +244,19 @@ func NewBlockValidator(self peer.ID, chain *store.ChainStore, cns consensus.Cons
}
func (bv *BlockValidator) flagPeer(p peer.ID) {
v, ok := bv.peers.Get(p)
val, ok := bv.peers.Get(p)
if !ok {
bv.peers.Add(p, int(1))
bv.peers.Add(p, 1)
return
}
val := v.(int)
if val >= bv.killThresh {
log.Warnf("blacklisting peer %s", p)
bv.blacklist(p)
return
}
bv.peers.Add(p, v.(int)+1)
bv.peers.Add(p, val+1)
}
func (bv *BlockValidator) Validate(ctx context.Context, pid peer.ID, msg *pubsub.Message) (res pubsub.ValidationResult) {
@@ -293,11 +291,11 @@ func (bv *BlockValidator) Validate(ctx context.Context, pid peer.ID, msg *pubsub
}
type blockReceiptCache struct {
blocks *lru.TwoQueueCache
blocks *lru.TwoQueueCache[cid.Cid, int]
}
func newBlockReceiptCache() *blockReceiptCache {
c, _ := lru.New2Q(8192)
c, _ := lru.New2Q[cid.Cid, int](8192)
return &blockReceiptCache{
blocks: c,
@@ -307,12 +305,12 @@ func newBlockReceiptCache() *blockReceiptCache {
func (brc *blockReceiptCache) add(bcid cid.Cid) int {
val, ok := brc.blocks.Get(bcid)
if !ok {
brc.blocks.Add(bcid, int(1))
brc.blocks.Add(bcid, 1)
return 0
}
brc.blocks.Add(bcid, val.(int)+1)
return val.(int)
brc.blocks.Add(bcid, val+1)
return val
}
type MessageValidator struct {
@@ -466,13 +464,13 @@ type peerMsgInfo struct {
type IndexerMessageValidator struct {
self peer.ID
peerCache *lru.TwoQueueCache
peerCache *lru.TwoQueueCache[address.Address, *peerMsgInfo]
chainApi full.ChainModuleAPI
stateApi full.StateModuleAPI
}
func NewIndexerMessageValidator(self peer.ID, chainApi full.ChainModuleAPI, stateApi full.StateModuleAPI) *IndexerMessageValidator {
peerCache, _ := lru.New2Q(8192)
peerCache, _ := lru.New2Q[address.Address, *peerMsgInfo](8192)
return &IndexerMessageValidator{
self: self,
@@ -515,15 +513,12 @@ func (v *IndexerMessageValidator) Validate(ctx context.Context, pid peer.ID, msg
return pubsub.ValidationReject
}
minerID := minerAddr.String()
msgCid := idxrMsg.Cid
var msgInfo *peerMsgInfo
val, ok := v.peerCache.Get(minerID)
msgInfo, ok := v.peerCache.Get(minerAddr)
if !ok {
msgInfo = &peerMsgInfo{}
} else {
msgInfo = val.(*peerMsgInfo)
}
// Lock this peer's message info.
@@ -544,7 +539,7 @@ func (v *IndexerMessageValidator) Validate(ctx context.Context, pid peer.ID, msg
// Check that the miner ID maps to the peer that sent the message.
err = v.authenticateMessage(ctx, minerAddr, originPeer)
if err != nil {
log.Warnw("cannot authenticate messsage", "err", err, "peer", originPeer, "minerID", minerID)
log.Warnw("cannot authenticate messsage", "err", err, "peer", originPeer, "minerID", minerAddr)
stats.Record(ctx, metrics.IndexerMessageValidationFailure.M(1))
return pubsub.ValidationReject
}
@@ -554,7 +549,7 @@ func (v *IndexerMessageValidator) Validate(ctx context.Context, pid peer.ID, msg
// messages from the same peer are handled concurrently, there is a
// small chance that one msgInfo could replace the other here when
// the info is first cached. This is OK, so no need to prevent it.
v.peerCache.Add(minerID, msgInfo)
v.peerCache.Add(minerAddr, msgInfo)
}
}