Skip to content

Commit 35eebab

Browse files
liam-laiwjrjerome
authored andcommitted
xin-106 add generated message into its pool (ethereum#32)
* add debug log and change to contain or add for cache * add generated message into its pool
1 parent e0d66d4 commit 35eebab

File tree

6 files changed

+50
-19
lines changed

6 files changed

+50
-19
lines changed

consensus/XDPoS/engines/engine_v2/engine.go

Lines changed: 25 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -120,7 +120,7 @@ func (x *XDPoS_v2) Prepare(chain consensus.ChainReader, header *types.Header) er
120120
currentRound := x.currentRound
121121
highestQC := x.highestQuorumCert
122122
x.lock.Unlock()
123-
//parentRound := highestQC.ProposedBlockInfo.Round
123+
124124
if (highestQC == nil) || (header.ParentHash != highestQC.ProposedBlockInfo.Hash) {
125125
return consensus.ErrNotReadyToPropose
126126
}
@@ -447,6 +447,16 @@ func (x *XDPoS_v2) VerifyHeader(chain consensus.ChainReader, header *types.Heade
447447
return nil
448448
}
449449

450+
// Utils for test to get current Pool size
451+
func (x *XDPoS_v2) GetVotePoolSize(vote *utils.Vote) int {
452+
return x.votePool.Size(vote)
453+
}
454+
455+
// Utils for test to get Timeout Pool Size
456+
func (x *XDPoS_v2) GetTimeoutPoolSize(timeout *utils.Timeout) int {
457+
return x.timeoutPool.Size(timeout)
458+
}
459+
450460
/*
451461
SyncInfo workflow
452462
*/
@@ -504,6 +514,10 @@ func (x *XDPoS_v2) VerifyVoteMessage(vote *utils.Vote) (bool, error) {
504514
func (x *XDPoS_v2) VoteHandler(chain consensus.ChainReader, voteMsg *utils.Vote) error {
505515
x.lock.Lock()
506516
defer x.lock.Unlock()
517+
return x.voteHandler(chain, voteMsg)
518+
}
519+
520+
func (x *XDPoS_v2) voteHandler(chain consensus.ChainReader, voteMsg *utils.Vote) error {
507521

508522
// 1. checkRoundNumber
509523
if voteMsg.ProposedBlockInfo.Round != x.currentRound {
@@ -516,7 +530,7 @@ func (x *XDPoS_v2) VoteHandler(chain consensus.ChainReader, voteMsg *utils.Vote)
516530
log.Debug("Vote pool threashold reached: %v, number of items in the pool: %v", thresholdReached, numberOfVotesInPool)
517531
err := x.onVotePoolThresholdReached(chain, pooledVotes, voteMsg)
518532
if err != nil {
519-
return nil
533+
return err
520534
}
521535
}
522536

@@ -570,7 +584,10 @@ func (x *XDPoS_v2) VerifyTimeoutMessage(timeoutMsg *utils.Timeout) (bool, error)
570584
func (x *XDPoS_v2) TimeoutHandler(timeout *utils.Timeout) error {
571585
x.lock.Lock()
572586
defer x.lock.Unlock()
587+
return x.timeoutHandler(timeout)
588+
}
573589

590+
func (x *XDPoS_v2) timeoutHandler(timeout *utils.Timeout) error {
574591
// 1. checkRoundNumber
575592
if timeout.Round != x.currentRound {
576593
return &utils.ErrIncomingMessageRoundNotEqualCurrentRound{
@@ -666,7 +683,7 @@ func (x *XDPoS_v2) ProposedBlockHandler(blockChainReader consensus.ChainReader,
666683
return err
667684
}
668685
if verified {
669-
return x.sendVote(blockInfo)
686+
return x.sendVote(blockChainReader, blockInfo)
670687
} else {
671688
log.Info("Failed to pass the voting rule verification", "ProposeBlockHash", blockInfo.Hash)
672689
}
@@ -806,7 +823,7 @@ func (x *XDPoS_v2) verifyVotingRule(blockChainReader consensus.ChainReader, bloc
806823
}
807824

808825
// Once Hot stuff voting rule has verified, this node can then send vote
809-
func (x *XDPoS_v2) sendVote(blockInfo *utils.BlockInfo) error {
826+
func (x *XDPoS_v2) sendVote(chainReader consensus.ChainReader, blockInfo *utils.BlockInfo) error {
810827
// First step: Update the highest Voted round
811828
// Second step: Generate the signature by using node's private key(The signature is the blockInfo signature)
812829
// Third step: Construct the vote struct with the above signature & blockinfo struct
@@ -822,6 +839,8 @@ func (x *XDPoS_v2) sendVote(blockInfo *utils.BlockInfo) error {
822839
ProposedBlockInfo: blockInfo,
823840
Signature: signedHash,
824841
}
842+
843+
x.voteHandler(chainReader, voteMsg)
825844
x.broadcastToBftChannel(voteMsg)
826845
return nil
827846
}
@@ -841,6 +860,8 @@ func (x *XDPoS_v2) sendTimeout() error {
841860
Round: x.currentRound,
842861
Signature: signedHash,
843862
}
863+
864+
x.timeoutHandler(timeoutMsg)
844865
x.broadcastToBftChannel(timeoutMsg)
845866
return nil
846867
}

consensus/XDPoS/utils/pool.go

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,14 @@ func (p *Pool) Add(obj PoolObj) (bool, int, map[common.Hash]PoolObj) {
3636
}
3737
return false, numOfItems, objListKeyed
3838
}
39+
func (p *Pool) Size(obj PoolObj) int {
40+
poolKey := obj.PoolKey()
41+
objListKeyed, ok := p.objList[poolKey]
42+
if !ok {
43+
return 0
44+
}
45+
return len(objListKeyed)
46+
}
3947

4048
func (p *Pool) Clear() {
4149
p.objList = make(map[string]map[common.Hash]PoolObj)

consensus/tests/countdown_test.go

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,8 @@ func TestCountdownTimeoutToSendTimeoutMessage(t *testing.T) {
1616
engineV2.SetNewRoundFaker(utils.Round(1), true)
1717

1818
timeoutMsg := <-engineV2.BroadcastCh
19+
poolSize := engineV2.GetTimeoutPoolSize(timeoutMsg.(*utils.Timeout))
20+
assert.Equal(t, poolSize, 1)
1921
assert.NotNil(t, timeoutMsg)
2022

2123
valid, err := engineV2.VerifyTimeoutMessage(timeoutMsg.(*utils.Timeout))

consensus/tests/proposed_block_test.go

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,9 @@ func TestProcessFirstV2BlockAndSendVoteMsg(t *testing.T) {
2727
}
2828

2929
voteMsg := <-engineV2.BroadcastCh
30+
poolSize := engineV2.GetVotePoolSize(voteMsg.(*utils.Vote))
31+
32+
assert.Equal(t, poolSize, 1)
3033
assert.NotNil(t, voteMsg)
3134
assert.Equal(t, currentBlock.Hash(), voteMsg.(*utils.Vote).ProposedBlockInfo.Hash)
3235

eth/bft/bft_handler.go

Lines changed: 11 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -26,9 +26,9 @@ type Bfter struct {
2626
broadcast BroadcastFns
2727

2828
// Message Cache
29-
knownVotes *lru.ARCCache
30-
knownSyncInfos *lru.ARCCache
31-
knownTimeouts *lru.ARCCache
29+
knownVotes *lru.Cache
30+
knownSyncInfos *lru.Cache
31+
knownTimeouts *lru.Cache
3232
}
3333

3434
type ConsensusFns struct {
@@ -49,9 +49,9 @@ type BroadcastFns struct {
4949
}
5050

5151
func New(broadcasts BroadcastFns, blockCahinReader *core.BlockChain) *Bfter {
52-
knownVotes, _ := lru.NewARC(messageLimit)
53-
knownSyncInfos, _ := lru.NewARC(messageLimit)
54-
knownTimeouts, _ := lru.NewARC(messageLimit)
52+
knownVotes, _ := lru.New(messageLimit)
53+
knownSyncInfos, _ := lru.New(messageLimit)
54+
knownTimeouts, _ := lru.New(messageLimit)
5555
return &Bfter{
5656
quit: make(chan struct{}),
5757
broadcastCh: make(chan interface{}),
@@ -79,9 +79,9 @@ func (b *Bfter) SetConsensusFuns(engine consensus.Engine) {
7979

8080
// TODO: rename
8181
func (b *Bfter) Vote(vote *utils.Vote) error {
82-
log.Info("Receive Vote", "voted block hash", vote.ProposedBlockInfo.Hash.Hex(), "number", vote.ProposedBlockInfo.Number, "round", vote.ProposedBlockInfo.Round)
83-
if b.knownVotes.Contains(vote.Hash()) {
84-
log.Info("Discarded vote, known vote", "voted block hash", vote.ProposedBlockInfo.Hash.Hex(), "number", vote.ProposedBlockInfo.Number, "round", vote.ProposedBlockInfo.Round)
82+
log.Trace("Receive Vote", "vote hash", vote.Hash(), "voted block hash", vote.ProposedBlockInfo.Hash.Hex(), "number", vote.ProposedBlockInfo.Number, "round", vote.ProposedBlockInfo.Round, "signature", vote.Signature)
83+
if exist, _ := b.knownVotes.ContainsOrAdd(vote.Hash(), true); exist {
84+
log.Info("Discarded vote, known vote", "vote hash", vote.Hash(), "voted block hash", vote.ProposedBlockInfo.Hash.Hex(), "number", vote.ProposedBlockInfo.Number, "round", vote.ProposedBlockInfo.Round)
8585
return nil
8686
}
8787

@@ -90,7 +90,6 @@ func (b *Bfter) Vote(vote *utils.Vote) error {
9090
log.Error("Verify BFT Vote", "error", err)
9191
return err
9292
}
93-
b.knownVotes.Add(vote.Hash(), true)
9493
b.broadcastCh <- vote
9594

9695
err = b.consensus.voteHandler(b.blockCahinReader, vote)
@@ -102,7 +101,7 @@ func (b *Bfter) Vote(vote *utils.Vote) error {
102101
}
103102
func (b *Bfter) Timeout(timeout *utils.Timeout) error {
104103
log.Trace("Receive Timeout", "timeout", timeout)
105-
if b.knownVotes.Contains(timeout.Hash()) {
104+
if exist, _ := b.knownTimeouts.ContainsOrAdd(timeout.Hash(), true); exist {
106105
log.Trace("Discarded Timeout, known Timeout", "Signature", timeout.Signature, "hash", timeout.Hash(), "round", timeout.Round)
107106
return nil
108107
}
@@ -111,7 +110,6 @@ func (b *Bfter) Timeout(timeout *utils.Timeout) error {
111110
log.Error("Verify BFT Timeout", "error", err)
112111
return err
113112
}
114-
b.knownTimeouts.Add(timeout.Hash(), true)
115113
b.broadcastCh <- timeout
116114

117115
err = b.consensus.timeoutHandler(timeout)
@@ -127,7 +125,7 @@ func (b *Bfter) Timeout(timeout *utils.Timeout) error {
127125
}
128126
func (b *Bfter) SyncInfo(syncInfo *utils.SyncInfo) error {
129127
log.Trace("Receive SyncInfo", "syncInfo", syncInfo)
130-
if b.knownVotes.Contains(syncInfo.Hash()) {
128+
if exist, _ := b.knownSyncInfos.ContainsOrAdd(syncInfo.Hash(), true); exist {
131129
log.Trace("Discarded SyncInfo, known SyncInfo", "hash", syncInfo.Hash())
132130
return nil
133131
}
@@ -137,7 +135,6 @@ func (b *Bfter) SyncInfo(syncInfo *utils.SyncInfo) error {
137135
return err
138136
}
139137

140-
b.knownSyncInfos.Add(syncInfo.Hash(), true)
141138
b.broadcastCh <- syncInfo
142139

143140
err = b.consensus.syncInfoHandler(b.blockCahinReader, syncInfo)

eth/handler.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -909,7 +909,7 @@ func (pm *ProtocolManager) BroadcastVote(vote *utils.Vote) {
909909
for _, peer := range peers {
910910
peer.SendVote(vote)
911911
}
912-
log.Info("Propagated Vote", "voted block hash", vote.ProposedBlockInfo.Hash.Hex(), "number", vote.ProposedBlockInfo.Number, "round", vote.ProposedBlockInfo.Round, "recipients", len(peers))
912+
log.Info("Propagated Vote", "vote hash", vote.Hash(), "voted block hash", vote.ProposedBlockInfo.Hash.Hex(), "number", vote.ProposedBlockInfo.Number, "round", vote.ProposedBlockInfo.Round, "recipients", len(peers))
913913
}
914914

915915
// BroadcastTimeout will propagate a Timeout to all peers which are not known to

0 commit comments

Comments
 (0)