Merge pull request #133 from getamis/feature/protocol-consensus-interface

cmd, consensus, eth, ethstats: modify consensus interface to handle different consensus messages

Author: markya0616

cmd/geth/misccmd.go

@@ -102,7 +102,6 @@ func version(ctx *cli.Context) error {
 		fmt.Println("Git Commit:", gitCommit)
 	}
 	fmt.Println("Architecture:", runtime.GOARCH)
-	fmt.Println("Protocol Versions:", eth.ProtocolVersions)
 	fmt.Println("Network Id:", eth.DefaultConfig.NetworkId)
 	fmt.Println("Go Version:", runtime.Version())
 	fmt.Println("Operating System:", runtime.GOOS)

consensus/clique/clique.go

@@ -654,3 +654,8 @@ func (c *Clique) APIs(chain consensus.ChainReader) []rpc.API {
 		Public:    false,
 	}}
 }
+
+// Protocol implements consensus.Engine.Protocol
+func (c *Clique) Protocol() consensus.Protocol {
+	return consensus.EthProtocol
+}

consensus/consensus.go

@@ -21,6 +21,7 @@ import (
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/state"
 	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/params"
 	"github.com/ethereum/go-ethereum/rpc"
 )
@@ -90,6 +91,18 @@ type Engine interface {
 
 	// APIs returns the RPC APIs this consensus engine provides.
 	APIs(chain ChainReader) []rpc.API
+
+	// Protocol returns the protocol for this consensus
+	Protocol() Protocol
+}
+
+// Handler should be implemented is the consensus needs to handle and send peer's message
+type Handler interface {
+	// HandleMsg handles a message from peer
+	HandleMsg(address common.Address, data p2p.Msg) (bool, error)
+
+	// SetBroadcaster sets the broadcaster to send message to peers
+	SetBroadcaster(Broadcaster)
 }
 
 // PoW is a consensus engine based on proof-of-work.
@@ -104,12 +117,6 @@ type PoW interface {
 type Istanbul interface {
 	Engine
 
-	// HandleMsg handles a message from peer
-	HandleMsg(addr common.Address, data []byte) error
-
-	// NewChainHead is called if a new chain head block comes
-	NewChainHead(block *types.Block) error
-
 	// Start starts the engine
 	Start(chain ChainReader, inserter func(types.Blocks) (int, error)) error
 

consensus/ethash/ethash.go

@@ -597,3 +597,8 @@ func (ethash *Ethash) APIs(chain consensus.ChainReader) []rpc.API {
 func SeedHash(block uint64) []byte {
 	return seedHash(block)
 }
+
+// Protocol implements consensus.Engine.Protocol
+func (ethash *Ethash) Protocol() consensus.Protocol {
+	return consensus.EthProtocol
+}

consensus/istanbul/backend.go

@@ -17,7 +17,6 @@
 package istanbul
 
 import (
-	"math/big"
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -59,15 +58,9 @@ type Backend interface {
 	// the given validator
 	CheckSignature(data []byte, addr common.Address, sig []byte) error
 
-	// HasBlock checks if the combination of the given hash and height matches any existing blocks
-	HasBlock(hash common.Hash, number *big.Int) bool
-
 	// GetProposer returns the proposer of the given block height
 	GetProposer(number uint64) common.Address
 
-	// ParentValidators returns the validator set of the given proposal's parent block
-	ParentValidators(proposal Proposal) ValidatorSet
-
 	// LastProposal retrieves latest committed proposal and the address of proposer
 	LastProposal() (Proposal, common.Address)
 }

consensus/istanbul/backend/backend.go

@@ -18,7 +18,6 @@ package backend
 
 import (
 	"crypto/ecdsa"
-	"math/big"
 	"sync"
 	"time"
 
@@ -40,6 +39,8 @@ import (
 func New(config *istanbul.Config, eventMux *event.TypeMux, privateKey *ecdsa.PrivateKey, db ethdb.Database) consensus.Istanbul {
 	// Allocate the snapshot caches and create the engine
 	recents, _ := lru.NewARC(inmemorySnapshots)
+	recentMessages, _ := lru.NewARC(inmemoryPeers)
+	knownMessages, _ := lru.NewARC(inmemoryMessages)
 	backend := &backend{
 		config:           config,
 		eventMux:         eventMux,
@@ -52,6 +53,8 @@ func New(config *istanbul.Config, eventMux *event.TypeMux, privateKey *ecdsa.Pri
 		recents:          recents,
 		candidates:       make(map[common.Address]bool),
 		coreStarted:      false,
+		recentMessages:   recentMessages,
+		knownMessages:    knownMessages,
 	}
 	backend.core = istanbulCore.New(backend, backend.config)
 	return backend
@@ -84,6 +87,13 @@ type backend struct {
 	candidatesLock sync.RWMutex
 	// Snapshots for recent block to speed up reorgs
 	recents *lru.ARCCache
+
+	// event subscription for ChainHeadEvent event
+	eventSub    *event.TypeMuxSubscription
+	broadcaster consensus.Broadcaster
+
+	recentMessages *lru.ARCCache // the cache of peer's messages
+	knownMessages  *lru.ARCCache // the cache of self messages
 }
 
 // Address implements istanbul.Backend.Address
@@ -110,18 +120,36 @@ func (sb *backend) Broadcast(valSet istanbul.ValidatorSet, payload []byte) error
 
 // Broadcast implements istanbul.Backend.Gossip
 func (sb *backend) Gossip(valSet istanbul.ValidatorSet, payload []byte) error {
+	hash := istanbul.RLPHash(payload)
+	sb.knownMessages.Add(hash, true)
+
 	targets := make(map[common.Address]bool)
 	for _, val := range valSet.List() {
 		if val.Address() != sb.Address() {
 			targets[val.Address()] = true
 		}
 	}
 
-	if len(targets) > 0 {
-		go sb.eventMux.Post(istanbul.ConsensusDataEvent{
-			Targets: targets,
-			Data:    payload,
-		})
+	if sb.broadcaster != nil && len(targets) > 0 {
+		ps := sb.broadcaster.FindPeers(targets)
+		for addr, p := range ps {
+			ms, ok := sb.recentMessages.Get(addr)
+			var m *lru.ARCCache
+			if ok {
+				m, _ = ms.(*lru.ARCCache)
+				if _, k := m.Get(hash); k {
+					// This peer had this event, skip it
+					continue
+				}
+			} else {
+				m, _ = lru.NewARC(inmemoryMessages)
+			}
+
+			m.Add(hash, true)
+			sb.recentMessages.Add(addr, m)
+
+			go p.Send(istanbulMsg, payload)
+		}
 	}
 	return nil
 }
@@ -162,10 +190,10 @@ func (sb *backend) Commit(proposal istanbul.Proposal, seals [][]byte) error {
 	if _, err := sb.inserter(types.Blocks{block}); err != nil {
 		return err
 	}
-	msg := istanbul.NewCommittedEvent{
-		Block: block,
+
+	if sb.broadcaster != nil {
+		go sb.broadcaster.BroadcastBlock(block, false)
 	}
-	go sb.eventMux.Post(msg)
 	return nil
 }
 
@@ -222,11 +250,6 @@ func (sb *backend) CheckSignature(data []byte, address common.Address, sig []byt
 	return nil
 }
 
-// HasBlock implements istanbul.Backend.HashBlock
-func (sb *backend) HasBlock(hash common.Hash, number *big.Int) bool {
-	return sb.chain.GetHeader(hash, number.Uint64()) != nil
-}
-
 // GetProposer implements istanbul.Backend.GetProposer
 func (sb *backend) GetProposer(number uint64) common.Address {
 	if h := sb.chain.GetHeaderByNumber(number); h != nil {
@@ -236,14 +259,6 @@ func (sb *backend) GetProposer(number uint64) common.Address {
 	return common.Address{}
 }
 
-// ParentValidators implements istanbul.Backend.GetParentValidators
-func (sb *backend) ParentValidators(proposal istanbul.Proposal) istanbul.ValidatorSet {
-	if block, ok := proposal.(*types.Block); ok {
-		return sb.getValidators(block.Number().Uint64()-1, block.ParentHash())
-	}
-	return validator.NewSet(nil, sb.config.ProposerPolicy)
-}
-
 func (sb *backend) getValidators(number uint64, hash common.Hash) istanbul.ValidatorSet {
 	snap, err := sb.snapshot(sb.chain, number, hash, nil)
 	if err != nil {

consensus/istanbul/backend/backend_test.go

@@ -177,19 +177,6 @@ func TestCommit(t *testing.T) {
 	}
 }
 
-func TestHasBlock(t *testing.T) {
-	chain, engine := newBlockChain(1)
-	block := makeBlockWithoutSeal(chain, engine, chain.Genesis())
-	finalBlock, _ := engine.Seal(chain, block, nil)
-	chain.InsertChain(types.Blocks{finalBlock})
-	if engine.HasBlock(block.Hash(), finalBlock.Number()) {
-		t.Errorf("error mismatch: have true, want false")
-	}
-	if !engine.HasBlock(finalBlock.Hash(), finalBlock.Number()) {
-		t.Errorf("error mismatch: have false, want true")
-	}
-}
-
 func TestGetProposer(t *testing.T) {
 	chain, engine := newBlockChain(1)
 	block := makeBlock(chain, engine, chain.Genesis())
@@ -201,20 +188,6 @@ func TestGetProposer(t *testing.T) {
 	}
 }
 
-func TestParentValidators(t *testing.T) {
-	chain, engine := newBlockChain(1)
-	block := makeBlock(chain, engine, chain.Genesis())
-	chain.InsertChain(types.Blocks{block})
-	expected := engine.Validators(block).List()
-	//Block without seal will make empty validator set
-	block = makeBlockWithoutSeal(chain, engine, block)
-	chain.InsertChain(types.Blocks{block})
-	actual := engine.ParentValidators(block).List()
-	if len(expected) != len(actual) || expected[0] != actual[0] {
-		t.Errorf("validator set mismatch: have %v, want %v", actual, expected)
-	}
-}
-
 /**
  * SimpleBackend
  * Private key: bb047e5940b6d83354d9432db7c449ac8fca2248008aaa7271369880f9f11cc1

consensus/istanbul/backend/engine.go

@@ -29,6 +29,7 @@ import (
 	"github.com/ethereum/go-ethereum/consensus/istanbul"
 	istanbulCore "github.com/ethereum/go-ethereum/consensus/istanbul/core"
 	"github.com/ethereum/go-ethereum/consensus/istanbul/validator"
+	"github.com/ethereum/go-ethereum/core"
 	"github.com/ethereum/go-ethereum/core/state"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/crypto/sha3"
@@ -41,6 +42,8 @@ import (
 const (
 	checkpointInterval = 1024 // Number of blocks after which to save the vote snapshot to the database
 	inmemorySnapshots  = 128  // Number of recent vote snapshots to keep in memory
+	inmemoryPeers      = 40
+	inmemoryMessages   = 100
 )
 
 var (
@@ -482,39 +485,6 @@ func (sb *backend) APIs(chain consensus.ChainReader) []rpc.API {
 	}}
 }
 
-// HandleMsg implements consensus.Istanbul.HandleMsg
-func (sb *backend) HandleMsg(addr common.Address, data []byte) error {
-	sb.coreMu.Lock()
-	defer sb.coreMu.Unlock()
-	if !sb.coreStarted {
-		return istanbul.ErrStoppedEngine
-	}
-
-	go sb.istanbulEventMux.Post(istanbul.MessageEvent{
-		Payload: data,
-	})
-	return nil
-}
-
-// NewChainHead implements consensus.Istanbul.NewChainHead
-func (sb *backend) NewChainHead(block *types.Block) error {
-	sb.coreMu.Lock()
-	defer sb.coreMu.Unlock()
-	if !sb.coreStarted {
-		return istanbul.ErrStoppedEngine
-	}
-	p, err := sb.Author(block.Header())
-	if err != nil {
-		sb.logger.Error("Failed to get block proposer", "err", err)
-		return err
-	}
-	go sb.istanbulEventMux.Post(istanbul.FinalCommittedEvent{
-		Proposal: block,
-		Proposer: p,
-	})
-	return nil
-}
-
 // Start implements consensus.Istanbul.Start
 func (sb *backend) Start(chain consensus.ChainReader, inserter func(types.Blocks) (int, error)) error {
 	sb.coreMu.Lock()
@@ -550,6 +520,11 @@ func (sb *backend) Start(chain consensus.ChainReader, inserter func(types.Blocks
 	if err := sb.core.Start(lastSequence, lastProposer, lastProposal); err != nil {
 		return err
 	}
+
+	// subscribe for chain head event
+	sb.eventSub = sb.eventMux.Subscribe(core.ChainHeadEvent{})
+	go sb.eventLoop()
+
 	sb.coreStarted = true
 	return nil
 }
@@ -565,6 +540,7 @@ func (sb *backend) Stop() error {
 		return err
 	}
 	sb.coreStarted = false
+	sb.eventSub.Unsubscribe()
 	return nil
 }