Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 12 additions & 8 deletions dot/sync/chain_sync.go
Original file line number Diff line number Diff line change
Expand Up @@ -162,7 +162,9 @@ type chainSync struct {
maxWorkerRetries uint16
slotDuration time.Duration

logSyncPeriod time.Duration
logSyncTicker *time.Ticker
logSyncTickerC <-chan time.Time // channel as field for unit testing
logSyncDone chan struct{}
}

type chainSyncConfig struct {
Expand All @@ -178,6 +180,8 @@ func newChainSync(cfg *chainSyncConfig) *chainSync {
ctx, cancel := context.WithCancel(context.Background())
const syncSamplesToKeep = 30
const logSyncPeriod = 5 * time.Second
logSyncTicker := time.NewTicker(logSyncPeriod)

return &chainSync{
ctx: ctx,
cancel: cancel,
Expand All @@ -197,7 +201,9 @@ func newChainSync(cfg *chainSyncConfig) *chainSync {
minPeers: cfg.minPeers,
maxWorkerRetries: uint16(cfg.maxPeers),
slotDuration: cfg.slotDuration,
logSyncPeriod: logSyncPeriod,
logSyncTicker: logSyncTicker,
logSyncTickerC: logSyncTicker.C,
logSyncDone: make(chan struct{}),
}
}

Expand Down Expand Up @@ -227,6 +233,7 @@ func (cs *chainSync) stop() {
close(cs.pendingBlockDoneCh)
}
cs.cancel()
<-cs.logSyncDone
}

func (cs *chainSync) syncState() chainSyncState {
Expand Down Expand Up @@ -333,8 +340,8 @@ func (cs *chainSync) setPeerHead(p peer.ID, hash common.Hash, number uint) error
}

func (cs *chainSync) logSyncSpeed() {
t := time.NewTicker(cs.logSyncPeriod)
defer t.Stop()
defer close(cs.logSyncDone)
defer cs.logSyncTicker.Stop()

for {
before, err := cs.blockState.BestBlockHeader()
Expand All @@ -347,10 +354,7 @@ func (cs *chainSync) logSyncSpeed() {
}

select {
case <-t.C:
if cs.ctx.Err() != nil {
return
}
case <-cs.logSyncTickerC: // channel of cs.logSyncTicker
case <-cs.ctx.Done():
return
}
Expand Down
83 changes: 36 additions & 47 deletions dot/sync/chain_sync_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -1118,7 +1118,7 @@ func Test_chainSync_logSyncSpeed(t *testing.T) {

type fields struct {
blockStateBuilder func(ctrl *gomock.Controller) BlockState
networkBuilder func(ctrl *gomock.Controller, done chan struct{}) Network
networkBuilder func(ctrl *gomock.Controller, cancel context.CancelFunc) Network
Comment thread
qdm12 marked this conversation as resolved.
Outdated
state chainSyncState
benchmarker *syncBenchmarker
}
Expand All @@ -1131,16 +1131,13 @@ func Test_chainSync_logSyncSpeed(t *testing.T) {
fields: fields{
blockStateBuilder: func(ctrl *gomock.Controller) BlockState {
mockBlockState := NewMockBlockState(ctrl)
mockBlockState.EXPECT().BestBlockHeader().Return(&types.Header{}, nil).AnyTimes()
mockBlockState.EXPECT().BestBlockHeader().Return(&types.Header{}, nil).Times(3)
mockBlockState.EXPECT().GetHighestFinalisedHeader().Return(&types.Header{}, nil)
return mockBlockState
},
networkBuilder: func(ctrl *gomock.Controller, done chan struct{}) Network {
networkBuilder: func(ctrl *gomock.Controller, cancel context.CancelFunc) Network {
Comment thread
qdm12 marked this conversation as resolved.
Outdated
mockNetwork := NewMockNetwork(ctrl)
mockNetwork.EXPECT().Peers().DoAndReturn(func() error {
close(done)
return nil
})
mockNetwork.EXPECT().Peers().Return(nil)
return mockNetwork
},
benchmarker: newSyncBenchmarker(10),
Expand All @@ -1152,16 +1149,13 @@ func Test_chainSync_logSyncSpeed(t *testing.T) {
fields: fields{
blockStateBuilder: func(ctrl *gomock.Controller) BlockState {
mockBlockState := NewMockBlockState(ctrl)
mockBlockState.EXPECT().BestBlockHeader().Return(&types.Header{}, nil).AnyTimes()
mockBlockState.EXPECT().BestBlockHeader().Return(&types.Header{}, nil).Times(3)
mockBlockState.EXPECT().GetHighestFinalisedHeader().Return(&types.Header{}, nil)
return mockBlockState
},
networkBuilder: func(ctrl *gomock.Controller, done chan struct{}) Network {
networkBuilder: func(ctrl *gomock.Controller, cancel context.CancelFunc) Network {
Comment thread
qdm12 marked this conversation as resolved.
Outdated
mockNetwork := NewMockNetwork(ctrl)
mockNetwork.EXPECT().Peers().DoAndReturn(func() error {
close(done)
return nil
})
mockNetwork.EXPECT().Peers().Return(nil)
return mockNetwork
},
benchmarker: newSyncBenchmarker(10),
Expand All @@ -1175,19 +1169,24 @@ func Test_chainSync_logSyncSpeed(t *testing.T) {
t.Parallel()
ctrl := gomock.NewController(t)
ctx, cancel := context.WithCancel(context.Background())
done := make(chan struct{})
tickerChannel := make(chan time.Time)
cs := &chainSync{
ctx: ctx,
cancel: cancel,
blockState: tt.fields.blockStateBuilder(ctrl),
network: tt.fields.networkBuilder(ctrl, done),
state: tt.fields.state,
benchmarker: tt.fields.benchmarker,
logSyncPeriod: time.Millisecond,
ctx: ctx,
cancel: cancel,
blockState: tt.fields.blockStateBuilder(ctrl),
network: tt.fields.networkBuilder(ctrl, cancel),
Comment thread
qdm12 marked this conversation as resolved.
Outdated
state: tt.fields.state,
benchmarker: tt.fields.benchmarker,
logSyncTickerC: tickerChannel,
logSyncTicker: time.NewTicker(time.Hour), // just here to be stopped
logSyncDone: make(chan struct{}),
}

go cs.logSyncSpeed()
<-done
cancel()

tickerChannel <- time.Time{}
cs.cancel()
<-cs.logSyncDone
})
}
}
Expand All @@ -1197,10 +1196,8 @@ func Test_chainSync_start(t *testing.T) {

type fields struct {
blockStateBuilder func(ctrl *gomock.Controller) BlockState
disjointBlockSetBuilder func(ctrl *gomock.Controller) DisjointBlockSet
networkBuilder func(ctrl *gomock.Controller, done chan struct{}) Network
disjointBlockSetBuilder func(ctrl *gomock.Controller, called chan<- struct{}) DisjointBlockSet
benchmarker *syncBenchmarker
slotDuration time.Duration
}
tests := []struct {
name string
Expand All @@ -1211,26 +1208,18 @@ func Test_chainSync_start(t *testing.T) {
fields: fields{
blockStateBuilder: func(ctrl *gomock.Controller) BlockState {
mockBlockState := NewMockBlockState(ctrl)
mockBlockState.EXPECT().BestBlockHeader().Return(&types.Header{}, nil).AnyTimes()
mockBlockState.EXPECT().GetHighestFinalisedHeader().Return(&types.Header{}, nil)
mockBlockState.EXPECT().BestBlockHeader().Return(&types.Header{}, nil).AnyTimes()
mockBlockState.EXPECT().BestBlockHeader().Return(&types.Header{}, nil)
return mockBlockState
},
disjointBlockSetBuilder: func(ctrl *gomock.Controller) DisjointBlockSet {
disjointBlockSetBuilder: func(ctrl *gomock.Controller, called chan<- struct{}) DisjointBlockSet {
mockDisjointBlockSet := NewMockDisjointBlockSet(ctrl)
mockDisjointBlockSet.EXPECT().run(gomock.Any())
mockDisjointBlockSet.EXPECT().run(gomock.AssignableToTypeOf(make(<-chan struct{}))).
DoAndReturn(func(stop <-chan struct{}) {
close(called) // test glue, ideally we would use a ready chan struct passed to run().
})
return mockDisjointBlockSet
},
networkBuilder: func(ctrl *gomock.Controller, done chan struct{}) Network {
mockNetwork := NewMockNetwork(ctrl)
mockNetwork.EXPECT().Peers().DoAndReturn(func() []common.PeerInfo {
close(done)
return nil
})
return mockNetwork
},
slotDuration: defaultSlotDuration,
benchmarker: newSyncBenchmarker(1),
benchmarker: newSyncBenchmarker(1),
},
},
}
Expand All @@ -1240,19 +1229,19 @@ func Test_chainSync_start(t *testing.T) {
t.Parallel()
ctrl := gomock.NewController(t)
ctx, cancel := context.WithCancel(context.Background())
done := make(chan struct{})
disjointBlockSetCalled := make(chan struct{})
cs := &chainSync{
ctx: ctx,
cancel: cancel,
blockState: tt.fields.blockStateBuilder(ctrl),
pendingBlocks: tt.fields.disjointBlockSetBuilder(ctrl),
network: tt.fields.networkBuilder(ctrl, done),
pendingBlocks: tt.fields.disjointBlockSetBuilder(ctrl, disjointBlockSetCalled),
benchmarker: tt.fields.benchmarker,
slotDuration: tt.fields.slotDuration,
logSyncPeriod: time.Second,
slotDuration: time.Hour,
logSyncTicker: time.NewTicker(time.Hour), // just here to be closed
logSyncDone: make(chan struct{}),
}
cs.start()
<-done
<-disjointBlockSetCalled
cs.stop()
})
}
Expand Down