Task #10118: Enforce OMP_NUM_THREADS for all recipes

CEP/Pipeline/framework/lofarpipe/support/remotecommand.py

@@ -276,19 +276,24 @@ class ComputeJob(object):
                 self.results['returncode'] = 1
                 error.set()
                 return 1
-            cmdarray = run_remote_command(
-                config,
-                logger,
-                self.host,
-                self.command,
-                {
+
+            environment = {
                     "PATH": os.environ.get('PATH'),
                     "PYTHONPATH": os.environ.get('PYTHONPATH'),
                     "LD_LIBRARY_PATH": os.environ.get('LD_LIBRARY_PATH'),
                     "LOFARROOT" : os.environ.get('LOFARROOT'),
                     "LOFARENV" : os.environ.get('LOFARENV',''),
                     "QUEUE_PREFIX" : os.environ.get('QUEUE_PREFIX','')
-                },
+                }
+            if "cores" in self.resources:
+                environment["OMP_NUM_THREADS"] = str(self.resources["cores"])
+
+            cmdarray = run_remote_command(
+                config,
+                logger,
+                self.host,
+                self.command,
+                environment,
                 arguments = [id, jobhost, jobport],
                 resources = self.resources
             )

CEP/Pipeline/recipes/sip/nodes/dppp.py

@@ -111,10 +111,6 @@ class dppp(LOFARnodeTCP):
                 )
                 shutil.copytree(infile, tmpfile)
 
-            # Limit number of threads used.
-            self.environment['OMP_NUM_THREADS'] = str(nthreads)
-            self.logger.debug("Using %s threads for NDPPP" % nthreads)
-
             # *****************************************************************
             # 3. Update the parset with locally calculate information
 

CEP/Pipeline/recipes/sip/nodes/imager_prepare.py

@@ -220,9 +220,6 @@ class imager_prepare(LOFARnodeTCP):
         Muckable function running the dppp executable.
         Wraps dppp with catchLog4CPLus and catch_segfaults
         """
-        # TODO: cpu limited is static at this location
-        environment['OMP_NUM_THREADS'] = str(8)
-        self.logger.debug("Using %s threads for ndppp" % 8)
         with CatchLog4CPlus(working_dir, self.logger.name +
              "." + os.path.basename("imager_prepare_ndppp"),
                   os.path.basename(ndppp)) as logger:

CEP/Pipeline/recipes/sip/nodes/selfcal_awimager.py

@@ -174,11 +174,6 @@ class selfcal_awimager(LOFARnodeTCP):
             # *****************************************************************
             # 5. Run the awimager with the parameterset
 
-            # TODO: FIXME: manually Limit number of threads used.
-            omp_num_threads = 8
-            self.environment['OMP_NUM_THREADS'] = str(omp_num_threads)
-            self.logger.debug("Using %s threads for swimager" % omp_num_threads)
-
             cmd = [executable, calculated_parset_path]
             self.logger.debug("Parset used for awimager run:")
             self.logger.debug(cmd)