Fix warnings flagged by SonarQube

Author: stijnh

kernel_tuner/core.py

@@ -224,7 +224,8 @@ def instantiate_observer(observer, args):
     if isinstance(observer, BenchmarkObserver):
         return observer
     elif callable(observer):
-        return instantiate_observer(observer(args), args) # Check again if BenchmarkObserver
+        # Check again if BenchmarkObserver
+        return instantiate_observer(observer(args), args)
     else:
         raise TypeError(f"Invalid observer: {observer!r} does not extend BenchmarkObserver")
 

kernel_tuner/runners/parallel.py

@@ -28,6 +28,7 @@
 
 @ray.remote(num_gpus=1)
 class WorkerActor:
+    """A Ray actor that owns a single GPU and benchmarks kernel configurations on it."""
     def __init__(
         self, kernel_source, kernel_options, device_options, tuning_options, iterations, observers
     ):
@@ -67,7 +68,7 @@ def get_environment(self):
         return env
 
     def run(self, params):
-        # TODO: logging.debug("sequential runner started for " + self.kernel_options.kernel_name)
+        # logging.debug("sequential runner started for " + self.kernel_options.kernel_name)
         result = None
 
         # attempt to warmup the GPU by running the first config in the parameter space and ignoring the result
@@ -86,17 +87,20 @@ def run(self, params):
         params["ray_actor_id"] = ray.get_runtime_context().get_actor_id()
         params["host_name"] = socket.gethostname()
 
-        # all visited configurations are added to results to provide a trace for optimization strategies
         return params
 
 
 class Worker:
+    """Local handle for a ``WorkerActor`` running in a remote Ray worker process."""
+
     def __init__(self, index, actor):
         self.index = index
         self.running_jobs = []
         self.maximum_running_jobs = 2
         self.is_running = True
         self.actor = actor
+
+        # Note: This will block until the environment is available locally
         self.env = ray.get(actor.get_environment.remote())
 
     def __repr__(self):
@@ -108,9 +112,14 @@ def __repr__(self):
             return f"{self.index}"
 
     def shutdown(self):
+        """Request the remote actor to exit and mark this handle as stopped."""
         if not self.is_running:
             return
 
+        # Wait until running jobs complete
+        if self.running_jobs:
+            ray.wait(self.running_jobs)
+
         self.is_running = False
 
         try:
@@ -119,11 +128,14 @@ def shutdown(self):
             logger.exception("failed to request actor shutdown: worker %s", self)
 
     def submit(self, config):
+        """Submit a kernel configuration for benchmarking."""
         job = self.actor.run.remote(config)
         self.running_jobs.append(job)
         return job
 
     def is_available(self):
+        """Return True if this worker can accept another job right now."""
+
         if not self.is_running:
             return False
 
@@ -139,26 +151,26 @@ def launch_workers(n, *args):
     workers = []
 
     try:
-        # Start all actors in parallel
+        # Start all actors in parallel since `WorkerActor.remote` does not block
         for _ in range(n):
             actors.append(WorkerActor.remote(*args))
 
-        # Create `Worker` objects. This blocks until each worker is ready
+        # Create local `Worker` objects. This blocks until each worker is ready
         for index, actor in enumerate(actors):
             worker = Worker(index, actor)
             workers.append(worker)
             logging.info("connected: worker %s", worker)
-
-        return workers
-    except:
-        # Attempt to shut down actors
+    except Exception:
+        # Attempt to shut down the running actors
         for actor in actors:
             try:
                 actor.shutdown.remote()
             except:
                 logger.exception("failed to request actor shutdown: %s", actor)
         raise
 
+    return workers
+
 
 class ParallelRunner(Runner):
     def __init__(
@@ -228,7 +240,7 @@ def shutdown(self):
         for worker in self.workers:
             try:
                 worker.shutdown()
-            except Exception as err:
+            except Exception:
                 logger.exception(f"error while shutting down worker {worker}")
 
     def available_parallelism(self):
@@ -350,8 +362,7 @@ def run(self, parameter_space, tuning_options) -> List[Optional[dict]]:
                 result = process_metrics(result, metrics)
             else:
                 logging.error(
-                    "kernel configuration {key} was skipped silently due to compile or runtime failure",
-                    key,
+                    f"kernel configuration {key} was skipped silently due to compile or runtime failure"
                 )
 
             # print configuration to the console

kernel_tuner/runners/runner.py

@@ -18,6 +18,7 @@ def add_strategy_time(self, seconds: float):
         self.accumulated_strategy_time += seconds
 
     def shutdown(self):
+        """ Signal to this runner that we are about to shut down. """
         pass
 
     def available_parallelism(self):

kernel_tuner/runners/sequential.py

@@ -122,7 +122,8 @@ def run(self, parameter_space, tuning_options):
             # all visited configurations are added to results to provide a trace for optimization strategies
             results.append(params)
 
-        num_valid_results = sum(bool(r) for r in results)  # Count the number of valid results
+        # Count the number of valid results
+        num_valid_results = sum(bool(r) for r in results)
 
         if num_valid_results > 0:
             strategy_time = self.accumulated_strategy_time

kernel_tuner/runners/simulation.py

@@ -135,7 +135,8 @@ def run(self, parameter_space, tuning_options):
             params_dict = dict(zip(tuning_options['tune_params'].keys(), element))
             check = util.check_restrictions(tuning_options.restrictions, params_dict, True)
             if not check:
-                result = util.copy_without_benchmark_timings(params_dict) # Set timings to zero
+                # Set timings to zero
+                result = util.copy_without_benchmark_timings(params_dict)
                 result[tuning_options.objective] = util.InvalidConfig()
                 results.append(result)
                 warn(f"Configuration {element} not in cache, does not pass restrictions. Will be treated as an InvalidConfig, but make sure you are evaluating the correct cache file.")

kernel_tuner/strategies/common.py

@@ -145,7 +145,6 @@ def _run_configs(self, xs, check_restrictions=True):
         for index, x in enumerate(xs):
             config, is_legal = self._normalize_and_validate_config(x, check_restrictions=check_restrictions)
             logging.debug("normalize config: %s -> %s (legal: %s)", str(x), str(config), is_legal)
-            key = ",".join([str(i) for i in config])
 
             # Not legal, just return `InvalidConfig`
             if not is_legal: