dbft: implement WithRequestTx callback. Close #59

Author: chenquanyu

consensus/dbft/dbft.go

@@ -201,6 +201,10 @@ type DBFT struct {
 	// about a new consensus payload to be sent.
 	broadcast func(m *dbftproto.Message) error
 
+	// requestTxs is a callback which is called to request the missing
+	// transactions from neighbor nodes.
+	requestTxs func(hashed []common.Hash)
+
 	// various chain/mempool events and subscription management:
 	chainHeadSub    event.Subscription
 	chainHeadEvents chan core.ChainHeadEvent
@@ -465,6 +469,14 @@ func New(config *params.DBFTConfig, db ethdb.Database) (*DBFT, error) {
 			return res
 		}),
 		dbft.WithRequestTx(func(h ...util.Uint256) {
+			if len(h) == 0 {
+				return
+			}
+			hashes := make([]common.Hash, len(h))
+			for i := range h {
+				hashes = append(hashes, common.Hash(h[i]))
+			}
+			c.requestTxs(hashes)
 		}),
 		dbft.WithGetConsensusAddress(func(keys ...dbftCrypto.PublicKey) util.Uint160 {
 			// NextConsensus is filled manually in NewBlockFromContext.
@@ -607,6 +619,11 @@ func (c *DBFT) WithBroadcast(f func(m *dbftproto.Message) error) {
 	c.broadcast = f
 }
 
+// WithRequestTxs sets callback to request the missing transactions from neighbor nodese.
+func (c *DBFT) WithRequestTxs(f func(hashed []common.Hash)) {
+	c.requestTxs = f
+}
+
 // WithTxPool initializes transaction pool API for DBFT interactions with memory pool
 // (fetching unknown transactions).
 func (c *DBFT) WithTxPool(pool txPool) {

eth/backend.go

@@ -279,6 +279,7 @@ func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
 		bft.WithEthAPI(ethAPI)
 		bft.WithBroadcast(eth.dbftSrv.BroadcastMessage)
 		bft.WithTxPool(eth.txPool)
+		bft.WithRequestTxs(eth.handler.BroadcastRequestTxs)
 	}
 
 	// Setup DNS discovery iterators.

eth/handler.go

@@ -53,6 +53,9 @@ const (
 	// All transactions with a higher size will be announced and need to be fetched
 	// by the peer.
 	txMaxBroadcastSize = 4096
+
+	// maxHashCount is the max count of hashes in one batch to RequestTxs
+	maxHashesCount = 500
 )
 
 var (
@@ -685,3 +688,29 @@ func (h *handler) enableSyncedFeatures() {
 		h.chain.TrieDB().SetBufferSize(pathdb.DefaultBufferSize)
 	}
 }
+
+// BroadcastRequestTxs will send GetPooledTransactionsMsg to neighbor peers
+func (h *handler) BroadcastRequestTxs(txHashes []common.Hash) {
+	if len(txHashes) == 0 {
+		return
+	}
+	for i := 0; i <= len(txHashes)/maxHashesCount; i++ {
+		start := i * maxHashesCount
+		stop := (i + 1) * maxHashesCount
+		if stop > len(txHashes) {
+			stop = len(txHashes)
+		}
+		if start == stop {
+			break
+		}
+		// Broadcast RequestTxs
+		for key, peer := range h.peers.peers {
+			err := peer.RequestTxs(txHashes)
+			if err != nil {
+				log.Error("BroadcastRequestTxs", "txHashes", txHashes,
+					"peer", key,
+					"error", err)
+			}
+		}
+	}
+}