logos-messaging/go-libp2p-pubsub
1
0
Fork 0
mirror of https://github.com/logos-messaging/go-libp2p-pubsub.git synced 2026-02-19 12:03:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf: use msgio pooled buffers for received msgs (#500)

Browse Source
This commit is contained in:
Hlib Kanunnikov 2022-11-19 14:19:10 +01:00 committed by GitHub
parent 1e161006c4
commit 9c56b2deca
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 16 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
comm.go
View File

@ -6,14 +6,14 @@ import (
"io" "io"
"time" "time"
"github.com/gogo/protobuf/proto"
"github.com/libp2p/go-libp2p/core/network" "github.com/libp2p/go-libp2p/core/network"
"github.com/libp2p/go-libp2p/core/peer" "github.com/libp2p/go-libp2p/core/peer"
"github.com/libp2p/go-msgio"
pb "github.com/libp2p/go-libp2p-pubsub/pb"
"github.com/libp2p/go-msgio/protoio" "github.com/libp2p/go-msgio/protoio"
"github.com/gogo/protobuf/proto" pb "github.com/libp2p/go-libp2p-pubsub/pb"
) )
// get the initial RPC containing all of our subscriptions to send to new peers // get the initial RPC containing all of our subscriptions to send to new peers
@ -60,11 +60,11 @@ func (p *PubSub) handleNewStream(s network.Stream) {
p.inboundStreamsMx.Unlock() p.inboundStreamsMx.Unlock()
}() }()
r := protoio.NewDelimitedReader(s, p.maxMessageSize) r := msgio.NewVarintReaderSize(s, p.maxMessageSize)
for { for {
rpc := new(RPC) msgbytes, err := r.ReadMsg()
err := r.ReadMsg(&rpc.RPC)
if err != nil { if err != nil {
r.ReleaseMsg(msgbytes)
if err != io.EOF { if err != io.EOF {
s.Reset() s.Reset()
log.Debugf("error reading rpc from %s: %s", s.Conn().RemotePeer(), err) log.Debugf("error reading rpc from %s: %s", s.Conn().RemotePeer(), err)
@ -77,6 +77,15 @@ func (p *PubSub) handleNewStream(s network.Stream) {
return return
} }
rpc := new(RPC)
err = rpc.Unmarshal(msgbytes)
r.ReleaseMsg(msgbytes)
if err != nil {
s.Reset()
log.Warnf("bogus rpc from %s: %s", s.Conn().RemotePeer(), err)
return
}
rpc.from = peer rpc.from = peer
select { select {
case p.incoming <- rpc: case p.incoming <- rpc: