From: Perry Gagne <pgagne(a)redhat.com>
Added support for a `Tool` wrapper around [
neper](https://github.com/google/neper).
The primary use case for this is to use neper's `tcp_rr`, `tcp_crr`, and `udp_rr`
tests
for `ShortLivedConnections` and other latency tests. Neper also supports other tests like
`tcp_stream`, 'udp_stream', etc but we are using them at this time.
Including `Measurement` and `MeasurementGenerator` related code.
Signed-off-by: Perry Gagne <pgagne(a)redhat.com>
---
.../Perf/Measurements/NeperFlowMeasurement.py | 206 ++++++++++++++++++
.../Perf/Measurements/__init__.py | 1 +
.../NeperMeasurementGenerator.py | 109 +++++++++
lnst/Tests/Neper.py | 128 +++++++++++
4 files changed, 444 insertions(+)
create mode 100644 lnst/RecipeCommon/Perf/Measurements/NeperFlowMeasurement.py
create mode 100644 lnst/Recipes/ENRT/MeasurementGenerators/NeperMeasurementGenerator.py
create mode 100644 lnst/Tests/Neper.py
diff --git a/lnst/RecipeCommon/Perf/Measurements/NeperFlowMeasurement.py
b/lnst/RecipeCommon/Perf/Measurements/NeperFlowMeasurement.py
new file mode 100644
index 0000000..84a00c1
--- /dev/null
+++ b/lnst/RecipeCommon/Perf/Measurements/NeperFlowMeasurement.py
@@ -0,0 +1,206 @@
+import time
+from typing import List, Dict, Tuple
+from lnst.Common.IpAddress import ipaddress
+from lnst.Controller.Job import Job
+from lnst.Common.Utils import pairwise
+from lnst.Controller.Recipe import RecipeError
+from lnst.Controller.RecipeResults import ResultLevel
+from lnst.RecipeCommon.Perf.Measurements.BaseFlowMeasurement import BaseFlowMeasurement,
NetworkFlowTest, \
+ FlowMeasurementResults, Flow
+from lnst.RecipeCommon.Perf.Measurements.MeasurementError import MeasurementError
+from lnst.RecipeCommon.Perf.Results import PerfInterval, SequentialPerfResult,
ParallelPerfResult
+from lnst.Tests.Neper import NeperServer, NeperClient
+
+
+class NeperFlowMeasurement(BaseFlowMeasurement):
+ _MEASUREMENT_VERSION = 1
+
+ def __init__(self, flows: List[Flow], recipe_conf=None):
+ super(NeperFlowMeasurement, self).__init__(recipe_conf)
+ self._flows = flows
+ self._running_measurements = []
+ self._finished_measurements = []
+
+ @property
+ def flows(self) -> List[Flow]:
+ return self._flows
+
+ def start(self):
+ if len(self._running_measurements) > 0:
+ raise MeasurementError("Measurement already running!")
+
+ test_flows = self._prepare_test_flows(self.flows)
+
+ for flow in test_flows:
+ flow.server_job.start(bg=True)
+
+ for flow in test_flows:
+ flow.client_job.start(bg=True)
+
+ self._running_measurements = test_flows
+
+ def finish(self):
+ test_flows = self._running_measurements
+ try:
+ for flow in test_flows:
+ client_neper = flow.client_job.what
+ flow.client_job.wait(timeout=client_neper.runtime_estimate())
+ flow.server_job.wait(timeout=5)
+ finally:
+ for flow in test_flows:
+ flow.server_job.kill()
+ flow.client_job.kill()
+
+ self._running_measurements = []
+ self._finished_measurements = test_flows
+
+ def _prepare_test_flows(self, flows: List[Flow]):
+ test_flows = []
+ for flow in flows:
+ server_job = self._prepare_server(flow)
+ client_job = self._prepare_client(flow)
+ test_flow = NetworkFlowTest(flow, server_job, client_job)
+ test_flows.append(test_flow)
+ return test_flows
+
+ def _prepare_server(self, flow: Flow) -> Job:
+ host = flow.receiver
+ server_params = dict(workload = flow.type,
+ bind = ipaddress(flow.receiver_bind),
+ test_length = flow.duration)
+
+ if flow.cpupin is not None and flow.cpupin >= 0:
+ if flow.parallel_streams == 1:
+ server_params["cpu_bind"] = flow.cpupin
+ else:
+ raise RecipeError("Unsupported combination of single cpupin "
+ "with parallel perf streams.")
+ elif flow.cpupin is not None:
+ raise RecipeError("Negative perf cpupin value provided.")
+
+ return host.prepare_job(NeperServer(**server_params),
+ job_level=ResultLevel.NORMAL)
+
+ def _prepare_client(self, flow: Flow) -> Job:
+ host = flow.generator
+ client_params = dict(workload = flow.type,
+ server = ipaddress(flow.receiver_bind),
+ test_length = flow.duration)
+
+ if flow.cpupin is not None and flow.cpupin >= 0:
+ if flow.parallel_streams == 1:
+ client_params["cpu_bind"] = flow.cpupin
+ else:
+ raise RecipeError("Unsupported combination of single cpupin "
+ "with parallel perf streams.")
+ elif flow.cpupin is not None:
+ raise RecipeError("Negative perf cpupin value provided.")
+
+ #TODO Figure out what to do about parallel_streams
+ # (num_treads? num_flows? possible 2 instances runing at once?)
+ # Added NeperBase options to configure num_threads, num_flows but
+ # it appears parallel streams is not needed for now.
+ # The legacy lnst doesnt seem to use the paralellism even when
+ # setting perf_parallel_steams
+ #if flow.parallel_streams > 1:
+ # client_params["parallel"] = flow.parallel_streams
+
+ if flow.msg_size:
+ client_params["request_size"] = flow.msg_size
+ client_params["response_size"] = flow.msg_size
+
+ return host.prepare_job(NeperClient(**client_params),
+ job_level=ResultLevel.NORMAL)
+
+ def collect_results(self) -> List[FlowMeasurementResults]:
+ test_flows = self._finished_measurements
+
+ results = []
+ for test_flow in test_flows:
+ flow_results = FlowMeasurementResults(
+ measurement=self,
+ flow=test_flow.flow)
+ generator_stats = self._parse_job_samples(test_flow.client_job)
+ flow_results.generator_results = generator_stats[0]
+ flow_results.generator_cpu_stats = generator_stats[1]
+
+ receiver_stats = self._parse_job_samples(test_flow.server_job)
+ flow_results.receiver_results = receiver_stats[0]
+ flow_results.receiver_cpu_stats = receiver_stats[1]
+
+ results.append(flow_results)
+
+ return results
+
+ def _parse_job_samples(self, job: Job) ->\
+ Tuple[ParallelPerfResult, ParallelPerfResult]:
+ """
+ each perfinterval is samples.csv line #2 (l2) - line #1 (l1) to get #
transactions and duration
+ timestamp is time of l1, but we need to convert it from CLOCK_MONOTONIC time to
unix time.
+ samples.csv looks like this:
+ ```
+
tid,flow_id,time,transactions,utime,stime,maxrss,minflt,majflt,nvcsw,nivcsw,latency_min,latency_mean,latency_max,latency_stddev
+
0,0,1898645.747723502,1,0.000371,0.000000,1144,39,0,2,0,0.000000,0.000000,0.000000,-nan
+
0,0,1898647.747733162,59322,0.185458,0.241758,1144,43,0,59320,0,0.000000,0.000000,0.000000,0.000000
+
0,0,1898648.747757407,89210,0.281500,0.354934,1144,43,0,89207,0,0.000000,0.000000,0.000000,0.000000
+
0,0,1898649.747737156,118790,0.281500,0.354934,1144,43,0,89207,0,0.000000,0.000000,0.000000,0.000000
+ ```
+ :param job:
+ :type job:
+ :return:
+ :rtype:
+ """
+
+ results = SequentialPerfResult()
+ cpu_results = SequentialPerfResult()
+
+ if not job.passed:
+ results.append(PerfInterval(0, 0, "transactions", time.time()))
+ cpu_results.append(PerfInterval(0, 0, "cpu_percent", time.time()))
+ else:
+ job_start = job.result['start_time']
+ samples = job.result['samples']
+ if samples is not None:
+ neper_start_time = float(samples[0]['time'])
+ for s_start, s_end in pairwise(samples):
+ flow, cpu = get_interval(s_start, s_end,
+ job_start, neper_start_time)
+ results.append(flow)
+ cpu_results.append(cpu)
+
+ #Wrap in ParallelPerfResult for now for easier graphing
+ #TODO When we add support for multiple flows and threads
+ #We want to update this accordingly.
+ p_results = ParallelPerfResult()
+ p_results.append(results)
+ p_cpu_results = ParallelPerfResult()
+ p_cpu_results.append(cpu_results)
+ return p_results, p_cpu_results
+
+
+def get_interval(s_start: Dict, s_end: Dict, job_start: float,
+ neper_start: float) -> Tuple[PerfInterval, PerfInterval]:
+
+ transactions = int(s_end['transactions']) -
int(s_start['transactions'])
+ s_start_time = float(s_start['time'])
+ s_start_utime = float(s_start['utime'])
+ s_start_stime = float(s_start['stime'])
+
+ s_end_time = float(s_end['time'])
+ s_end_utime = float(s_end['utime'])
+ s_end_stime = float(s_end['stime'])
+
+ # cpu_usage_percent = (utime_delta + stime_delta) / duration
+ utime_delta = s_end_utime - s_start_utime
+ stime_delta = s_end_stime - s_start_stime
+
+ # neper uses CLOCK_MONOTONIC, need to convert to
+ # unix time using job_start as reference
+ timestamp = job_start + (s_start_time - neper_start)
+ duration = s_end_time - s_start_time
+ cpu_usage = (utime_delta + stime_delta) / duration
+
+ interval = PerfInterval(transactions, duration, 'transactions', timestamp)
+ cpu_interval = PerfInterval(cpu_usage, duration, 'cpu_percent', timestamp)
+
+ return interval, cpu_interval
diff --git a/lnst/RecipeCommon/Perf/Measurements/__init__.py
b/lnst/RecipeCommon/Perf/Measurements/__init__.py
index 0b98cd6..2e6da11 100644
--- a/lnst/RecipeCommon/Perf/Measurements/__init__.py
+++ b/lnst/RecipeCommon/Perf/Measurements/__init__.py
@@ -2,3 +2,4 @@ from lnst.RecipeCommon.Perf.Measurements.BaseFlowMeasurement import Flow
from lnst.RecipeCommon.Perf.Measurements.IperfFlowMeasurement import
IperfFlowMeasurement
from lnst.RecipeCommon.Perf.Measurements.TRexFlowMeasurement import TRexFlowMeasurement
from lnst.RecipeCommon.Perf.Measurements.StatCPUMeasurement import StatCPUMeasurement
+from lnst.RecipeCommon.Perf.Measurements.NeperFlowMeasurement import
NeperFlowMeasurement
diff --git a/lnst/Recipes/ENRT/MeasurementGenerators/NeperMeasurementGenerator.py
b/lnst/Recipes/ENRT/MeasurementGenerators/NeperMeasurementGenerator.py
new file mode 100644
index 0000000..435c5f9
--- /dev/null
+++ b/lnst/Recipes/ENRT/MeasurementGenerators/NeperMeasurementGenerator.py
@@ -0,0 +1,109 @@
+from lnst.Common.IpAddress import AF_INET, AF_INET6
+
+from lnst.Recipes.ENRT.MeasurementGenerators.BaseMeasurementGenerator import
BaseMeasurementGenerator
+from lnst.RecipeCommon.Perf.Measurements import NeperFlowMeasurement
+from lnst.RecipeCommon.Perf.Measurements import Flow as PerfFlow
+
+from lnst.Common.Parameters import (
+ Param,
+ IntParam,
+ ListParam,
+)
+
+class NeperMeasurementGenerator(BaseMeasurementGenerator):
+
+ perf_tests = Param(default=("tcp_rr", "tcp_crr",
"udp_rr"))
+ perf_duration = IntParam(default=60)
+ perf_iterations = IntParam(default=5)
+ perf_tool_cpu = IntParam(mandatory=False)
+ perf_parallel_streams = IntParam(default=1)
+ perf_msg_sizes = ListParam(default=[123])
+
+
+ net_perf_tool = Param(default=NeperFlowMeasurement)
+
+
+ def generate_perf_measurements_combinations(self, config):
+ combinations = super().generate_perf_measurements_combinations(config)
+ for flow_combination in self.generate_flow_combinations(config):
+ combinations.append([self.params.net_perf_tool(flow_combination)])
+ return combinations
+
+ def generate_flow_combinations(self, config):
+ """Base flow combination generator
+
+ The generator loops over all endpoint pairs to test performance between
+ (generated by the :any:`generate_perf_endpoints` method) then over all
+ the selected :any:`ip_versions` and uses the first IP address fitting
+ these criteria. Then the generator loops over the selected performance
+ tests as selected via :any:`perf_tests`, then message sizes from
+ :any:`msg_sizes`.
+
+ :return: list of Flow combinations to measure in parallel
+ :rtype: List[:any:`PerfFlow`]
+ """
+ for client_nic, server_nic in self.generate_perf_endpoints(config):
+ for ipv in self.params.ip_versions:
+ ip_filter = {}
+ if ipv == "ipv4":
+ ip_filter.update(family=AF_INET)
+ elif ipv == "ipv6":
+ ip_filter.update(family=AF_INET6)
+ ip_filter.update(is_link_local=False)
+
+ client_bind = client_nic.ips_filter(**ip_filter)[0]
+ server_bind = server_nic.ips_filter(**ip_filter)[0]
+
+ for perf_test in self.params.perf_tests:
+ for size in self.params.perf_msg_sizes:
+ yield [
+ self._create_perf_flow(
+ perf_test,
+ client_nic,
+ client_bind,
+ server_nic,
+ server_bind,
+ size,
+ )
+ ]
+
+ def generate_perf_endpoints(self, config):
+ """Generator for perf endpoints
+
+ To be overriden by a derived class.
+
+ :return: list of device pairs
+ :rtype: List[Tuple[:any:`Device`, :any:`Device`]]
+ """
+ return []
+
+ def _create_perf_flow(
+ self,
+ perf_test,
+ client_nic,
+ client_bind,
+ server_nic,
+ server_bind,
+ msg_size,
+ ) -> PerfFlow:
+ """
+ Wrapper to create a PerfFlow. Mixins that want to change this behavior (for
example, to reverse the direction)
+ can override this method as an alternative to overriding
:any:`generate_flow_combinations`
+ """
+ return PerfFlow(
+ type=perf_test,
+ generator=client_nic.netns,
+ generator_bind=client_bind,
+ generator_nic=client_nic,
+ receiver=server_nic.netns,
+ receiver_bind=server_bind,
+ receiver_nic=server_nic,
+ msg_size=msg_size,
+ duration=self.params.perf_duration,
+ parallel_streams=self.params.perf_parallel_streams,
+ cpupin=(
+ self.params.perf_tool_cpu
+ if "perf_tool_cpu" in self.params
+ else None
+ ),
+ )
\ No newline at end of file
diff --git a/lnst/Tests/Neper.py b/lnst/Tests/Neper.py
new file mode 100644
index 0000000..69d96ad
--- /dev/null
+++ b/lnst/Tests/Neper.py
@@ -0,0 +1,128 @@
+import csv
+import logging
+import os
+import pathlib
+import re
+import subprocess
+import time
+import tempfile
+from typing import List, Dict
+
+from lnst.Common.Parameters import HostnameOrIpParam, StrParam, IntParam, IpParam,
ChoiceParam
+from lnst.Tests.BaseTestModule import BaseTestModule, TestModuleError
+
+NEPER_OUT_RE = re.compile(r"^(?P<key>.*)=(?P<value>.*)$",
flags=re.M)
+NEPER_PATH = pathlib.Path('/root/neper')
+
+
+class NeperBase(BaseTestModule):
+ _supported_workloads = set(['tcp_rr', 'tcp_crr', 'udp_rr'])
+ workload = ChoiceParam(type=StrParam, choices=_supported_workloads,
+ mandatory=True)
+ port = IntParam()
+ control_port = IntParam()
+ cpu_bind = IntParam()
+ num_flows = IntParam()
+ num_threads = IntParam()
+ test_length = IntParam()
+ request_size = IntParam()
+ response_size = IntParam()
+ opts = StrParam()
+
+ def __init__(self, **kwargs):
+ self._samples_file = None
+ super(NeperBase, self).__init__(**kwargs)
+
+ def _parse_result(self, res: subprocess.CompletedProcess) -> Dict:
+ data = {}
+ for match in NEPER_OUT_RE.finditer(res.stdout):
+ if match is not None:
+ k, v = match.groups()
+ if v == '':
+ v = None
+ data[k] = v
+ return data
+
+ def run(self):
+ self._res_data = {}
+ if not NEPER_PATH.joinpath(self.params.workload).exists():
+ self._res_data['msg'] = f"neper workload
{self.params.workload}" \
+ f" is not installed on this machine!"
+ logging.error(self._res_data['msg'])
+ return False
+
+ with tempfile.NamedTemporaryFile('r', prefix='neper-samples-',
+ suffix='.csv', newline='') as
sf:
+
+ cmd = self._compose_cmd(sf.name)
+ logging.debug(f"compiled command: {cmd}")
+ logging.debug(f"running as {self._role}")
+
+ self._res_data["start_time"] = time.time()
+ res = subprocess.run(cmd,
+ stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+ universal_newlines=True, shell=True,
+ close_fds=True, cwd=NEPER_PATH)
+
+ if res.stderr != "":
+ self._res_data["msg"] = f"errors reported by
{self.params.workload}"
+ logging.error(self._res_data["msg"])
+ logging.error(self._res_data["stderr"])
+
+ if res.returncode > 0:
+ self._res_data["msg"] = "{} returncode = {}".format(
+ self._role, res.returncode)
+ logging.error(self._res_data["msg"])
+ return False
+
+ self._res_data["data"] = self._parse_result(res)
+ self._res_data["stderr"] = res.stderr
+ self._res_data["samples"] = [r for r in csv.DictReader(sf)]
+
+ return True
+
+ def _compose_cmd(self, samples_path:str) -> str:
+ cmd = [f"./{self.params.workload}",
+ f"--all-samples={samples_path}"]
+
+ if self._role == "client":
+ cmd.append("-c")
+ if "num_threads" in self.params:
+ cmd.append(f"-T {self.params.num_threads}")
+ if "num_flows" in self.params:
+ cmd.append(f"-F {self.params.num_flows}")
+ if "port" in self.params:
+ cmd.append(f"-P {self.params.port}")
+ if "control_port" in self.params:
+ cmd.append(f"-C {self.params.control_port}")
+ if "bind" in self.params:
+ cmd.append(f"-H {self.params.bind}")
+ if "server" in self.params:
+ cmd.append(f"-H {self.params.server}")
+ if "request_size" in self.params:
+ cmd.append(f"-Q {self.params.request_size}")
+ if "response_size" in self.params:
+ cmd.append(f"-R {self.params.response_size}")
+ if "test_length" in self.params:
+ cmd.append(f"-l {self.params.test_length}")
+ if "opts" in self.params:
+ cmd.append(self.params.opts)
+
+ if "cpu_bind" in self.params:
+ cmd.insert(0, f"taskset -c {self.params.cpu_bind}")
+
+ return " ".join(cmd)
+
+
+class NeperServer(NeperBase):
+ _role = "server"
+ bind = IpParam()
+
+
+class NeperClient(NeperBase):
+ _role = "client"
+ server = HostnameOrIpParam(mandatory=True)
+
+ def runtime_estimate(self):
+ _overhead = 5
+ return self.params.test_length + _overhead
--
2.26.2