Skip to content
This repository was archived by the owner on Feb 1, 2023. It is now read-only.

Commit 6099047

Browse files
authored
Merge pull request #349 from ipfs/refactor/simplify-mq-onSent
refactor: simplify messageQueue onSent
2 parents 906b2fb + b6a8a73 commit 6099047

File tree

1 file changed

+15
-26
lines changed

1 file changed

+15
-26
lines changed

internal/messagequeue/messagequeue.go

Lines changed: 15 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -113,8 +113,8 @@ func (r *recallWantlist) RemoveType(c cid.Cid, wtype pb.Message_Wantlist_WantTyp
113113
r.pending.RemoveType(c, wtype)
114114
}
115115

116-
// Sent moves the want from the pending to the sent list
117-
func (r *recallWantlist) Sent(e bsmsg.Entry) {
116+
// MarkSent moves the want from the pending to the sent list
117+
func (r *recallWantlist) MarkSent(e wantlist.Entry) {
118118
r.pending.RemoveType(e.Cid, e.WantType)
119119
r.sent.Add(e.Cid, e.Priority, e.WantType)
120120
}
@@ -439,7 +439,7 @@ func (mq *MessageQueue) sendMessage() {
439439
for i := 0; i < maxRetries; i++ {
440440
if mq.attemptSendAndRecovery(message) {
441441
// We were able to send successfully.
442-
onSent(wantlist)
442+
onSent()
443443

444444
mq.simulateDontHaveWithTimeout(wantlist)
445445

@@ -540,7 +540,7 @@ func (mq *MessageQueue) pendingWorkCount() int {
540540
}
541541

542542
// Convert the lists of wants into a Bitswap message
543-
func (mq *MessageQueue) extractOutgoingMessage(supportsHave bool) (bsmsg.BitSwapMessage, func([]bsmsg.Entry)) {
543+
func (mq *MessageQueue) extractOutgoingMessage(supportsHave bool) (bsmsg.BitSwapMessage, func()) {
544544
mq.wllock.Lock()
545545
defer mq.wllock.Unlock()
546546

@@ -566,6 +566,7 @@ func (mq *MessageQueue) extractOutgoingMessage(supportsHave bool) (bsmsg.BitSwap
566566
}
567567

568568
// Add each regular want-have / want-block to the message
569+
peerSent := make([]wantlist.Entry, 0, len(peerEntries))
569570
for i := 0; i < len(peerEntries) && msgSize < mq.maxMessageSize; i++ {
570571
e := peerEntries[i]
571572
// If the remote peer doesn't support HAVE / DONT_HAVE messages,
@@ -574,11 +575,13 @@ func (mq *MessageQueue) extractOutgoingMessage(supportsHave bool) (bsmsg.BitSwap
574575
mq.peerWants.RemoveType(e.Cid, pb.Message_Wantlist_Have)
575576
} else {
576577
msgSize += mq.msg.AddEntry(e.Cid, e.Priority, e.WantType, true)
578+
peerSent = append(peerSent, e)
577579
}
578580
}
579581

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

@@ -588,41 +591,27 @@ func (mq *MessageQueue) extractOutgoingMessage(supportsHave bool) (bsmsg.BitSwap
588591
wantType = pb.Message_Wantlist_Block
589592
}
590593

591-
e := bcstEntries[i]
594+
e := bcstEntries[bcstSentCount]
592595
msgSize += mq.msg.AddEntry(e.Cid, e.Priority, wantType, false)
593596
}
594597

595598
// Called when the message has been successfully sent.
596-
onMessageSent := func(wantlist []bsmsg.Entry) {
597-
bcst := keysToSet(bcstEntries)
598-
prws := keysToSet(peerEntries)
599-
599+
onMessageSent := func() {
600600
mq.wllock.Lock()
601601
defer mq.wllock.Unlock()
602602

603603
// Move the keys from pending to sent
604-
for _, e := range wantlist {
605-
if _, ok := bcst[e.Cid]; ok {
606-
mq.bcstWants.Sent(e)
607-
}
608-
if _, ok := prws[e.Cid]; ok {
609-
mq.peerWants.Sent(e)
610-
}
604+
for i := 0; i < bcstSentCount; i++ {
605+
mq.bcstWants.MarkSent(bcstEntries[i])
606+
}
607+
for _, e := range peerSent {
608+
mq.peerWants.MarkSent(e)
611609
}
612610
}
613611

614612
return mq.msg, onMessageSent
615613
}
616614

617-
// Convert wantlist entries into a set of cids
618-
func keysToSet(wl []wantlist.Entry) map[cid.Cid]struct{} {
619-
set := make(map[cid.Cid]struct{}, len(wl))
620-
for _, e := range wl {
621-
set[e.Cid] = struct{}{}
622-
}
623-
return set
624-
}
625-
626615
func (mq *MessageQueue) initializeSender() error {
627616
if mq.sender != nil {
628617
return nil

0 commit comments

Comments
 (0)