remove debug messages that caused quadratic runtime

johanneskoester · johanneskoester · commit 774d785d3e70 · 2024-09-20T10:25:49.000+02:00
diff --git a/snakemake/dag.py b/snakemake/dag.py
@@ -481,8 +481,6 @@ async def check_incomplete(self):
             incomplete = await self.incomplete_files()
             if incomplete:
                 if self.workflow.dag_settings.force_incomplete:
-                    logger.debug("Forcing incomplete files:")
-                    logger.debug("\t" + "\n\t".join(incomplete))
                     self.forcefiles.update(incomplete)
                 else:
                     raise IncompleteFilesException(incomplete)
@@ -767,11 +765,6 @@ def unshadow_output(self, job, only_log=False):
 
             if os.path.realpath(shadow_output) == os.path.realpath(real_output):
                 continue
-            logger.debug(
-                "Moving shadow output {} to destination {}".format(
-                    shadow_output, real_output
-                )
-            )
             shutil.move(shadow_output, real_output)
         shutil.rmtree(job.shadow_dir)
 
@@ -1851,7 +1844,6 @@ async def update_checkpoint_dependencies(self, jobs=None):
                 depending = list(self.depending[job])
                 all_depending.extend(depending)
         for j in all_depending:
-            logger.debug(f"Updating job {j}.")
             newjob = await j.updated()
             await self.replace_job(j, newjob, recursive=False)
             updated = True
@@ -2044,9 +2036,7 @@ async def replace_job(self, job, newjob, recursive=True):
 
         await self.update([newjob])
 
-        logger.debug(f"Replace {job} with {newjob}")
         for job_, files in depending:
-            logger.debug(f"updating depending job {job_}")
             self._dependencies[job_][newjob].update(files)
             self.depending[newjob][job_].update(files)
 
diff --git a/snakemake/scheduler.py b/snakemake/scheduler.py
@@ -263,20 +263,14 @@ def schedule(self):
                         job.reset_params_and_resources()
 
                     logger.debug(f"Resources before job selection: {self.resources}")
-                    logger.debug(
-                        f"Ready jobs ({len(needrun)})",
-                        # + "\n\t".join(map(str, needrun))
-                    )
+                    logger.debug(f"Ready jobs: {len(needrun)}")
 
                     if not self._last_job_selection_empty:
                         logger.info("Select jobs to execute...")
                     run = self.job_selector(needrun)
                     self._last_job_selection_empty = not run
 
-                    logger.debug(
-                        f"Selected jobs ({len(run)})"
-                        # + "\n\t".join(map(str, run))
-                    )
+                    logger.debug(f"Selected jobs: {len(run)}")
                     logger.debug(f"Resources after job selection: {self.resources}")
 
                 # update running jobs
@@ -335,6 +329,9 @@ def _finish_jobs(self):
         # clear the global tofinish such that parallel calls do not interfere
         async def postprocess():
             for job in self._tofinish:
+                # IMPORTANT: inside of this loop, there may be no calls that have
+                # a complexity of at least the number of jobs.
+                # Otherwise the function would be quadratic in the number of jobs.
                 if not self.workflow.dryrun:
                     try:
                         if self.workflow.exec_mode == ExecMode.DEFAULT:
@@ -373,9 +370,6 @@ async def postprocess():
                 if self.update_resources:
                     # normal jobs have len=1, group jobs have len>1
                     self.finished_jobs += len(job)
-                    logger.debug(
-                        f"jobs registered as running before removal {self.running}"
-                    )
                     self.running.remove(job)
                     self._free_resources(job)
 
@@ -429,7 +423,6 @@ def _free_resources(self, job):
     def _proceed(self, job):
         """Do stuff after job is finished."""
         with self._lock:
-            logger.debug(f"Completion of job {job.rules} reported to scheduler.")
             self._tofinish.append(job)
 
             if self.dryrun:
@@ -506,12 +499,7 @@ def job_selector_ilp(self, jobs):
         import pulp
         from pulp import lpSum
 
-        logger.debug("Selecting jobs to run using ILP solver.")
-
         if len(jobs) == 1:
-            logger.debug(
-                "Switching to greedy selector because only one job has to be scheduled."
-            )
             return self.job_selector_greedy(jobs)
 
         with self._lock:
@@ -626,7 +614,6 @@ async def get_temp_sizes_gb():
                 )
 
         status = self._solve_ilp(prob, time_limit=10)
-        logger.debug(f"Problem is {pulp.LpStatus[status]}")
         if pulp.LpStatus[status] != "Optimal":
             if pulp.LpStatus[status] == "Not Solved":
                 logger.warning(
@@ -635,7 +622,7 @@ async def get_temp_sizes_gb():
             elif pulp.LpStatus[status] == "Infeasible":
                 logger.warning("Failed to solve scheduling problem with ILP solver.")
 
-            logger.debug("Falling back to greedy solver.")
+            logger.warning("Falling back to greedy solver.")
             return self.job_selector_greedy(jobs)
 
         selected_jobs = set(
@@ -841,7 +828,4 @@ def get_free_jobs(self):
         for _ in range(index):
             self._jobs.popleft()
         n_free = max(self._limit.max_jobs - len(self._jobs), 0)
-        logger.debug(
-            f"Free jobs: {n_free}, jobs in timespan: {len(self._jobs)}, limit: {self._limit.max_jobs}, timespan: {self._limit.timespan}"
-        )
         return n_free
