storage: raise MemTableStopWritesThreshold when memory permits

When there's sufficient memory available, raise the MemTableStopWritesThreshold
up to as high as 16. We've observed instances of memtable write stalls in
practice, especially in the presence of high latencies in cloud networked block
devices. If the higher levels are not writing to the storage engine too fast,
but flushes are unable to keep up because of disk slowness, we'd rather
accumulate additional memtables that forcibly stall writes.

Epic: none
Fixes: cockroachdb#153673.
Release note: None

Author: jbowens

pkg/server/BUILD.bazel

@@ -613,13 +613,15 @@ go_test(
         "//pkg/util/tracing",
         "//pkg/util/tracing/tracingpb",
         "//pkg/util/uuid",
+        "@com_github_cockroachdb_crlib//crstrings",
         "@com_github_cockroachdb_datadriven//:datadriven",
         "@com_github_cockroachdb_errors//:errors",
         "@com_github_cockroachdb_logtags//:logtags",
         "@com_github_cockroachdb_pebble//:pebble",
         "@com_github_cockroachdb_pebble//vfs",
         "@com_github_cockroachdb_redact//:redact",
         "@com_github_dustin_go_humanize//:go-humanize",
+        "@com_github_ghemawat_stream//:stream",
         "@com_github_gogo_protobuf//jsonpb",
         "@com_github_gogo_protobuf//proto",
         "@com_github_gorilla_mux//:mux",

pkg/server/config.go

@@ -743,9 +743,26 @@ func (cfg *Config) CreateEngines(ctx context.Context) (Engines, error) {
 	if err != nil {
 		return Engines{}, err
 	}
-
 	log.Event(ctx, "initializing engines")
 
+	// The (pebble.Options).MemTableStopWritesThreshold configures the number of
+	// memtables that may be queued before Pebble induces a write stall.
+	// Queueing memtables consume memory from the block cache, evicting resident
+	// blocks. If flushes are not keeping up and the count of queued memtables
+	// grows too large, read performance will degrade severely:
+	//
+	// - Every read needs to seek in every queued memtable.
+	// - Memtables take memory from the block cache, meaning that block
+	//   cache effectiveness decreases the more memtables that are queued.
+	//
+	// We constrain the count of queued memtables to be between 4 and 16. Within
+	// those bounds, we'll grow it to use up to half of the block cache. If
+	// there are multiple stores, we need to divide that half by the count of
+	// stores.
+	stopWritesThreshold := int(cfg.CacheSize/2/storage.DefaultMemtableSize) / len(cfg.Stores.Specs)
+	stopWritesThreshold = max(stopWritesThreshold, 4)
+	stopWritesThreshold = min(stopWritesThreshold, 16)
+
 	var fileCache *pebble.FileCache
 	// TODO(radu): use the fileCache for in-memory stores as well.
 	if physicalStores > 0 {
@@ -783,6 +800,7 @@ func (cfg *Config) CreateEngines(ctx context.Context) (Engines, error) {
 			storage.Attributes(roachpb.Attributes{Attrs: spec.Attributes}),
 			storage.If(storeKnobs.SmallEngineBlocks, storage.BlockSize(1)),
 			storage.BlockConcurrencyLimitDivisor(len(cfg.Stores.Specs)),
+			storage.MemTableStopWritesThreshold(stopWritesThreshold),
 		}
 		if len(storeKnobs.EngineKnobs) > 0 {
 			storageConfigOpts = append(storageConfigOpts, storeKnobs.EngineKnobs...)

pkg/server/config_test.go

@@ -6,10 +6,14 @@
 package server
 
 import (
+	"bytes"
 	"context"
+	"fmt"
+	"io"
 	"net"
 	"os"
 	"reflect"
+	"strings"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -27,7 +31,10 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/netutil"
 	"github.com/cockroachdb/cockroach/pkg/util/netutil/addr"
+	"github.com/cockroachdb/crlib/crstrings"
+	"github.com/cockroachdb/datadriven"
 	"github.com/cockroachdb/pebble/vfs"
+	"github.com/ghemawat/stream"
 	"github.com/kr/pretty"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -368,3 +375,76 @@ func TestIdProviderServerIdentityString(t *testing.T) {
 		})
 	}
 }
+
+func TestCreateEngines(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	specs := map[string]base.StoreSpec{}
+
+	datadriven.RunTest(t, "testdata/create_engines", func(t *testing.T, d *datadriven.TestData) string {
+		switch d.Cmd {
+		case "store-spec":
+			var name string
+			d.ScanArgs(t, "name", &name)
+			spec := base.StoreSpec{}
+			spec.Path = name
+			for _, line := range crstrings.Lines(d.Input) {
+				parts := strings.SplitN(line, "=", 2)
+				switch parts[0] {
+				case "in-memory":
+					spec.InMemory = true
+				case "attrs":
+					spec.Attributes = strings.Split(parts[1], ":")
+				default:
+					return fmt.Sprintf("unknown field: %q", parts[0])
+				}
+			}
+			specs[name] = spec
+			return ""
+		case "create-engines":
+			var cfg Config
+			cfg.Settings = cluster.MakeTestingClusterSettings()
+
+			var pattern string
+			d.ScanArgs(t, "pattern", &pattern)
+			var specNames []string
+			d.ScanArgs(t, "specs", &specNames)
+			for _, specName := range specNames {
+				spec, ok := specs[specName]
+				if !ok {
+					return fmt.Sprintf("unknown store: %q", specName)
+				}
+				cfg.Stores.Specs = append(cfg.Stores.Specs, spec)
+			}
+			d.MaybeScanArgs(t, "CacheSize", &cfg.CacheSize)
+
+			engines, err := cfg.CreateEngines(context.Background())
+			if err != nil {
+				return fmt.Sprintf("failed to create engines: %v", err)
+			}
+			defer engines.Close()
+
+			var buf bytes.Buffer
+			for _, e := range engines {
+				buf.WriteString(strings.TrimSpace(e.GetPebbleOptions().String()))
+				buf.WriteString("\n")
+			}
+			return grepStr(&buf, pattern)
+		default:
+			return fmt.Sprintf("unknown command: %q", d.Cmd)
+		}
+	})
+}
+
+func grepStr(r io.Reader, pattern string) string {
+	var buf bytes.Buffer
+	if err := stream.Run(stream.Sequence(
+		stream.ReadLines(r),
+		stream.Grep(pattern),
+		stream.WriteLines(&buf),
+	)); err != nil {
+		return err.Error()
+	}
+	return buf.String()
+}

pkg/server/testdata/create_engines

@@ -0,0 +1,40 @@
+store-spec name=a
+in-memory
+----
+
+store-spec name=b
+in-memory
+----
+
+# 64 MiB cache; should result in 4 memtable stop writes threshold.
+
+create-engines specs=(a,b) pattern=mem_table_stop_writes_threshold CacheSize=67108864
+----
+  mem_table_stop_writes_threshold=4
+  mem_table_stop_writes_threshold=4
+
+create-engines specs=(a) pattern=mem_table_stop_writes_threshold CacheSize=67108864
+----
+  mem_table_stop_writes_threshold=4
+
+# 1 GiB cache; should result in 8 memtable stop writes threshold with a single
+# store (with 64 MiB memtables).
+
+create-engines specs=(b) pattern=mem_table_stop_writes_threshold CacheSize=1073741824
+----
+  mem_table_stop_writes_threshold=8
+
+# But if we add another store, the memtable stop writes threshold should be
+# halved back to the minimum of 4.
+
+create-engines specs=(a, b) pattern=mem_table_stop_writes_threshold CacheSize=1073741824
+----
+  mem_table_stop_writes_threshold=4
+  mem_table_stop_writes_threshold=4
+
+# 64 GiB cache; should result in 16 memtable stop writes threshold.
+
+create-engines specs=(a,b) pattern=mem_table_stop_writes_threshold CacheSize=68719476736
+----
+  mem_table_stop_writes_threshold=16
+  mem_table_stop_writes_threshold=16

pkg/storage/open.go

@@ -215,6 +215,16 @@ func MemtableSize(bytes uint64) ConfigOption {
 	}
 }
 
+// MemTableStopWritesThreshold configures the number of memtables that can be
+// queued before Pebble will induce a write stall, preventing all batch commits
+// from proceeding until a flush completes.
+func MemTableStopWritesThreshold(n int) ConfigOption {
+	return func(cfg *engineConfig) error {
+		cfg.opts.MemTableStopWritesThreshold = n
+		return nil
+	}
+}
+
 // L0CompactionThreshold configures the amount of L0 read-amplification
 // necessary to trigger an L0 compaction.
 func L0CompactionThreshold(n int) ConfigOption {

pkg/storage/pebble.go

@@ -519,6 +519,9 @@ var MVCCMerger = &pebble.Merger{
 	},
 }
 
+// DefaultMemtableSize is the default size of a memtable.
+const DefaultMemtableSize = 64 << 20 // 64 MB
+
 const mvccWallTimeIntervalCollector = "MVCCTimeInterval"
 
 // DefaultPebbleOptions returns the default pebble options.
@@ -532,7 +535,7 @@ func DefaultPebbleOptions() *pebble.Options {
 		L0CompactionThreshold:       2,
 		L0StopWritesThreshold:       1000,
 		LBaseMaxBytes:               64 << 20, // 64 MB
-		MemTableSize:                64 << 20, // 64 MB
+		MemTableSize:                DefaultMemtableSize,
 		MemTableStopWritesThreshold: 4,
 		Merger:                      MVCCMerger,
 		BlockPropertyCollectors:     cockroachkvs.BlockPropertyCollectors,