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

Address review

Browse Source
This commit is contained in:
Aayush Rajasekaran 2021-05-31 18:12:42 -04:00
parent 9ceee6028b
commit 621e4eab0d
6 changed files with 35 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
chain/messagepool/messagepool.go
View File

@ -390,7 +390,7 @@ func New(api Provider, ds dtypes.MetadataDS, netName dtypes.NetworkName, j journ
// enable initial prunes // enable initial prunes
mp.pruneCooldown <- struct{}{} mp.pruneCooldown <- struct{}{}
ctx, cancel := context.WithCancel(context.Background()) ctx, cancel := context.WithCancel(context.TODO())
// load the current tipset and subscribe to head changes _before_ loading local messages // load the current tipset and subscribe to head changes _before_ loading local messages
mp.curTs = api.SubscribeHeadChanges(func(rev, app []*types.TipSet) error { mp.curTs = api.SubscribeHeadChanges(func(rev, app []*types.TipSet) error {
@ -431,7 +431,7 @@ func (mp *MessagePool) resolveToKey(ctx context.Context, addr address.Address) (
} }
// resolve the address // resolve the address
ka, err := mp.api.StateAccountKey(ctx, addr, mp.curTs) ka, err := mp.api.StateAccountKeyAtFinality(ctx, addr, mp.curTs)
if err != nil { if err != nil {
return address.Undef, err return address.Undef, err
} }
@ -875,6 +875,7 @@ func (mp *MessagePool) addLocked(ctx context.Context, m *types.SignedMessage, st
return err return err
} }
// Note: If performance becomes an issue, making this getOrCreatePendingMset will save some work
mset, ok, err := mp.getPendingMset(ctx, m.Message.From) mset, ok, err := mp.getPendingMset(ctx, m.Message.From)
if err != nil { if err != nil {
log.Debug(err) log.Debug(err)

2
chain/messagepool/messagepool_test.go
View File

@ -150,7 +150,7 @@ func (tma *testMpoolAPI) GetActorAfter(addr address.Address, ts *types.TipSet) (
}, nil }, nil
} }
func (tma *testMpoolAPI) StateAccountKey(ctx context.Context, addr address.Address, ts *types.TipSet) (address.Address, error) { func (tma *testMpoolAPI) StateAccountKeyAtFinality(ctx context.Context, addr address.Address, ts *types.TipSet) (address.Address, error) {
if addr.Protocol() != address.BLS && addr.Protocol() != address.SECP256K1 { if addr.Protocol() != address.BLS && addr.Protocol() != address.SECP256K1 {
return address.Undef, fmt.Errorf("given address was not a key addr") return address.Undef, fmt.Errorf("given address was not a key addr")
} }

6
chain/messagepool/provider.go
View File

@ -25,7 +25,7 @@ type Provider interface {
PutMessage(m types.ChainMsg) (cid.Cid, error) PutMessage(m types.ChainMsg) (cid.Cid, error)
PubSubPublish(string, []byte) error PubSubPublish(string, []byte) error
GetActorAfter(address.Address, *types.TipSet) (*types.Actor, error) GetActorAfter(address.Address, *types.TipSet) (*types.Actor, error)
StateAccountKey(context.Context, address.Address, *types.TipSet) (address.Address, error) StateAccountKeyAtFinality(context.Context, address.Address, *types.TipSet) (address.Address, error)
MessagesForBlock(*types.BlockHeader) ([]*types.Message, []*types.SignedMessage, error) MessagesForBlock(*types.BlockHeader) ([]*types.Message, []*types.SignedMessage, error)
MessagesForTipset(*types.TipSet) ([]types.ChainMsg, error) MessagesForTipset(*types.TipSet) ([]types.ChainMsg, error)
LoadTipSet(tsk types.TipSetKey) (*types.TipSet, error) LoadTipSet(tsk types.TipSetKey) (*types.TipSet, error)
@ -74,8 +74,8 @@ func (mpp *mpoolProvider) GetActorAfter(addr address.Address, ts *types.TipSet)
return st.GetActor(addr) return st.GetActor(addr)
} }
func (mpp *mpoolProvider) StateAccountKey(ctx context.Context, addr address.Address, ts *types.TipSet) (address.Address, error) { func (mpp *mpoolProvider) StateAccountKeyAtFinality(ctx context.Context, addr address.Address, ts *types.TipSet) (address.Address, error) {
return mpp.sm.ResolveToKeyAddressReorgProof(ctx, addr, ts) return mpp.sm.ResolveToKeyAddressAtFinality(ctx, addr, ts)
} }
func (mpp *mpoolProvider) MessagesForBlock(h *types.BlockHeader) ([]*types.Message, []*types.SignedMessage, error) { func (mpp *mpoolProvider) MessagesForBlock(h *types.BlockHeader) ([]*types.Message, []*types.SignedMessage, error) {

2
chain/messagepool/pruning.go
View File

@ -57,7 +57,7 @@ func (mp *MessagePool) pruneMessages(ctx context.Context, ts *types.TipSet) erro
mpCfg := mp.getConfig() mpCfg := mp.getConfig()
// we never prune priority addresses // we never prune priority addresses
for _, actor := range mpCfg.PriorityAddrs { for _, actor := range mpCfg.PriorityAddrs {
pk, err := mp.api.StateAccountKey(ctx, actor, mp.curTs) pk, err := mp.resolveToKey(ctx, actor)
if err != nil { if err != nil {
log.Debugf("pruneMessages failed to resolve priority address: %s", err) log.Debugf("pruneMessages failed to resolve priority address: %s", err)
} }

2
chain/messagepool/selection.go
View File

@ -543,7 +543,7 @@ func (mp *MessagePool) selectPriorityMessages(ctx context.Context, pending map[a
var chains []*msgChain var chains []*msgChain
priority := mpCfg.PriorityAddrs priority := mpCfg.PriorityAddrs
for _, actor := range priority { for _, actor := range priority {
pk, err := mp.api.StateAccountKey(ctx, actor, mp.curTs) pk, err := mp.resolveToKey(ctx, actor)
if err != nil { if err != nil {
log.Debugf("mpooladdlocal failed to resolve sender: %s", err) log.Debugf("mpooladdlocal failed to resolve sender: %s", err)
return nil, gasLimit return nil, gasLimit

27
chain/stmgr/stmgr.go
View File

@ -89,6 +89,7 @@ type StateManager struct {
expensiveUpgrades map[abi.ChainEpoch]struct{} expensiveUpgrades map[abi.ChainEpoch]struct{}
stCache map[string][]cid.Cid stCache map[string][]cid.Cid
tCache treeCache
compWait map[string]chan struct{} compWait map[string]chan struct{}
stlk sync.Mutex stlk sync.Mutex
genesisMsigLk sync.Mutex genesisMsigLk sync.Mutex
@ -101,6 +102,12 @@ type StateManager struct {
genesisMarketFunds abi.TokenAmount genesisMarketFunds abi.TokenAmount
} }
// Caches a single state tree
type treeCache struct {
root cid.Cid
tree *state.StateTree
}
func NewStateManager(cs *store.ChainStore) *StateManager { func NewStateManager(cs *store.ChainStore) *StateManager {
sm, err := NewStateManagerWithUpgradeSchedule(cs, DefaultUpgradeSchedule()) sm, err := NewStateManagerWithUpgradeSchedule(cs, DefaultUpgradeSchedule())
if err != nil { if err != nil {
@ -153,6 +160,10 @@ func NewStateManagerWithUpgradeSchedule(cs *store.ChainStore, us UpgradeSchedule
newVM: vm.NewVM, newVM: vm.NewVM,
cs: cs, cs: cs,
stCache: make(map[string][]cid.Cid), stCache: make(map[string][]cid.Cid),
tCache: treeCache{
root: cid.Undef,
tree: nil,
},
compWait: make(map[string]chan struct{}), compWait: make(map[string]chan struct{}),
}, nil }, nil
} }
@ -544,9 +555,9 @@ func (sm *StateManager) ResolveToKeyAddress(ctx context.Context, addr address.Ad
return vm.ResolveToKeyAddr(tree, cst, addr) return vm.ResolveToKeyAddr(tree, cst, addr)
} }
// ResolveToKeyAddressReorgProof is similar to stmgr.ResolveToKeyAddress but fails if the ID address being resolved isn't reorg-stable yet. // ResolveToKeyAddressAtFinality is similar to stmgr.ResolveToKeyAddress but fails if the ID address being resolved isn't reorg-stable yet.
// It should not be used for consensus-critical subsystems. // It should not be used for consensus-critical subsystems.
func (sm *StateManager) ResolveToKeyAddressReorgProof(ctx context.Context, addr address.Address, ts *types.TipSet) (address.Address, error) { func (sm *StateManager) ResolveToKeyAddressAtFinality(ctx context.Context, addr address.Address, ts *types.TipSet) (address.Address, error) {
switch addr.Protocol() { switch addr.Protocol() {
case address.BLS, address.SECP256K1: case address.BLS, address.SECP256K1:
return addr, nil return addr, nil
@ -559,8 +570,8 @@ func (sm *StateManager) ResolveToKeyAddressReorgProof(ctx context.Context, addr
ts = sm.cs.GetHeaviestTipSet() ts = sm.cs.GetHeaviestTipSet()
} }
if ts.Height() > policy.ChainFinality {
var err error var err error
if ts.Height() > policy.ChainFinality {
ts, err = sm.ChainStore().GetTipsetByHeight(ctx, ts.Height()-policy.ChainFinality, ts, true) ts, err = sm.ChainStore().GetTipsetByHeight(ctx, ts.Height()-policy.ChainFinality, ts, true)
if err != nil { if err != nil {
return address.Undef, xerrors.Errorf("failed to load lookback tipset: %w", err) return address.Undef, xerrors.Errorf("failed to load lookback tipset: %w", err)
@ -568,12 +579,20 @@ func (sm *StateManager) ResolveToKeyAddressReorgProof(ctx context.Context, addr
} }
cst := cbor.NewCborStore(sm.cs.StateBlockstore()) cst := cbor.NewCborStore(sm.cs.StateBlockstore())
tree := sm.tCache.tree
tree, err := state.LoadStateTree(cst, ts.ParentState()) if tree == nil || sm.tCache.root != ts.ParentState() {
tree, err = state.LoadStateTree(cst, ts.ParentState())
if err != nil { if err != nil {
return address.Undef, xerrors.Errorf("failed to load parent state tree: %w", err) return address.Undef, xerrors.Errorf("failed to load parent state tree: %w", err)
} }
sm.tCache = treeCache{
root: ts.ParentState(),
tree: tree,
}
}
resolved, err := vm.ResolveToKeyAddr(tree, cst, addr) resolved, err := vm.ResolveToKeyAddr(tree, cst, addr)
if err == nil { if err == nil {
return resolved, nil return resolved, nil