mongodb_stress workload enhancements (GoogleCloudPlatform#110)

* added flags to override mongodb cache size and fill memory so that it isn't available to mongodb or file cache * added flag to YCSB to bind instances to NUMA nodes: ycsb_numa_bind (default: False) * added flag to YCSB to skip loading phase: ycsb_load_database (default: True) * added flags MongoDB_stress to control mongodb NUMA node affinity * important note: default behavior has changed for NUMA. Set the mongodb_numa_bind_cpu flag to maintain previous behavior.
keerockl · Mar 14, 2019 · de9b52a · de9b52a
1 parent f738e94
commit de9b52a
Show file tree

Hide file tree

Showing 3 changed files with 147 additions and 23 deletions.
diff --git a/perfkitbenchmarker/linux_benchmarks/mongodb_stress_benchmark.py b/perfkitbenchmarker/linux_benchmarks/mongodb_stress_benchmark.py
@@ -39,6 +39,25 @@
 
 flags.DEFINE_integer('mongodb_ycsb_processes', 1,
                      'Number of total ycsb processes across all clients.')
+flags.DEFINE_boolean('mongodb_numa_bind_cpu', False,
+                     'Bind mongodb instances to CPUs. The CPUs will '
+                     'be allocated as evenly as possible across instances.')
+flags.DEFINE_boolean('mongodb_numa_bind_node', False,
+                     'Bind mongodb instances to alternating NUMA nodes.')
+flags.DEFINE_boolean('mongodb_numa_memory_interleave', False,
+                     'MongoDB memory allocations will be made from '
+                     'alternating NUMA nodes. That is, memory will be '
+                     'allocated evenly across NUMA nodes.')
+flags.DEFINE_boolean('mongodb_numa_memory_bind', False,
+                     'MongoDB instance memory allocations will only be '
+                     'made from a single NUMA node.')
+flags.DEFINE_integer('mongodb_ycsb_load_threads', 4,
+                     'Number of threads used per YCSB instance to in load phase.')
+flags.DEFINE_integer('mongodb_cache_size', None,
+                     'Override default cache size per database instance. Units: GB')
+flags.DEFINE_integer('mongodb_fill_memory', 0,
+                     'Prevent mongodb and file cache from using this memory. '
+                     'Units: GB')
 FLAGS = flags.FLAGS
 
 BENCHMARK_NAME = 'mongodb_stress'
@@ -170,22 +189,86 @@ def _GetSiblingHyperThreadCpuId(physical_cpu_id):
   return ','.join(str(s) for s in instance_physical_cpu_ids + instance_hyperthread_cpu_ids)
 
 
+def _GetNodeForInstance(instance_index, numa_node_count):
+  return instance_index % numa_node_count
+
+
+def _RemoteBackgroundCommand(vm, command):
+  vm.RemoteCommand('nohup {0} 1> /dev/null 2> /dev/null &'.format(command))
+
+
+def _FillMemory(vm):
+  assert FLAGS.mongodb_fill_memory
+  vm.Install('stress_ng')
+  gb_free = vm.total_free_memory_kb / 1024 / 1024
+  if FLAGS.mongodb_fill_memory > gb_free:
+    raise errors.PrepareException("mongodb_fill_memory value set higher "
+                                  "than free memory %dGB" % (gb_free))
+  num_nodes = _GetNumaNodeCount(vm)
+  assert num_nodes > 0
+  cmd = 'stress-ng --vm-bytes {0}g --vm-keep --vm-hang 0 --vm {1}'.format(
+                                FLAGS.mongodb_fill_memory // num_nodes, vm.num_cpus // num_nodes)
+  if num_nodes:
+    vm.InstallPackages('numactl')
+    for node in range(num_nodes):
+      nc_cmd = 'numactl --cpunodebind={0} --membind={0} -- {1}'.format(
+                                              node, cmd)
+      _RemoteBackgroundCommand(vm, nc_cmd)
+  else:
+    _RemoteBackgroundCommand(vm, cmd)
+
+
+def _UnfillMemory(vm):
+  assert FLAGS.mongodb_fill_memory
+  vm.RemoteCommand('sudo pkill stress-ng')
+
+
 def _PrepareServer(vm):
-  """Installs MongoDB on the server."""
+  """Installs, Configures, Starts MongoDB instance(s) on the server."""
+  # install
   vm.Install('mongodb')
-  mongodb.Configure(vm)
-  # start all instances on VM
-  if _GetNumaNodeCount(vm) > 1:
+  # fill a block of memory (maybe)
+  if FLAGS.mongodb_fill_memory:
+    _FillMemory(vm)
+  # configure and start mongodb
+  if not FLAGS.mongodb_numa_bind_cpu and \
+     not FLAGS.mongodb_numa_bind_node and \
+     not FLAGS.mongodb_numa_memory_interleave and \
+     not FLAGS.mongodb_numa_memory_bind:
+    mongodb.Configure(vm, FLAGS.mongodb_cache_size, FLAGS.mongodb_fill_memory)
+    mongodb.Start(vm)
+  else:
     vm.InstallPackages('numactl')
+    numa_node_count = _GetNumaNodeCount(vm)
+    # special case for memory node binding a single MongoDB instance
+    if not FLAGS.mongodb_cache_size and \
+      FLAGS.mongodb_numa_memory_bind and \
+      FLAGS.mongodb_total_num_processes == 1:
+      # we assume memory is equally distributed across NUMA nodes
+      # MongoDB recommends cache size of 50% of (RAM - 1GB)
+      total_ram_gb = vm.total_memory_kb / 1000 / 1000
+      # take away filled memory (if any)
+      total_ram_gb -= FLAGS.mongodb_fill_memory
+      numa_node_ram_gb = total_ram_gb / numa_node_count
+      instance_cache_size = (numa_node_ram_gb - 1) / 2
+    else:
+      instance_cache_size = FLAGS.mongodb_cache_size
+    mongodb.Configure(vm, instance_cache_size)
+    vm.RemoteCommand('sudo sync')
+    vm.DropCaches()
     for i in range(FLAGS.mongodb_total_num_processes):
-      start_command = mongodb.GetStartCommand(vm, i)
-      full_command = 'sudo numactl --interleave=all --physcpubind={0} -- {1}'.format(_GetCpusForInstance(vm, i), start_command)
-      logging.info('Starting MongoDB: {0}'.format(full_command))
-      vm.RemoteCommand(full_command)
-  else:
-    # TODO: for completeness, when we have a single NUMA node, we could use taskset to
-    # affinitize processes to CPUs.
-    mongodb.Start(vm)
+      cmd = 'sudo numactl'
+      if FLAGS.mongodb_numa_bind_cpu:
+        cmd += ' --physcpubind={0}'.format(_GetCpusForInstance(vm, i))
+      elif FLAGS.mongodb_numa_bind_node:
+        cmd += ' --cpunodebind={0}'.format(_GetNodeForInstance(i, numa_node_count))
+      if FLAGS.mongodb_numa_memory_interleave:
+        cmd += ' --interleave=all'
+      elif FLAGS.mongodb_numa_memory_bind:
+        cmd += ' --membind={0}'.format(_GetNodeForInstance(i, numa_node_count))
+      cmd += ' -- {0}'.format(mongodb.GetStartCommand(vm, i))
+      logging.info('Starting MongoDB: {0}'.format(cmd))
+      vm.RemoteCommand(cmd)
 
 
 def _PrepareClient(vm):
@@ -204,6 +287,7 @@ def Prepare(benchmark_spec):
     benchmark_spec: The benchmark specification. Contains all data that is
         required to run the benchmark.
   """
+  benchmark_spec.always_call_cleanup = True
   mongodb_vm = benchmark_spec.vm_groups['workers'][0]
   server_partials = [functools.partial(_PrepareServer, mongodb_vm)]
   client_partials = [functools.partial(_PrepareClient, client)
@@ -228,7 +312,7 @@ def Prepare(benchmark_spec):
           'perclientparam': server_metadata})
   # load records into database(s)
   benchmark_spec.executor.Load(_GetClientVms(benchmark_spec), 
-                               load_kwargs={'threads': 4})
+                               load_kwargs={'threads': FLAGS.mongodb_ycsb_load_threads})
 
 
 def _GetClientVms(benchmark_spec):
@@ -274,4 +358,7 @@ def Cleanup(benchmark_spec):
     benchmark_spec: The benchmark specification. Contains all data that is
         required to run the benchmark.
   """
-  mongodb.Stop(benchmark_spec.vm_groups['workers'][0])
+  vm = benchmark_spec.vm_groups['workers'][0]
+  mongodb.Stop(vm)
+  if FLAGS.mongodb_fill_memory:
+    _UnfillMemory(vm)
diff --git a/perfkitbenchmarker/linux_packages/mongodb.py b/perfkitbenchmarker/linux_packages/mongodb.py
@@ -13,6 +13,9 @@
 # limitations under the License.
 
 """Module containing mongodb installation and cleanup functions."""
+import time
+import logging
+
 from perfkitbenchmarker import errors
 from perfkitbenchmarker.linux_packages import INSTALL_DIR
 from perfkitbenchmarker import flags
@@ -21,6 +24,10 @@
 flags.DEFINE_integer('mongodb_total_num_processes', 1,
                      'Total number of mongodb server processes.',
                      lower_bound=1)
+flags.DEFINE_string('mongodb_pre_generated', None,
+                    'Path to root of pre-generated database directories. '
+                    'Directory names within must match this format: '
+                    'mongod-<port>-data.')
 
 MONGODB_DIR = '%s/mongodb-folder' % (INSTALL_DIR)
 MONGODB_FIRST_PORT = 27017
@@ -45,15 +52,17 @@ def _Install(vm, mongodb_dl_url,mongodb_tar, mongodb_dl_dir):
   vm.RemoteCommand('cd %s && tar xvfz %s' % (INSTALL_DIR, mongodb_tar)) 
   vm.RemoteCommand('mv %s %s' %(mongodb_dl_dir, MONGODB_DIR))
 
-def Configure(vm):
+def Configure(vm, cache_size=None, reserved_memory=0):
   """Configure mongodb server."""
-  # mongodb.conf from: https://github.com/mongodb/mongo/blob/v4.0/debian/mongod.conf
-
-  total_ram_gb = vm.total_memory_kb / 1000 / 1000
-  cache_size = (total_ram_gb - FLAGS.mongodb_total_num_processes) / FLAGS.mongodb_total_num_processes / 2
+  if not cache_size:
+    # MongoDB recommends cache size of 50% of (RAM - 1GB)
+    total_ram_gb = (vm.total_memory_kb / 1000 / 1000) - reserved_memory
+    useable_ram_gb = (total_ram_gb - FLAGS.mongodb_total_num_processes) / 2
+    cache_size = useable_ram_gb / FLAGS.mongodb_total_num_processes
   for i in range(FLAGS.mongodb_total_num_processes):
     port = MONGODB_FIRST_PORT + i
     dbpath = '{0}/mongod-{1}-data'.format(vm.GetScratchDir(), port)
+    # mongodb.conf from: https://github.com/mongodb/mongo/blob/v4.0/debian/mongod.conf
     config_str = """\
 # mongod.conf
 # for documentation of all options, see:
@@ -91,6 +100,9 @@ def Configure(vm):
 """.format(port=port, bindip=vm.internal_ip, cache_size=cache_size, dbpath=dbpath)
     vm.RemoteCommand('echo "{0}" > {1}/mongod-{2}.conf'.format(config_str, MONGODB_DIR, port)) 
     vm.RemoteCommand('mkdir -p {0}'.format(dbpath))
+    if FLAGS.mongodb_pre_generated:
+      cmd = 'sudo cp -r {0}/mongod-{1}-data/* {2}/mongod-{1}-data/'.format(FLAGS.mongodb_pre_generated, port, vm.GetScratchDir())
+      vm.RemoteCommand(cmd)
 
 
 def GetStartCommand(vm, instance_index):
@@ -100,14 +112,25 @@ def GetStartCommand(vm, instance_index):
 
 def Start(vm):
   """Start mongodb server process."""
+  vm.RemoteCommand('sudo sync')
+  vm.DropCaches()
   for i in range(FLAGS.mongodb_total_num_processes):
     vm.RemoteCommand(GetStartCommand(vm, i))
 
 
 def Stop(vm):
   """Stop all instances of mongod."""
+  vm.RemoteCommand('sudo sync')
+  vm.DropCaches()
   vm.RemoteCommand('sudo pkill mongod')
-  vm.RemoteCommand('sudo rm -r {0}/mongod-*-data'.format(vm.GetScratchDir()))
+  logging.info("Wait a few seconds after killing mongod before deleting data files.")
+  time.sleep(5)
+  # remove data files
+  for i in range(FLAGS.mongodb_total_num_processes):
+    vm.RemoteCommand('sudo rm -r {0}/mongod-{1}-data/*'.format(vm.GetScratchDir(), MONGODB_FIRST_PORT + i))
+    vm.RemoteCommand('sudo rmdir {0}/mongod-{1}-data'.format(vm.GetScratchDir(), MONGODB_FIRST_PORT + i),
+      ignore_failure=True)
+
 
 def Uninstall(vm):
   """Remove mongodb."""

diff --git a/perfkitbenchmarker/linux_packages/ycsb.py b/perfkitbenchmarker/linux_packages/ycsb.py
@@ -114,6 +114,12 @@
                      'Reload database, othewise skip load stage. '
                      'Note, this flag is only used if the database '
                      'is already loaded.')
+flags.DEFINE_boolean('ycsb_load_database', True,
+                     'Load database, otherwise skip load stage. Note '
+                     'if this flag is False, load will not be run.')
+flags.DEFINE_boolean('ycsb_numa_bind', False,
+                     'Bind multiple instances of YCSB to alternating '
+                     'NUMA nodes.')
 flags.DEFINE_integer('ycsb_client_vms', 1, 'Number of YCSB client VMs.')
 flags.DEFINE_list('ycsb_workload_files', ['workloada', 'workloadb'],
                   'Path to YCSB workload file to use during *run* '
@@ -208,7 +214,8 @@ def _Install(vm):
   """Installs the YCSB and, if needed, hdrhistogram package on the VM."""
   vm.Install('openjdk')
   vm.Install('curl')
-  vm.InstallPackages('numactl')
+  if FLAGS.ycsb_numa_bind:
+    vm.InstallPackages('numactl')
   ycsb_url = ('https://github.com/brianfrankcooper/YCSB/releases/'
               'download/{0}/ycsb-{0}.tar.gz').format(FLAGS.ycsb_version)
   install_cmd = ('mkdir -p {0} && curl -L {1} | '
@@ -921,6 +928,10 @@ def _Load(loader_index):
       results.append(self._Load(vms[loader_index], **kw))
       logging.info('VM %d (%s) finished', loader_index, vms[loader_index])
 
+    if not FLAGS.ycsb_load_database:
+      logging.info('ycsb_load_database flag is False. Skipping load.')
+      return []
+
     start = time.time()
     vm_util.RunThreaded(_Load, range(len(vms)))
     events.record_event.send(
@@ -957,8 +968,11 @@ def _Run(self, vm, **kwargs):
     # bind ycsb instances to alternating NUMA nodes
     # this will work best if the number of instances is a multiple
     # of the number of NUMA nodes
-    if kwargs['duplicate_count'] > 1:
-      node = kwargs['duplicate_index'] % _GetNumaNodeCount(vm)
+    numa_node_count = _GetNumaNodeCount(vm)
+    if FLAGS.ycsb_numa_bind and \
+        numa_node_count > 1 and \
+        kwargs['duplicate_count'] > 1:
+      node = kwargs['duplicate_index'] % numa_node_count
     else:
       node = None
     command = self._BuildCommand('run', numa_node=node, **kwargs)