Add support for specifying concurrency levels for plaintext test (#3068)

* Replace hardcoded concurrency values with field in benchmarks.cfg

* If the field is not specified, fall back to previous default (to
maintain backwards compatibility)

benchmark.cfg.example

@@ -17,6 +17,7 @@ install_strategy=unified
 install_only=False
 list_tests=False
 concurrency_levels=[16, 32, 64, 128, 256, 512]
+pipeline_concurrency_levels=[256,1024,4096,16384]
 query_levels=[1,5,10,15,20]
 cached_query_levels=[1,10,20,50,100]
 mode=benchmark

deployment/vagrant/bootstrap.sh

@@ -55,6 +55,7 @@ install_strategy=unified
 install_only=False
 list_tests=False
 concurrency_levels=[8, 16, 32, 64, 128, 256]
+pipeline_concurrency_levels=[256,1024,4096,16384]
 query_levels=[1, 5,10,15,20]
 cached_query_levels=[1,10,20,50,100]
 threads=8

toolset/benchmark/benchmarker.py

@@ -931,6 +931,8 @@ class Benchmarker:
         del args['type']
 
         args['max_concurrency'] = max(args['concurrency_levels'])
+        if 'pipeline_concurrency_levels' not in args:
+            args['pipeline_concurrency_levels'] = [256,1024,4096,16384]
 
         self.__dict__.update(args)
         # pprint(self.__dict__)
@@ -986,6 +988,7 @@ class Benchmarker:
             self.results['startTime'] = int(round(time.time() * 1000))
             self.results['completionTime'] = None
             self.results['concurrencyLevels'] = self.concurrency_levels
+            self.results['pipelineConcurrencyLevels'] = self.pipeline_concurrency_levels
             self.results['queryIntervals'] = self.query_levels
             self.results['cachedQueryIntervals'] = self.cached_query_levels
             self.results['frameworks'] = [t.name for t in self.__gather_tests]

toolset/benchmark/framework_test.py

@@ -657,9 +657,9 @@ class FrameworkTest:
   ############################################################
   def __generate_pipeline_script(self, url, port, accept_header, wrk_command="wrk"):
     headers = self.headers_template.format(server_host=self.benchmarker.server_host, accept=accept_header)
-    return self.pipeline_template.format(max_concurrency=16384,
+    return self.pipeline_template.format(max_concurrency=max(self.benchmarker.pipeline_concurrency_levels),
       name=self.name, duration=self.benchmarker.duration,
-      levels=" ".join("{}".format(item) for item in [256,1024,4096,16384]),
+      levels=" ".join("{}".format(item) for item in self.benchmarker.pipeline_concurrency_levels),
       server_host=self.benchmarker.server_host, port=port, url=url, headers=headers, wrk=wrk_command,
       pipeline=16)