Improve M3DB session performance part 2: Dont clone IDs if they are IsNoFinalize()

glide.yaml

@@ -1,7 +1,7 @@
 package: github.com/m3db/m3
 import:
 - package: github.com/m3db/m3x
-  version: 3ab0cd67ff3c6e63a40893bad1ec4d993b5422c2
+  version: 659975cd3b5b69f112f1b74244b4d8b440888cf4
   vcs: git
   subpackages:
   - checked

src/dbnode/client/session.go

@@ -972,8 +972,8 @@ func (s *session) writeAttemptWithRLock(
 	// use in the various queues. Tracking per writeAttempt isn't sufficient as
 	// we may enqueue multiple writeStates concurrently depending on retries
 	// and consistency level checks.
-	nsID := s.pools.id.Clone(namespace)
-	tsID := s.pools.id.Clone(id)
+	nsID := s.cloneFinalizable(namespace)
+	tsID := s.cloneFinalizable(id)
 	var tagEncoder serialize.TagEncoder
 	if wType == taggedWriteAttemptType {
 		tagEncoder = s.pools.tagEncoder.Get()
@@ -3024,6 +3024,13 @@ func (s *session) verifyFetchedBlock(block *rpc.Block) error {
 	return nil
 }
 
+func (s *session) cloneFinalizable(id ident.ID) ident.ID {
+	if id.IsNoFinalize() {
+		return id
+	}
+	return s.pools.id.Clone(id)
+}
+
 type reason int
 
 const (

src/dbnode/client/session_write_tagged_test.go

@@ -33,6 +33,7 @@ import (
 	"github.com/m3db/m3/src/dbnode/serialize"
 	"github.com/m3db/m3/src/dbnode/topology"
 	xmetrics "github.com/m3db/m3/src/dbnode/x/metrics"
+	xm3test "github.com/m3db/m3/src/x/test"
 	"github.com/m3db/m3x/checked"
 	xerrors "github.com/m3db/m3x/errors"
 	"github.com/m3db/m3x/ident"
@@ -157,6 +158,62 @@ func TestSessionWriteTagged(t *testing.T) {
 	assert.NoError(t, session.Close())
 }
 
+func TestSessionWriteTaggedDoesNotCloneNoFinalize(t *testing.T) {
+	ctrl := gomock.NewController(xtest.Reporter{t})
+	defer ctrl.Finish()
+
+	w := newWriteTaggedStub()
+	session := newDefaultTestSession(t).(*session)
+	mockEncoder := serialize.NewMockTagEncoder(ctrl)
+	mockEncoderPool := serialize.NewMockTagEncoderPool(ctrl)
+	session.pools.tagEncoder = mockEncoderPool
+
+	gomock.InOrder(
+		mockEncoderPool.EXPECT().Get().Return(mockEncoder),
+		mockEncoder.EXPECT().Encode(gomock.Any()).Return(nil),
+		mockEncoder.EXPECT().Data().Return(testEncodeTags(w.tags), true),
+		mockEncoder.EXPECT().Finalize(),
+	)
+
+	var completionFn completionFn
+	enqueueWg := mockHostQueues(ctrl, session, sessionTestReplicas, []testEnqueueFn{func(idx int, op op) {
+		completionFn = op.CompletionFn()
+		write, ok := op.(*writeTaggedOperation)
+		require.True(t, ok)
+		require.True(t,
+			xm3test.ByteSlicesBackedBySameData(
+				w.ns.Bytes(),
+				write.namespace.Bytes()))
+		require.True(t,
+			xm3test.ByteSlicesBackedBySameData(
+				w.id.Bytes(),
+				write.request.ID))
+	}})
+
+	require.NoError(t, session.Open())
+	// Begin write
+	var resultErr error
+	var writeWg sync.WaitGroup
+	writeWg.Add(1)
+	go func() {
+		resultErr = session.WriteTagged(w.ns, w.id, ident.NewTagsIterator(w.tags),
+			w.t, w.value, w.unit, w.annotation)
+		writeWg.Done()
+	}()
+
+	// Callback
+	enqueueWg.Wait()
+	for i := 0; i < session.state.topoMap.Replicas(); i++ {
+		completionFn(session.state.topoMap.Hosts()[0], nil)
+	}
+
+	// Wait for write to complete
+	writeWg.Wait()
+	require.NoError(t, resultErr)
+
+	require.NoError(t, session.Close())
+}
+
 func TestSessionWriteTaggedBadUnitErr(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()

src/dbnode/client/session_write_test.go

@@ -32,6 +32,7 @@ import (
 	"github.com/m3db/m3/src/dbnode/generated/thrift/rpc"
 	"github.com/m3db/m3/src/dbnode/topology"
 	xmetrics "github.com/m3db/m3/src/dbnode/x/metrics"
+	xtest "github.com/m3db/m3/src/x/test"
 	xerrors "github.com/m3db/m3x/errors"
 	"github.com/m3db/m3x/ident"
 	xretry "github.com/m3db/m3x/retry"
@@ -101,6 +102,49 @@ func TestSessionWrite(t *testing.T) {
 	assert.NoError(t, session.Close())
 }
 
+func TestSessionWriteDoesNotCloneNoFinalize(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	session := newDefaultTestSession(t).(*session)
+	w := newWriteStub()
+	var completionFn completionFn
+	enqueueWg := mockHostQueues(ctrl, session, sessionTestReplicas, []testEnqueueFn{func(idx int, op op) {
+		completionFn = op.CompletionFn()
+		write, ok := op.(*writeOperation)
+		require.True(t, ok)
+		require.True(t,
+			xtest.ByteSlicesBackedBySameData(
+				w.ns.Bytes(),
+				write.namespace.Bytes()))
+		require.True(t,
+			xtest.ByteSlicesBackedBySameData(
+				w.id.Bytes(),
+				write.request.ID))
+	}})
+
+	require.NoError(t, session.Open())
+
+	// Begin write
+	var resultErr error
+	var writeWg sync.WaitGroup
+	writeWg.Add(1)
+	go func() {
+		resultErr = session.Write(w.ns, w.id, w.t, w.value, w.unit, w.annotation)
+		writeWg.Done()
+	}()
+
+	// Callback
+	enqueueWg.Wait()
+	for i := 0; i < session.state.topoMap.Replicas(); i++ {
+		completionFn(session.state.topoMap.Hosts()[0], nil)
+	}
+
+	writeWg.Wait()
+	require.NoError(t, resultErr)
+	require.NoError(t, session.Close())
+}
+
 func TestSessionWriteBadUnitErr(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	defer ctrl.Finish()

src/dbnode/persist/fs/persist_manager_test.go

@@ -33,6 +33,7 @@ import (
 	"github.com/m3db/m3/src/m3ninx/index/segment"
 	m3ninxfs "github.com/m3db/m3/src/m3ninx/index/segment/fst"
 	m3ninxpersist "github.com/m3db/m3/src/m3ninx/persist"
+	m3test "github.com/m3db/m3/src/x/test"
 	"github.com/m3db/m3x/checked"
 	"github.com/m3db/m3x/ident"
 	xtest "github.com/m3db/m3x/test"
@@ -92,7 +93,7 @@ func TestPersistenceManagerPrepareDataFileExistsWithDelete(t *testing.T) {
 			BlockStart: blockStart,
 		},
 		BlockSize: testBlockSize,
-	})
+	}, m3test.IdentTransformer)
 	writer.EXPECT().Open(writerOpts).Return(nil)
 
 	shardDir := createDataShardDir(t, pm.filePathPrefix, testNs1ID, shard)
@@ -142,7 +143,7 @@ func TestPersistenceManagerPrepareOpenError(t *testing.T) {
 			BlockStart: blockStart,
 		},
 		BlockSize: testBlockSize,
-	})
+	}, m3test.IdentTransformer)
 	writer.EXPECT().Open(writerOpts).Return(expectedErr)
 
 	flush, err := pm.StartDataPersist()
@@ -179,7 +180,7 @@ func TestPersistenceManagerPrepareSuccess(t *testing.T) {
 			BlockStart: blockStart,
 		},
 		BlockSize: testBlockSize,
-	})
+	}, m3test.IdentTransformer)
 	writer.EXPECT().Open(writerOpts).Return(nil)
 
 	var (
@@ -277,8 +278,8 @@ func TestPersistenceManagerPrepareIndexFileExists(t *testing.T) {
 				Namespace:          testNs1ID,
 				VolumeIndex:        1,
 			},
-		},
-	)).Return(nil)
+		}, m3test.IdentTransformer),
+	).Return(nil)
 	prepared, err := flush.PrepareIndex(prepareOpts)
 	require.NoError(t, err)
 	require.NotNil(t, prepared.Persist)
@@ -303,7 +304,7 @@ func TestPersistenceManagerPrepareIndexOpenError(t *testing.T) {
 			BlockStart:         blockStart,
 		},
 		BlockSize: testBlockSize,
-	})
+	}, m3test.IdentTransformer)
 	writer.EXPECT().Open(writerOpts).Return(expectedErr)
 
 	flush, err := pm.StartIndexPersist()
@@ -340,7 +341,7 @@ func TestPersistenceManagerPrepareIndexSuccess(t *testing.T) {
 		},
 		BlockSize: testBlockSize,
 	}
-	writer.EXPECT().Open(xtest.CmpMatcher(writerOpts)).Return(nil)
+	writer.EXPECT().Open(xtest.CmpMatcher(writerOpts, m3test.IdentTransformer)).Return(nil)
 
 	flush, err := pm.StartIndexPersist()
 	require.NoError(t, err)
@@ -369,7 +370,7 @@ func TestPersistenceManagerPrepareIndexSuccess(t *testing.T) {
 
 	reader.EXPECT().Open(xtest.CmpMatcher(IndexReaderOpenOptions{
 		Identifier: writerOpts.Identifier,
-	})).Return(IndexReaderOpenResult{}, nil)
+	}, m3test.IdentTransformer)).Return(IndexReaderOpenResult{}, nil)
 
 	file := NewMockIndexSegmentFile(ctrl)
 	gomock.InOrder(
@@ -408,7 +409,7 @@ func TestPersistenceManagerNoRateLimit(t *testing.T) {
 			BlockStart: blockStart,
 		},
 		BlockSize: testBlockSize,
-	})
+	}, m3test.IdentTransformer)
 	writer.EXPECT().Open(writerOpts).Return(nil)
 
 	var (
@@ -487,7 +488,7 @@ func TestPersistenceManagerWithRateLimit(t *testing.T) {
 			BlockStart: blockStart,
 		},
 		BlockSize: testBlockSize,
-	})
+	}, m3test.IdentTransformer)
 	writer.EXPECT().Open(writerOpts).Return(nil).Times(iter)
 	writer.EXPECT().WriteAll(id, ident.Tags{}, pm.dataPM.segmentHolder, checksum).Return(nil).AnyTimes()
 	writer.EXPECT().Close().Times(iter)
@@ -577,7 +578,7 @@ func TestPersistenceManagerNamespaceSwitch(t *testing.T) {
 			BlockStart: blockStart,
 		},
 		BlockSize: testBlockSize,
-	})
+	}, m3test.IdentTransformer)
 	writer.EXPECT().Open(writerOpts).Return(nil)
 	prepareOpts := persist.DataPrepareOptions{
 		NamespaceMetadata: testNs1Metadata(t),
@@ -596,7 +597,7 @@ func TestPersistenceManagerNamespaceSwitch(t *testing.T) {
 			BlockStart: blockStart,
 		},
 		BlockSize: testBlockSize,
-	})
+	}, m3test.IdentTransformer)
 	writer.EXPECT().Open(writerOpts).Return(nil)
 	prepareOpts = persist.DataPrepareOptions{
 		NamespaceMetadata: testNs2Metadata(t),

src/query/ts/m3db/storage/local.go

@@ -30,7 +30,6 @@ import (
 	"github.com/m3db/m3/src/query/storage"
 	"github.com/m3db/m3/src/query/storage/m3"
 	m3block "github.com/m3db/m3/src/query/ts/m3db"
-	"github.com/m3db/m3x/ident"
 	"github.com/m3db/m3x/pool"
 )
 
@@ -44,7 +43,7 @@ var (
 		return m3tsz.NewReaderIterator(r, m3tsz.DefaultIntOptimizationEnabled, encoding.NewOptions())
 	}
 
-	emptySeriesMap map[ident.ID][]m3block.SeriesBlocks
+	emptySeriesMap map[string][]m3block.SeriesBlocks
 )
 
 // nolint: deadcode
@@ -63,12 +62,14 @@ func (s *localStorage) fetchRaw(
 	return session.FetchTagged(namespaceID, query, opts)
 }
 
-// todo(braskin): merge this with Fetch()
+// todo(braskin): merge this with Fetch() and use genny to generate
+// a map that can handle ident.ID as a key so we don't need to
+// allocate a string for each entry.
 func (s *localStorage) fetchBlocks(
 	_ context.Context,
 	query *storage.FetchQuery,
 	options *storage.FetchOptions,
-) (map[ident.ID][]m3block.SeriesBlocks, error) {
+) (map[string][]m3block.SeriesBlocks, error) {
 
 	m3query, err := storage.FetchQueryToM3Query(query)
 	if err != nil {
@@ -107,13 +108,14 @@ func (s *localStorage) Close() error {
 	return nil
 }
 
-func fromNamespaceListToSeriesList(nsList []m3block.NamespaceSeriesList) map[ident.ID][]m3block.SeriesBlocks {
-	seriesList := make(map[ident.ID][]m3block.SeriesBlocks)
+func fromNamespaceListToSeriesList(nsList []m3block.NamespaceSeriesList) map[string][]m3block.SeriesBlocks {
+	seriesList := make(map[string][]m3block.SeriesBlocks)
 
 	for _, ns := range nsList {
 		for _, series := range ns.SeriesList {
-			if _, ok := seriesList[series.ID]; !ok {
-				seriesList[series.ID] = []m3block.SeriesBlocks{
+			idStr := series.ID.String()
+			if _, ok := seriesList[idStr]; !ok {
+				seriesList[idStr] = []m3block.SeriesBlocks{
 					{
 						Blocks:    series.Blocks,
 						ID:        series.ID,
@@ -122,7 +124,7 @@ func fromNamespaceListToSeriesList(nsList []m3block.NamespaceSeriesList) map[ide
 					},
 				}
 			} else {
-				seriesList[series.ID] = append(seriesList[series.ID], series)
+				seriesList[idStr] = append(seriesList[idStr], series)
 			}
 		}
 	}

src/query/ts/m3db/storage/local_test.go

@@ -107,7 +107,7 @@ func TestLocalRead(t *testing.T) {
 	assert.NoError(t, err)
 
 	for id, seriesBlocks := range results {
-		assert.Equal(t, "id", id.String())
+		assert.Equal(t, "id", id)
 		for _, blocks := range seriesBlocks {
 			assert.Equal(t, "namespace", blocks.Namespace.String())
 			blockTags, err := storage.FromIdentTagIteratorToTags(blocks.Tags)

src/x/test/ident_cmp_matcher.go

@@ -0,0 +1,31 @@
+// 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 xtest
+
+import (
+	"github.com/m3db/m3x/ident"
+
+	"github.com/google/go-cmp/cmp"
+)
+
+// IdentTransformer transforms any ident.ID into ident.BytesID to make it easier for comparison.
+var IdentTransformer = cmp.Transformer("",
+	func(id ident.ID) ident.BytesID { return ident.BytesID(id.Bytes()) })