Improve the performance of downloading headers (#1340)

* Add a new message: BlockHeadersMessage.

* Add a new message: BlockHeadersMessage.

* Send a lot of headers as a single message instead of many small messages.

* Keep a short queue of blockHeadersMessages so that there's never a moment when the node is not validating and inserting headers

* Add a missing return statement.

* Remove MsgBlockHeader from payloads.

app/appmessage/message.go

@@ -59,6 +59,7 @@ const (
 	CmdIBDRootHash
 	CmdIBDBlockLocator
 	CmdIBDBlockLocatorHighestHash
+	CmdBlockHeaders
 
 	// rpc
 	CmdGetCurrentNetworkRequestMessage
@@ -153,6 +154,7 @@ var ProtocolMessageCommandToString = map[MessageCommand]string{
 	CmdIBDRootHash:                   "IBDIBDRootHash",
 	CmdIBDBlockLocator:               "IBDBlockLocator",
 	CmdIBDBlockLocatorHighestHash:    "IBDBlockLocatorHighestHash",
+	CmdBlockHeaders:                  "BlockHeaders",
 }
 
 // RPCMessageCommandToString maps all MessageCommands to their string representation

app/appmessage/p2p_blockheaders.go

@@ -0,0 +1,19 @@
+package appmessage
+
+// BlockHeadersMessage represents a kaspa BlockHeaders message
+type BlockHeadersMessage struct {
+	baseMessage
+	BlockHeaders []*MsgBlockHeader
+}
+
+// Command returns the protocol command string for the message
+func (msg *BlockHeadersMessage) Command() MessageCommand {
+	return CmdBlockHeaders
+}
+
+// NewBlockHeadersMessage returns a new kaspa BlockHeaders message
+func NewBlockHeadersMessage(blockHeaders []*MsgBlockHeader) *BlockHeadersMessage {
+	return &BlockHeadersMessage{
+		BlockHeaders: blockHeaders,
+	}
+}

app/protocol/flows/blockrelay/handle_request_headers.go

@@ -43,9 +43,9 @@ func (flow *handleRequestBlocksFlow) start() error {
 			return err
 		}
 
-		batchBlockHeaders := make([]*appmessage.MsgBlockHeader, 0, ibdBatchSize)
 		for !lowHash.Equal(highHash) {
 			log.Debugf("Getting block hashes between %s and %s to %s", lowHash, highHash, flow.peer)
+
 			// GetHashesBetween is a relatively heavy operation so we limit it
 			// in order to avoid locking the consensus for too long
 			const maxBlueScoreDifference = 1 << 10
@@ -53,36 +53,22 @@ func (flow *handleRequestBlocksFlow) start() error {
 			if err != nil {
 				return err
 			}
-			log.Debugf("Got %d headers hashes lowHash %s", len(blockHashes), lowHash)
+			log.Debugf("Got %d header hashes above lowHash %s", len(blockHashes), lowHash)
 
-			offset := 0
+			blockHeaders := make([]*appmessage.MsgBlockHeader, len(blockHashes))
-			for offset < len(blockHashes) {
+			for i, blockHash := range blockHashes {
-				for len(batchBlockHeaders) < ibdBatchSize {
+				blockHeader, err := flow.Domain().Consensus().GetBlockHeader(blockHash)
-					hashAtOffset := blockHashes[offset]
-					blockHeader, err := flow.Domain().Consensus().GetBlockHeader(hashAtOffset)
 				if err != nil {
 					return err
 				}
-					blockHeaderMessage := appmessage.DomainBlockHeaderToBlockHeader(blockHeader)
+				blockHeaders[i] = appmessage.DomainBlockHeaderToBlockHeader(blockHeader)
-					batchBlockHeaders = append(batchBlockHeaders, blockHeaderMessage)
-
-					offset++
-					if offset == len(blockHashes) {
-						break
-					}
 			}
 
-				if len(batchBlockHeaders) < ibdBatchSize {
+			blockHeadersMessage := appmessage.NewBlockHeadersMessage(blockHeaders)
-					break
+			err = flow.outgoingRoute.Enqueue(blockHeadersMessage)
-				}
-
-				err = flow.sendHeaders(batchBlockHeaders)
 			if err != nil {
-					return nil
+				return err
 			}
-				log.Debugf("Sent %d headers to peer %s", len(batchBlockHeaders), flow.peer)
-
-				batchBlockHeaders = make([]*appmessage.MsgBlockHeader, 0, ibdBatchSize)
 
 			message, err := flow.incomingRoute.Dequeue()
 			if err != nil {
@@ -92,20 +78,11 @@ func (flow *handleRequestBlocksFlow) start() error {
 				return protocolerrors.Errorf(true, "received unexpected message type. "+
 					"expected: %s, got: %s", appmessage.CmdRequestNextHeaders, message.Command())
 			}
-			}
 
 			// The next lowHash is the last element in blockHashes
 			lowHash = blockHashes[len(blockHashes)-1]
 		}
 
-		if len(batchBlockHeaders) > 0 {
-			err = flow.sendHeaders(batchBlockHeaders)
-			if err != nil {
-				return nil
-			}
-			log.Debugf("Sent %d headers to peer %s", len(batchBlockHeaders), flow.peer)
-		}
-
 		err = flow.outgoingRoute.Enqueue(appmessage.NewMsgDoneHeaders())
 		if err != nil {
 			return err
@@ -124,13 +101,3 @@ func receiveRequestHeaders(incomingRoute *router.Route) (lowHash *externalapi.Do
 
 	return msgRequestIBDBlocks.LowHash, msgRequestIBDBlocks.HighHash, nil
 }
-
-func (flow *handleRequestBlocksFlow) sendHeaders(headers []*appmessage.MsgBlockHeader) error {
-	for _, msgBlockHeader := range headers {
-		err := flow.outgoingRoute.Enqueue(msgBlockHeader)
-		if err != nil {
-			return err
-		}
-	}
-	return nil
-}

app/protocol/flows/blockrelay/ibd.go

@@ -184,28 +184,48 @@ func (flow *handleRelayInvsFlow) downloadHeaders(highestSharedBlockHash *externa
 		return err
 	}
 
-	blocksReceived := 0
+	// Keep a short queue of blockHeadersMessages so that there's
+	// never a moment when the node is not validating and inserting
+	// headers
+	blockHeadersMessageChan := make(chan *appmessage.BlockHeadersMessage, 2)
+	errChan := make(chan error)
+	doneChan := make(chan interface{})
+	spawn("handleRelayInvsFlow-downloadHeaders", func() {
 		for {
-		msgBlockHeader, doneIBD, err := flow.receiveHeader()
+			blockHeadersMessage, doneIBD, err := flow.receiveHeaders()
 			if err != nil {
-			return err
+				errChan <- err
+				return
 			}
 			if doneIBD {
-			return nil
+				doneChan <- struct{}{}
+				return
 			}
 
-		err = flow.processHeader(msgBlockHeader)
+			blockHeadersMessageChan <- blockHeadersMessage
-		if err != nil {
-			return err
-		}
 
-		blocksReceived++
-		if blocksReceived%ibdBatchSize == 0 {
 			err = flow.outgoingRoute.Enqueue(appmessage.NewMsgRequestNextHeaders())
+			if err != nil {
+				errChan <- err
+				return
+			}
+		}
+	})
+
+	for {
+		select {
+		case blockHeadersMessage := <-blockHeadersMessageChan:
+			for _, header := range blockHeadersMessage.BlockHeaders {
+				err = flow.processHeader(header)
 				if err != nil {
 					return err
 				}
 			}
+		case err := <-errChan:
+			return err
+		case <-doneChan:
+			return nil
+		}
 	}
 }
 
@@ -216,13 +236,13 @@ func (flow *handleRelayInvsFlow) sendRequestHeaders(highestSharedBlockHash *exte
 	return flow.outgoingRoute.Enqueue(msgGetBlockInvs)
 }
 
-func (flow *handleRelayInvsFlow) receiveHeader() (msgIBDBlock *appmessage.MsgBlockHeader, doneIBD bool, err error) {
+func (flow *handleRelayInvsFlow) receiveHeaders() (msgIBDBlock *appmessage.BlockHeadersMessage, doneIBD bool, err error) {
 	message, err := flow.dequeueIncomingMessageAndSkipInvs(common.DefaultTimeout)
 	if err != nil {
 		return nil, false, err
 	}
 	switch message := message.(type) {
-	case *appmessage.MsgBlockHeader:
+	case *appmessage.BlockHeadersMessage:
 		return message, false, nil
 	case *appmessage.MsgDoneHeaders:
 		return nil, true, nil

app/protocol/protocol.go

@@ -136,7 +136,7 @@ func (m *Manager) registerBlockRelayFlows(router *routerpkg.Router, isStopping *
 		m.registerFlow("HandleRelayInvs", router, []appmessage.MessageCommand{
 			appmessage.CmdInvRelayBlock, appmessage.CmdBlock, appmessage.CmdBlockLocator, appmessage.CmdIBDBlock,
 			appmessage.CmdDoneHeaders, appmessage.CmdIBDRootNotFound, appmessage.CmdIBDRootUTXOSetAndBlock,
-			appmessage.CmdHeader, appmessage.CmdIBDRootHash, appmessage.CmdIBDBlockLocatorHighestHash},
+			appmessage.CmdBlockHeaders, appmessage.CmdIBDRootHash, appmessage.CmdIBDBlockLocatorHighestHash},
 			isStopping, errChan, func(incomingRoute *routerpkg.Route, peer *peerpkg.Peer) error {
 				return blockrelay.HandleRelayInvs(m.context, incomingRoute,
 					outgoingRoute, peer)

infrastructure/network/netadapter/server/grpcserver/protowire/messages.proto

@@ -25,7 +25,6 @@ message KaspadMessage {
     VersionMessage version = 20;
     TransactionNotFoundMessage transactionNotFound = 21;
     RejectMessage reject = 22;
-    BlockHeaderMessage blockHeader = 23;
     RequestIBDRootUTXOSetAndBlockMessage requestIBDRootUTXOSetAndBlock = 24;
     IBDRootUTXOSetAndBlockMessage ibdRootUTXOSetAndBlock = 25;
     RequestIBDBlocksMessage requestIBDBlocks = 26;
@@ -34,6 +33,7 @@ message KaspadMessage {
     IBDRootHashMessage ibdRootHash = 29;
     IbdBlockLocatorMessage ibdBlockLocator = 30;
     IbdBlockLocatorHighestHashMessage ibdBlockLocatorHighestHash = 31;
+    BlockHeadersMessage blockHeaders = 32;
 
     GetCurrentNetworkRequestMessage getCurrentNetworkRequest = 1001;
     GetCurrentNetworkResponseMessage getCurrentNetworkResponse = 1002;
@@ -325,6 +325,10 @@ message IbdBlockLocatorHighestHashMessage {
 }
 // IbdBlockLocatorHighestHashMessage end
 
+message BlockHeadersMessage {
+  repeated BlockHeaderMessage blockHeaders = 1;
+}
+
 service P2P {
   rpc MessageStream (stream KaspadMessage) returns (stream KaspadMessage) {}
 }

infrastructure/network/netadapter/server/grpcserver/protowire/p2p_block_headers.go

@@ -0,0 +1,34 @@
+package protowire
+
+import "github.com/kaspanet/kaspad/app/appmessage"
+
+func (x *KaspadMessage_BlockHeaders) toAppMessage() (appmessage.Message, error) {
+	blockHeaders := make([]*appmessage.MsgBlockHeader, len(x.BlockHeaders.BlockHeaders))
+	for i, blockHeader := range x.BlockHeaders.BlockHeaders {
+		var err error
+		blockHeaders[i], err = blockHeader.toAppMessage()
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return &appmessage.BlockHeadersMessage{
+		BlockHeaders: blockHeaders,
+	}, nil
+}
+
+func (x *KaspadMessage_BlockHeaders) fromAppMessage(blockHeadersMessage *appmessage.BlockHeadersMessage) error {
+	blockHeaders := make([]*BlockHeaderMessage, len(blockHeadersMessage.BlockHeaders))
+	for i, blockHeader := range blockHeadersMessage.BlockHeaders {
+		blockHeaders[i] = &BlockHeaderMessage{}
+		err := blockHeaders[i].fromAppMessage(blockHeader)
+		if err != nil {
+			return err
+		}
+	}
+
+	x.BlockHeaders = &BlockHeadersMessage{
+		BlockHeaders: blockHeaders,
+	}
+	return nil
+}

infrastructure/network/netadapter/server/grpcserver/protowire/p2p_header.go

@@ -6,15 +6,6 @@ import (
 	"github.com/pkg/errors"
 )
 
-func (x *KaspadMessage_BlockHeader) toAppMessage() (appmessage.Message, error) {
-	return x.BlockHeader.toAppMessage()
-}
-
-func (x *KaspadMessage_BlockHeader) fromAppMessage(msgBlockHeader *appmessage.MsgBlockHeader) error {
-	x.BlockHeader = new(BlockHeaderMessage)
-	return x.BlockHeader.fromAppMessage(msgBlockHeader)
-}
-
 func (x *BlockHeaderMessage) toAppMessage() (*appmessage.MsgBlockHeader, error) {
 	if len(x.ParentHashes) > appmessage.MaxBlockParents {
 		return nil, errors.Errorf("block header has %d parents, but the maximum allowed amount "+

infrastructure/network/netadapter/server/grpcserver/protowire/wire.go

@@ -195,13 +195,6 @@ func toP2PPayload(message appmessage.Message) (isKaspadMessage_Payload, error) {
 			return nil, err
 		}
 		return payload, nil
-	case *appmessage.MsgBlockHeader:
-		payload := new(KaspadMessage_BlockHeader)
-		err := payload.fromAppMessage(message)
-		if err != nil {
-			return nil, err
-		}
-		return payload, nil
 	case *appmessage.MsgRequestIBDRootUTXOSetAndBlock:
 		payload := new(KaspadMessage_RequestIBDRootUTXOSetAndBlock)
 		err := payload.fromAppMessage(message)
@@ -258,6 +251,13 @@ func toP2PPayload(message appmessage.Message) (isKaspadMessage_Payload, error) {
 			return nil, err
 		}
 		return payload, nil
+	case *appmessage.BlockHeadersMessage:
+		payload := new(KaspadMessage_BlockHeaders)
+		err := payload.fromAppMessage(message)
+		if err != nil {
+			return nil, err
+		}
+		return payload, nil
 	default:
 		return nil, nil
 	}