Merge pull request #143 from ipfs/fix/free-memory

aggressively free memory

Author: Stebalien

go.mod

@@ -22,12 +22,13 @@ require (
 	github.com/ipfs/go-metrics-interface v0.0.1
 	github.com/ipfs/go-peertaskqueue v0.1.1
 	github.com/jbenet/goprocess v0.1.3
+	github.com/libp2p/go-buffer-pool v0.0.2
 	github.com/libp2p/go-libp2p v0.1.1
 	github.com/libp2p/go-libp2p-core v0.0.3
 	github.com/libp2p/go-libp2p-loggables v0.1.0
 	github.com/libp2p/go-libp2p-netutil v0.1.0
 	github.com/libp2p/go-libp2p-testing v0.0.4
-	github.com/libp2p/go-msgio v0.0.3 // indirect
+	github.com/libp2p/go-msgio v0.0.4
 	github.com/mattn/go-colorable v0.1.2 // indirect
 	github.com/multiformats/go-multiaddr v0.0.4
 	github.com/opentracing/opentracing-go v1.1.0 // indirect

go.sum

@@ -185,8 +185,8 @@ github.com/libp2p/go-mplex v0.1.0 h1:/nBTy5+1yRyY82YaO6HXQRnO5IAGsXTjEJaR3LdTPc0
 github.com/libp2p/go-mplex v0.1.0/go.mod h1:SXgmdki2kwCUlCCbfGLEgHjC4pFqhTp0ZoV6aiKgxDU=
 github.com/libp2p/go-msgio v0.0.2 h1:ivPvEKHxmVkTClHzg6RXTYHqaJQ0V9cDbq+6lKb3UV0=
 github.com/libp2p/go-msgio v0.0.2/go.mod h1:63lBBgOTDKQL6EWazRMCwXsEeEeK9O2Cd+0+6OOuipQ=
-github.com/libp2p/go-msgio v0.0.3 h1:VsOlWispTivSsOMg70e0W77y6oiSBSRCyP6URrWvE04=
-github.com/libp2p/go-msgio v0.0.3/go.mod h1:63lBBgOTDKQL6EWazRMCwXsEeEeK9O2Cd+0+6OOuipQ=
+github.com/libp2p/go-msgio v0.0.4 h1:agEFehY3zWJFUHK6SEMR7UYmk2z6kC3oeCM7ybLhguA=
+github.com/libp2p/go-msgio v0.0.4/go.mod h1:63lBBgOTDKQL6EWazRMCwXsEeEeK9O2Cd+0+6OOuipQ=
 github.com/libp2p/go-nat v0.0.3 h1:l6fKV+p0Xa354EqQOQP+d8CivdLM4kl5GxC1hSc/UeI=
 github.com/libp2p/go-nat v0.0.3/go.mod h1:88nUEt0k0JD45Bk93NIwDqjlhiOwOoV36GchpcVc1yI=
 github.com/libp2p/go-reuseport v0.0.1 h1:7PhkfH73VXfPJYKQ6JwS5I/eVcoyYi9IMNGc6FWpFLw=

message/message.go

@@ -1,15 +1,17 @@
 package message
 
 import (
+	"encoding/binary"
 	"fmt"
 	"io"
 
 	pb "github.com/ipfs/go-bitswap/message/pb"
 	wantlist "github.com/ipfs/go-bitswap/wantlist"
 	blocks "github.com/ipfs/go-block-format"
 
-	ggio "github.com/gogo/protobuf/io"
 	cid "github.com/ipfs/go-cid"
+	pool "github.com/libp2p/go-buffer-pool"
+	msgio "github.com/libp2p/go-msgio"
 
 	"github.com/libp2p/go-libp2p-core/network"
 )
@@ -170,18 +172,25 @@ func (m *impl) AddBlock(b blocks.Block) {
 
 // FromNet generates a new BitswapMessage from incoming data on an io.Reader.
 func FromNet(r io.Reader) (BitSwapMessage, error) {
-	pbr := ggio.NewDelimitedReader(r, network.MessageSizeMax)
-	return FromPBReader(pbr)
+	reader := msgio.NewVarintReaderSize(r, network.MessageSizeMax)
+	return FromMsgReader(reader)
 }
 
 // FromPBReader generates a new Bitswap message from a gogo-protobuf reader
-func FromPBReader(pbr ggio.Reader) (BitSwapMessage, error) {
-	pb := new(pb.Message)
-	if err := pbr.ReadMsg(pb); err != nil {
+func FromMsgReader(r msgio.Reader) (BitSwapMessage, error) {
+	msg, err := r.ReadMsg()
+	if err != nil {
 		return nil, err
 	}
 
-	return newMessageFromProto(*pb)
+	var pb pb.Message
+	err = pb.Unmarshal(msg)
+	r.ReleaseMsg(msg)
+	if err != nil {
+		return nil, err
+	}
+
+	return newMessageFromProto(pb)
 }
 
 func (m *impl) ToProtoV0() *pb.Message {
@@ -228,15 +237,29 @@ func (m *impl) ToProtoV1() *pb.Message {
 }
 
 func (m *impl) ToNetV0(w io.Writer) error {
-	pbw := ggio.NewDelimitedWriter(w)
-
-	return pbw.WriteMsg(m.ToProtoV0())
+	return write(w, m.ToProtoV0())
 }
 
 func (m *impl) ToNetV1(w io.Writer) error {
-	pbw := ggio.NewDelimitedWriter(w)
+	return write(w, m.ToProtoV1())
+}
+
+func write(w io.Writer, m *pb.Message) error {
+	size := m.Size()
+
+	buf := pool.Get(size + binary.MaxVarintLen64)
+	defer pool.Put(buf)
+
+	n := binary.PutUvarint(buf, uint64(size))
+
+	written, err := m.MarshalTo(buf[n:])
+	if err != nil {
+		return err
+	}
+	n += written
 
-	return pbw.WriteMsg(m.ToProtoV1())
+	_, err = w.Write(buf[:n])
+	return err
 }
 
 func (m *impl) Loggable() map[string]interface{} {

network/ipfs_impl.go

@@ -10,7 +10,6 @@ import (
 	bsmsg "github.com/ipfs/go-bitswap/message"
 	"github.com/libp2p/go-libp2p-core/helpers"
 
-	ggio "github.com/gogo/protobuf/io"
 	cid "github.com/ipfs/go-cid"
 	logging "github.com/ipfs/go-log"
 	"github.com/libp2p/go-libp2p-core/connmgr"
@@ -19,6 +18,7 @@ import (
 	"github.com/libp2p/go-libp2p-core/peer"
 	peerstore "github.com/libp2p/go-libp2p-core/peerstore"
 	"github.com/libp2p/go-libp2p-core/routing"
+	msgio "github.com/libp2p/go-msgio"
 	ma "github.com/multiformats/go-multiaddr"
 )
 
@@ -178,9 +178,9 @@ func (bsnet *impl) handleNewStream(s network.Stream) {
 		return
 	}
 
-	reader := ggio.NewDelimitedReader(s, network.MessageSizeMax)
+	reader := msgio.NewVarintReaderSize(s, network.MessageSizeMax)
 	for {
-		received, err := bsmsg.FromPBReader(reader)
+		received, err := bsmsg.FromMsgReader(reader)
 		if err != nil {
 			if err != io.EOF {
 				s.Reset()