Merge pull request #255 from IntelPython/fix/multiprocessing

Fix multiprocessing

Author: ZzEeKkAa

.github/workflows/build_and_run.yml

@@ -71,7 +71,7 @@ jobs:
 
           dpbench \
           -i python,numpy,dpnp,sycl,numba_n,numba_np,numba_npr,numba_dpex_k,numba_dpex_n,numba_dpex_p \
-          run -r2 --no-print-results
+          run --no-validate -r2 --no-print-results
 
       - name: Generate report
         shell: bash -l {0}

.pre-commit-config.yaml

@@ -50,7 +50,7 @@ repos:
     hooks:
     -   id: pydocstyle
         # TODO: add packages one by one to enforce pydocstyle eventually
-        files: (^dpbench/config/|^scripts/|^dpbench/console/)
+        files: (^dpbench/config/|^scripts/|^dpbench/console/|^dpbench/infrastructure/benchmark_runner.py)
         args: ["--convention=google"]
         # D417 does not work properly:
         # https://github.com/PyCQA/pydocstyle/issues/459

dpbench/console/_namespace.py

@@ -12,7 +12,7 @@ class Namespace(argparse.Namespace):
     """Namespace class for parsed arguments."""
 
     benchmarks: set[str]
-    implementations: set[str]
+    implementations: list[str]
     all_implementations: bool
     preset: str
     sycl_device: str
@@ -31,6 +31,7 @@ class Namespace(argparse.Namespace):
     program: str
     color: str
     comparisons: list[str]
+    skip_expected_failures: bool
 
 
 class CommaSeparateStringAction(argparse.Action):

dpbench/console/entry.py

@@ -7,7 +7,11 @@
 import argparse
 from importlib.metadata import version
 
-from ._namespace import CommaSeparateStringAction, Namespace
+from ._namespace import (
+    CommaSeparateStringAction,
+    CommaSeparateStringListAction,
+    Namespace,
+)
 from .config import add_config_arguments, execute_config
 from .report import add_report_arguments, execute_report
 from .run import add_run_arguments, execute_run
@@ -31,7 +35,7 @@ def parse_args() -> Namespace:
         "-i",
         "--implementations",
         type=str,
-        action=CommaSeparateStringAction,
+        action=CommaSeparateStringListAction,
         nargs="?",
         default={"python", "numpy"},
         help="Comma separated list of implementations. Use "
@@ -100,7 +104,8 @@ def main():
     """Main function to run on dpbench console tool."""
     args = parse_args()
 
-    if args.program in {"run", "report"}:
+    conn = None
+    if args.program == "report" or args.program == "run" and args.save:
         import dpbench.infrastructure as dpbi
         from dpbench.infrastructure.reporter import update_run_id
 

dpbench/console/run.py

@@ -5,9 +5,14 @@
 """Run subcommand package."""
 
 import argparse
+import logging
 
 import sqlalchemy
 
+import dpbench.config as cfg
+import dpbench.infrastructure as dpbi
+from dpbench.infrastructure.benchmark_runner import BenchmarkRunner, RunConfig
+
 from ._namespace import Namespace
 
 
@@ -94,6 +99,23 @@ def add_run_arguments(parser: argparse.ArgumentParser):
         default=None,
         help="Sycl device to overwrite for framework configurations.",
     )
+    parser.add_argument(
+        "--skip-expected-failures",
+        action=argparse.BooleanOptionalAction,
+        default=True,
+        help="Either to save execution into database.",
+    )
+
+
+def _find_framework_config(implementation: str) -> cfg.Framework:
+    framework = None
+
+    for f in cfg.GLOBAL.frameworks:
+        for impl in f.postfixes:
+            if impl.postfix == implementation:
+                framework = f
+
+    return framework
 
 
 def execute_run(args: Namespace, conn: sqlalchemy.Engine):
@@ -105,33 +127,63 @@ def execute_run(args: Namespace, conn: sqlalchemy.Engine):
         args: object with all input arguments.
         conn: database connection.
     """
-    import dpbench.config as cfg
-    import dpbench.infrastructure as dpbi
-    from dpbench.infrastructure.runner import run_benchmarks
-
     cfg.GLOBAL = cfg.read_configs(
         benchmarks=args.benchmarks,
-        implementations=args.implementations,
+        implementations=set(args.implementations),
         no_dpbench=not args.dpbench,
         with_npbench=args.npbench,
         with_polybench=args.polybench,
     )
 
+    if args.all_implementations:
+        args.implementations = {
+            impl.postfix for impl in cfg.GLOBAL.implementations
+        }
+
     if args.sycl_device:
         for framework in cfg.GLOBAL.frameworks:
             framework.sycl_device = args.sycl_device
 
-    if args.run_id is None:
+    if args.save and args.run_id is None:
         args.run_id = dpbi.create_run(conn)
 
-    run_benchmarks(
-        conn=conn,
-        preset=args.preset,
-        repeat=args.repeat,
-        validate=args.validate,
-        timeout=args.timeout,
-        precision=args.precision,
-        print_results=args.print_results,
-        run_id=args.run_id,
-        implementations=list(args.implementations),
-    )
+    runner = BenchmarkRunner()
+
+    for benchmark in cfg.GLOBAL.benchmarks:
+        print("")
+        print(
+            f"================ Benchmark {benchmark.name} ({benchmark.module_name}) ========================"
+        )
+        print("")
+
+        for implementation in args.implementations:
+            framework = _find_framework_config(implementation)
+
+            if not framework:
+                logging.error(
+                    f"Could not find framework for {implementation} implementation"
+                )
+                continue
+
+            logging.info(
+                f"Running {benchmark.module_name} ({implementation}) on {framework.simple_name}"
+            )
+
+            runner.run_benchmark_and_save(
+                RunConfig(
+                    conn=conn,
+                    benchmark=benchmark,
+                    framework=framework,
+                    implementation=implementation,
+                    preset=args.preset,
+                    repeat=args.repeat,
+                    validate=args.validate,
+                    timeout=args.timeout,
+                    precision=args.precision,
+                    print_results=args.print_results,
+                    run_id=args.run_id,
+                    skip_expected_failures=args.skip_expected_failures,
+                )
+            )
+
+    runner.close_connections()

dpbench/infrastructure/benchmark.py

@@ -366,12 +366,10 @@ def _set_reference_implementation(self) -> BenchmarkImplFn:
         reference_implementations = [
             impl
             for impl in self.impl_fnlist
-            if self.info.reference_implementation_postfix in impl.name
+            if self.info.reference_implementation_postfix
+            and self.info.reference_implementation_postfix in impl.name
         ]
 
-        print(self.impl_fnlist)
-        print(reference_implementations)
-
         if len(reference_implementations) == 0:
             raise RuntimeError("No reference implementation")
 
@@ -714,67 +712,3 @@ def get_data_init(
                         )
                     }
                 )
-
-    def run(
-        self,
-        implementation_postfix: str = None,
-        preset: str = "S",
-        repeat: int = 10,
-        validate: bool = True,
-        timeout: float = 200.0,
-        precision: str = None,
-        conn: sqlite3.Connection = None,
-        run_id: int = None,
-    ) -> list[BenchmarkResults]:
-        results: list[BenchmarkResults] = []
-
-        implementation_postfixes = []
-
-        if implementation_postfix:
-            implementation_postfixes.append(implementation_postfix)
-        else:
-            for impl in self.impl_fnlist:
-                impl_postfix = impl.name[
-                    (len(self.bname) - len(impl.name) + 1) :  # noqa: E203
-                ]
-
-                implementation_postfixes.append(impl_postfix)
-
-        # TODO: do we call ref benchmark function twice?
-        for implementation_postfix in implementation_postfixes:
-            # copy_output is true only if validation is needed.
-            runner = BenchmarkRunner(
-                bench=self,
-                impl_postfix=implementation_postfix,
-                preset=preset,
-                repeat=repeat,
-                timeout=timeout,
-                precision=precision,
-                copy_output=validate,
-            )
-            result = runner.results
-            if validate and result.error_state == ErrorCodes.SUCCESS:
-                if self._validate_results(
-                    preset, runner.fmwrk, runner.output, precision
-                ):
-                    result.validation_state = ValidationStatusCodes.SUCCESS
-                else:
-                    result.validation_state = ValidationStatusCodes.FAILURE
-                    result.error_state = ErrorCodes.FAILED_VALIDATION
-                    result.error_msg = "Validation failed"
-            if conn:
-                store_results(
-                    conn,
-                    result.Result(
-                        run_id=run_id,
-                        benchmark_name=self.bname,
-                        framework_version=runner.fmwrk.fname
-                        + " "
-                        + runner.fmwrk.version()
-                        if runner.fmwrk
-                        else "n/a",
-                    ),
-                )
-            results.append((result, runner.fmwrk))
-
-        return results