peer, main: allow for queuing multiple invvects

For utreexo tx inv messages, the invvects will be related to each other
as the invvect for the tx will be matched with the position invs. It'll
look like so:

[InvTypeUtreexoTx][TXID1]
[InvTypeProofHash][packed positions1.1] (4 positions)
[InvTypeProofHash][packed positions1.2] (1 position)

At the moment the invvects can only be queued one at a time which may
result in the InvTypeProofHash to be separated from the
InvTypeUtreexoTx. Allowing for queuing multiple invvects eliminates this
problem.

Author: kcalvinalvin

peer/peer.go

@@ -476,7 +476,7 @@ type Peer struct {
 	outputQueue   chan outMsg
 	sendQueue     chan outMsg
 	sendDoneQueue chan struct{}
-	outputInvChan chan *wire.InvVect
+	outputInvChan chan []*wire.InvVect
 	inQuit        chan struct{}
 	queueQuit     chan struct{}
 	outQuit       chan struct{}
@@ -1609,23 +1609,27 @@ out:
 			val := pendingMsgs.Remove(next)
 			p.sendQueue <- val.(outMsg)
 
-		case iv := <-p.outputInvChan:
-			// No handshake?  They'll find out soon enough.
-			if p.VersionKnown() {
-				// If this is a new block, then we'll blast it
-				// out immediately, sipping the inv trickle
-				// queue.
-				if iv.Type == wire.InvTypeBlock ||
-					iv.Type == wire.InvTypeUtreexoBlock ||
-					iv.Type == wire.InvTypeWitnessBlock ||
-					iv.Type == wire.InvTypeWitnessUtreexoBlock {
-
-					invMsg := wire.NewMsgInvSizeHint(1)
-					invMsg.AddInvVect(iv)
-					waiting = queuePacket(outMsg{msg: invMsg},
-						pendingMsgs, waiting)
-				} else {
-					invSendQueue.PushBack(iv)
+		case ivs := <-p.outputInvChan:
+			for i := range ivs {
+				iv := ivs[i]
+
+				// No handshake?  They'll find out soon enough.
+				if p.VersionKnown() {
+					// If this is a new block, then we'll blast it
+					// out immediately, sipping the inv trickle
+					// queue.
+					if iv.Type == wire.InvTypeBlock ||
+						iv.Type == wire.InvTypeUtreexoBlock ||
+						iv.Type == wire.InvTypeWitnessBlock ||
+						iv.Type == wire.InvTypeWitnessUtreexoBlock {
+
+						invMsg := wire.NewMsgInvSizeHint(1)
+						invMsg.AddInvVect(iv)
+						waiting = queuePacket(outMsg{msg: invMsg},
+							pendingMsgs, waiting)
+					} else {
+						invSendQueue.PushBack(iv)
+					}
 				}
 			}
 
@@ -1849,11 +1853,13 @@ func (p *Peer) QueueMessageWithEncoding(msg wire.Message, doneChan chan<- struct
 // Inventory that the peer is already known to have is ignored.
 //
 // This function is safe for concurrent access.
-func (p *Peer) QueueInventory(invVect *wire.InvVect) {
-	// Don't add the inventory to the send queue if the peer is already
-	// known to have it.
-	if p.knownInventory.Contains(invVect) {
-		return
+func (p *Peer) QueueInventory(invVects []*wire.InvVect) {
+	for i := 0; i < len(invVects); i++ {
+		// Don't add the inventory to the send queue if the peer is already
+		// known to have it.
+		if p.knownInventory.Contains(invVects[i]) {
+			invVects = append(invVects[:i], invVects[i+1:]...)
+		}
 	}
 
 	// Avoid risk of deadlock if goroutine already exited.  The goroutine
@@ -1863,7 +1869,7 @@ func (p *Peer) QueueInventory(invVect *wire.InvVect) {
 		return
 	}
 
-	p.outputInvChan <- invVect
+	p.outputInvChan <- invVects
 }
 
 // Connected returns whether or not the peer is currently connected.
@@ -2261,7 +2267,7 @@ func newPeerBase(origCfg *Config, inbound bool) *Peer {
 		outputQueue:     make(chan outMsg, outputBufferSize),
 		sendQueue:       make(chan outMsg, 1),   // nonblocking sync
 		sendDoneQueue:   make(chan struct{}, 1), // nonblocking sync
-		outputInvChan:   make(chan *wire.InvVect, outputBufferSize),
+		outputInvChan:   make(chan []*wire.InvVect, outputBufferSize),
 		inQuit:          make(chan struct{}),
 		queueQuit:       make(chan struct{}),
 		outQuit:         make(chan struct{}),

peer/peer_test.go

@@ -664,9 +664,9 @@ func TestOutboundPeer(t *testing.T) {
 	fakeInv := wire.NewInvVect(wire.InvTypeBlock, fakeBlockHash)
 
 	// Should be noops as the peer could not connect.
-	p.QueueInventory(fakeInv)
+	p.QueueInventory([]*wire.InvVect{fakeInv})
 	p.AddKnownInventory(fakeInv)
-	p.QueueInventory(fakeInv)
+	p.QueueInventory([]*wire.InvVect{fakeInv})
 
 	fakeMsg := wire.NewMsgVerAck()
 	p.QueueMessage(fakeMsg, nil)
@@ -710,7 +710,7 @@ func TestOutboundPeer(t *testing.T) {
 	}
 
 	// Test Queue Inv after connection
-	p1.QueueInventory(fakeInv)
+	p1.QueueInventory([]*wire.InvVect{fakeInv})
 	p1.Disconnect()
 
 	// Test regression

server.go

@@ -2035,7 +2035,7 @@ 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)
+		sp.QueueInventory([]*wire.InvVect{msg.invVect})
 	})
 }