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

PopRelated should pop all

Browse Source 
Signed-off-by: Jakub Sztandera <kubuxu@protocol.ai>
This commit is contained in:
Jakub Sztandera 2020-10-23 00:33:35 +02:00
parent 24fc7d4cbd
commit 810feee5a1
No known key found for this signature in database
GPG Key ID: 9A9AF56F8B3879BA
2 changed files with 30 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
chain/sync_manager.go
View File

@ -5,7 +5,6 @@ import (
"sort"
"strings"
"sync"
"time"
"github.com/filecoin-project/lotus/chain/types"
peer "github.com/libp2p/go-libp2p-core/peer"
@ -212,16 +211,12 @@ func (sbs *syncBucketSet) removeBucket(toremove *syncTargetBucket) {
sbs.buckets = nbuckets
}
func (sbs *syncBucketSet) PopRelated(ts *types.TipSet) *syncTargetBucket {
var bOut *syncTargetBucket
func (sbs *syncBucketSet) PopRelated(ts *types.TipSet) []*syncTargetBucket {
var bOut []*syncTargetBucket
for _, b := range sbs.buckets {
if b.sameChainAs(ts) {
if bOut == nil {
sbs.removeBucket(b)
bOut = b
} else {
log.Errorf("REPORT THIS more that one related bucket for %s", ts)
}
sbs.removeBucket(b)
bOut = append(bOut, b)
}
}
return bOut
@ -312,8 +307,6 @@ func (sm *syncManager) selectSyncTarget() (*types.TipSet, error) {
}
func (sm *syncManager) syncScheduler() {
t := time.NewTicker(10 * time.Second)
for {
select {
case ts, ok := <-sm.incomingTipSets:
@ -330,16 +323,6 @@ func (sm *syncManager) syncScheduler() {
case <-sm.stop:
log.Info("sync scheduler shutting down")
return
case <-t.C:
activeSyncs := make([]types.TipSetKey, 0, len(sm.activeSyncs))
for tsk := range sm.activeSyncs {
activeSyncs = append(activeSyncs, tsk)
}
sort.Slice(activeSyncs, func(i, j int) bool {
return string(activeSyncs[i].Bytes()) < string(activeSyncs[j].Bytes())
})
log.Infof("activeSyncs: %v, activeSyncTips: %s ", activeSyncs, sm.activeSyncTips.String())
}
}
}
@ -399,14 +382,17 @@ func (sm *syncManager) scheduleProcessResult(res *syncResult) {
}
delete(sm.activeSyncs, res.ts.Key())
relbucket := sm.activeSyncTips.PopRelated(res.ts)
if relbucket != nil {
relbuckets := sm.activeSyncTips.PopRelated(res.ts)
if len(relbuckets) != 0 {
if res.success {
if sm.nextSyncTarget == nil {
sm.nextSyncTarget = relbucket
sm.nextSyncTarget = relbuckets[0]
sm.workerChan = sm.syncTargets
} else {
for _, t := range relbucket.tips {
relbuckets = relbuckets[1:]
}
for _, b := range relbuckets {
for _, t := range b.tips {
sm.syncQueue.Insert(t)
}
}

34
chain/sync_manager_test.go
View File

@ -80,12 +80,14 @@ func TestSyncManagerEdgeCase(t *testing.T) {
t.Logf("c1: %s", c1)
c2 := mock.TipSet(mock.MkBlock(b2, 1, 5))
t.Logf("c2: %s", c2)
d1 := mock.TipSet(mock.MkBlock(c1, 1, 6))
t.Logf("d1: %s", d1)
e1 := mock.TipSet(mock.MkBlock(d1, 1, 7))
t.Logf("e1: %s", e1)
runSyncMgrTest(t, "edgeCase", 1, func(t *testing.T, sm *syncManager, stc chan *syncOp) {
sm.SetPeerHead(ctx, "peer1", a)
assertGetSyncOp(t, stc, a)
time.Sleep(10 * time.Millisecond)
t.Logf("bootstate: %d", sm.bootstrapState)
sm.SetPeerHead(ctx, "peer1", b1)
sm.SetPeerHead(ctx, "peer1", b2)
@ -96,7 +98,6 @@ func TestSyncManagerEdgeCase(t *testing.T) {
if !b1op.ts.Equals(b1) {
b1op, b2op = b2op, b1op
}
t.Logf("activeSyncs: %s: activeSyncTips: %s", sm.activeSyncs, sm.activeSyncTips.String())
sm.SetPeerHead(ctx, "peer2", c2) // c2 is put into activeSyncTips at index 0
sm.SetPeerHead(ctx, "peer2", c1) // c1 is put into activeSyncTips at index 1
@ -106,28 +107,29 @@ func TestSyncManagerEdgeCase(t *testing.T) {
b1op.done() // b1 completes first, is related to a, so it pops activeSyncTips index 0
// even though correct one is index 1
time.Sleep(100 * time.Millisecond)
t.Logf("activeSyncs: %s: activeSyncTips: %s", sm.activeSyncs, sm.activeSyncTips.String())
b2op.done()
// b2 completes and is not related to c1, so it leaves activeSyncTips as it is
for i := 0; i < 10; {
select {
case so := <-stc:
so.done()
default:
i++
time.Sleep(10 * time.Millisecond)
}
}
time.Sleep(10 * time.Millisecond)
waitUntilAllWorkersAreDone(stc)
if len(sm.activeSyncTips.buckets) != 0 {
t.Errorf("activeSyncTips expected empty but got: %s", sm.activeSyncTips.String())
}
t.Logf("activeSyncs: %s: activeSyncTips: %s", sm.activeSyncs, sm.activeSyncTips.String())
})
}
func waitUntilAllWorkersAreDone(stc chan *syncOp) {
for i := 0; i < 10; {
select {
case so := <-stc:
so.done()
default:
i++
time.Sleep(10 * time.Millisecond)
}
}
}
func TestSyncManager(t *testing.T) {
ctx := context.Background()