Don't allow msgpack library to allocate bufio.NewReader() and mmap summaries and bloom filter as files

src/cmd/services/m3dbnode/config/config_test.go

@@ -110,6 +110,8 @@ db:
         seekReadBufferSize: 4096
         throughputLimitMbps: 100.0
         throughputCheckEvery: 128
+        force_index_summaries_mmap_memory: true
+        force_bloom_filter_mmap_memory: true
 
     repair:
         enabled: false
@@ -391,6 +393,8 @@ db:
     newFileMode: null
     newDirectoryMode: null
     mmap: null
+    force_index_summaries_mmap_memory: true
+    force_bloom_filter_mmap_memory: true
   commitlog:
     flushMaxBytes: 524288
     flushEvery: 1s

src/cmd/services/m3dbnode/config/fs.go

@@ -75,6 +75,14 @@ type FilesystemConfiguration struct {
 
 	// Mmap is the mmap options which features are primarily platform dependent
 	Mmap *MmapConfiguration `yaml:"mmap"`
+
+	// ForceIndexSummariesMmapMemory forces the mmap that stores the index lookup bytes
+	// to be an anonymous region in memory as opposed to a file-based mmap.
+	ForceIndexSummariesMmapMemory bool `yaml:"force_index_summaries_mmap_memory"`
+
+	// ForceBloomFilterMmapMemory forces the mmap that stores the index lookup bytes
+	// to be an anonymous region in memory as opposed to a file-based mmap.
+	ForceBloomFilterMmapMemory bool `yaml:"force_bloom_filter_mmap_memory"`
 }
 
 // MmapConfiguration is the mmap configuration.

src/dbnode/persist/fs/bloom_filter.go

@@ -73,34 +73,16 @@ func newManagedConcurrentBloomFilterFromFile(
 	expectedDigest uint32,
 	numElementsM uint,
 	numHashesK uint,
+	forceMmapMemory bool,
 ) (*ManagedConcurrentBloomFilter, error) {
 	// Determine how many bytes to request for the mmap'd region
 	bloomFilterFdWithDigest.Reset(bloomFilterFd)
-	stat, err := bloomFilterFd.Stat()
-	if err != nil {
-		return nil, err
-	}
-	numBytes := stat.Size()
-
-	// Request an anonymous (non-file-backed) mmap region. Note that we're going
-	// to use the mmap'd region to create a read-only bloom filter, but the mmap
-	// region itself needs to be writable so we can copy the bytes from disk
-	// into it
-	result, err := mmap.Bytes(numBytes, mmap.Options{Read: true, Write: true})
-	if err != nil {
-		return nil, err
-	}
-	anonMmap := result.Result
 
-	// Validate the bytes on disk using the digest, and read them into
-	// the mmap'd region
-	_, err = bloomFilterFdWithDigest.ReadAllAndValidate(
-		anonMmap, expectedDigest)
+	bloomFilterMmap, err := validateAndMmap(bloomFilterFdWithDigest, expectedDigest, forceMmapMemory)
 	if err != nil {
-		mmap.Munmap(anonMmap)
 		return nil, err
 	}
 
-	bloomFilter := bloom.NewConcurrentReadOnlyBloomFilter(numElementsM, numHashesK, anonMmap)
-	return newManagedConcurrentBloomFilter(bloomFilter, anonMmap), nil
+	bloomFilter := bloom.NewConcurrentReadOnlyBloomFilter(numElementsM, numHashesK, bloomFilterMmap)
+	return newManagedConcurrentBloomFilter(bloomFilter, bloomFilterMmap), nil
 }

src/dbnode/persist/fs/index_lookup.go

@@ -56,7 +56,7 @@ func newNearestIndexOffsetLookup(
 		summaryIDsOffsets: summaryIDsOffsets,
 		summariesMmap:     summariesMmap,
 		decoderStream:     decoderStream,
-		msgpackDecoder:    msgpack.NewDecoder(nil),
+		msgpackDecoder:    msgpack.NewDecoder(decoderStream),
 		isClone:           false,
 	}
 }
@@ -66,11 +66,12 @@ func (il *nearestIndexOffsetLookup) concurrentClone() (*nearestIndexOffsetLookup
 		return nil, errCloneShouldNotBeCloned
 	}
 
+	decoderStream := xmsgpack.NewDecoderStream(nil)
 	return &nearestIndexOffsetLookup{
 		summaryIDsOffsets: il.summaryIDsOffsets,
 		summariesMmap:     il.summariesMmap,
-		decoderStream:     xmsgpack.NewDecoderStream(nil),
-		msgpackDecoder:    msgpack.NewDecoder(nil),
+		decoderStream:     decoderStream,
+		msgpackDecoder:    msgpack.NewDecoder(decoderStream),
 		isClone:           true,
 	}, nil
 }
@@ -160,32 +161,13 @@ func (il *nearestIndexOffsetLookup) close() error {
 // the summaries file is sorted (which it always should be).
 func newNearestIndexOffsetLookupFromSummariesFile(
 	summariesFdWithDigest digest.FdWithDigestReader,
-	expectedSummariesDigest uint32,
+	expectedDigest uint32,
 	decoder *xmsgpack.Decoder,
 	numEntries int,
+	forceMmapMemory bool,
 ) (*nearestIndexOffsetLookup, error) {
-	summariesFd := summariesFdWithDigest.Fd()
-	stat, err := summariesFd.Stat()
-	if err != nil {
-		return nil, err
-	}
-	numBytes := stat.Size()
-
-	// Request an anonymous (non-file-backed) mmap region. Note that we're going
-	// to use the mmap'd region to store the read-only summaries data, but the mmap
-	// region itself needs to be writable so we can copy the bytes from disk
-	// into it
-	mmapResult, err := mmap.Bytes(numBytes, mmap.Options{Read: true, Write: true})
-	if err != nil {
-		return nil, err
-	}
-	summariesMmap := mmapResult.Result
-
-	// Validate the bytes on disk using the digest, and read them into
-	// the mmap'd region
-	_, err = summariesFdWithDigest.ReadAllAndValidate(summariesMmap, expectedSummariesDigest)
+	summariesMmap, err := validateAndMmap(summariesFdWithDigest, expectedDigest, forceMmapMemory)
 	if err != nil {
-		mmap.Munmap(summariesMmap)
 		return nil, err
 	}
 
@@ -210,7 +192,7 @@ func newNearestIndexOffsetLookupFromSummariesFile(
 		// if they're not. This should never happen as files should be sorted on disk.
 		if lastReadID != nil && bytes.Compare(lastReadID, entry.ID) != -1 {
 			mmap.Munmap(summariesMmap)
-			return nil, fmt.Errorf("summaries file is not sorted: %s", summariesFd.Name())
+			return nil, fmt.Errorf("summaries file is not sorted: %s", summariesFdWithDigest.Fd().Name())
 		}
 		summaryTokens = append(summaryTokens, summaryToken)
 		lastReadID = entry.ID

src/dbnode/persist/fs/index_lookup_prop_test.go

@@ -118,7 +118,7 @@ func TestIndexLookupWriteRead(t *testing.T) {
 		expectedSummariesDigest := calculateExpectedChecksum(t, summariesFilePath)
 		decoder := msgpack.NewDecoder(options.DecodingOptions())
 		indexLookup, err := newNearestIndexOffsetLookupFromSummariesFile(
-			summariesFdWithDigest, expectedSummariesDigest, decoder, len(writes))
+			summariesFdWithDigest, expectedSummariesDigest, decoder, len(writes), input.forceMmapMemory)
 		if err != nil {
 			return false, fmt.Errorf("err reading index lookup from summaries file: %v, ", err)
 		}
@@ -173,6 +173,9 @@ type propTestInput struct {
 	realWrites []generatedWrite
 	// Shard number to use for the files
 	shard uint32
+	// Whether the summaries file bytes should be mmap'd as an
+	// anonymous region or file.
+	forceMmapMemory bool
 }
 
 type generatedWrite struct {
@@ -196,10 +199,12 @@ func genPropTestInput(numRealWrites int) gopter.Gen {
 	return gopter.CombineGens(
 		gen.SliceOfN(numRealWrites, genWrite()),
 		gen.UInt32(),
+		gen.Bool(),
 	).Map(func(vals []interface{}) propTestInput {
 		return propTestInput{
-			realWrites: vals[0].([]generatedWrite),
-			shard:      vals[1].(uint32),
+			realWrites:      vals[0].([]generatedWrite),
+			shard:           vals[1].(uint32),
+			forceMmapMemory: vals[2].(bool),
 		}
 	})
 }

src/dbnode/persist/fs/index_lookup_test.go

@@ -79,6 +79,7 @@ func TestNewNearestIndexOffsetDetectsUnsortedFiles(t *testing.T) {
 		expectedDigest,
 		msgpack.NewDecoder(nil),
 		len(outOfOrderSummaries),
+		false,
 	)
 	expectedErr := fmt.Errorf("summaries file is not sorted: %s", file.Name())
 	require.Equal(t, expectedErr, err)
@@ -109,7 +110,7 @@ func TestClosingCloneDoesNotAffectParent(t *testing.T) {
 		},
 	}
 
-	indexLookup := newIndexLookupWithSummaries(t, indexSummaries)
+	indexLookup := newIndexLookupWithSummaries(t, indexSummaries, false)
 	clone, err := indexLookup.concurrentClone()
 	require.NoError(t, err)
 	require.NoError(t, clone.close())
@@ -125,6 +126,14 @@ func TestClosingCloneDoesNotAffectParent(t *testing.T) {
 }
 
 func TestParentAndClonesSafeForConcurrentUse(t *testing.T) {
+	testParentAndClonesSafeForConcurrentUse(t, false)
+}
+
+func TestParentAndClonesSafeForConcurrentUseForceMmapMemory(t *testing.T) {
+	testParentAndClonesSafeForConcurrentUse(t, true)
+}
+
+func testParentAndClonesSafeForConcurrentUse(t *testing.T, forceMmapMemory bool) {
 	numSummaries := 1000
 	numClones := 10
 
@@ -140,7 +149,7 @@ func TestParentAndClonesSafeForConcurrentUse(t *testing.T) {
 	sort.Sort(sortableSummaries(indexSummaries))
 
 	// Create indexLookup and associated clones
-	indexLookup := newIndexLookupWithSummaries(t, indexSummaries)
+	indexLookup := newIndexLookupWithSummaries(t, indexSummaries, forceMmapMemory)
 	clones := []*nearestIndexOffsetLookup{}
 	for i := 0; i < numClones; i++ {
 		clone, err := indexLookup.concurrentClone()
@@ -184,7 +193,8 @@ func TestParentAndClonesSafeForConcurrentUse(t *testing.T) {
 
 // newIndexLookupWithSummaries will return a new index lookup that is backed by the provided
 // indexSummaries (in the order that they are provided).
-func newIndexLookupWithSummaries(t *testing.T, indexSummaries []schema.IndexSummary) *nearestIndexOffsetLookup {
+func newIndexLookupWithSummaries(
+	t *testing.T, indexSummaries []schema.IndexSummary, forceMmapMemory bool) *nearestIndexOffsetLookup {
 	// Create a temp file
 	file, err := ioutil.TempFile("", "index-lookup-sort")
 	require.NoError(t, err)
@@ -211,6 +221,7 @@ func newIndexLookupWithSummaries(t *testing.T, indexSummaries []schema.IndexSumm
 		expectedDigest,
 		msgpack.NewDecoder(nil),
 		len(indexSummaries),
+		forceMmapMemory,
 	)
 	require.NoError(t, err)
 	return indexLookup

src/dbnode/persist/fs/mmap_util.go

@@ -0,0 +1,95 @@
+// Copyright (c) 2019 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 fs
+
+import (
+	"fmt"
+
+	"github.com/m3db/m3/src/dbnode/digest"
+	"github.com/m3db/m3/src/x/mmap"
+)
+
+func validateAndMmap(
+	fdWithDigest digest.FdWithDigestReader,
+	expectedDigest uint32,
+	forceMmapMemory bool,
+) ([]byte, error) {
+
+	if forceMmapMemory {
+		return validateAndMmapMemory(fdWithDigest, expectedDigest)
+	}
+
+	return validateAndMmapFile(fdWithDigest, expectedDigest)
+
+}
+
+func validateAndMmapMemory(
+	fdWithDigest digest.FdWithDigestReader,
+	expectedDigest uint32,
+) ([]byte, error) {
+	fd := fdWithDigest.Fd()
+	stat, err := fd.Stat()
+	if err != nil {
+		return nil, err
+	}
+	numBytes := stat.Size()
+
+	// Request an anonymous (non-file-backed) mmap region. Note that we're going
+	// to use the mmap'd region to store the read-only summaries data, but the mmap
+	// region itself needs to be writable so we can copy the bytes from disk
+	// into it.
+	mmapResult, err := mmap.Bytes(numBytes, mmap.Options{Read: true, Write: true})
+	if err != nil {
+		return nil, err
+	}
+
+	mmapBytes := mmapResult.Result
+
+	// Validate the bytes on disk using the digest, and read them into
+	// the mmap'd region
+	_, err = fdWithDigest.ReadAllAndValidate(mmapBytes, expectedDigest)
+	if err != nil {
+		mmap.Munmap(mmapBytes)
+		return nil, err
+	}
+
+	return mmapBytes, nil
+}
+
+func validateAndMmapFile(
+	fdWithDigest digest.FdWithDigestReader,
+	expectedDigest uint32,
+) ([]byte, error) {
+	fd := fdWithDigest.Fd()
+	mmapResult, err := mmap.File(fd, mmap.Options{Read: true, Write: false})
+	if err != nil {
+		return nil, err
+	}
+
+	mmapBytes := mmapResult.Result
+	if calculatedDigest := digest.Checksum(mmapBytes); calculatedDigest != expectedDigest {
+		mmap.Munmap(mmapBytes)
+		return nil, fmt.Errorf("expected summaries file digest was: %d, but got: %d",
+			expectedDigest, calculatedDigest)
+	}
+
+	return mmapBytes, nil
+}