messagequeue.go 17.3 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"
dirkmc's avatar
dirkmc committed
12 13
	bswl "github.com/ipfs/go-bitswap/wantlist"
	cid "github.com/ipfs/go-cid"
14
	logging "github.com/ipfs/go-log"
Raúl Kripalani's avatar
Raúl Kripalani committed
15
	peer "github.com/libp2p/go-libp2p-core/peer"
16
	"github.com/libp2p/go-libp2p/p2p/protocol/ping"
Dirk McCormick's avatar
Dirk McCormick committed
17
	"go.uber.org/zap"
18 19 20
)

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

23 24
const (
	defaultRebroadcastInterval = 30 * time.Second
dirkmc's avatar
dirkmc committed
25 26 27 28 29 30 31 32 33 34 35 36
	// maxRetries is the number of times to attempt to send a message before
	// giving up
	maxRetries = 10
	// 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
37 38
	// when we reach sendMessageCutoff wants/cancels, we'll send the message immediately.
	sendMessageCutoff = 256
Steven Allen's avatar
Steven Allen committed
39 40
	// when we debounce for more than sendMessageMaxDelay, we'll send the
	// message immediately.
41
	sendMessageMaxDelay = 20 * time.Millisecond
42
)
43

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

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

Steven Allen's avatar
Steven Allen committed
63 64
	outgoingWork chan time.Time
	done         chan struct{}
dirkmc's avatar
dirkmc committed
65 66 67 68 69 70 71 72 73

	// Take lock whenever any of these variables are modified
	wllock    sync.Mutex
	bcstWants recallWantlist
	peerWants recallWantlist
	cancels   *cid.Set
	priority  int

	// Dont touch any of these variables outside of run loop
74 75 76 77
	sender                bsnet.MessageSender
	rebroadcastIntervalLk sync.RWMutex
	rebroadcastInterval   time.Duration
	rebroadcastTimer      *time.Timer
78 79
}

dirkmc's avatar
dirkmc committed
80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114
// recallWantlist keeps a list of pending wants, and a list of all wants that
// have ever been requested
type recallWantlist struct {
	// The list of all wants that have been requested, including wants that
	// have been sent and wants that have not yet been sent
	allWants *bswl.Wantlist
	// The list of wants that have not yet been sent
	pending *bswl.Wantlist
}

func newRecallWantList() recallWantlist {
	return recallWantlist{
		allWants: bswl.New(),
		pending:  bswl.New(),
	}
}

// Add want to both the pending list and the list of all wants
func (r *recallWantlist) Add(c cid.Cid, priority int, wtype pb.Message_Wantlist_WantType) {
	r.allWants.Add(c, priority, wtype)
	r.pending.Add(c, priority, wtype)
}

// Remove wants from both the pending list and the list of all wants
func (r *recallWantlist) Remove(c cid.Cid) {
	r.allWants.Remove(c)
	r.pending.Remove(c)
}

// Remove wants by type from both the pending list and the list of all wants
func (r *recallWantlist) RemoveType(c cid.Cid, wtype pb.Message_Wantlist_WantType) {
	r.allWants.RemoveType(c, wtype)
	r.pending.RemoveType(c, wtype)
}

115 116 117 118 119 120 121 122 123 124 125 126 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
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
154
		log.Infow("Bitswap: timeout waiting for blocks", "cids", ks, "peer", p)
155 156 157 158
		onDontHaveTimeout(p, ks)
	}
	dhTimeoutMgr := newDontHaveTimeoutMgr(ctx, newPeerConnection(p, network), onTimeout)
	return newMessageQueue(ctx, p, network, maxMessageSize, sendErrorBackoff, dhTimeoutMgr)
dirkmc's avatar
dirkmc committed
159 160 161
}

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

dirkmc's avatar
dirkmc committed
165
	mq := &MessageQueue{
166 167
		ctx:                 ctx,
		p:                   p,
dirkmc's avatar
dirkmc committed
168
		network:             network,
169
		dhTimeoutMgr:        dhTimeoutMgr,
dirkmc's avatar
dirkmc committed
170 171 172 173
		maxMessageSize:      maxMsgSize,
		bcstWants:           newRecallWantList(),
		peerWants:           newRecallWantList(),
		cancels:             cid.NewSet(),
Steven Allen's avatar
Steven Allen committed
174
		outgoingWork:        make(chan time.Time, 1),
175 176
		done:                make(chan struct{}),
		rebroadcastInterval: defaultRebroadcastInterval,
dirkmc's avatar
dirkmc committed
177 178
		sendErrorBackoff:    sendErrorBackoff,
		priority:            maxPriority,
179
	}
dirkmc's avatar
dirkmc committed
180 181

	return mq
182 183
}

dirkmc's avatar
dirkmc committed
184 185 186
// Add want-haves that are part of a broadcast to all connected peers
func (mq *MessageQueue) AddBroadcastWantHaves(wantHaves []cid.Cid) {
	if len(wantHaves) == 0 {
187 188
		return
	}
dirkmc's avatar
dirkmc committed
189 190 191 192 193 194 195 196 197 198 199

	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)
200
	}
dirkmc's avatar
dirkmc committed
201 202 203

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

dirkmc's avatar
dirkmc committed
206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241
// 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
	}

242 243 244
	// Cancel any outstanding DONT_HAVE timers
	mq.dhTimeoutMgr.CancelPending(cancelKs)

dirkmc's avatar
dirkmc committed
245 246 247
	mq.wllock.Lock()
	defer mq.wllock.Unlock()

248
	// Remove keys from broadcast and peer wants, and add to cancels
dirkmc's avatar
dirkmc committed
249 250 251 252 253 254 255 256
	for _, c := range cancelKs {
		mq.bcstWants.Remove(c)
		mq.peerWants.Remove(c)
		mq.cancels.Add(c)
	}

	// Schedule a message send
	mq.signalWorkReady()
257 258 259 260 261 262
}

// 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
263 264 265
	if mq.rebroadcastTimer != nil {
		mq.rebroadcastTimer.Reset(delay)
	}
266
	mq.rebroadcastIntervalLk.Unlock()
267
}
268

dirkmc's avatar
dirkmc committed
269
// Startup starts the processing of messages and rebroadcasting.
270
func (mq *MessageQueue) Startup() {
271 272 273
	mq.rebroadcastIntervalLk.RLock()
	mq.rebroadcastTimer = time.NewTimer(mq.rebroadcastInterval)
	mq.rebroadcastIntervalLk.RUnlock()
274
	go mq.runQueue()
275 276
}

277
// Shutdown stops the processing of messages for a message queue.
278 279 280
func (mq *MessageQueue) Shutdown() {
	close(mq.done)
}
281

282 283 284 285 286
func (mq *MessageQueue) onShutdown() {
	// Shut down the DONT_HAVE timeout manager
	mq.dhTimeoutMgr.Shutdown()
}

287
func (mq *MessageQueue) runQueue() {
288 289
	defer mq.onShutdown()

Steven Allen's avatar
Steven Allen committed
290 291 292
	// Create a timer for debouncing scheduled work.
	scheduleWork := time.NewTimer(0)
	if !scheduleWork.Stop() {
293 294
		// Need to drain the timer if Stop() returns false
		// See: https://golang.org/pkg/time/#Timer.Stop
Steven Allen's avatar
Steven Allen committed
295 296 297 298
		<-scheduleWork.C
	}

	var workScheduled time.Time
299 300
	for {
		select {
301 302
		case <-mq.rebroadcastTimer.C:
			mq.rebroadcastWantlist()
Steven Allen's avatar
Steven Allen committed
303 304 305 306 307 308 309 310
		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() {
311
				// Need to drain the timer if Stop() returns false
Steven Allen's avatar
Steven Allen committed
312 313 314 315 316
				<-scheduleWork.C
			}

			// If we have too many updates and/or we've waited too
			// long, send immediately.
317
			if mq.pendingWorkCount() > sendMessageCutoff ||
Steven Allen's avatar
Steven Allen committed
318 319 320 321 322 323 324 325 326 327 328
				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
329
			mq.sendIfReady()
330 331 332 333 334
		case <-mq.done:
			if mq.sender != nil {
				mq.sender.Close()
			}
			return
335
		case <-mq.ctx.Done():
336
			if mq.sender != nil {
Steven Allen's avatar
Steven Allen committed
337
				_ = mq.sender.Reset()
338 339 340 341 342 343
			}
			return
		}
	}
}

dirkmc's avatar
dirkmc committed
344
// Periodically resend the list of wants to the peer
345 346 347 348 349
func (mq *MessageQueue) rebroadcastWantlist() {
	mq.rebroadcastIntervalLk.RLock()
	mq.rebroadcastTimer.Reset(mq.rebroadcastInterval)
	mq.rebroadcastIntervalLk.RUnlock()

dirkmc's avatar
dirkmc committed
350 351 352 353 354
	// If some wants were transferred from the rebroadcast list
	if mq.transferRebroadcastWants() {
		// Send them out
		mq.sendMessage()
	}
355 356
}

dirkmc's avatar
dirkmc committed
357 358 359 360
// Transfer wants from the rebroadcast lists into the pending lists.
func (mq *MessageQueue) transferRebroadcastWants() bool {
	mq.wllock.Lock()
	defer mq.wllock.Unlock()
361

dirkmc's avatar
dirkmc committed
362 363 364
	// Check if there are any wants to rebroadcast
	if mq.bcstWants.allWants.Len() == 0 && mq.peerWants.allWants.Len() == 0 {
		return false
365
	}
dirkmc's avatar
dirkmc committed
366 367 368 369 370 371

	// Copy all wants into pending wants lists
	mq.bcstWants.pending.Absorb(mq.bcstWants.allWants)
	mq.peerWants.pending.Absorb(mq.peerWants.allWants)

	return true
372 373
}

Steven Allen's avatar
Steven Allen committed
374
func (mq *MessageQueue) signalWorkReady() {
dirkmc's avatar
dirkmc committed
375
	select {
Steven Allen's avatar
Steven Allen committed
376
	case mq.outgoingWork <- time.Now():
dirkmc's avatar
dirkmc committed
377 378
	default:
	}
379 380
}

dirkmc's avatar
dirkmc committed
381 382 383
func (mq *MessageQueue) sendIfReady() {
	if mq.hasPendingWork() {
		mq.sendMessage()
384
	}
dirkmc's avatar
dirkmc committed
385
}
386

dirkmc's avatar
dirkmc committed
387
func (mq *MessageQueue) sendMessage() {
388
	err := mq.initializeSender()
389 390 391
	if err != nil {
		log.Infof("cant open message sender to peer %s: %s", mq.p, err)
		// TODO: cant connect, what now?
dirkmc's avatar
dirkmc committed
392 393
		// TODO: should we stop using this connection and clear the want list
		// to avoid using up memory?
394
		return
395 396
	}

397
	// Make sure the DONT_HAVE timeout manager has started
398 399
	// Note: Start is idempotent
	mq.dhTimeoutMgr.Start()
400

dirkmc's avatar
dirkmc committed
401 402 403 404 405 406
	// Convert want lists to a Bitswap Message
	message, onSent := mq.extractOutgoingMessage(mq.sender.SupportsHave())
	if message == nil || message.Empty() {
		return
	}

Dirk McCormick's avatar
Dirk McCormick committed
407
	mq.logOutgoingMessage(message)
dirkmc's avatar
dirkmc committed
408 409 410

	// Try to send this message repeatedly
	for i := 0; i < maxRetries; i++ {
411
		if mq.attemptSendAndRecovery(message) {
dirkmc's avatar
dirkmc committed
412 413 414
			// We were able to send successfully.
			onSent()

415 416
			mq.simulateDontHaveWithTimeout(message)

dirkmc's avatar
dirkmc committed
417 418 419 420 421 422 423
			// 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()
			}

424 425
			return
		}
426 427
	}
}
428

429 430 431 432
// 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.
433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455
func (mq *MessageQueue) simulateDontHaveWithTimeout(msg bsmsg.BitSwapMessage) {
	mq.wllock.Lock()

	// Get the CID of each want-block that expects a DONT_HAVE response
	wantlist := msg.Wantlist()
	wants := make([]cid.Cid, 0, len(wantlist))
	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
			if _, ok := mq.peerWants.allWants.Contains(c); ok {
				wants = append(wants, c)
			}
		}
	}

	mq.wllock.Unlock()

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

Dirk McCormick's avatar
Dirk McCormick committed
456
func (mq *MessageQueue) logOutgoingMessage(msg bsmsg.BitSwapMessage) {
Dirk McCormick's avatar
Dirk McCormick committed
457 458 459 460 461
	// Save some CPU cycles and allocations if log level is higher than debug
	if ce := sflog.Check(zap.DebugLevel, "Bitswap -> send wants"); ce == nil {
		return
	}

Dirk McCormick's avatar
Dirk McCormick committed
462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479
	self := mq.network.Self()
	entries := msg.Wantlist()
	for _, e := range entries {
		if e.Cancel {
			if e.WantType == pb.Message_Wantlist_Have {
				log.Debugw("Bitswap -> cancel-have", "local", self, "to", mq.p, "cid", e.Cid)
			} else {
				log.Debugw("Bitswap -> cancel-block", "local", self, "to", mq.p, "cid", e.Cid)
			}
		} else {
			if e.WantType == pb.Message_Wantlist_Have {
				log.Debugw("Bitswap -> want-have", "local", self, "to", mq.p, "cid", e.Cid)
			} else {
				log.Debugw("Bitswap -> want-block", "local", self, "to", mq.p, "cid", e.Cid)
			}
		}
	}
}
dirkmc's avatar
dirkmc committed
480 481

func (mq *MessageQueue) hasPendingWork() bool {
Steven Allen's avatar
Steven Allen committed
482 483 484 485
	return mq.pendingWorkCount() > 0
}

func (mq *MessageQueue) pendingWorkCount() int {
dirkmc's avatar
dirkmc committed
486 487 488
	mq.wllock.Lock()
	defer mq.wllock.Unlock()

Steven Allen's avatar
Steven Allen committed
489
	return mq.bcstWants.pending.Len() + mq.peerWants.pending.Len() + mq.cancels.Len()
dirkmc's avatar
dirkmc committed
490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558
}

func (mq *MessageQueue) extractOutgoingMessage(supportsHave bool) (bsmsg.BitSwapMessage, func()) {
	// Create a new message
	msg := bsmsg.New(false)

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

	// Get broadcast and regular wantlist entries
	bcstEntries := mq.bcstWants.pending.SortedEntries()
	peerEntries := mq.peerWants.pending.SortedEntries()

	// Size of the message so far
	msgSize := 0

	// Add each broadcast want-have to the message
	for i := 0; i < len(bcstEntries) && msgSize < mq.maxMessageSize; i++ {
		// Broadcast wants are sent as want-have
		wantType := pb.Message_Wantlist_Have

		// If the remote peer doesn't support HAVE / DONT_HAVE messages,
		// send a want-block instead
		if !supportsHave {
			wantType = pb.Message_Wantlist_Block
		}

		e := bcstEntries[i]
		msgSize += msg.AddEntry(e.Cid, e.Priority, wantType, false)
	}

	// Add each regular want-have / want-block to the message
	for i := 0; i < len(peerEntries) && msgSize < mq.maxMessageSize; i++ {
		e := peerEntries[i]
		// If the remote peer doesn't support HAVE / DONT_HAVE messages,
		// don't send want-haves (only send want-blocks)
		if !supportsHave && e.WantType == pb.Message_Wantlist_Have {
			mq.peerWants.RemoveType(e.Cid, pb.Message_Wantlist_Have)
		} else {
			msgSize += msg.AddEntry(e.Cid, e.Priority, e.WantType, true)
		}
	}

	// Add each cancel to the message
	cancels := mq.cancels.Keys()
	for i := 0; i < len(cancels) && msgSize < mq.maxMessageSize; i++ {
		c := cancels[i]

		msgSize += msg.Cancel(c)

		// Clear the cancel - we make a best effort to let peers know about
		// cancels but won't save them to resend if there's a failure.
		mq.cancels.Remove(c)
	}

	// Called when the message has been successfully sent.
	// Remove the sent keys from the broadcast and regular wantlists.
	onSent := func() {
		mq.wllock.Lock()
		defer mq.wllock.Unlock()

		for _, e := range msg.Wantlist() {
			mq.bcstWants.pending.Remove(e.Cid)
			mq.peerWants.pending.RemoveType(e.Cid, e.WantType)
		}
	}

	return msg, onSent
}
559

560
func (mq *MessageQueue) initializeSender() error {
561 562 563
	if mq.sender != nil {
		return nil
	}
564
	nsender, err := openSender(mq.ctx, mq.network, mq.p)
565 566 567 568 569 570
	if err != nil {
		return err
	}
	mq.sender = nsender
	return nil
}
571

572 573
func (mq *MessageQueue) attemptSendAndRecovery(message bsmsg.BitSwapMessage) bool {
	err := mq.sender.SendMsg(mq.ctx, message)
574 575 576
	if err == nil {
		return true
	}
577

578
	log.Infof("bitswap send error: %s", err)
Steven Allen's avatar
Steven Allen committed
579
	_ = mq.sender.Reset()
580 581 582 583 584
	mq.sender = nil

	select {
	case <-mq.done:
		return true
585
	case <-mq.ctx.Done():
586
		return true
dirkmc's avatar
dirkmc committed
587 588
	case <-time.After(mq.sendErrorBackoff):
		// wait 100ms in case disconnect notifications are still propagating
589
		log.Warn("SendMsg errored but neither 'done' nor context.Done() were set")
590
	}
591

592
	err = mq.initializeSender()
593 594 595
	if err != nil {
		log.Infof("couldnt open sender again after SendMsg(%s) failed: %s", mq.p, err)
		return true
596
	}
597 598 599 600 601 602 603 604 605 606 607 608

	// TODO: Is this the same instance for the remote peer?
	// If its not, we should resend our entire wantlist to them
	/*
		if mq.sender.InstanceID() != mq.lastSeenInstanceID {
			wlm = mq.getFullWantlistMessage()
		}
	*/
	return false
}

func openSender(ctx context.Context, network MessageNetwork, p peer.ID) (bsnet.MessageSender, error) {
609 610 611 612 613
	// allow ten minutes for connections this includes looking them up in the
	// dht dialing them, and handshaking
	conctx, cancel := context.WithTimeout(ctx, time.Minute*10)
	defer cancel()

614
	err := network.ConnectTo(conctx, p)
615
	if err != nil {
616
		return nil, err
617 618
	}

619
	nsender, err := network.NewMessageSender(ctx, p)
620
	if err != nil {
621
		return nil, err
622 623
	}

624
	return nsender, nil
625
}