Optimize captive-core pipe decoding

Also, remove `UnmarshalFramed` which wasn't used

ingest/ledgerbackend/buffered_meta_pipe_reader.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/pkg/errors"
+	xdr3 "github.com/stellar/go-xdr/xdr3"
 
 	"github.com/stellar/go/support/log"
 	"github.com/stellar/go/xdr"
@@ -63,16 +64,19 @@ type metaResult struct {
 // until xdr.LedgerCloseMeta objects channel is empty. This prevents memory
 // exhaustion when network closes a series a large ledgers.
 type bufferedLedgerMetaReader struct {
-	r *bufio.Reader
-	c chan metaResult
+	r       *bufio.Reader
+	c       chan metaResult
+	decoder *xdr3.Decoder
 }
 
 // newBufferedLedgerMetaReader creates a new meta reader that will shutdown
 // when stellar-core terminates.
 func newBufferedLedgerMetaReader(reader io.Reader) *bufferedLedgerMetaReader {
+	r := bufio.NewReaderSize(reader, metaPipeBufferSize)
 	return &bufferedLedgerMetaReader{
-		c: make(chan metaResult, ledgerReadAheadBufferSize),
-		r: bufio.NewReaderSize(reader, metaPipeBufferSize),
+		c:       make(chan metaResult, ledgerReadAheadBufferSize),
+		r:       r,
+		decoder: xdr3.NewDecoder(r),
 	}
 }
 
@@ -82,7 +86,7 @@ func newBufferedLedgerMetaReader(reader io.Reader) *bufferedLedgerMetaReader {
 //   * The next ledger available in the buffer exceeds the meta pipe buffer size.
 //     In such case the method will block until LedgerCloseMeta buffer is empty.
 func (b *bufferedLedgerMetaReader) readLedgerMetaFromPipe() (*xdr.LedgerCloseMeta, error) {
-	frameLength, err := xdr.ReadFrameLength(b.r)
+	frameLength, err := xdr.ReadFrameLength(b.decoder)
 	if err != nil {
 		return nil, errors.Wrap(err, "error reading frame length")
 	}
@@ -93,7 +97,7 @@ func (b *bufferedLedgerMetaReader) readLedgerMetaFromPipe() (*xdr.LedgerCloseMet
 	}
 
 	var xlcm xdr.LedgerCloseMeta
-	_, err = xdr.Unmarshal(b.r, &xlcm)
+	_, err = xlcm.DecodeFrom(b.decoder)
 	if err != nil {
 		return nil, errors.Wrap(err, "unmarshalling framed LedgerCloseMeta")
 	}

xdr/main.go

@@ -259,9 +259,8 @@ func MarshalFramed(w io.Writer, v interface{}) error {
 }
 
 // ReadFrameLength returns a length of a framed XDR object.
-func ReadFrameLength(r io.Reader) (uint32, error) {
-	var frameLen uint32
-	n, e := Unmarshal(r, &frameLen)
+func ReadFrameLength(d *xdr.Decoder) (uint32, error) {
+	frameLen, n, e := d.DecodeUint()
 	if e != nil {
 		return 0, errors.Wrap(e, "unmarshalling XDR frame header")
 	}
@@ -275,24 +274,6 @@ func ReadFrameLength(r io.Reader) (uint32, error) {
 	return frameLen, nil
 }
 
-// XDR and RPC define a (minimal) framing format which our metadata arrives in: a 4-byte
-// big-endian length header that has the high bit set, followed by that length worth of
-// XDR data. Decoding this involves just a little more work than xdr.Unmarshal.
-func UnmarshalFramed(r io.Reader, v interface{}) (int, error) {
-	frameLen, err := ReadFrameLength(r)
-	if err != nil {
-		return 0, errors.Wrap(err, "unmarshalling XDR frame header")
-	}
-	m, err := xdr.Unmarshal(r, v)
-	if err != nil {
-		return 0, errors.Wrap(err, "unmarshalling framed XDR")
-	}
-	if int64(m) != int64(frameLen) {
-		return 0, errors.New("bad length of XDR frame body")
-	}
-	return m + 4 /* frame size: uint32 */, nil
-}
-
 type countWriter struct {
 	Count int
 }