feat(memory): improve memory usage (#110)

reduce copies of temporary buffers in message write and unverified block store

go.mod

@@ -31,6 +31,7 @@ require (
 	github.com/ipld/go-ipld-prime-proto v0.1.0
 	github.com/jbenet/go-random v0.0.0-20190219211222-123a90aedc0c
 	github.com/jbenet/goprocess v0.1.4 // indirect
+	github.com/libp2p/go-buffer-pool v0.0.2
 	github.com/libp2p/go-libp2p v0.6.0
 	github.com/libp2p/go-libp2p-core v0.5.0
 	github.com/libp2p/go-libp2p-netutil v0.1.0

message/message.go

@@ -1,14 +1,15 @@
 package message
 
 import (
+	"encoding/binary"
 	"errors"
 	"fmt"
 	"io"
 
-	ggio "github.com/gogo/protobuf/io"
 	blocks "github.com/ipfs/go-block-format"
 	cid "github.com/ipfs/go-cid"
 	"github.com/ipld/go-ipld-prime"
+	pool "github.com/libp2p/go-buffer-pool"
 	"github.com/libp2p/go-libp2p-core/network"
 	"github.com/libp2p/go-msgio"
 	"google.golang.org/protobuf/proto"
@@ -346,12 +347,19 @@ func (gsm *graphSyncMessage) ToProto() (*pb.Message, error) {
 }
 
 func (gsm *graphSyncMessage) ToNet(w io.Writer) error {
-	pbw := ggio.NewDelimitedWriter(w)
 	msg, err := gsm.ToProto()
+	size := proto.Size(msg)
+	buf := pool.Get(size + binary.MaxVarintLen64)
+	defer pool.Put(buf)
+
+	n := binary.PutUvarint(buf, uint64(size))
+
+	out, err := proto.MarshalOptions{}.MarshalAppend(buf[:n], msg)
 	if err != nil {
 		return err
 	}
-	return pbw.WriteMsg(msg)
+	_, err = w.Write(out)
+	return err
 }
 
 func (gsm *graphSyncMessage) Loggable() map[string]interface{} {

requestmanager/asyncloader/unverifiedblockstore/unverifiedblockstore.go

@@ -6,6 +6,10 @@ import (
 	ipld "github.com/ipld/go-ipld-prime"
 )
 
+type settableWriter interface {
+	SetBytes([]byte) error
+}
+
 // UnverifiedBlockStore holds an in memory cache of receied blocks from the network
 // that have not been verified to be part of a traversal
 type UnverifiedBlockStore struct {
@@ -55,7 +59,11 @@ func (ubs *UnverifiedBlockStore) VerifyBlock(lnk ipld.Link) ([]byte, error) {
 	if err != nil {
 		return nil, err
 	}
-	_, err = buffer.Write(data)
+	if settable, ok := buffer.(settableWriter); ok {
+		err = settable.SetBytes(data)
+	} else {
+		_, err = buffer.Write(data)
+	}
 	if err != nil {
 		return nil, err
 	}

storeutil/storeutil.go

@@ -31,7 +31,7 @@ func LoaderForBlockstore(bs bstore.Blockstore) ipld.Loader {
 // from an IPFS blockstore
 func StorerForBlockstore(bs bstore.Blockstore) ipld.Storer {
 	return func(lnkCtx ipld.LinkContext) (io.Writer, ipld.StoreCommitter, error) {
-		var buffer bytes.Buffer
+		var buffer settableBuffer
 		committer := func(lnk ipld.Link) error {
 			asCidLink, ok := lnk.(cidlink.Link)
 			if !ok {
@@ -46,3 +46,22 @@ func StorerForBlockstore(bs bstore.Blockstore) ipld.Storer {
 		return &buffer, committer, nil
 	}
 }
+
+type settableBuffer struct {
+	bytes.Buffer
+	didSetData bool
+	data       []byte
+}
+
+func (sb *settableBuffer) SetBytes(data []byte) error {
+	sb.didSetData = true
+	sb.data = data
+	return nil
+}
+
+func (sb *settableBuffer) Bytes() []byte {
+	if sb.didSetData {
+		return sb.data
+	}
+	return sb.Buffer.Bytes()
+}