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

fix incoming tipset bucketing

Browse Source
This commit is contained in:
whyrusleeping 2019-12-11 21:41:24 +01:00
parent f16cc0322e
commit f141849df5
4 changed files with 35 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
chain/sub/incoming.go
View File

@ -33,6 +33,9 @@ func HandleIncomingBlocks(ctx context.Context, bsub *pubsub.Subscription, s *cha
} }
go func() { go func() {
log.Infof("New block over pubsub: %s", blk.Cid())
start := time.Now()
log.Debug("about to fetch messages for block from pubsub") log.Debug("about to fetch messages for block from pubsub")
bmsgs, err := s.Bsync.FetchMessagesByCids(context.TODO(), blk.BlsMessages) bmsgs, err := s.Bsync.FetchMessagesByCids(context.TODO(), blk.BlsMessages)
if err != nil { if err != nil {
@ -46,7 +49,8 @@ func HandleIncomingBlocks(ctx context.Context, bsub *pubsub.Subscription, s *cha
return return
} }
log.Debugw("new block over pubsub", "cid", blk.Header.Cid(), "source", msg.GetFrom()) took := time.Since(start)
log.Infow("new block over pubsub", "cid", blk.Header.Cid(), "source", msg.GetFrom(), "msgfetch", took)
if delay := time.Now().Unix() - int64(blk.Header.Timestamp); delay > 5 { if delay := time.Now().Unix() - int64(blk.Header.Timestamp); delay > 5 {
log.Warnf("Received block with large delay %d from miner %s", delay, blk.Header.Miner) log.Warnf("Received block with large delay %d from miner %s", delay, blk.Header.Miner)
} }

1
chain/sync.go
View File

@ -362,6 +362,7 @@ func (syncer *Syncer) tryLoadFullTipSet(cids []cid.Cid) (*store.FullTipSet, erro
} }
func (syncer *Syncer) Sync(ctx context.Context, maybeHead *types.TipSet) error { func (syncer *Syncer) Sync(ctx context.Context, maybeHead *types.TipSet) error {
log.Info("SYNC TIME: ", maybeHead.Cids())
ctx, span := trace.StartSpan(ctx, "chain.Sync") ctx, span := trace.StartSpan(ctx, "chain.Sync")
defer span.End() defer span.End()

6
chain/sync_manager.go
View File

@ -203,9 +203,6 @@ func (stb *syncTargetBucket) sameChainAs(ts *types.TipSet) bool {
if types.CidArrsEqual(ts.Parents(), t.Cids()) { if types.CidArrsEqual(ts.Parents(), t.Cids()) {
return true return true
} }
if types.CidArrsEqual(ts.Parents(), t.Parents()) {
return true
}
} }
return false return false
} }
@ -283,6 +280,7 @@ func (sm *SyncManager) syncScheduler() {
} }
func (sm *SyncManager) scheduleIncoming(ts *types.TipSet) { func (sm *SyncManager) scheduleIncoming(ts *types.TipSet) {
log.Info("scheduling incoming tipset sync: ", ts.Cids())
if sm.getBootstrapState() == BSStateSelected { if sm.getBootstrapState() == BSStateSelected {
sm.setBootstrapState(BSStateScheduled) sm.setBootstrapState(BSStateScheduled)
sm.syncTargets <- ts sm.syncTargets <- ts
@ -295,7 +293,7 @@ func (sm *SyncManager) scheduleIncoming(ts *types.TipSet) {
break break
} }
if types.CidArrsEqual(ts.Parents(), acts.Cids()) || types.CidArrsEqual(ts.Parents(), acts.Parents()) { if types.CidArrsEqual(ts.Parents(), acts.Cids()) {
// sync this next, after that sync process finishes // sync this next, after that sync process finishes
relatedToActiveSync = true relatedToActiveSync = true
} }

27
chain/sync_manager_test.go
View File

@ -74,6 +74,7 @@ func TestSyncManager(t *testing.T) {
b := mock.TipSet(mock.MkBlock(a, 1, 2)) b := mock.TipSet(mock.MkBlock(a, 1, 2))
c1 := mock.TipSet(mock.MkBlock(b, 1, 3)) c1 := mock.TipSet(mock.MkBlock(b, 1, 3))
c2 := mock.TipSet(mock.MkBlock(b, 2, 4)) c2 := mock.TipSet(mock.MkBlock(b, 2, 4))
c3 := mock.TipSet(mock.MkBlock(b, 3, 5))
d := mock.TipSet(mock.MkBlock(c1, 4, 5)) d := mock.TipSet(mock.MkBlock(c1, 4, 5))
runSyncMgrTest(t, "testBootstrap", 1, func(t *testing.T, sm *SyncManager, stc chan *syncOp) { runSyncMgrTest(t, "testBootstrap", 1, func(t *testing.T, sm *SyncManager, stc chan *syncOp) {
@ -120,4 +121,30 @@ func TestSyncManager(t *testing.T) {
assertGetSyncOp(t, stc, d) assertGetSyncOp(t, stc, d)
}) })
runSyncMgrTest(t, "testSyncIncomingTipset", 1, func(t *testing.T, sm *SyncManager, stc chan *syncOp) {
sm.SetPeerHead(ctx, "peer1", a)
assertGetSyncOp(t, stc, a)
sm.SetPeerHead(ctx, "peer2", b)
op := <-stc
op.done()
sm.SetPeerHead(ctx, "peer2", c1)
op1 := <-stc
fmt.Println("op1: ", op1.ts.Cids())
sm.SetPeerHead(ctx, "peer2", c2)
sm.SetPeerHead(ctx, "peer2", c3)
op1.done()
op2 := <-stc
fmt.Println("op2: ", op2.ts.Cids())
op2.done()
op3 := <-stc
fmt.Println("op3: ", op3.ts.Cids())
op3.done()
})
} }