The NetworkFlowTest now holds a list of client/server proceses so
the IperfFlowMeasurement class needs to be updated accordingly.
This also adds one layer of parallel results in the results generated by
multiple client/server processes. However, at this point this is a
transparent change.
Signed-off-by: Jan Tluka <jtluka(a)redhat.com>
---
.../Perf/Measurements/IperfFlowMeasurement.py | 65 ++++++++++++-------
1 file changed, 43 insertions(+), 22 deletions(-)
diff --git a/lnst/RecipeCommon/Perf/Measurements/IperfFlowMeasurement.py
b/lnst/RecipeCommon/Perf/Measurements/IperfFlowMeasurement.py
index 10088f45..c1217130 100644
--- a/lnst/RecipeCommon/Perf/Measurements/IperfFlowMeasurement.py
+++ b/lnst/RecipeCommon/Perf/Measurements/IperfFlowMeasurement.py
@@ -60,11 +60,13 @@ class IperfFlowMeasurement(BaseFlowMeasurement):
result = None
for flow in test_flows:
- flow.server_job.start(bg=True)
+ for server_job in flow.server_jobs:
+ server_job.start(bg=True)
time.sleep(2)
for flow in test_flows:
- flow.client_job.start(bg=True)
+ for client_job in flow.client_jobs:
+ client_job.start(bg=True)
self._running_measurements = test_flows
@@ -72,13 +74,17 @@ class IperfFlowMeasurement(BaseFlowMeasurement):
test_flows = self._running_measurements
try:
for flow in test_flows:
- client_iperf = flow.client_job.what
- flow.client_job.wait(timeout=client_iperf.runtime_estimate())
- flow.server_job.wait(timeout=5)
+ for client_job in flow.client_jobs:
+ client_iperf = client_job.what
+ client_job.wait(timeout=client_iperf.runtime_estimate())
+ for server_job in flow.server_jobs:
+ server_job.wait(timeout=5)
finally:
for flow in test_flows:
- flow.server_job.kill()
- flow.client_job.kill()
+ for server_job in flow.server_jobs:
+ server_job.kill()
+ for client_job in flow.client_jobs:
+ client_job.kill()
self._running_measurements = []
self._finished_measurements = test_flows
@@ -91,15 +97,26 @@ class IperfFlowMeasurement(BaseFlowMeasurement):
flow_results = FlowMeasurementResults(
measurement=self,
flow=test_flow.flow)
- flow_results.generator_results = self._parse_job_streams(
- test_flow.client_job)
- flow_results.generator_cpu_stats = self._parse_job_cpu(
- test_flow.client_job)
- flow_results.receiver_results = self._parse_job_streams(
- test_flow.server_job)
- flow_results.receiver_cpu_stats = self._parse_job_cpu(
- test_flow.server_job)
+ generator_results = flow_results.generator_results = ParallelPerfResult()
+ for client_job in test_flow.client_jobs:
+ job_results = self._parse_job_streams(client_job)
+ generator_results.append(job_results)
+
+ generator_cpu_stats = flow_results.generator_cpu_stats =
ParallelPerfResult()
+ for client_job in test_flow.client_jobs:
+ job_results = self._parse_job_cpu(client_job)
+ generator_cpu_stats.append(job_results)
+
+ receiver_results = flow_results.receiver_results = ParallelPerfResult()
+ for server_job in test_flow.server_jobs:
+ job_result = self._parse_job_streams(server_job)
+ receiver_results.append(job_result)
+
+ receiver_cpu_stats = flow_results.receiver_cpu_stats = ParallelPerfResult()
+ for server_job in test_flow.server_jobs:
+ job_results = self._parse_job_cpu(server_job)
+ receiver_cpu_stats.append(job_results)
results.append(flow_results)
@@ -108,9 +125,9 @@ class IperfFlowMeasurement(BaseFlowMeasurement):
def _prepare_test_flows(self, flows):
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)
+ server_jobs = self._prepare_server(flow)
+ client_jobs = self._prepare_client(flow)
+ test_flow = NetworkFlowTest(flow, server_jobs, client_jobs)
test_flows.append(test_flow)
return test_flows
@@ -128,8 +145,10 @@ class IperfFlowMeasurement(BaseFlowMeasurement):
elif flow.cpupin is not None:
raise RecipeError("Negative perf cpupin value provided.")
- return host.prepare_job(IperfServer(**server_params),
- job_level=ResultLevel.NORMAL)
+ server_jobs = [host.prepare_job(IperfServer(**server_params),
+ job_level=ResultLevel.NORMAL)]
+
+ return server_jobs
def _prepare_client(self, flow):
host = flow.generator
@@ -161,8 +180,10 @@ class IperfFlowMeasurement(BaseFlowMeasurement):
if flow.msg_size:
client_params["blksize"] = flow.msg_size
- return host.prepare_job(IperfClient(**client_params),
- job_level=ResultLevel.NORMAL)
+ client_jobs = [host.prepare_job(IperfClient(**client_params),
+ job_level=ResultLevel.NORMAL)]
+
+ return client_jobs
def _parse_job_streams(self, job):
result = ParallelPerfResult()
--
2.26.2