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

fix a potential race with chain reorgs notifees.

Browse Source
This commit is contained in:
Raúl Kripalani 2020-06-12 19:16:54 +01:00
parent 9fbc7aafb4
commit 4e9293ba04
1 changed files with 17 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
chain/store/store.go
View File

@ -48,6 +48,9 @@ var log = logging.Logger("chainstore")
var chainHeadKey = dstore.NewKey("head") var chainHeadKey = dstore.NewKey("head")
// ReorgNotifee represents a callback that gets called upon reorgs.
type ReorgNotifee func(rev, app []*types.TipSet) error
type ChainStore struct { type ChainStore struct {
bs bstore.Blockstore bs bstore.Blockstore
ds dstore.Datastore ds dstore.Datastore
@ -63,8 +66,8 @@ type ChainStore struct {
cindex *ChainIndex cindex *ChainIndex
reorgCh chan<- reorg reorgCh chan<- reorg
headChangeNotifs []func(rev, app []*types.TipSet) error reorgNotifeeCh chan ReorgNotifee
mmCache *lru.ARCCache mmCache *lru.ARCCache
tsCache *lru.ARCCache tsCache *lru.ARCCache
@ -89,8 +92,6 @@ func NewChainStore(bs bstore.Blockstore, ds dstore.Batching, vmcalls runtime.Sys
cs.cindex = ci cs.cindex = ci
cs.reorgCh = cs.reorgWorker(context.TODO())
hcnf := func(rev, app []*types.TipSet) error { hcnf := func(rev, app []*types.TipSet) error {
cs.pubLk.Lock() cs.pubLk.Lock()
defer cs.pubLk.Unlock() defer cs.pubLk.Unlock()
@ -122,7 +123,8 @@ func NewChainStore(bs bstore.Blockstore, ds dstore.Batching, vmcalls runtime.Sys
return nil return nil
} }
cs.headChangeNotifs = append(cs.headChangeNotifs, hcnf, hcmetric) cs.reorgNotifeeCh = make(chan ReorgNotifee)
cs.reorgCh = cs.reorgWorker(context.TODO(), []ReorgNotifee{hcnf, hcmetric})
return cs return cs
} }
@ -211,8 +213,8 @@ func (cs *ChainStore) SubHeadChanges(ctx context.Context) chan []*api.HeadChange
return out return out
} }
func (cs *ChainStore) SubscribeHeadChanges(f func(rev, app []*types.TipSet) error) { func (cs *ChainStore) SubscribeHeadChanges(f ReorgNotifee) {
cs.headChangeNotifs = append(cs.headChangeNotifs, f) cs.reorgNotifeeCh <- f
} }
func (cs *ChainStore) SetGenesis(b *types.BlockHeader) error { func (cs *ChainStore) SetGenesis(b *types.BlockHeader) error {
@ -273,13 +275,19 @@ type reorg struct {
new *types.TipSet new *types.TipSet
} }
func (cs *ChainStore) reorgWorker(ctx context.Context) chan<- reorg { func (cs *ChainStore) reorgWorker(ctx context.Context, initialNotifees []ReorgNotifee) chan<- reorg {
out := make(chan reorg, 32) out := make(chan reorg, 32)
notifees := make([]ReorgNotifee, len(initialNotifees))
copy(notifees, initialNotifees)
go func() { go func() {
defer log.Warn("reorgWorker quit") defer log.Warn("reorgWorker quit")
for { for {
select { select {
case n := <-cs.reorgNotifeeCh:
notifees = append(notifees, n)
case r := <-out: case r := <-out:
revert, apply, err := cs.ReorgOps(r.old, r.new) revert, apply, err := cs.ReorgOps(r.old, r.new)
if err != nil { if err != nil {
@ -293,7 +301,7 @@ func (cs *ChainStore) reorgWorker(ctx context.Context) chan<- reorg {
apply[i], apply[opp] = apply[opp], apply[i] apply[i], apply[opp] = apply[opp], apply[i]
} }
for _, hcf := range cs.headChangeNotifs { for _, hcf := range notifees {
if err := hcf(revert, apply); err != nil { if err := hcf(revert, apply); err != nil {
log.Error("head change func errored (BAD): ", err) log.Error("head change func errored (BAD): ", err)
} }