Add experimental mempurge policy flag to db_stress. (#8588)

Summary:
Add `experimental_mempurge_policy` flag to `db_stress` and `db_crashtest.py`.
This flag is only read if the `experimental_allow_mempurge` flag is set to `true`. This flag can take the following values: `kAlways`, and `kAlternate` (default).
- `kAlways`: a flush is always redirected to a mempurge. If the mempurge aborts, the a regular flush proceeds.
- `kAlternate`: if one or more of the flush input memtables is an mempurge output memtable, then a flush is performed, else a mempurge is carried out. Similar to kAlways, if a mempurge aborts, the FlushJob proceeds to a regular flush to storage.

Pull Request resolved: https://github.com/facebook/rocksdb/pull/8588

Reviewed By: pdillinger

Differential Revision: D29934251

Pulled By: bjlemaire

fbshipit-source-id: 90c1debed2029b9915d066914556547507c33dae

db_stress_tool/db_stress_common.h

@@ -142,6 +142,7 @@ DECLARE_uint64(periodic_compaction_seconds);
 DECLARE_uint64(compaction_ttl);
 DECLARE_bool(allow_concurrent_memtable_write);
 DECLARE_bool(experimental_allow_mempurge);
+DECLARE_string(experimental_mempurge_policy);
 DECLARE_bool(enable_write_thread_adaptive_yield);
 DECLARE_int32(reopen);
 DECLARE_double(bloom_bits);
@@ -340,6 +341,18 @@ inline enum ROCKSDB_NAMESPACE::CompressionType StringToCompressionType(
   return ret_compression_type;
 }
 
+inline enum ROCKSDB_NAMESPACE::MemPurgePolicy StringToMemPurgePolicy(
+    const char* mpolicy) {
+  assert(mpolicy);
+  if (!strcasecmp(mpolicy, "kAlways")) {
+    return ROCKSDB_NAMESPACE::MemPurgePolicy::kAlways;
+  } else if (!strcasecmp(mpolicy, "kAlternate")) {
+    return ROCKSDB_NAMESPACE::MemPurgePolicy::kAlternate;
+  }
+  fprintf(stderr, "Cannot parse mempurge policy: '%s'\n", mpolicy);
+  return ROCKSDB_NAMESPACE::MemPurgePolicy::kAlternate;
+}
+
 inline enum ROCKSDB_NAMESPACE::ChecksumType StringToChecksumType(
     const char* ctype) {
   assert(ctype);

db_stress_tool/db_stress_gflags.cc

@@ -329,6 +329,9 @@ DEFINE_bool(allow_concurrent_memtable_write, false,
 DEFINE_bool(experimental_allow_mempurge, false,
             "Allow mempurge process to collect memtable garbage bytes.");
 
+DEFINE_string(experimental_mempurge_policy, "kAlternate",
+              "Set mempurge (MemTable Garbage Collection) policy.");
+
 DEFINE_bool(enable_write_thread_adaptive_yield, true,
             "Use a yielding spin loop for brief writer thread waits.");
 

db_stress_tool/db_stress_test_base.cc

@@ -2268,6 +2268,8 @@ void StressTest::Open() {
     options_.allow_concurrent_memtable_write =
         FLAGS_allow_concurrent_memtable_write;
     options_.experimental_allow_mempurge = FLAGS_experimental_allow_mempurge;
+    options_.experimental_mempurge_policy =
+        StringToMemPurgePolicy(FLAGS_experimental_mempurge_policy.c_str());
     options_.periodic_compaction_seconds = FLAGS_periodic_compaction_seconds;
     options_.ttl = FLAGS_compaction_ttl;
     options_.enable_pipelined_write = FLAGS_enable_pipelined_write;

tools/db_crashtest.py

@@ -220,7 +220,8 @@ whitebox_default_params = {
 simple_default_params = {
     "allow_concurrent_memtable_write": lambda: random.randint(0, 1),
     "column_families": 1,
-    "experimental_allow_mempurge": 0,
+    "experimental_allow_mempurge": lambda: random.randint(0, 1),
+    "experimental_mempurge_policy": lambda: random.choice(["kAlways", "kAlternate"]),
     "max_background_compactions": 1,
     "max_bytes_for_level_base": 67108864,
     "memtablerep": "skip_list",