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

avoid copying messages and improve logging #326

Merged
merged 3 commits into from
Mar 26, 2020
Merged

avoid copying messages and improve logging #326

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
3895cc0
chore: address todo in engine.go
Stebalien Mar 26, 2020
484399b
fix: avoid copying messages multiple times on send
Stebalien Mar 26, 2020
8c7bf92
feat: normalize message logging
Stebalien Mar 26, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions bitswap.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@ import (
)

var log = logging.Logger("bitswap")
var sflog = log.Desugar()

var _ exchange.SessionExchange = (*Bitswap)(nil)

Expand Down
3 changes: 1 addition & 2 deletions internal/decision/engine.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -733,8 +733,7 @@ func (e *Engine) MessageSent(p peer.ID, m bsmsg.BitSwapMessage) {

// Remove sent block presences from the want list for the peer
for _, bp := range m.BlockPresences() {
// TODO: record block presence bytes as well?
// l.SentBytes(?)
// Don't record sent data. We reserve that for data blocks.
if bp.Type == pb.Message_Have {
l.wantList.RemoveType(bp.Cid, pb.Message_Wantlist_Have)
}
Expand Down
30 changes: 25 additions & 5 deletions internal/messagequeue/messagequeue.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -466,23 +466,43 @@ func (mq *MessageQueue) simulateDontHaveWithTimeout(wantlist []bsmsg.Entry) {

func (mq *MessageQueue) logOutgoingMessage(wantlist []bsmsg.Entry) {
// Save some CPU cycles and allocations if log level is higher than debug
if ce := sflog.Check(zap.DebugLevel, "Bitswap -> send wants"); ce == nil {
if ce := sflog.Check(zap.DebugLevel, "sent message"); ce == nil {
return
}

self := mq.network.Self()
for _, e := range wantlist {
if e.Cancel {
if e.WantType == pb.Message_Wantlist_Have {
log.Debugw("Bitswap -> cancel-have", "local", self, "to", mq.p, "cid", e.Cid)
log.Debugw("sent message",
"type", "CANCEL_WANT_HAVE",
"cid", e.Cid,
"local", self,
"to", mq.p,
)
} else {
log.Debugw("Bitswap -> cancel-block", "local", self, "to", mq.p, "cid", e.Cid)
log.Debugw("sent message",
"type", "CANCEL_WANT_BLOCK",
"cid", e.Cid,
"local", self,
"to", mq.p,
)
}
} else {
if e.WantType == pb.Message_Wantlist_Have {
log.Debugw("Bitswap -> want-have", "local", self, "to", mq.p, "cid", e.Cid)
log.Debugw("sent message",
"type", "WANT_HAVE",
"cid", e.Cid,
"local", self,
"to", mq.p,
)
} else {
log.Debugw("Bitswap -> want-block", "local", self, "to", mq.p, "cid", e.Cid)
log.Debugw("sent message",
"type", "WANT_BLOCK",
"cid", e.Cid,
"local", self,
"to", mq.p,
)
}
}
}
Expand Down
93 changes: 53 additions & 40 deletions workers.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,11 +5,11 @@ import (
"fmt"

engine "github.com/ipfs/go-bitswap/internal/decision"
bsmsg "github.com/ipfs/go-bitswap/message"
pb "github.com/ipfs/go-bitswap/message/pb"
cid "github.com/ipfs/go-cid"
process "github.com/jbenet/goprocess"
procctx "github.com/jbenet/goprocess/context"
"go.uber.org/zap"
)

// TaskWorkerCount is the total number of simultaneous threads sending
Expand Down Expand Up @@ -52,29 +52,11 @@ func (bs *Bitswap) taskWorker(ctx context.Context, id int) {
continue
}

// update the BS ledger to reflect sent message
// TODO: Should only track *useful* messages in ledger
outgoing := bsmsg.New(false)
for _, block := range envelope.Message.Blocks() {
log.Debugw("Bitswap.TaskWorker.Work",
"Target", envelope.Peer,
"Block", block.Cid(),
)
outgoing.AddBlock(block)
}
for _, blockPresence := range envelope.Message.BlockPresences() {
outgoing.AddBlockPresence(blockPresence.Cid, blockPresence.Type)
}
// TODO: Only record message as sent if there was no error?
bs.engine.MessageSent(envelope.Peer, outgoing)

// Ideally, yes. But we'd need some way to trigger a retry and/or drop
// the peer.
bs.engine.MessageSent(envelope.Peer, envelope.Message)
bs.sendBlocks(ctx, envelope)
bs.counterLk.Lock()
for _, block := range envelope.Message.Blocks() {
bs.counters.blocksSent++
bs.counters.dataSent += uint64(len(block.RawData()))
}
bs.counterLk.Unlock()
case <-ctx.Done():
return
}
Expand All @@ -84,41 +66,72 @@ func (bs *Bitswap) taskWorker(ctx context.Context, id int) {
}
}

func (bs *Bitswap) sendBlocks(ctx context.Context, env *engine.Envelope) {
// Blocks need to be sent synchronously to maintain proper backpressure
// throughout the network stack
defer env.Sent()
func (bs *Bitswap) logOutgoingBlocks(env *engine.Envelope) {
if ce := sflog.Check(zap.DebugLevel, "sent message"); ce == nil {
return
}

msgSize := 0
msg := bsmsg.New(false)
self := bs.network.Self()

for _, blockPresence := range env.Message.BlockPresences() {
c := blockPresence.Cid
switch blockPresence.Type {
case pb.Message_Have:
log.Infof("Sending HAVE %s to %s", c.String()[2:8], env.Peer)
log.Debugw("sent message",
"type", "HAVE",
"cid", c,
"local", self,
"to", env.Peer,
)
case pb.Message_DontHave:
log.Infof("Sending DONT_HAVE %s to %s", c.String()[2:8], env.Peer)
log.Debugw("sent message",
"type", "DONT_HAVE",
"cid", c,
"local", self,
"to", env.Peer,
)
default:
panic(fmt.Sprintf("unrecognized BlockPresence type %v", blockPresence.Type))
}

msgSize += bsmsg.BlockPresenceSize(c)
msg.AddBlockPresence(c, blockPresence.Type)
}
for _, block := range env.Message.Blocks() {
msgSize += len(block.RawData())
msg.AddBlock(block)
log.Infof("Sending block %s to %s", block, env.Peer)
log.Debugw("sent message",
"type", "BLOCK",
"cid", block.Cid(),
"local", self,
"to", env.Peer,
)
}
}

bs.sentHistogram.Observe(float64(msgSize))
err := bs.network.SendMessage(ctx, env.Peer, msg)
func (bs *Bitswap) sendBlocks(ctx context.Context, env *engine.Envelope) {
// Blocks need to be sent synchronously to maintain proper backpressure
// throughout the network stack
defer env.Sent()

err := bs.network.SendMessage(ctx, env.Peer, env.Message)
if err != nil {
// log.Infof("sendblock error: %s", err)
log.Errorf("SendMessage error: %s. size: %d. block-presence length: %d", err, msg.Size(), len(env.Message.BlockPresences()))
log.Debugw("failed to send blocks message",
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should this be logged at error level?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Failing to send to a peer isn't an error for us. They may just not be reachable.

(this is actually why I started digging into this code, I saw this error in a test)

"peer", env.Peer,
"error", err,
)
return
}

bs.logOutgoingBlocks(env)

dataSent := 0
blocks := env.Message.Blocks()
for _, b := range blocks {
dataSent += len(b.RawData())
}
log.Infof("Sent message to %s", env.Peer)
bs.counterLk.Lock()
bs.counters.blocksSent += uint64(len(blocks))
bs.counters.dataSent += uint64(dataSent)
bs.counterLk.Unlock()
bs.sentHistogram.Observe(float64(env.Message.Size()))
log.Debugw("sent message", "peer", env.Peer)
}

func (bs *Bitswap) provideWorker(px process.Process) {
Expand Down