consensus, eth: Istanbuil interface proposal

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"
 	"math/big"
@@ -97,6 +98,34 @@ type Engine interface {
 	APIs(chain ChainReader) []rpc.API
 }
 
+// Handler defines the interface to implement if a consensus engine needs to send/receive
+// messages from peers.
+type Handler interface {
+	// AddPeer adds a P2P peer
+	AddPeer(id string, peer Peer)
+
+	// RemovePeer removes a P2P peer
+	RemovePeer(id string)
+
+	// HandleMsg handles a message from peer
+	HandleMsg(data p2p.Msg) error
+
+	// SetBlockScheduler sets the scheduler for inserting a block after consensus is achieved.
+	SetBlockScheduler(scheduler BlockScheduler)
+}
+
+// BlockScheduler provides the interface that consensus engine needs to
+// schedule for a future import of a block.
+type BlockScheduler interface {
+	Enqueue(id string, block *types.Block) error
+}
+
+// Peer defines the interface to communicate with peers of the consensus engine.
+type Peer interface {
+	// Send sends the message to this peer
+	Send(msgCode uint64, data interface{}) error
+}
+
 // PoW is a consensus engine based on proof-of-work.
 type PoW interface {
 	Engine

eth/backend.go

@@ -130,7 +130,7 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) {
 		stopDbUpgrade:  stopDbUpgrade,
 		networkId:      config.NetworkId,
 		gasPrice:       config.GasPrice,
-		etherbase:      config.Etherbase,
+		etherbase:      determineEtherbase(ctx, config, chainConfig),
 		bloomRequests:  make(chan chan *bloombits.Retrieval),
 		bloomIndexer:   NewBloomIndexer(chainDb, params.BloomBitsBlocks),
 	}
@@ -216,6 +216,10 @@ func CreateConsensusEngine(ctx *node.ServiceContext, config *ethash.Config, chai
 	if chainConfig.Clique != nil {
 		return clique.New(chainConfig.Clique, db)
 	}
+	// If Istanbul is requested, set it up
+	//if chainConfig.Istanbul != nil {
+	//	// Do something
+	//}
 	// Otherwise assume proof-of-work
 	switch {
 	case config.PowMode == ethash.ModeFake:
@@ -241,6 +245,14 @@ func CreateConsensusEngine(ctx *node.ServiceContext, config *ethash.Config, chai
 	}
 }
 
+func determineEtherbase(ctx *node.ServiceContext, config *Config, chainConfig *params.ChainConfig) common.Address {
+	// Force etherbase to node key address when using Istanbul
+	//if chainConfig.Istanbul != nil {
+	//	return crypto.PubkeyToAddress(ctx.NodeKey().PublicKey)
+	//}
+	return config.Etherbase
+}
+
 // APIs returns the collection of RPC services the ethereum package offers.
 // NOTE, some of these services probably need to be moved to somewhere else.
 func (s *Ethereum) APIs() []rpc.API {
@@ -328,6 +340,12 @@ func (s *Ethereum) Etherbase() (eb common.Address, err error) {
 // set in js console via admin interface or wrapper from cli flags
 func (self *Ethereum) SetEtherbase(etherbase common.Address) {
 	self.lock.Lock()
+	// Disallow re-setting etherbase when using Istanbul
+	//if _, ok := self.engine.(consensus.Istanbul); ok {
+	//	self.lock.Unlock()
+	//	log.Error("Cannot set etherbase in Istanbul consensus")
+	//	return
+	//}
 	self.etherbase = etherbase
 	self.lock.Unlock()
 

eth/handler.go

@@ -94,6 +94,8 @@ type ProtocolManager struct {
 	// wait group is used for graceful shutdowns during downloading
 	// and processing
 	wg sync.WaitGroup
+
+	engine consensus.Engine
 }
 
 // NewProtocolManager returns a new ethereum sub protocol manager. The Ethereum sub protocol manages peers capable
@@ -111,6 +113,7 @@ func NewProtocolManager(config *params.ChainConfig, mode downloader.SyncMode, ne
 		noMorePeers: make(chan struct{}),
 		txsyncCh:    make(chan *txsync),
 		quitSync:    make(chan struct{}),
+		engine:      engine,
 	}
 	// Figure out whether to allow fast sync or not
 	if mode == downloader.FastSync && blockchain.CurrentBlock().NumberU64() > 0 {
@@ -177,6 +180,9 @@ func NewProtocolManager(config *params.ChainConfig, mode downloader.SyncMode, ne
 		return manager.blockchain.InsertChain(blocks)
 	}
 	manager.fetcher = fetcher.New(blockchain.GetBlockByHash, validator, manager.BroadcastBlock, heighter, inserter, manager.removePeer)
+	if handler, ok := manager.engine.(consensus.Handler); ok {
+		handler.SetBlockScheduler(manager.fetcher)
+	}
 
 	return manager, nil
 }
@@ -191,6 +197,10 @@ func (pm *ProtocolManager) removePeer(id string) {
 
 	// Unregister the peer from the downloader and Ethereum peer set
 	pm.downloader.UnregisterPeer(id)
+	// Remove from the consensus engine's peer set
+	if handler, ok := pm.engine.(consensus.Handler); ok {
+		handler.RemovePeer(id)
+	}
 	if err := pm.peers.Unregister(id); err != nil {
 		log.Error("Peer removal failed", "peer", id, "err", err)
 	}
@@ -275,6 +285,10 @@ func (pm *ProtocolManager) handle(p *peer) error {
 		p.Log().Error("Ethereum peer registration failed", "err", err)
 		return err
 	}
+	// Register the peer with the consensus engine
+	if handler, ok := pm.engine.(consensus.Handler); ok {
+		handler.AddPeer(p.id, p)
+	}
 	defer pm.removePeer(p.id)
 
 	// Register the peer in the downloader. If the downloader considers it banned, we disconnect
@@ -673,6 +687,13 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 		}
 		pm.txpool.AddRemotes(txs)
 
+	case msg.Code == ConsensusMsg:
+		if handler, ok := pm.engine.(consensus.Handler); ok {
+			if err := handler.HandleMsg(msg); err != nil {
+				return err
+			}
+		}
+
 	default:
 		return errResp(ErrInvalidMsgCode, "%v", msg.Code)
 	}

eth/peer.go

@@ -130,6 +130,12 @@ func (p *peer) MarkTransaction(hash common.Hash) {
 	p.knownTxs.Add(hash)
 }
 
+// Send writes an RLP-encoded message with the given code.
+// data should encode as an RLP list.
+func (p *peer) Send(msgCode uint64, data interface{}) error {
+	return p2p.Send(p.rw, msgCode, data)
+}
+
 // SendTransactions sends transactions to the peer and includes the hashes
 // in its transaction hash set for future reference.
 func (p *peer) SendTransactions(txs types.Transactions) error {

eth/protocol.go

@@ -32,16 +32,17 @@ import (
 const (
 	eth62 = 62
 	eth63 = 63
+	eth64 = 64
 )
 
 // Official short name of the protocol used during capability negotiation.
 var ProtocolName = "eth"
 
 // Supported versions of the eth protocol (first is primary).
-var ProtocolVersions = []uint{eth63, eth62}
+var ProtocolVersions = []uint{eth64, eth63, eth62}
 
 // Number of implemented message corresponding to different protocol versions.
-var ProtocolLengths = []uint64{17, 8}
+var ProtocolLengths = []uint64{18, 17, 8}
 
 const ProtocolMaxMsgSize = 10 * 1024 * 1024 // Maximum cap on the size of a protocol message
 
@@ -62,6 +63,9 @@ const (
 	NodeDataMsg    = 0x0e
 	GetReceiptsMsg = 0x0f
 	ReceiptsMsg    = 0x10
+
+	// Protocol messages belonging to eth/64
+	ConsensusMsg = 0x11
 )
 
 type errCode int