multi: Fast relay checked tip blocks.

blockchain/accept.go

@@ -173,6 +173,25 @@ func (b *BlockChain) maybeAcceptBlock(block *dcrutil.Block, flags BehaviorFlags)
 		return 0, err
 	}
 
+	// Notify the caller when the block intends to extend the main chain,
+	// the chain believes it is current, and the block has passed all of the
+	// sanity and contextual checks, such as having valid proof of work,
+	// valid merkle and stake roots, and only containing allowed votes and
+	// revocations.
+	//
+	// This allows the block to be relayed before doing the more expensive
+	// connection checks, because even though the block might still fail
+	// to connect and becomes the new main chain tip, that is quite rare in
+	// practice since a lot of work was expended to create a block that
+	// satisifies the proof of work requirement.
+	//
+	// Notice that the chain lock is not released before sending the
+	// notification.  This is intentional and must not be changed without
+	// understanding why!
+	if b.isCurrent() && b.bestChain.Tip() == prevNode {
+		b.sendNotification(NTNewTipBlockChecked, block)
+	}
+
 	// Fetching a stake node could enable a new DoS vector, so restrict
 	// this only to blocks that are recent in history.
 	if newNode.height < b.bestChain.Tip().height-minMemoryNodes {
@@ -200,7 +219,8 @@ func (b *BlockChain) maybeAcceptBlock(block *dcrutil.Block, flags BehaviorFlags)
 
 	// Notify the caller that the new block was accepted into the block
 	// chain.  The caller would typically want to react by relaying the
-	// inventory to other peers.
+	// inventory to other peers unless it was already relayed above
+	// via NTNewTipBlockChecked.
 	bestHeight := b.bestChain.Tip().height
 	b.chainLock.Unlock()
 	b.sendNotification(NTBlockAccepted, &BlockAcceptedNtfnsData{

blockchain/notifications.go

@@ -21,10 +21,31 @@ type NotificationCallback func(*Notification)
 
 // Constants for the type of a notification message.
 const (
+	// NTNewTipBlockChecked indicates the associated block intends to extend
+	// the current main chain and has passed all of the sanity and
+	// contextual checks such as having valid proof of work, valid merkle
+	// and stake roots, and only containing allowed votes and revocations.
+	//
+	// It should be noted that the block might still ultimately fail to
+	// become the new main chain tip if it contains invalid scripts, double
+	// spends, etc.  However, this is quite rare in practice because a lot
+	// of work was expended to create a block which satisifies the proof of
+	// work requirement.
+	//
+	// Finally, this notification is only sent if the the chain is believed
+	// to be current and the chain lock is NOT released, so consumers must
+	// take care to avoid calling blockchain functions to avoid potential
+	// deadlock.
+	//
+	// Typically, a caller would want to use this notification to relay the
+	// block to the rest of the network without needing to wait for the more
+	// time consuming full connection to take place.
+	NTNewTipBlockChecked NotificationType = iota
+
 	// NTBlockAccepted indicates the associated block was accepted into
 	// the block chain.  Note that this does not necessarily mean it was
 	// added to the main chain.  For that, use NTBlockConnected.
-	NTBlockAccepted NotificationType = iota
+	NTBlockAccepted
 
 	// NTBlockConnected indicates the associated block was connected to the
 	// main chain.
@@ -50,6 +71,7 @@ const (
 // notificationTypeStrings is a map of notification types back to their constant
 // names for pretty printing.
 var notificationTypeStrings = map[NotificationType]string{
+	NTNewTipBlockChecked:    "NTNewTipBlockChecked",
 	NTBlockAccepted:         "NTBlockAccepted",
 	NTBlockConnected:        "NTBlockConnected",
 	NTBlockDisconnected:     "NTBlockDisconnected",
@@ -111,6 +133,7 @@ type TicketNotificationsData struct {
 // Notification defines notification that is sent to the caller via the callback
 // function provided during the call to New and consists of a notification type
 // as well as associated data that depends on the type as follows:
+// 	- NTNewTipBlockChecked:    *dcrutil.Block
 // 	- NTBlockAccepted:         *BlockAcceptedNtfnsData
 // 	- NTBlockConnected:        []*dcrutil.Block of len 2
 // 	- NTBlockDisconnected:     []*dcrutil.Block of len 2

blockmanager.go

@@ -1852,8 +1852,26 @@ func (b *blockManager) notifiedWinningTickets(hash *chainhash.Hash) bool {
 // as request orphan block parents and relay accepted blocks to connected peers.
 func (b *blockManager) handleNotifyMsg(notification *blockchain.Notification) {
 	switch notification.Type {
+	// A block that intends to extend the main chain has passed all sanity and
+	// contextual checks and the chain is believed to be current.  Relay it to
+	// other peers.
+	case blockchain.NTNewTipBlockChecked:
+		// WARNING: The chain lock is not released before sending this
+		// notification, so care must be taken to avoid calling chain functions
+		// which could result in a deadlock.
+		block, ok := notification.Data.(*dcrutil.Block)
+		if !ok {
+			bmgrLog.Warnf("New tip block checkedd notification is not a block.")
+			break
+		}
+
+		// Generate the inventory vector and relay it immediately.
+		iv := wire.NewInvVect(wire.InvTypeBlock, block.Hash())
+		b.server.RelayInventory(iv, block.MsgBlock().Header, true)
+
 	// A block has been accepted into the block chain.  Relay it to other peers
-	// and possibly notify RPC clients with the winning tickets.
+	// (will be ignored if already relayed via NTNewTipBlockChecked) and
+	// possibly notify RPC clients with the winning tickets.
 	case blockchain.NTBlockAccepted:
 		// Don't relay or notify RPC clients with winning tickets if we
 		// are not current. Other peers that are current should already
@@ -1927,9 +1945,9 @@ func (b *blockManager) handleNotifyMsg(notification *blockchain.Notification) {
 			}
 		}
 
-		// Generate the inventory vector and relay it.
+		// Generate the inventory vector and relay it immediately.
 		iv := wire.NewInvVect(wire.InvTypeBlock, blockHash)
-		b.server.RelayInventory(iv, block.MsgBlock().Header)
+		b.server.RelayInventory(iv, block.MsgBlock().Header, true)
 
 	// A block has been connected to the main block chain.
 	case blockchain.NTBlockConnected:

go.mod

@@ -11,7 +11,7 @@ require (
 	github.com/dchest/siphash v1.2.0
 	github.com/decred/base58 v1.0.0
 	github.com/decred/dcrd/addrmgr v1.0.2
-	github.com/decred/dcrd/blockchain v1.0.2
+	github.com/decred/dcrd/blockchain v1.1.0
 	github.com/decred/dcrd/blockchain/stake v1.0.2
 	github.com/decred/dcrd/certgen v1.0.1
 	github.com/decred/dcrd/chaincfg v1.1.1
@@ -27,7 +27,7 @@ require (
 	github.com/decred/dcrd/hdkeychain v1.1.0
 	github.com/decred/dcrd/mempool v1.0.1
 	github.com/decred/dcrd/mining v1.0.1
-	github.com/decred/dcrd/peer v1.0.2
+	github.com/decred/dcrd/peer v1.1.0
 	github.com/decred/dcrd/rpcclient v1.0.1
 	github.com/decred/dcrd/txscript v1.0.1
 	github.com/decred/dcrd/wire v1.1.0

peer/peer.go

@@ -1724,6 +1724,33 @@ func (p *Peer) QueueInventory(invVect *wire.InvVect) {
 	p.outputInvChan <- invVect
 }
 
+// QueueInventoryImmediate adds the passed inventory to the send queue to be
+// sent immediately.  This should typically only be used for inventory that is
+// time sensitive such as new tip blocks or votes.  Normal inventory should be
+// announced via QueueInventory which instead trickles it to the peer in
+// batches.  Inventory that the peer is already known to have is ignored.
+//
+// This function is safe for concurrent access.
+func (p *Peer) QueueInventoryImmediate(invVect *wire.InvVect) {
+	// Don't announce the inventory if the peer is already known to have it.
+	if p.knownInventory.Exists(invVect) {
+		return
+	}
+
+	// Avoid risk of deadlock if goroutine already exited.  The goroutine
+	// we will be sending to hangs around until it knows for a fact that
+	// it is marked as disconnected and *then* it drains the channels.
+	if !p.Connected() {
+		return
+	}
+
+	// Generate and queue a single inv message with the inventory vector.
+	invMsg := wire.NewMsgInvSizeHint(1)
+	invMsg.AddInvVect(invVect)
+	p.AddKnownInventory(invVect)
+	p.outputQueue <- outMsg{msg: invMsg, doneChan: nil}
+}
+
 // Connected returns whether or not the peer is currently connected.
 //
 // This function is safe for concurrent access.

server.go

@@ -91,10 +91,13 @@ type broadcastInventoryDel *wire.InvVect
 type broadcastPruneInventory struct{}
 
 // relayMsg packages an inventory vector along with the newly discovered
-// inventory so the relay has access to that information.
+// inventory and a flag that determines if the relay should happen immediately
+// (it will be put into a trickle queue if false) so the relay has access to
+// that information.
 type relayMsg struct {
-	invVect *wire.InvVect
-	data    interface{}
+	invVect   *wire.InvVect
+	data      interface{}
+	immediate bool
 }
 
 // updatePeerHeightsMsg is a message sent from the blockmanager to the server
@@ -1124,7 +1127,7 @@ func (s *server) AnnounceNewTransactions(newTxs []*dcrutil.Tx) {
 	for _, tx := range newTxs {
 		// Generate the inventory vector and relay it.
 		iv := wire.NewInvVect(wire.InvTypeTx, tx.Hash())
-		s.RelayInventory(iv, tx)
+		s.RelayInventory(iv, tx, false)
 
 		if s.rpcServer != nil {
 			// Notify websocket clients about mempool transactions.
@@ -1389,10 +1392,16 @@ func (s *server) handleRelayInvMsg(state *peerState, msg relayMsg) {
 			}
 		}
 
-		// Queue the inventory to be relayed with the next batch.
-		// It will be ignored if the peer is already known to
-		// have the inventory.
-		sp.QueueInventory(msg.invVect)
+		// Either queue the inventory to be relayed immediately or with
+		// the next batch depending on the immediate flag.
+		//
+		// It will be ignored in either case if the peer is already
+		// known to have the inventory.
+		if msg.immediate {
+			sp.QueueInventoryImmediate(msg.invVect)
+		} else {
+			sp.QueueInventory(msg.invVect)
+		}
 	})
 }
 
@@ -1780,8 +1789,8 @@ func (s *server) BanPeer(sp *serverPeer) {
 
 // RelayInventory relays the passed inventory vector to all connected peers
 // that are not already known to have it.
-func (s *server) RelayInventory(invVect *wire.InvVect, data interface{}) {
-	s.relayInv <- relayMsg{invVect: invVect, data: data}
+func (s *server) RelayInventory(invVect *wire.InvVect, data interface{}, immediate bool) {
+	s.relayInv <- relayMsg{invVect: invVect, data: data, immediate: immediate}
 }
 
 // BroadcastMessage sends msg to all peers currently connected to the server
@@ -2008,7 +2017,7 @@ out:
 			// yet. We periodically resubmit them until they have.
 			for iv, data := range pendingInvs {
 				ivCopy := iv
-				s.RelayInventory(&ivCopy, data)
+				s.RelayInventory(&ivCopy, data, false)
 			}
 
 			// Process at a random time up to 30mins (in seconds)