[PATCH 0/8] RFC: Refactor IperfFlowMeasurement for parallel iperf testing

[Jan Tluka]

This patchset refactors IperfFlowMeasurement and
IperfFlowMeasurementGenerator classes to allow parallel iperf testing.

The current parallel implementation that can be achieved by specifying
the perf_parallel_streams recipe parameter works correctly however the
limitation is that there's only one iperf process that creates multiple
connections and that process (and all the connections) can be handled by
a single CPU at the same time. In our internal testing this proved to
report very variable CPU utilization numbers.

This patchset extends the IperfFlowMeasurementGenerator with additional
recipe parameters, the perf_parallel_processes and
perf_parallel_processes_cpus, to support paralell iperf testing.

The patch set includes also update of DevInterruptHWConfigMixin that is
required for this test scenario to provide reproducible results.

Jan Tluka (8):
  Perf.Measurements.BaseFlowMeasurement.NetworkFlowTest: change flow to
    contain a list of server/client jobs
  Perf.Measurements.IperfFlowMeasurement: adapt to changes of
    NetworkFlowTest
  TRexFlowMeasurement: adapt to changes of NetworkFlowTest
  Recipes.ENRT.MeasurementGenerators.IperfMeasurementGenerator: add
    perf_parallel_processes parameter
  Perf.Measurements.IperfFlowMeasurement: use parallel_perf_processes
    parameter
  Recipes.ENRT.MeasurementGenerators.IperfMeasurementGenerator: add
    parallel_perf_processes_cpus parameter
  Perf.Measurements.IperfFlowMeasurement: use parallel_processes_cpus
    parameter
  Recipes.ENRT.ConfigMixins.DevInterruptHWConfigMixin: change
    dev_intr_cpu to dev_intr_cpus

 .../Perf/Measurements/BaseFlowMeasurement.py  | 28 +++++--
 .../Perf/Measurements/IperfFlowMeasurement.py | 82 +++++++++++++------
 .../Perf/Measurements/TRexFlowMeasurement.py  | 31 ++++---
 .../ConfigMixins/DevInterruptHWConfigMixin.py | 38 +++++----
 .../IperfMeasurementGenerator.py              | 19 +++++
 5 files changed, 138 insertions(+), 60 deletions(-)

-- 
2.26.2

[PATCH 1/8] Perf.Measurements.BaseFlowMeasurement.NetworkFlowTest: change flow to contain a list of server/client jobs

[Jan Tluka]

To be able to run multiple processes within a NetworkFlowTest the object now
holds a list of server and client jobs instead of just one job for each.

Signed-off-by: Jan Tluka <jtluka(a)redhat.com&gt;
---
 .../Perf/Measurements/BaseFlowMeasurement.py       | 14 +++++++-------
 1 file changed, 7 insertions(+), 7 deletions(-)

diff --git a/lnst/RecipeCommon/Perf/Measurements/BaseFlowMeasurement.py
b/lnst/RecipeCommon/Perf/Measurements/BaseFlowMeasurement.py
index dcb0d2a7..2d1f911e 100644
--- a/lnst/RecipeCommon/Perf/Measurements/BaseFlowMeasurement.py
+++ b/lnst/RecipeCommon/Perf/Measurements/BaseFlowMeasurement.py
@@ -101,22 +101,22 @@ class Flow(object):
         return string
 
 class NetworkFlowTest(object):
-    def __init__(self, flow, server_job, client_job):
+    def __init__(self, flow, server_jobs, client_jobs):
         self._flow = flow
-        self._server_job = server_job
-        self._client_job = client_job
+        self._server_jobs = server_jobs
+        self._client_jobs = client_jobs
 
     @property
     def flow(self):
         return self._flow
 
     @property
-    def server_job(self):
-        return self._server_job
+    def server_jobs(self):
+        return self._server_jobs
 
     @property
-    def client_job(self):
-        return self._client_job
+    def client_jobs(self):
+        return self._client_jobs
 
 class FlowMeasurementResults(BaseMeasurementResults):
     def __init__(self, measurement, flow):
-- 
2.26.2

[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

[PATCH 3/8] TRexFlowMeasurement: adapt to changes of NetworkFlowTest

[Jan Tluka]

The NetworkFlowTest now holds a list of client/server proceses so
the TrexFlowMeasurement 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/TRexFlowMeasurement.py  | 31 ++++++++++++-------
 1 file changed, 20 insertions(+), 11 deletions(-)

diff --git a/lnst/RecipeCommon/Perf/Measurements/TRexFlowMeasurement.py
b/lnst/RecipeCommon/Perf/Measurements/TRexFlowMeasurement.py
index b7808426..4cc1f09b 100644
--- a/lnst/RecipeCommon/Perf/Measurements/TRexFlowMeasurement.py
+++ b/lnst/RecipeCommon/Perf/Measurements/TRexFlowMeasurement.py
@@ -59,13 +59,15 @@ class TRexFlowMeasurement(BaseFlowMeasurement):
 
         result = None
         for test in tests:
-            test.server_job.start(bg=True)
+            for server_job in test.server_jobs:
+                server_job.start(bg=True)
 
         #wait for Trex server to start
         time.sleep(15)
 
         for test in tests:
-            test.client_job.start(bg=True)
+            for client_job in test.client_jobs:
+                client_job.start(bg=True)
 
         self._running_measurements = tests
 
@@ -73,15 +75,19 @@ class TRexFlowMeasurement(BaseFlowMeasurement):
         tests = self._running_measurements
         try:
             for test in tests:
-                client_test = test.client_job.what
-                test.client_job.wait(timeout=client_test.runtime_estimate())
+                for client_job in test.client_jobs:
+                    client_test = client_job.what
+                    client_job.wait(timeout=client_test.runtime_estimate())
 
-                test.server_job.kill(signal.SIGINT)
-                test.server_job.wait(5)
+                for server_job in test.server_jobs:
+                    server_job.kill(signal.SIGINT)
+                    server_job.wait(5)
         finally:
             for test in tests:
-                test.server_job.kill()
-                test.client_job.kill()
+                for server_job in test.server_jobs:
+                    server_job.kill()
+                for client_job in test.client_jobs:
+                    client_job.kill()
 
         self._running_measurements = []
         self._finished_measurements = tests
@@ -107,7 +113,7 @@ class TRexFlowMeasurement(BaseFlowMeasurement):
                         duration=flows[0].duration,
                         msg_size=flows[0].msg_size))
 
-            test = NetworkFlowTest(flows, server_job, client_job)
+            test = NetworkFlowTest(flows, [server_job], [client_job])
             tests.append(test)
         return tests
 
@@ -117,8 +123,11 @@ class TRexFlowMeasurement(BaseFlowMeasurement):
         results = []
         for test in tests:
             for port, flow in enumerate(test.flow):
-                flow_results = self._parse_results_by_port(
-                        test.client_job, port, flow)
+                flow_results = ParallelPerfResult()
+                for client_job in test.client_jobs:
+                    job_results = self._parse_results_by_port(
+                            client_job, port, flow)
+                    flow_results.append(job_results)
                 results.append(flow_results)
 
         return results
-- 
2.26.2

[PATCH 4/8] Recipes.ENRT.MeasurementGenerators.IperfMeasurementGenerator: add perf_parallel_processes parameter

[Jan Tluka]

This extends the IperfMeasurementGenerator with perf_parallel_processes parameter.

This is to allow user to run multiple processes of iperf server/client as an
alternative to running multiple streams within one iperf client/server process
specified by perf_parallel_streams.

This also updates the Flow object with this configuration.

Signed-off-by: Jan Tluka <jtluka(a)redhat.com&gt;
---
 lnst/RecipeCommon/Perf/Measurements/BaseFlowMeasurement.py | 7 ++++++-
 .../MeasurementGenerators/IperfMeasurementGenerator.py     | 2 ++
 2 files changed, 8 insertions(+), 1 deletion(-)

diff --git a/lnst/RecipeCommon/Perf/Measurements/BaseFlowMeasurement.py
b/lnst/RecipeCommon/Perf/Measurements/BaseFlowMeasurement.py
index 2d1f911e..6973ae73 100644
--- a/lnst/RecipeCommon/Perf/Measurements/BaseFlowMeasurement.py
+++ b/lnst/RecipeCommon/Perf/Measurements/BaseFlowMeasurement.py
@@ -11,7 +11,7 @@ class Flow(object):
                  type,
                  generator, generator_bind, generator_nic,
                  receiver, receiver_bind, receiver_nic,
-                 msg_size, duration, parallel_streams, cpupin):
+                 msg_size, duration, parallel_streams, parallel_processes, cpupin):
         self._type = type
 
         self._generator = generator
@@ -24,6 +24,7 @@ class Flow(object):
         self._msg_size = msg_size
         self._duration = duration
         self._parallel_streams = parallel_streams
+        self._parallel_processes = parallel_processes
         self._cpupin = cpupin
 
     @property
@@ -66,6 +67,10 @@ class Flow(object):
     def parallel_streams(self):
         return self._parallel_streams
 
+    @property
+    def parallel_processes(self):
+        return self._parallel_processes
+
     @property
     def cpupin(self):
         return self._cpupin
diff --git a/lnst/Recipes/ENRT/MeasurementGenerators/IperfMeasurementGenerator.py
b/lnst/Recipes/ENRT/MeasurementGenerators/IperfMeasurementGenerator.py
index 4ccf9227..9d757ca5 100644
--- a/lnst/Recipes/ENRT/MeasurementGenerators/IperfMeasurementGenerator.py
+++ b/lnst/Recipes/ENRT/MeasurementGenerators/IperfMeasurementGenerator.py
@@ -56,6 +56,7 @@ class IperfMeasurementGenerator(BaseMeasurementGenerator):
     perf_duration = IntParam(default=60)
     perf_iterations = IntParam(default=5)
     perf_parallel_streams = IntParam(default=1)
+    perf_parallel_processes = IntParam(default=1)
     perf_msg_sizes = ListParam(default=[123])
 
     net_perf_tool = Param(default=IperfFlowMeasurement)
@@ -138,6 +139,7 @@ class IperfMeasurementGenerator(BaseMeasurementGenerator):
             msg_size=msg_size,
             duration=self.params.perf_duration,
             parallel_streams=self.params.perf_parallel_streams,
+            parallel_processes=self.params.perf_parallel_processes,
             cpupin=(
                 self.params.perf_tool_cpu
                 if "perf_tool_cpu" in self.params
-- 
2.26.2

[PATCH 5/8] Perf.Measurements.IperfFlowMeasurement: use parallel_perf_processes parameter

[Jan Tluka]

This implements parallel iperf processes measurement.

I extended the arguments of the _prepare_client() method with the server_jobs
list to be able to use the ports the servers listen on when creating the
client jobs.

Signed-off-by: Jan Tluka <jtluka(a)redhat.com&gt;
---
 .../Perf/Measurements/IperfFlowMeasurement.py | 21 +++++++++++++------
 1 file changed, 15 insertions(+), 6 deletions(-)

diff --git a/lnst/RecipeCommon/Perf/Measurements/IperfFlowMeasurement.py
b/lnst/RecipeCommon/Perf/Measurements/IperfFlowMeasurement.py
index c1217130..6b884975 100644
--- a/lnst/RecipeCommon/Perf/Measurements/IperfFlowMeasurement.py
+++ b/lnst/RecipeCommon/Perf/Measurements/IperfFlowMeasurement.py
@@ -126,7 +126,7 @@ class IperfFlowMeasurement(BaseFlowMeasurement):
         test_flows = []
         for flow in flows:
             server_jobs = self._prepare_server(flow)
-            client_jobs = self._prepare_client(flow)
+            client_jobs = self._prepare_client(flow, server_jobs)
             test_flow = NetworkFlowTest(flow, server_jobs, client_jobs)
             test_flows.append(test_flow)
         return test_flows
@@ -145,12 +145,17 @@ class IperfFlowMeasurement(BaseFlowMeasurement):
         elif flow.cpupin is not None:
             raise RecipeError("Negative perf cpupin value provided.")
 
-        server_jobs = [host.prepare_job(IperfServer(**server_params),
-                                        job_level=ResultLevel.NORMAL)]
+        server_jobs = []
+        offset=12000
+        for i in range(flow.parallel_processes):
+            server_params["port"] = offset+i
+            server_job = host.prepare_job(IperfServer(**server_params),
+                                           job_level=ResultLevel.NORMAL)
+            server_jobs.append(server_job)
 
         return server_jobs
 
-    def _prepare_client(self, flow):
+    def _prepare_client(self, flow, server_jobs):
         host = flow.generator
         client_params = dict(server = ipaddress(flow.receiver_bind),
                              duration = flow.duration)
@@ -180,8 +185,12 @@ class IperfFlowMeasurement(BaseFlowMeasurement):
         if flow.msg_size:
             client_params["blksize"] = flow.msg_size
 
-        client_jobs = [host.prepare_job(IperfClient(**client_params),
-                                        job_level=ResultLevel.NORMAL)]
+        client_jobs = []
+        for i in range(0, flow.parallel_processes):
+            client_params["port"] = server_jobs[i].what.params.port
+            client_job = host.prepare_job(IperfClient(**client_params),
+                                           job_level=ResultLevel.NORMAL)
+            client_jobs.append(client_job)
 
         return client_jobs
 
-- 
2.26.2

[PATCH 6/8] Recipes.ENRT.MeasurementGenerators.IperfMeasurementGenerator: add parallel_perf_processes_cpus parameter

[Jan Tluka]

This extends the IperfMeasurementGenerator with parallel_perf_processes_cpus
parameter. This parameter specifies a list of cpus where each cpu in this
list will be used to pin one process of the processes specified by
the perf_parallel_process parameter. If there are more processes than
the number of cpus in the parameter, the assignment will continue from
beginning of the list.

The original cpupin parameter has a different meaning. That is being able
to pin one or more iperf processes to a cpu (currently only one cpu is
supported, but this could be changed to a list of cpus in the future).
So, one iperf process can be eventually pinned to multiple cpus.
On the other hand, the perf_parallel_processes_cpus is a list from which
only ONE cpu is assigned to one iperf process.

Signed-off-by: Jan Tluka <jtluka(a)redhat.com&gt;
---
 .../Perf/Measurements/BaseFlowMeasurement.py    |  9 ++++++++-
 .../IperfMeasurementGenerator.py                | 17 +++++++++++++++++
 2 files changed, 25 insertions(+), 1 deletion(-)

diff --git a/lnst/RecipeCommon/Perf/Measurements/BaseFlowMeasurement.py
b/lnst/RecipeCommon/Perf/Measurements/BaseFlowMeasurement.py
index 6973ae73..6183d249 100644
--- a/lnst/RecipeCommon/Perf/Measurements/BaseFlowMeasurement.py
+++ b/lnst/RecipeCommon/Perf/Measurements/BaseFlowMeasurement.py
@@ -11,7 +11,9 @@ class Flow(object):
                  type,
                  generator, generator_bind, generator_nic,
                  receiver, receiver_bind, receiver_nic,
-                 msg_size, duration, parallel_streams, parallel_processes, cpupin):
+                 msg_size, duration, parallel_streams,
+                 parallel_processes, parallel_processes_cpus,
+                 cpupin):
         self._type = type
 
         self._generator = generator
@@ -25,6 +27,7 @@ class Flow(object):
         self._duration = duration
         self._parallel_streams = parallel_streams
         self._parallel_processes = parallel_processes
+        self._parallel_processes_cpus = parallel_processes_cpus
         self._cpupin = cpupin
 
     @property
@@ -71,6 +74,10 @@ class Flow(object):
     def parallel_processes(self):
         return self._parallel_processes
 
+    @property
+    def parallel_processes_cpus(self):
+        return self._parallel_processes_cpus
+
     @property
     def cpupin(self):
         return self._cpupin
diff --git a/lnst/Recipes/ENRT/MeasurementGenerators/IperfMeasurementGenerator.py
b/lnst/Recipes/ENRT/MeasurementGenerators/IperfMeasurementGenerator.py
index 9d757ca5..b2afec06 100644
--- a/lnst/Recipes/ENRT/MeasurementGenerators/IperfMeasurementGenerator.py
+++ b/lnst/Recipes/ENRT/MeasurementGenerators/IperfMeasurementGenerator.py
@@ -57,6 +57,7 @@ class IperfMeasurementGenerator(BaseMeasurementGenerator):
     perf_iterations = IntParam(default=5)
     perf_parallel_streams = IntParam(default=1)
     perf_parallel_processes = IntParam(default=1)
+    perf_parallel_processes_cpus = ListParam(mandatory=False)
     perf_msg_sizes = ListParam(default=[123])
 
     net_perf_tool = Param(default=IperfFlowMeasurement)
@@ -140,6 +141,22 @@ class IperfMeasurementGenerator(BaseMeasurementGenerator):
             duration=self.params.perf_duration,
             parallel_streams=self.params.perf_parallel_streams,
             parallel_processes=self.params.perf_parallel_processes,
+            # TODO: this is a new parameter,
+            #       the purpose is to be able to provide a list of cpus where
+            #       each cpu in this list will be used to pin one process from
+            #       parallel_processes, if there are more processes the
+            #       assignment will continue from beginning of the list
+            #       NOTE: the original cpupin should have a different meaning
+            #       that is being able to pin one (or more iperf processes) to
+            #       a set of cpus (currently only one value is supported, but
+            #       could be changed to a list in the future), whereas
+            #       parallel_processes_cpus is a list from which only ONE
+            #       cpu is assigned to an iperf process
+            parallel_processes_cpus=(
+                self.params.perf_parallel_processes_cpus
+                if "perf_parallel_processes_cpus" in self.params
+                else None
+                ),
             cpupin=(
                 self.params.perf_tool_cpu
                 if "perf_tool_cpu" in self.params
-- 
2.26.2

[PATCH 7/8] Perf.Measurements.IperfFlowMeasurement: use parallel_processes_cpus parameter

[Jan Tluka]

This implements pinning of parallel processes to cpus specified
by the parallel_processes_cpus.

Signed-off-by: Jan Tluka <jtluka(a)redhat.com&gt;
---
 lnst/RecipeCommon/Perf/Measurements/IperfFlowMeasurement.py | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/lnst/RecipeCommon/Perf/Measurements/IperfFlowMeasurement.py
b/lnst/RecipeCommon/Perf/Measurements/IperfFlowMeasurement.py
index 6b884975..618c0453 100644
--- a/lnst/RecipeCommon/Perf/Measurements/IperfFlowMeasurement.py
+++ b/lnst/RecipeCommon/Perf/Measurements/IperfFlowMeasurement.py
@@ -147,8 +147,11 @@ class IperfFlowMeasurement(BaseFlowMeasurement):
 
         server_jobs = []
         offset=12000
+        cpus=flow.parallel_processes_cpus
         for i in range(flow.parallel_processes):
             server_params["port"] = offset+i
+            if cpus:
+                server_params["cpu_bind"] =  cpus[i % len(cpus)]
             server_job = host.prepare_job(IperfServer(**server_params),
                                            job_level=ResultLevel.NORMAL)
             server_jobs.append(server_job)
@@ -186,8 +189,11 @@ class IperfFlowMeasurement(BaseFlowMeasurement):
             client_params["blksize"] = flow.msg_size
 
         client_jobs = []
+        cpus=flow.parallel_processes_cpus
         for i in range(0, flow.parallel_processes):
             client_params["port"] = server_jobs[i].what.params.port
+            if cpus:
+                client_params["cpu_bind"] = cpus[i % len(cpus)]
             client_job = host.prepare_job(IperfClient(**client_params),
                                            job_level=ResultLevel.NORMAL)
             client_jobs.append(client_job)
-- 
2.26.2

[PATCH 8/8] Recipes.ENRT.ConfigMixins.DevInterruptHWConfigMixin: change dev_intr_cpu to dev_intr_cpus

[Jan Tluka]

This is a change required by parallel iperf test. To achieve reproducible
measurements it's required to turn off irqbalance and spread the NIC IRQs
evenly on the available cpus.

The current recipe parameter dev_intr_cpu can specify only one cpu.

This patch renames the dev_intr_cpu to dev_intr_cpus and changes it to ListParam.
The value is a list of cpus that will be used for pinning the NIC IRQs.

For the original behaviour a user simply needs to change
	dev_intr_cpu=0
to
	dev_intr_cpus=[0]
and the recipe will run as before.

Signed-off-by: Jan Tluka <jtluka(a)redhat.com&gt;
---
 .../ConfigMixins/DevInterruptHWConfigMixin.py | 38 ++++++++++---------
 1 file changed, 20 insertions(+), 18 deletions(-)

diff --git a/lnst/Recipes/ENRT/ConfigMixins/DevInterruptHWConfigMixin.py
b/lnst/Recipes/ENRT/ConfigMixins/DevInterruptHWConfigMixin.py
index 1f8f5e8e..d9ba81fc 100644
--- a/lnst/Recipes/ENRT/ConfigMixins/DevInterruptHWConfigMixin.py
+++ b/lnst/Recipes/ENRT/ConfigMixins/DevInterruptHWConfigMixin.py
@@ -1,6 +1,6 @@
 import re
 
-from lnst.Common.Parameters import IntParam
+from lnst.Common.Parameters import ListParam
 from lnst.Controller.Recipe import RecipeError
 from lnst.Controller.RecipeResults import ResultLevel
 from lnst.Recipes.ENRT.ConfigMixins.BaseHWConfigMixin import BaseHWConfigMixin
@@ -15,11 +15,11 @@ class DevInterruptHWConfigMixin(BaseHWConfigMixin):
      .. note::
         Note that this Mixin also stops the irqbalance service.
 
-    :param dev_intr_cpu:
+    :param dev_intr_cpus:
         (optional test parameter) CPU id to which the device IRQs should be pinned
     """
 
-    dev_intr_cpu = IntParam(mandatory=False)
+    dev_intr_cpus = ListParam(mandatory=False)
 
     @property
     def dev_interrupt_hw_config_dev_list(self):
@@ -34,8 +34,8 @@ class DevInterruptHWConfigMixin(BaseHWConfigMixin):
 
         hw_config = config.hw_config
 
-        if "dev_intr_cpu" in self.params:
-            intr_cfg = hw_config["dev_intr_cpu_configuration"] = {}
+        if "dev_intr_cpus" in self.params:
+            intr_cfg = hw_config["dev_intr_cpus_configuration"] = {}
             intr_cfg["irq_devs"] = {}
             intr_cfg["irqbalance_hosts"] = []
 
@@ -49,11 +49,11 @@ class DevInterruptHWConfigMixin(BaseHWConfigMixin):
 
             for dev in self.dev_interrupt_hw_config_dev_list:
                 # TODO better service handling through HostAPI
-                self._pin_dev_interrupts(dev, self.params.dev_intr_cpu)
-                intr_cfg["irq_devs"][dev] = self.params.dev_intr_cpu
+                self._pin_dev_interrupts(dev, self.params.dev_intr_cpus)
+                intr_cfg["irq_devs"][dev] = self.params.dev_intr_cpus
 
     def hw_deconfig(self, config):
-        intr_config = config.hw_config.get("dev_intr_cpu_configuration", {})
+        intr_config = config.hw_config.get("dev_intr_cpus_configuration", {})
         for host in intr_config.get("irqbalance_hosts", []):
             host.run("service irqbalance start")
 
@@ -64,7 +64,7 @@ class DevInterruptHWConfigMixin(BaseHWConfigMixin):
 
         hw_config = config.hw_config
 
-        intr_cfg = hw_config.get("dev_intr_cpu_configuration", None)
+        intr_cfg = hw_config.get("dev_intr_cpus_configuration", None)
         if intr_cfg:
             desc += [
                 "{} irqbalance stopped".format(host.hostid)
@@ -80,19 +80,20 @@ class DevInterruptHWConfigMixin(BaseHWConfigMixin):
             desc.append("Device irq configuration skipped.")
         return desc
 
-    def _pin_dev_interrupts(self, dev, cpu):
+    def _pin_dev_interrupts(self, dev, cpus):
         netns = dev.netns
         cpu_info = netns.run("lscpu", job_level=ResultLevel.DEBUG).stdout
         regex = "CPU\(s\): *([0-9]*)"
         num_cpus = int(re.search(regex, cpu_info).groups()[0])
-        if cpu < 0 or cpu > num_cpus - 1:
-            raise RecipeError(
-                "Invalid CPU value given: %d. Accepted value %s."
-                % (
-                    cpu,
-                    "is: 0" if num_cpus == 1 else "are: 0..%d" %
(num_cpus - 1),
+        for cpu in cpus:
+            if cpu < 0 or cpu > num_cpus - 1:
+                raise RecipeError(
+                    "Invalid CPU value given: %d. Accepted value %s."
+                    % (
+                        cpu,
+                        "is: 0" if num_cpus == 1 else "are: 0..%d" %
(num_cpus - 1),
+                    )
                 )
-            )
 
         res = netns.run(
             "grep {} /proc/interrupts | cut -f1 -d: | sed 's/
//'".format(
@@ -110,9 +111,10 @@ class DevInterruptHWConfigMixin(BaseHWConfigMixin):
             )
             intrs = res.stdout
 
-        for intr in intrs.split("\n"):
+        for i, intr in enumerate(intrs.split("\n")):
             try:
                 int(intr)
+                cpu = cpus[i % len(cpus)]
                 netns.run(
                     "echo -n {} > /proc/irq/{}/smp_affinity_list".format(
                         cpu, intr.strip()
-- 
2.26.2

[Ondrej Lichtner]

On Fri, Feb 05, 2021 at 02:18:29PM +0100, Jan Tluka wrote:
>  This patchset refactors IperfFlowMeasurement and
>  IperfFlowMeasurementGenerator classes to allow parallel iperf testing.
>  
>  The current parallel implementation that can be achieved by specifying
>  the perf_parallel_streams recipe parameter works correctly however the
>  limitation is that there's only one iperf process that creates multiple
>  connections and that process (and all the connections) can be handled by
>  a single CPU at the same time. In our internal testing this proved to
>  report very variable CPU utilization numbers.
>  
>  This patchset extends the IperfFlowMeasurementGenerator with additional
>  recipe parameters, the perf_parallel_processes and
>  perf_parallel_processes_cpus, to support paralell iperf testing.
>  
>  The patch set includes also update of DevInterruptHWConfigMixin that is
>  required for this test scenario to provide reproducible results.
>  
>  Jan Tluka (8):
>    Perf.Measurements.BaseFlowMeasurement.NetworkFlowTest: change flow to
>      contain a list of server/client jobs
>    Perf.Measurements.IperfFlowMeasurement: adapt to changes of
>      NetworkFlowTest
>    TRexFlowMeasurement: adapt to changes of NetworkFlowTest
>    Recipes.ENRT.MeasurementGenerators.IperfMeasurementGenerator: add
>      perf_parallel_processes parameter
>    Perf.Measurements.IperfFlowMeasurement: use parallel_perf_processes
>      parameter
>    Recipes.ENRT.MeasurementGenerators.IperfMeasurementGenerator: add
>      parallel_perf_processes_cpus parameter
>    Perf.Measurements.IperfFlowMeasurement: use parallel_processes_cpus
>      parameter
>    Recipes.ENRT.ConfigMixins.DevInterruptHWConfigMixin: change
>      dev_intr_cpu to dev_intr_cpus
>  
>   .../Perf/Measurements/BaseFlowMeasurement.py  | 28 +++++--
>   .../Perf/Measurements/IperfFlowMeasurement.py | 82 +++++++++++++------
>   .../Perf/Measurements/TRexFlowMeasurement.py  | 31 ++++---
>   .../ConfigMixins/DevInterruptHWConfigMixin.py | 38 +++++----
>   .../IperfMeasurementGenerator.py              | 19 +++++
>   5 files changed, 138 insertions(+), 60 deletions(-)
>  
>  -- 
>  2.26.2
>  _______________________________________________
>  LNST-developers mailing list -- lnst-developers(a)lists.fedorahosted.org
>  To unsubscribe send an email to lnst-developers-leave(a)lists.fedorahosted.org
>  Fedora Code of Conduct: https://docs.fedoraproject.org/en-US/project/code-of-conduct/
>  List Guidelines: https://fedoraproject.org/wiki/Mailing_list_guidelines
>  List Archives:
> https://lists.fedorahosted.org/archives/list/[e-mail]...
we've discussed this over a video conference meeting, we agreed on
investigating some additional ideas, only including the main points here
instead of the full discussion as the email would be too long...

* instead of creating multiple server/client jobs per NetworkFlowTest,
  generate multiple parallel Flows to test (each translating into a
  NetworkFlowTest with single server+client).
* differentiate the Parallel flows by adding a source/destination port
  (completing the traditional 5-tuple that identifies a flow)
* extend the recipe parameter for cpu pinning to allow for a list of cpu
  cores and potentially add a parameter that defines the policy on how
  these cpus are used during generation of Flow combinations - if a
  single cpu is specified and a single Flow is generated, functionality
  stays as it is now. If multiple cpus are provided and multiple flows
  are generated we can for example "round-robin" assign cpus to each
  flow
* extend the "cpupin" property of a Flow to accept a list of integers
  (for multiple cpus that can be used for the specific flow), this would
  add the possibility of an additional policy for the previous point -
  run all flows on all provided cpus

-Ondrej