[PATCH 2/8] Perf.Measurements.IperfFlowMeasurement: adapt to changes of NetworkFlowTest

[Jan Tluka]

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&gt;
---
 .../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