messagequeue.go 18.7 KB
Newer Older
1 2 3 4
package messagequeue

import (
	"context"
dirkmc's avatar
dirkmc committed
5
	"math"
6
	"sync"
7 8 9
	"time"

	bsmsg "github.com/ipfs/go-bitswap/message"
dirkmc's avatar
dirkmc committed
10
	pb "github.com/ipfs/go-bitswap/message/pb"
11
	bsnet "github.com/ipfs/go-bitswap/network"
12
	"github.com/ipfs/go-bitswap/wantlist"
dirkmc's avatar
dirkmc committed
13 14
	bswl "github.com/ipfs/go-bitswap/wantlist"
	cid "github.com/ipfs/go-cid"
15
	logging "github.com/ipfs/go-log"
Raúl Kripalani's avatar
Raúl Kripalani committed
16
	peer "github.com/libp2p/go-libp2p-core/peer"
17
	"github.com/libp2p/go-libp2p/p2p/protocol/ping"
Dirk McCormick's avatar
Dirk McCormick committed
18
	"go.uber.org/zap"
19 20 21
)

var log = logging.Logger("bitswap")
Dirk McCormick's avatar
Dirk McCormick committed
22
var sflog = log.Desugar()
23

24 25
const (
	defaultRebroadcastInterval = 30 * time.Second
dirkmc's avatar
dirkmc committed
26 27
	// maxRetries is the number of times to attempt to send a message before
	// giving up
28 29
	maxRetries  = 3
	sendTimeout = 30 * time.Second
dirkmc's avatar
dirkmc committed
30 31 32 33 34 35 36 37 38
	// maxMessageSize is the maximum message size in bytes
	maxMessageSize = 1024 * 1024 * 2
	// sendErrorBackoff is the time to wait before retrying to connect after
	// an error when trying to send a message
	sendErrorBackoff = 100 * time.Millisecond
	// maxPriority is the max priority as defined by the bitswap protocol
	maxPriority = math.MaxInt32
	// sendMessageDebounce is the debounce duration when calling sendMessage()
	sendMessageDebounce = time.Millisecond
39 40
	// when we reach sendMessageCutoff wants/cancels, we'll send the message immediately.
	sendMessageCutoff = 256
Steven Allen's avatar
Steven Allen committed
41 42
	// when we debounce for more than sendMessageMaxDelay, we'll send the
	// message immediately.
43
	sendMessageMaxDelay = 20 * time.Millisecond
44
)
45

46
// MessageNetwork is any network that can connect peers and generate a message
47
// sender.
48 49
type MessageNetwork interface {
	ConnectTo(context.Context, peer.ID) error
50
	NewMessageSender(context.Context, peer.ID, *bsnet.MessageSenderOpts) (bsnet.MessageSender, error)
51 52
	Latency(peer.ID) time.Duration
	Ping(context.Context, peer.ID) ping.Result
53
	Self() peer.ID
54 55
}

56
// MessageQueue implements queue of want messages to send to peers.
57
type MessageQueue struct {
dirkmc's avatar
dirkmc committed
58
	ctx              context.Context
59
	shutdown         func()
dirkmc's avatar
dirkmc committed
60 61
	p                peer.ID
	network          MessageNetwork
62
	dhTimeoutMgr     DontHaveTimeoutManager
dirkmc's avatar
dirkmc committed
63 64 65
	maxMessageSize   int
	sendErrorBackoff time.Duration

Steven Allen's avatar
Steven Allen committed
66
	outgoingWork chan time.Time
dirkmc's avatar
dirkmc committed
67 68 69 70 71 72

	// Take lock whenever any of these variables are modified
	wllock    sync.Mutex
	bcstWants recallWantlist
	peerWants recallWantlist
	cancels   *cid.Set
73
	priority  int32
dirkmc's avatar
dirkmc committed
74 75

	// Dont touch any of these variables outside of run loop
76 77 78 79
	sender                bsnet.MessageSender
	rebroadcastIntervalLk sync.RWMutex
	rebroadcastInterval   time.Duration
	rebroadcastTimer      *time.Timer
80 81 82
	// For performance reasons we just clear out the fields of the message
	// instead of creating a new one every time.
	msg bsmsg.BitSwapMessage
83 84
}

85
// recallWantlist keeps a list of pending wants and a list of sent wants
dirkmc's avatar
dirkmc committed
86 87 88
type recallWantlist struct {
	// The list of wants that have not yet been sent
	pending *bswl.Wantlist
89 90
	// The list of wants that have been sent
	sent *bswl.Wantlist
dirkmc's avatar
dirkmc committed
91 92 93 94
}

func newRecallWantList() recallWantlist {
	return recallWantlist{
95 96
		pending: bswl.New(),
		sent:    bswl.New(),
dirkmc's avatar
dirkmc committed
97 98 99
	}
}

100
// Add want to the pending list
101
func (r *recallWantlist) Add(c cid.Cid, priority int32, wtype pb.Message_Wantlist_WantType) {
dirkmc's avatar
dirkmc committed
102 103 104
	r.pending.Add(c, priority, wtype)
}

105
// Remove wants from both the pending list and the list of sent wants
dirkmc's avatar
dirkmc committed
106
func (r *recallWantlist) Remove(c cid.Cid) {
107
	r.sent.Remove(c)
dirkmc's avatar
dirkmc committed
108 109 110
	r.pending.Remove(c)
}

111
// Remove wants by type from both the pending list and the list of sent wants
dirkmc's avatar
dirkmc committed
112
func (r *recallWantlist) RemoveType(c cid.Cid, wtype pb.Message_Wantlist_WantType) {
113
	r.sent.RemoveType(c, wtype)
dirkmc's avatar
dirkmc committed
114 115 116
	r.pending.RemoveType(c, wtype)
}

117
// MarkSent moves the want from the pending to the sent list
118 119 120 121 122 123 124
//
// Returns true if the want was marked as sent. Returns false if the want wasn't
// pending.
func (r *recallWantlist) MarkSent(e wantlist.Entry) bool {
	if !r.pending.RemoveType(e.Cid, e.WantType) {
		return false
	}
125
	r.sent.Add(e.Cid, e.Priority, e.WantType)
126
	return true
127 128
}

129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167
type peerConn struct {
	p       peer.ID
	network MessageNetwork
}

func newPeerConnection(p peer.ID, network MessageNetwork) *peerConn {
	return &peerConn{p, network}
}

func (pc *peerConn) Ping(ctx context.Context) ping.Result {
	return pc.network.Ping(ctx, pc.p)
}

func (pc *peerConn) Latency() time.Duration {
	return pc.network.Latency(pc.p)
}

// Fires when a timeout occurs waiting for a response from a peer running an
// older version of Bitswap that doesn't support DONT_HAVE messages.
type OnDontHaveTimeout func(peer.ID, []cid.Cid)

// DontHaveTimeoutManager pings a peer to estimate latency so it can set a reasonable
// upper bound on when to consider a DONT_HAVE request as timed out (when connected to
// a peer that doesn't support DONT_HAVE messages)
type DontHaveTimeoutManager interface {
	// Start the manager (idempotent)
	Start()
	// Shutdown the manager (Shutdown is final, manager cannot be restarted)
	Shutdown()
	// AddPending adds the wants as pending a response. If the are not
	// cancelled before the timeout, the OnDontHaveTimeout method will be called.
	AddPending([]cid.Cid)
	// CancelPending removes the wants
	CancelPending([]cid.Cid)
}

// New creates a new MessageQueue.
func New(ctx context.Context, p peer.ID, network MessageNetwork, onDontHaveTimeout OnDontHaveTimeout) *MessageQueue {
	onTimeout := func(ks []cid.Cid) {
Dirk McCormick's avatar
Dirk McCormick committed
168
		log.Infow("Bitswap: timeout waiting for blocks", "cids", ks, "peer", p)
169 170
		onDontHaveTimeout(p, ks)
	}
171
	dhTimeoutMgr := newDontHaveTimeoutMgr(newPeerConnection(p, network), onTimeout)
172
	return newMessageQueue(ctx, p, network, maxMessageSize, sendErrorBackoff, dhTimeoutMgr)
dirkmc's avatar
dirkmc committed
173 174 175
}

// This constructor is used by the tests
176 177 178
func newMessageQueue(ctx context.Context, p peer.ID, network MessageNetwork,
	maxMsgSize int, sendErrorBackoff time.Duration, dhTimeoutMgr DontHaveTimeoutManager) *MessageQueue {

179
	ctx, cancel := context.WithCancel(ctx)
dirkmc's avatar
dirkmc committed
180
	mq := &MessageQueue{
181
		ctx:                 ctx,
182
		shutdown:            cancel,
183
		p:                   p,
dirkmc's avatar
dirkmc committed
184
		network:             network,
185
		dhTimeoutMgr:        dhTimeoutMgr,
dirkmc's avatar
dirkmc committed
186 187 188 189
		maxMessageSize:      maxMsgSize,
		bcstWants:           newRecallWantList(),
		peerWants:           newRecallWantList(),
		cancels:             cid.NewSet(),
Steven Allen's avatar
Steven Allen committed
190
		outgoingWork:        make(chan time.Time, 1),
191
		rebroadcastInterval: defaultRebroadcastInterval,
dirkmc's avatar
dirkmc committed
192 193
		sendErrorBackoff:    sendErrorBackoff,
		priority:            maxPriority,
194 195 196
		// For performance reasons we just clear out the fields of the message
		// after using it, instead of creating a new one every time.
		msg: bsmsg.New(false),
197
	}
dirkmc's avatar
dirkmc committed
198 199

	return mq
200 201
}

dirkmc's avatar
dirkmc committed
202 203 204
// Add want-haves that are part of a broadcast to all connected peers
func (mq *MessageQueue) AddBroadcastWantHaves(wantHaves []cid.Cid) {
	if len(wantHaves) == 0 {
205 206
		return
	}
dirkmc's avatar
dirkmc committed
207 208 209 210 211 212 213 214 215 216 217

	mq.wllock.Lock()
	defer mq.wllock.Unlock()

	for _, c := range wantHaves {
		mq.bcstWants.Add(c, mq.priority, pb.Message_Wantlist_Have)
		mq.priority--

		// We're adding a want-have for the cid, so clear any pending cancel
		// for the cid
		mq.cancels.Remove(c)
218
	}
dirkmc's avatar
dirkmc committed
219 220 221

	// Schedule a message send
	mq.signalWorkReady()
222 223
}

dirkmc's avatar
dirkmc committed
224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259
// Add want-haves and want-blocks for the peer for this message queue.
func (mq *MessageQueue) AddWants(wantBlocks []cid.Cid, wantHaves []cid.Cid) {
	if len(wantBlocks) == 0 && len(wantHaves) == 0 {
		return
	}

	mq.wllock.Lock()
	defer mq.wllock.Unlock()

	for _, c := range wantHaves {
		mq.peerWants.Add(c, mq.priority, pb.Message_Wantlist_Have)
		mq.priority--

		// We're adding a want-have for the cid, so clear any pending cancel
		// for the cid
		mq.cancels.Remove(c)
	}
	for _, c := range wantBlocks {
		mq.peerWants.Add(c, mq.priority, pb.Message_Wantlist_Block)
		mq.priority--

		// We're adding a want-block for the cid, so clear any pending cancel
		// for the cid
		mq.cancels.Remove(c)
	}

	// Schedule a message send
	mq.signalWorkReady()
}

// Add cancel messages for the given keys.
func (mq *MessageQueue) AddCancels(cancelKs []cid.Cid) {
	if len(cancelKs) == 0 {
		return
	}

260 261 262
	// Cancel any outstanding DONT_HAVE timers
	mq.dhTimeoutMgr.CancelPending(cancelKs)

dirkmc's avatar
dirkmc committed
263 264
	mq.wllock.Lock()

265 266
	workReady := false

267
	// Remove keys from broadcast and peer wants, and add to cancels
dirkmc's avatar
dirkmc committed
268
	for _, c := range cancelKs {
269 270 271 272 273
		// Check if a want for the key was sent
		_, wasSentBcst := mq.bcstWants.sent.Contains(c)
		_, wasSentPeer := mq.peerWants.sent.Contains(c)

		// Remove the want from tracking wantlists
dirkmc's avatar
dirkmc committed
274 275
		mq.bcstWants.Remove(c)
		mq.peerWants.Remove(c)
276 277 278 279 280 281

		// Only send a cancel if a want was sent
		if wasSentBcst || wasSentPeer {
			mq.cancels.Add(c)
			workReady = true
		}
dirkmc's avatar
dirkmc committed
282 283
	}

284 285 286 287
	mq.wllock.Unlock()

	// Unlock first to be nice to the scheduler.

dirkmc's avatar
dirkmc committed
288
	// Schedule a message send
289 290 291
	if workReady {
		mq.signalWorkReady()
	}
292 293 294 295 296 297
}

// SetRebroadcastInterval sets a new interval on which to rebroadcast the full wantlist
func (mq *MessageQueue) SetRebroadcastInterval(delay time.Duration) {
	mq.rebroadcastIntervalLk.Lock()
	mq.rebroadcastInterval = delay
298 299 300
	if mq.rebroadcastTimer != nil {
		mq.rebroadcastTimer.Reset(delay)
	}
301
	mq.rebroadcastIntervalLk.Unlock()
302
}
303

dirkmc's avatar
dirkmc committed
304
// Startup starts the processing of messages and rebroadcasting.
305
func (mq *MessageQueue) Startup() {
306 307 308
	mq.rebroadcastIntervalLk.RLock()
	mq.rebroadcastTimer = time.NewTimer(mq.rebroadcastInterval)
	mq.rebroadcastIntervalLk.RUnlock()
309
	go mq.runQueue()
310 311
}

312
// Shutdown stops the processing of messages for a message queue.
313
func (mq *MessageQueue) Shutdown() {
314
	mq.shutdown()
315
}
316

317 318 319
func (mq *MessageQueue) onShutdown() {
	// Shut down the DONT_HAVE timeout manager
	mq.dhTimeoutMgr.Shutdown()
320 321 322 323 324

	// Reset the streamMessageSender
	if mq.sender != nil {
		_ = mq.sender.Reset()
	}
325 326
}

327
func (mq *MessageQueue) runQueue() {
328 329
	defer mq.onShutdown()

Steven Allen's avatar
Steven Allen committed
330 331 332
	// Create a timer for debouncing scheduled work.
	scheduleWork := time.NewTimer(0)
	if !scheduleWork.Stop() {
333 334
		// Need to drain the timer if Stop() returns false
		// See: https://golang.org/pkg/time/#Timer.Stop
Steven Allen's avatar
Steven Allen committed
335 336 337 338
		<-scheduleWork.C
	}

	var workScheduled time.Time
339
	for mq.ctx.Err() == nil {
340
		select {
341 342
		case <-mq.rebroadcastTimer.C:
			mq.rebroadcastWantlist()
Steven Allen's avatar
Steven Allen committed
343 344 345 346 347 348 349 350
		case when := <-mq.outgoingWork:
			// If we have work scheduled, cancel the timer. If we
			// don't, record when the work was scheduled.
			// We send the time on the channel so we accurately
			// track delay.
			if workScheduled.IsZero() {
				workScheduled = when
			} else if !scheduleWork.Stop() {
351
				// Need to drain the timer if Stop() returns false
Steven Allen's avatar
Steven Allen committed
352 353 354 355 356
				<-scheduleWork.C
			}

			// If we have too many updates and/or we've waited too
			// long, send immediately.
357
			if mq.pendingWorkCount() > sendMessageCutoff ||
Steven Allen's avatar
Steven Allen committed
358 359 360 361 362 363 364 365 366 367 368
				time.Since(workScheduled) >= sendMessageMaxDelay {
				mq.sendIfReady()
				workScheduled = time.Time{}
			} else {
				// Otherwise, extend the timer.
				scheduleWork.Reset(sendMessageDebounce)
			}
		case <-scheduleWork.C:
			// We have work scheduled and haven't seen any updates
			// in sendMessageDebounce. Send immediately.
			workScheduled = time.Time{}
dirkmc's avatar
dirkmc committed
369
			mq.sendIfReady()
370
		case <-mq.ctx.Done():
371 372 373 374 375
			return
		}
	}
}

dirkmc's avatar
dirkmc committed
376
// Periodically resend the list of wants to the peer
377 378 379 380 381
func (mq *MessageQueue) rebroadcastWantlist() {
	mq.rebroadcastIntervalLk.RLock()
	mq.rebroadcastTimer.Reset(mq.rebroadcastInterval)
	mq.rebroadcastIntervalLk.RUnlock()

dirkmc's avatar
dirkmc committed
382 383 384 385 386
	// If some wants were transferred from the rebroadcast list
	if mq.transferRebroadcastWants() {
		// Send them out
		mq.sendMessage()
	}
387 388
}

dirkmc's avatar
dirkmc committed
389 390 391 392
// Transfer wants from the rebroadcast lists into the pending lists.
func (mq *MessageQueue) transferRebroadcastWants() bool {
	mq.wllock.Lock()
	defer mq.wllock.Unlock()
393

dirkmc's avatar
dirkmc committed
394
	// Check if there are any wants to rebroadcast
395
	if mq.bcstWants.sent.Len() == 0 && mq.peerWants.sent.Len() == 0 {
dirkmc's avatar
dirkmc committed
396
		return false
397
	}
dirkmc's avatar
dirkmc committed
398

399 400 401
	// Copy sent wants into pending wants lists
	mq.bcstWants.pending.Absorb(mq.bcstWants.sent)
	mq.peerWants.pending.Absorb(mq.peerWants.sent)
dirkmc's avatar
dirkmc committed
402 403

	return true
404 405
}

Steven Allen's avatar
Steven Allen committed
406
func (mq *MessageQueue) signalWorkReady() {
dirkmc's avatar
dirkmc committed
407
	select {
Steven Allen's avatar
Steven Allen committed
408
	case mq.outgoingWork <- time.Now():
dirkmc's avatar
dirkmc committed
409 410
	default:
	}
411 412
}

dirkmc's avatar
dirkmc committed
413 414 415
func (mq *MessageQueue) sendIfReady() {
	if mq.hasPendingWork() {
		mq.sendMessage()
416
	}
dirkmc's avatar
dirkmc committed
417
}
418

dirkmc's avatar
dirkmc committed
419
func (mq *MessageQueue) sendMessage() {
420
	sender, err := mq.initializeSender()
421
	if err != nil {
422 423 424
		// If we fail to initialize the sender, the networking layer will
		// emit a Disconnect event and the MessageQueue will get cleaned up
		log.Infof("Could not open message sender to peer %s: %s", mq.p, err)
425
		mq.Shutdown()
426
		return
427 428
	}

429
	// Make sure the DONT_HAVE timeout manager has started
430 431
	// Note: Start is idempotent
	mq.dhTimeoutMgr.Start()
432

dirkmc's avatar
dirkmc committed
433
	// Convert want lists to a Bitswap Message
434
	message := mq.extractOutgoingMessage(mq.sender.SupportsHave())
435 436 437 438 439

	// After processing the message, clear out its fields to save memory
	defer mq.msg.Reset(false)

	if message.Empty() {
dirkmc's avatar
dirkmc committed
440 441 442
		return
	}

443 444
	wantlist := message.Wantlist()
	mq.logOutgoingMessage(wantlist)
dirkmc's avatar
dirkmc committed
445

446 447 448 449
	if err := sender.SendMsg(mq.ctx, message); err != nil {
		// If the message couldn't be sent, the networking layer will
		// emit a Disconnect event and the MessageQueue will get cleaned up
		log.Infof("Could not send message to peer %s: %s", mq.p, err)
450
		mq.Shutdown()
451 452
		return
	}
dirkmc's avatar
dirkmc committed
453

454 455
	// Set a timer to wait for responses
	mq.simulateDontHaveWithTimeout(wantlist)
dirkmc's avatar
dirkmc committed
456

457 458 459 460 461
	// If the message was too big and only a subset of wants could be
	// sent, schedule sending the rest of the wants in the next
	// iteration of the event loop.
	if mq.hasPendingWork() {
		mq.signalWorkReady()
462 463
	}
}
464

465 466 467 468
// If want-block times out, simulate a DONT_HAVE reponse.
// This is necessary when making requests to peers running an older version of
// Bitswap that doesn't support the DONT_HAVE response, and is also useful to
// mitigate getting blocked by a peer that takes a long time to respond.
469
func (mq *MessageQueue) simulateDontHaveWithTimeout(wantlist []bsmsg.Entry) {
470 471
	// Get the CID of each want-block that expects a DONT_HAVE response
	wants := make([]cid.Cid, 0, len(wantlist))
472 473 474

	mq.wllock.Lock()

475 476 477 478 479
	for _, entry := range wantlist {
		if entry.WantType == pb.Message_Wantlist_Block && entry.SendDontHave {
			// Unlikely, but just in case check that the block hasn't been
			// received in the interim
			c := entry.Cid
480
			if _, ok := mq.peerWants.sent.Contains(c); ok {
481 482 483 484 485 486 487 488 489 490 491
				wants = append(wants, c)
			}
		}
	}

	mq.wllock.Unlock()

	// Add wants to DONT_HAVE timeout manager
	mq.dhTimeoutMgr.AddPending(wants)
}

492
func (mq *MessageQueue) logOutgoingMessage(wantlist []bsmsg.Entry) {
Dirk McCormick's avatar
Dirk McCormick committed
493
	// Save some CPU cycles and allocations if log level is higher than debug
Steven Allen's avatar
Steven Allen committed
494
	if ce := sflog.Check(zap.DebugLevel, "sent message"); ce == nil {
Dirk McCormick's avatar
Dirk McCormick committed
495 496 497
		return
	}

Dirk McCormick's avatar
Dirk McCormick committed
498
	self := mq.network.Self()
499
	for _, e := range wantlist {
Dirk McCormick's avatar
Dirk McCormick committed
500 501
		if e.Cancel {
			if e.WantType == pb.Message_Wantlist_Have {
Steven Allen's avatar
Steven Allen committed
502 503 504 505 506 507
				log.Debugw("sent message",
					"type", "CANCEL_WANT_HAVE",
					"cid", e.Cid,
					"local", self,
					"to", mq.p,
				)
Dirk McCormick's avatar
Dirk McCormick committed
508
			} else {
Steven Allen's avatar
Steven Allen committed
509 510 511 512 513 514
				log.Debugw("sent message",
					"type", "CANCEL_WANT_BLOCK",
					"cid", e.Cid,
					"local", self,
					"to", mq.p,
				)
Dirk McCormick's avatar
Dirk McCormick committed
515 516 517
			}
		} else {
			if e.WantType == pb.Message_Wantlist_Have {
Steven Allen's avatar
Steven Allen committed
518 519 520 521 522 523
				log.Debugw("sent message",
					"type", "WANT_HAVE",
					"cid", e.Cid,
					"local", self,
					"to", mq.p,
				)
Dirk McCormick's avatar
Dirk McCormick committed
524
			} else {
Steven Allen's avatar
Steven Allen committed
525 526 527 528 529 530
				log.Debugw("sent message",
					"type", "WANT_BLOCK",
					"cid", e.Cid,
					"local", self,
					"to", mq.p,
				)
Dirk McCormick's avatar
Dirk McCormick committed
531 532 533 534
			}
		}
	}
}
dirkmc's avatar
dirkmc committed
535

536
// Whether there is work to be processed
dirkmc's avatar
dirkmc committed
537
func (mq *MessageQueue) hasPendingWork() bool {
Steven Allen's avatar
Steven Allen committed
538 539 540
	return mq.pendingWorkCount() > 0
}

541
// The amount of work that is waiting to be processed
Steven Allen's avatar
Steven Allen committed
542
func (mq *MessageQueue) pendingWorkCount() int {
dirkmc's avatar
dirkmc committed
543 544 545
	mq.wllock.Lock()
	defer mq.wllock.Unlock()

Steven Allen's avatar
Steven Allen committed
546
	return mq.bcstWants.pending.Len() + mq.peerWants.pending.Len() + mq.cancels.Len()
dirkmc's avatar
dirkmc committed
547 548
}

549
// Convert the lists of wants into a Bitswap message
550
func (mq *MessageQueue) extractOutgoingMessage(supportsHave bool) bsmsg.BitSwapMessage {
551
	// Get broadcast and regular wantlist entries.
552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573
	mq.wllock.Lock()
	peerEntries := mq.peerWants.pending.Entries()
	bcstEntries := mq.bcstWants.pending.Entries()
	cancels := mq.cancels.Keys()
	if !supportsHave {
		filteredPeerEntries := peerEntries[:0]
		// If the remote peer doesn't support HAVE / DONT_HAVE messages,
		// don't send want-haves (only send want-blocks)
		//
		// Doing this here under the lock makes everything else in this
		// function simpler.
		//
		// TODO: We should _try_ to avoid recording these in the first
		// place if possible.
		for _, e := range peerEntries {
			if e.WantType == pb.Message_Wantlist_Have {
				mq.peerWants.RemoveType(e.Cid, pb.Message_Wantlist_Have)
			} else {
				filteredPeerEntries = append(filteredPeerEntries, e)
			}
		}
		peerEntries = filteredPeerEntries
574
	}
575
	mq.wllock.Unlock()
dirkmc's avatar
dirkmc committed
576

577
	// We prioritize cancels, then regular wants, then broadcast wants.
dirkmc's avatar
dirkmc committed
578

579 580 581 582 583 584
	var (
		msgSize         = 0 // size of message so far
		sentCancels     = 0 // number of cancels in message
		sentPeerEntries = 0 // number of peer entries in message
		sentBcstEntries = 0 // number of broadcast entries in message
	)
dirkmc's avatar
dirkmc committed
585

586
	// Add each cancel to the message
587
	for _, c := range cancels {
588 589 590
		msgSize += mq.msg.Cancel(c)
		sentCancels++

591
		if msgSize >= mq.maxMessageSize {
592
			goto FINISH
593
		}
594
	}
595

596 597 598 599 600 601 602 603
	// Next, add the wants. If we have too many entries to fit into a single
	// message, sort by priority and include the high priority ones first.
	// However, avoid sorting till we really need to as this code is a
	// called frequently.

	// Add each regular want-have / want-block to the message.
	if msgSize+(len(peerEntries)*bsmsg.MaxEntrySize) > mq.maxMessageSize {
		bswl.SortEntries(peerEntries)
dirkmc's avatar
dirkmc committed
604 605
	}

606
	for _, e := range peerEntries {
607 608 609
		msgSize += mq.msg.AddEntry(e.Cid, e.Priority, e.WantType, true)
		sentPeerEntries++

610
		if msgSize >= mq.maxMessageSize {
611
			goto FINISH
612
		}
613
	}
614

615 616 617
	// Add each broadcast want-have to the message.
	if msgSize+(len(bcstEntries)*bsmsg.MaxEntrySize) > mq.maxMessageSize {
		bswl.SortEntries(bcstEntries)
dirkmc's avatar
dirkmc committed
618 619
	}

620
	// Add each broadcast want-have to the message
621
	for _, e := range bcstEntries {
622 623
		// Broadcast wants are sent as want-have
		wantType := pb.Message_Wantlist_Have
dirkmc's avatar
dirkmc committed
624

625 626 627 628 629
		// If the remote peer doesn't support HAVE / DONT_HAVE messages,
		// send a want-block instead
		if !supportsHave {
			wantType = pb.Message_Wantlist_Block
		}
dirkmc's avatar
dirkmc committed
630

631
		msgSize += mq.msg.AddEntry(e.Cid, e.Priority, wantType, false)
632
		sentBcstEntries++
dirkmc's avatar
dirkmc committed
633

634 635 636
		if msgSize >= mq.maxMessageSize {
			goto FINISH
		}
637 638
	}

639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665
FINISH:

	// Finally, re-take the lock, mark sent and remove any entries from our
	// message that we've decided to cancel at the last minute.
	mq.wllock.Lock()
	for _, e := range peerEntries[:sentPeerEntries] {
		if !mq.peerWants.MarkSent(e) {
			// It changed.
			mq.msg.Remove(e.Cid)
		}
	}

	for _, e := range bcstEntries[:sentBcstEntries] {
		if !mq.bcstWants.MarkSent(e) {
			mq.msg.Remove(e.Cid)
		}
	}

	for _, c := range cancels[:sentCancels] {
		if !mq.cancels.Has(c) {
			mq.msg.Remove(c)
		} else {
			mq.cancels.Remove(c)
		}
	}
	mq.wllock.Unlock()

666
	return mq.msg
667 668
}

669 670 671 672 673 674 675 676 677 678
func (mq *MessageQueue) initializeSender() (bsnet.MessageSender, error) {
	if mq.sender == nil {
		opts := &bsnet.MessageSenderOpts{
			MaxRetries:       maxRetries,
			SendTimeout:      sendTimeout,
			SendErrorBackoff: sendErrorBackoff,
		}
		nsender, err := mq.network.NewMessageSender(mq.ctx, mq.p, opts)
		if err != nil {
			return nil, err
679
		}
680

681
		mq.sender = nsender
682
	}
683
	return mq.sender, nil
684
}