Make commitlog msgpack encoder really, really fast.

src/dbnode/persist/fs/commitlog/writer.go

@@ -54,6 +54,8 @@ const (
 		chunkHeaderChecksumDataLen
 
 	defaultBitSetLength = 65536
+
+	defaultEncoderBuffSize = 16384
 )
 
 var (
@@ -95,21 +97,22 @@ type chunkWriter interface {
 type flushFn func(err error)
 
 type writer struct {
-	filePathPrefix     string
-	newFileMode        os.FileMode
-	newDirectoryMode   os.FileMode
-	nowFn              clock.NowFn
-	start              time.Time
-	duration           time.Duration
-	chunkWriter        chunkWriter
-	chunkReserveHeader []byte
-	buffer             *bufio.Writer
-	sizeBuffer         []byte
-	seen               *bitset.BitSet
-	logEncoder         *msgpack.Encoder
-	metadataEncoder    *msgpack.Encoder
-	tagEncoder         serialize.TagEncoder
-	tagSliceIter       ident.TagsIterator
+	filePathPrefix      string
+	newFileMode         os.FileMode
+	newDirectoryMode    os.FileMode
+	nowFn               clock.NowFn
+	start               time.Time
+	duration            time.Duration
+	chunkWriter         chunkWriter
+	chunkReserveHeader  []byte
+	buffer              *bufio.Writer
+	sizeBuffer          []byte
+	seen                *bitset.BitSet
+	logEncoder          *msgpack.Encoder
+	logEncoderBuff      []byte
+	metadataEncoderBuff []byte
+	tagEncoder          serialize.TagEncoder
+	tagSliceIter        ident.TagsIterator
 }
 
 func newCommitLogWriter(
@@ -119,19 +122,20 @@ func newCommitLogWriter(
 	shouldFsync := opts.Strategy() == StrategyWriteWait
 
 	return &writer{
-		filePathPrefix:     opts.FilesystemOptions().FilePathPrefix(),
-		newFileMode:        opts.FilesystemOptions().NewFileMode(),
-		newDirectoryMode:   opts.FilesystemOptions().NewDirectoryMode(),
-		nowFn:              opts.ClockOptions().NowFn(),
-		chunkWriter:        newChunkWriter(flushFn, shouldFsync),
-		chunkReserveHeader: make([]byte, chunkHeaderLen),
-		buffer:             bufio.NewWriterSize(nil, opts.FlushSize()),
-		sizeBuffer:         make([]byte, binary.MaxVarintLen64),
-		seen:               bitset.NewBitSet(defaultBitSetLength),
-		logEncoder:         msgpack.NewEncoder(),
-		metadataEncoder:    msgpack.NewEncoder(),
-		tagEncoder:         opts.FilesystemOptions().TagEncoderPool().Get(),
-		tagSliceIter:       ident.NewTagsIterator(ident.Tags{}),
+		filePathPrefix:      opts.FilesystemOptions().FilePathPrefix(),
+		newFileMode:         opts.FilesystemOptions().NewFileMode(),
+		newDirectoryMode:    opts.FilesystemOptions().NewDirectoryMode(),
+		nowFn:               opts.ClockOptions().NowFn(),
+		chunkWriter:         newChunkWriter(flushFn, shouldFsync),
+		chunkReserveHeader:  make([]byte, chunkHeaderLen),
+		buffer:              bufio.NewWriterSize(nil, opts.FlushSize()),
+		sizeBuffer:          make([]byte, binary.MaxVarintLen64),
+		seen:                bitset.NewBitSet(defaultBitSetLength),
+		logEncoder:          msgpack.NewEncoder(),
+		logEncoderBuff:      make([]byte, 0, defaultEncoderBuffSize),
+		metadataEncoderBuff: make([]byte, 0, defaultEncoderBuffSize),
+		tagEncoder:          opts.FilesystemOptions().TagEncoderPool().Get(),
+		tagSliceIter:        ident.NewTagsIterator(ident.Tags{}),
 	}
 }
 
@@ -140,6 +144,15 @@ func (w *writer) Open(start time.Time, duration time.Duration) (File, error) {
 		return File{}, errCommitLogWriterAlreadyOpen
 	}
 
+	// Reset buffers since they will grow 2x on demand so we want to make sure that
+	// one exceptionally large write does not cause them to remain oversized forever.
+	if cap(w.logEncoderBuff) != defaultEncoderBuffSize {
+		w.logEncoderBuff = make([]byte, 0, defaultEncoderBuffSize)
+	}
+	if cap(w.metadataEncoderBuff) != defaultEncoderBuffSize {
+		w.metadataEncoderBuff = make([]byte, 0, defaultEncoderBuffSize)
+	}
+
 	commitLogsDir := fs.CommitLogsDirPath(w.filePathPrefix)
 	if err := os.MkdirAll(commitLogsDir, w.newDirectoryMode); err != nil {
 		return File{}, err
@@ -224,22 +237,30 @@ func (w *writer) Write(
 		metadata.Namespace = series.Namespace.Bytes()
 		metadata.Shard = series.Shard
 		metadata.EncodedTags = encodedTags
-		w.metadataEncoder.Reset()
-		if err := w.metadataEncoder.EncodeLogMetadata(metadata); err != nil {
+
+		w.metadataEncoderBuff = w.metadataEncoderBuff[:0]
+		var err error
+
+		w.metadataEncoderBuff, err = msgpack.EncodeLogMetadataFast(w.metadataEncoderBuff, metadata)
+		if err != nil {
 			return err
 		}
-		logEntry.Metadata = w.metadataEncoder.Bytes()
+		logEntry.Metadata = w.metadataEncoderBuff
 	}
 
 	logEntry.Timestamp = datapoint.Timestamp.UnixNano()
 	logEntry.Value = datapoint.Value
 	logEntry.Unit = uint32(unit)
 	logEntry.Annotation = annotation
-	w.logEncoder.Reset()
-	if err := w.logEncoder.EncodeLogEntry(logEntry); err != nil {
+
+	w.logEncoderBuff = w.logEncoderBuff[:0]
+
+	var err error
+	w.logEncoderBuff, err = msgpack.EncodeLogEntryFast(w.logEncoderBuff, logEntry)
+	if err != nil {
 		return err
 	}
-	if err := w.write(w.logEncoder.Bytes()); err != nil {
+	if err := w.write(w.logEncoderBuff); err != nil {
 		return err
 	}
 

src/dbnode/persist/fs/msgpack/encoder_decoder_bench_test.go

@@ -0,0 +1,103 @@
+// Copyright (c) 2018 Uber Technologies, Inc
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE
+
+package msgpack
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func BenchmarkLogEntryDecoder(b *testing.B) {
+	var (
+		enc    = NewEncoder()
+		dec    = NewDecoder(nil)
+		stream = NewDecoderStream(nil)
+		err    error
+	)
+
+	require.NoError(b, enc.EncodeLogEntry(testLogEntry))
+	buf := enc.Bytes()
+	for n := 0; n < b.N; n++ {
+		stream.Reset(buf)
+		dec.Reset(stream)
+		_, err = dec.DecodeLogEntry()
+		if err != nil {
+			panic(err)
+		}
+	}
+}
+
+var benchmarkBuf []byte
+
+func BenchmarkLogEntryEncoderFast(b *testing.B) {
+	var err error
+	benchmarkBuf = []byte{}
+
+	for n := 0; n < b.N; n++ {
+		benchmarkBuf, err = EncodeLogEntryFast(benchmarkBuf[:0], testLogEntry)
+		if err != nil {
+			panic(err)
+		}
+	}
+}
+
+func BenchmarkLogEntryEncoder(b *testing.B) {
+	var (
+		enc = NewEncoder()
+		err error
+	)
+
+	for n := 0; n < b.N; n++ {
+		enc.EncodeLogEntry(testLogEntry)
+		if err != nil {
+			panic(err)
+		}
+		benchmarkBuf = enc.Bytes()
+	}
+}
+
+func BenchmarkLogMetadataEncoder(b *testing.B) {
+	var (
+		enc = NewEncoder()
+		err error
+	)
+
+	for n := 0; n < b.N; n++ {
+		enc.EncodeLogMetadata(testLogMetadata)
+		if err != nil {
+			panic(err)
+		}
+		benchmarkBuf = enc.Bytes()
+	}
+}
+
+func BenchmarkLogMetadataEncoderFast(b *testing.B) {
+	var err error
+	benchmarkBuf = []byte{}
+
+	for n := 0; n < b.N; n++ {
+		benchmarkBuf, err = EncodeLogMetadataFast(benchmarkBuf[:0], testLogMetadata)
+		if err != nil {
+			panic(err)
+		}
+	}
+}