Support create_graph

Author: george0st

README.md

@@ -54,11 +54,24 @@ generator = ParallelExecutor(prf_GIL_impact,
                              detail_output=True,
                              output_file="prf_gil_impact_01.txt")
 
+# Run setup, with test execution 20 seconds and zero delay before start 
+# (without waiting to other executors)
 setup=RunSetup(duration_second=20,start_delay=0)
 
-generator.run_bulk_executor(bulk_list=[[1, 1]],
-                            executor_list=[[16, 1, '1x thread'], [8, 2, '2x threads'],[4, 4,'4x threads']],
+# Run performance test with: 
+#  data bulk_list with two data sets 
+#    - first has 10 rows and 5 columns as [10, 5]
+#    - second has 1000 rows and 50 columns as [1000, 50]
+#  executor_list with six executor sets
+#    - first line has three executors with 2, 4 and 8 processes each with 2 threads 
+#    - second line has three executors with 2, 4 and 8 processes each with 4 threads
+generator.run_bulk_executor(bulk_list=[[10, 5], [1000, 50]],
+                            executor_list=[[2, 2, '2x thread'], [4, 2, '2x thread'],[8, 2,'2x thread'],
+                                           [2, 4, '4x thread'], [4, 4, '4x thread'],[8, 4,'4x thread']],
                             run_setup=setup)
+
+# We made 12 performance tests (two bulk_list x six executor_list) and write 
+# outputs to the file 'prf_gil_impact_01.txt'
 ```
 
 ## Outputs in text file 

cover.bat

@@ -1,7 +1,7 @@
-rem https://coverage.readthedocs.io/en/7.3.2/
+rem Setting based on 'https://coverage.readthedocs.io/en/7.3.2/'
 
 coverage erase
-coverage run -m unittest
+coverage run -m unittest discover
 coverage combine
 coverage report -m
 coverage-badge -f -o coverage.svg

publish.bat

@@ -1,5 +1,5 @@
-#pip install --upgrade setuptools wheel
-#pip install --upgrade twine
+rem pip install --upgrade setuptools wheel
+rem pip install --upgrade twine
 
 rmdir /S /Q dist
 rmdir /S /Q build

qgate_perf/parallel_executor.py

@@ -18,6 +18,9 @@
 from platform import python_version
 from packaging import version
 
+from qgate_graph.graph_performance import GraphPerformance
+from qgate_graph.graph_executor import GraphExecutor
+
 from contextlib import suppress
 
 
@@ -54,9 +57,9 @@ def __init__(self,
         self._output_file = output_file
         self._init_call = init_call
 
-        # technical point, how to close Process
+        # Technical point, how to close Process
         #   in python >= 3.7 Close() as soft closing
-        #   in python < 3.7 Terminate() as hard closing
+        #   in python < 3.7 Terminate() as hard closing (the Close method does not exist in python lower versions)
         self._process_close=True if version.parse(python_version()) >= version.parse("3.7") else False
 
     def _coreThreadClassPool(self, threads, return_key, return_dict, run_setup: RunSetup):
@@ -244,8 +247,6 @@ def _executeCore(self, run_setup: RunSetup, return_dict, processes=2, threads=2)
             for process_key in range(processes):
                 p = Process(target=self._coreThreadClassPool,
                             args=(threads, process_key, return_dict, run_setup))
-                # oldversion    p = Process(target=self._coreThreadClassPool,
-                #                p = Process(target=self._coreThreadClassPool,
                 # p = Process(target=self._coreThreadClass, args=(threads, process_key, return_dict, run_setup))
                 # p = Process(target=ParallelExecutor._coreThread, args=(self.func, threads, process_key, return_dict, run_setup))
                 # p = Process(target=ParallelExecutor._coreThreadPool, args=(self.func, threads, process_key, return_dict, run_setup))
@@ -271,7 +272,7 @@ def run_bulk_executor(self,
         """ Run cykle of bulks in cycle of sequences for function execution
 
         :param bulk_list:           list of bulks for execution in format [[rows, columns], ...]
-        :param executor_list:       list of executors for execution in format [[processes, threads], ...]
+        :param executor_list:       list of executors for execution in format [[processes, threads, 'label'], ...]
         :param run_setup:           setup of execution
         :param sleep_between_bulks: sleep between bulks
         :return:                    return state, True - all executions was without exceptions,
@@ -446,3 +447,19 @@ def test_run(self, run_setup: RunSetup=None, print_output=False):
             if ret.exception is not None:
                 return False
         return True
+
+    def create_graph(self, output_graph_dir="output", picture_dpi=100):
+        """
+        Generate graph based on output from performance tests
+
+        :param output_graph_dir:    directory for graph outputs
+        :param picture_dpi:         quality of picture (default is 100 DPI)
+        """
+
+        # TODO: add datetime to the path
+        #datetime.datetime.now().isoformat("")
+        graph = GraphPerformance(picture_dpi)
+        graph.generate_from_file(self._output_file,os.path.join(output_graph_dir,"graph-perf"))
+
+        graph = GraphExecutor(picture_dpi)
+        graph.generate_from_file(self._output_file,os.path.join(output_graph_dir,"graph-exec"))

requirements.txt

@@ -1,4 +1,6 @@
 numpy<=1.24.3
 psutil~=5.9
 packaging>=21.3
+qgate_graph>=1.3.6
+
 

tests/test_graph.py

@@ -0,0 +1,63 @@
+import os
+import unittest
+import logging
+from qgate_perf.parallel_executor import ParallelExecutor
+from qgate_perf.parallel_probe import ParallelProbe
+from qgate_perf.run_setup import RunSetup
+from qgate_perf.executor_helper import ExecutorHelper
+from qgate_perf.run_return import RunReturn
+import time
+from os import path
+import shutil
+
+def prf_test(run_return: RunReturn, run_setup: RunSetup):
+    """ Function for performance testing"""
+
+    # init (contain executor synchonization, if needed)
+    probe = ParallelProbe(run_setup)
+
+    if run_setup.is_init:
+        print(f"!!! INIT CALL !!!   {run_setup.bulk_row} x {run_setup.bulk_col} [{run_return.probe}]")
+
+    while (True):
+
+        # START - performance measure for specific part of code
+        probe.start()
+
+        for r in range(run_setup.bulk_row * run_setup.bulk_col):
+            time.sleep(0.001)
+
+        # STOP - performance measure specific part of code
+        if probe.stop():
+            break
+
+    if run_setup.param("generate_error"):
+        raise Exception('Simulated error')
+
+    # return outputs
+    run_return.probe=probe
+
+
+class TestCaseGraph(unittest.TestCase):
+    OUTPUT_ADR = "../output/test_graph/"
+
+    @classmethod
+    def setUpClass(cls):
+        shutil.rmtree(TestCaseGraph.OUTPUT_ADR,True)
+
+    @classmethod
+    def tearDownClass(cls):
+        pass
+
+    def test_graph(self):
+        generator = ParallelExecutor(prf_test,
+                                     label="test",
+                                     detail_output=True,
+                                     output_file=path.join(self.OUTPUT_ADR, "perf_test.txt"))
+
+        setup=RunSetup(duration_second=4, start_delay=2)
+        self.assertTrue(generator.run_bulk_executor([[10,10], [100,10]],
+                                                    [[1,2,'Austria perf'], [2,2,'Austria perf'], [4,2,'Austria perf']],
+#                                                     [1,4,'Germany perf'],[2,4,'Germany perf'],[4,4,'Germany perf']],
+                                                    setup))
+        generator.create_graph(self.OUTPUT_ADR)

tests/test_perf.py

@@ -47,10 +47,10 @@ def prf_GIL_impact(run_return: RunReturn, run_setup: RunSetup):
 
 class TestCasePerf(unittest.TestCase):
 
-    OUTPUT_PERF_ADR = "../output/test_perf/"
+    OUTPUT_ADR = "../output/test_perf/"
     @classmethod
     def setUpClass(cls):
-        shutil.rmtree(TestCasePerf.OUTPUT_PERF_ADR,True)
+        shutil.rmtree(TestCasePerf.OUTPUT_ADR, True)
 
     @classmethod
     def tearDownClass(cls):
@@ -60,15 +60,15 @@ def test_one_run(self):
         generator = ParallelExecutor(prf_GIL_impact,
                                      label="GIL_impact",
                                      detail_output=True,
-                                     output_file=path.join(self.OUTPUT_PERF_ADR,"perf_gil_impact_test.txt"))
+                                     output_file=path.join(self.OUTPUT_ADR, "perf_gil_impact_test.txt"))
 
         self.assertTrue(generator.one_run())
 
     def test_one_run_param(self):
         generator = ParallelExecutor(prf_GIL_impact,
                                      label="GIL_impact",
                                      detail_output=True,
-                                     output_file=path.join(self.OUTPUT_PERF_ADR,"perf_gil_impact_test.txt"))
+                                     output_file=path.join(self.OUTPUT_ADR, "perf_gil_impact_test.txt"))
 
         setting={"aa":10,
                "name": "Adam"}
@@ -79,7 +79,7 @@ def test_init_run(self):
         generator = ParallelExecutor(prf_GIL_impact,
                                      label="GIL_impact",
                                      detail_output=True,
-                                     output_file=path.join(self.OUTPUT_PERF_ADR,"perf_gil_impact_test.txt"))
+                                     output_file=path.join(self.OUTPUT_ADR, "perf_gil_impact_test.txt"))
 
         setting={"aa":10,
                "name": "Adam"}
@@ -91,7 +91,7 @@ def test_testrun_exception(self):
         generator = ParallelExecutor(prf_GIL_impact,
                                      label="GIL_impact",
                                      detail_output=True,
-                                     output_file=path.join(self.OUTPUT_PERF_ADR,"perf_gil_impact_test.txt"))
+                                     output_file=path.join(self.OUTPUT_ADR, "perf_gil_impact_test.txt"))
 
         setting={"generate_error": "yes"}
 
@@ -102,15 +102,15 @@ def test_testrun(self):
         generator = ParallelExecutor(prf_GIL_impact,
                                      label="GIL_impact",
                                      detail_output=True,
-                                     output_file=path.join(self.OUTPUT_PERF_ADR,"perf_gil_impact_test.txt"))
+                                     output_file=path.join(self.OUTPUT_ADR, "perf_gil_impact_test.txt"))
 
         self.assertTrue(generator.test_run(print_output=True))
 
     def test_testrun_setup(self):
         generator = ParallelExecutor(prf_GIL_impact,
                                      label="GIL_impact",
                                      detail_output=True,
-                                     output_file=path.join(self.OUTPUT_PERF_ADR,"perf_gil_impact_test.txt"))
+                                     output_file=path.join(self.OUTPUT_ADR, "perf_gil_impact_test.txt"))
 
         setting = {"aa": 10,
                    "name": "Adam"}
@@ -122,7 +122,7 @@ def test_run(self):
         generator = ParallelExecutor(prf_GIL_impact,
                                      label="GIL_impact",
                                      detail_output=True,
-                                     output_file=path.join(self.OUTPUT_PERF_ADR,"perf_gil_impact_test.txt"))
+                                     output_file=path.join(self.OUTPUT_ADR, "perf_gil_impact_test.txt"))
 
         setup=RunSetup(duration_second=4, start_delay=2)
         self.assertTrue(generator.run(2, 2, setup))
@@ -131,7 +131,7 @@ def test_run_exeption(self):
         generator = ParallelExecutor(prf_GIL_impact,
                                      label="GIL_impact",
                                      detail_output=True,
-                                     output_file=path.join(self.OUTPUT_PERF_ADR,"perf_gil_impact_test.txt"))
+                                     output_file=path.join(self.OUTPUT_ADR, "perf_gil_impact_test.txt"))
 
         setting = {"generate_error": "yes"}
 
@@ -143,7 +143,7 @@ def test_run_executor(self):
         generator = ParallelExecutor(prf_GIL_impact,
                                      label="GIL_impact",
                                      detail_output=True,
-                                     output_file=path.join(self.OUTPUT_PERF_ADR,"perf_gil_impact_test.txt"))
+                                     output_file=path.join(self.OUTPUT_ADR, "perf_gil_impact_test.txt"))
 
         setup=RunSetup(duration_second=4, start_delay=2)
         self.assertTrue(generator.run_executor([[1,1], [2,2]], setup))
@@ -152,7 +152,7 @@ def test_run_executor_exception(self):
         generator = ParallelExecutor(prf_GIL_impact,
                                      label="GIL_impact",
                                      detail_output=True,
-                                     output_file=path.join(self.OUTPUT_PERF_ADR,"perf_gil_impact_test.txt"))
+                                     output_file=path.join(self.OUTPUT_ADR, "perf_gil_impact_test.txt"))
 
         setting = {"generate_error": "yes"}
 
@@ -163,7 +163,7 @@ def test_run_bulk_executor(self):
         generator = ParallelExecutor(prf_GIL_impact,
                                      label="GIL_impact",
                                      detail_output=True,
-                                     output_file=path.join(self.OUTPUT_PERF_ADR,"perf_gil_impact_test.txt"))
+                                     output_file=path.join(self.OUTPUT_ADR, "perf_gil_impact_test.txt"))
 
         setup=RunSetup(duration_second=1, start_delay=0)
         self.assertTrue(generator.run_bulk_executor(bulk_list=[[1,2], [1,10]],
@@ -174,7 +174,7 @@ def test_run_bulk_executor_exception(self):
         generator = ParallelExecutor(prf_GIL_impact,
                                      label="GIL_impact",
                                      detail_output=True,
-                                     output_file=path.join(self.OUTPUT_PERF_ADR,"perf_gil_impact_test.txt"))
+                                     output_file=path.join(self.OUTPUT_ADR, "perf_gil_impact_test.txt"))
 
         setting={"generate_error": "yes"}
 
@@ -187,7 +187,7 @@ def test_run_bulk_executor_helpers(self):
         generator = ParallelExecutor(prf_GIL_impact,
                                      label="GIL_impact",
                                      detail_output=True,
-                                     output_file=path.join(self.OUTPUT_PERF_ADR,"perf_gil_impact_test.txt"))
+                                     output_file=path.join(self.OUTPUT_ADR, "perf_gil_impact_test.txt"))
 
         setup=RunSetup(duration_second=0, start_delay=0)
         self.assertTrue(generator.run_bulk_executor(bulk_list= BundleHelper.ROW_1_COL_10_100,
@@ -198,7 +198,7 @@ def test_run_bulk_executor_grow(self):
         generator = ParallelExecutor(prf_GIL_impact,
                                      label="GIL_impact",
                                      detail_output=True,
-                                     output_file=path.join(self.OUTPUT_PERF_ADR,"perf_gil_impact_test.txt"))
+                                     output_file=path.join(self.OUTPUT_ADR, "perf_gil_impact_test.txt"))
 
         setup=RunSetup(duration_second=1, start_delay=0)
         self.assertTrue(generator.run_bulk_executor(bulk_list=[[1,2]],
@@ -213,7 +213,7 @@ def test_run_bulk_executor_initcall(self):
         generator = ParallelExecutor(prf_GIL_impact,
                                      label="GIL_impact",
                                      detail_output=True,
-                                     output_file=path.join(self.OUTPUT_PERF_ADR,"perf_gil_impact_test.txt"),
+                                     output_file=path.join(self.OUTPUT_ADR, "perf_gil_impact_test.txt"),
                                      init_call=InitCallSetting.all())
 
         setup=RunSetup(duration_second=1, start_delay=0)