fix(resources): Embed executableFile in SrunWrapper

Author: aldbr

src/DIRAC/Resources/Computing/ComputingElement.py

@@ -79,7 +79,6 @@ def __init__(self, ceName):
         self.valid = None
         self.mandatoryParameters = []
         self.batchSystem = None
-        self.parallelLibrary = None
         self.taskResults = {}
         self.minProxyTime = gConfig.getValue("/Registry/MinProxyLifeTime", 10800)  # secs
         self.defaultProxyTime = gConfig.getValue("/Registry/DefaultProxyLifeTime", 43200)  # secs
@@ -197,52 +196,19 @@ def loadBatchSystem(self, batchSystemName):
 
         :param str batchSystemName: name of the batch system
         """
-        result = self._loadCEAttribute("BatchSystem", batchSystemName, "Resources.Computing.BatchSystems")
-        if not result["OK"]:
-            return result
-        self.batchSystem = result["Value"]
-        return S_OK()
-
-    def loadParallelLibrary(self, parallelLibraryName, workingDirectory="."):
-        """Instantiate object representing the parallel library that will generate a script to wrap the executable
-
-        :param str parallelLibraryName: name of the parallel library
-        """
-        result = self._loadCEAttribute(
-            "ParallelLibrary",
-            parallelLibraryName,
-            "Resources.Computing.ParallelLibraries",
-            workingDirectory=workingDirectory,
-        )
-        if not result["OK"]:
-            return result
-        self.parallelLibrary = result["Value"]
-        return S_OK()
-
-    def _loadCEAttribute(self, ceAttribute, typeAttribute, moduleLocation, **kwargs):
-        """Instantiate specific CE attributes
-
-        :param str ceAttribute: name of the CE attribute
-        :param str typeAttribute: class of the CE attribute
-        :param str moduleLocation: location of the class definition in the source code
-        :param dict kwargs: parameters to pass to the constructor of the ceAttribute instance
-        """
-        if typeAttribute is None:
-            typeAttribute = self.ceParameters[ceAttribute]
+        if batchSystemName is None:
+            batchSystemName = self.ceParameters['BatchSystem']
 
         objectLoader = ObjectLoader()
-        moduleLocation += ".%s" % typeAttribute
-        result = objectLoader.loadObject(moduleLocation, typeAttribute)
-        if not result["OK"]:
-            self.log.error("Failed to load %s object: %s" % (ceAttribute, result["Message"]))
+        result = objectLoader.loadObject('Resources.Computing.BatchSystems.%s' % batchSystemName, batchSystemName)
+        if not result['OK']:
+            self.log.error('Failed to load batch object: %s' % result['Message'])
             return result
-        ceAttributeClass = result["Value"]
-        ceAttributeModuleFile = result["ModuleFile"]
-
-        ceAttributeObject = ceAttributeClass(**kwargs)
-        self.log.info("Class from module: ", ceAttributeModuleFile)
-
-        return S_OK(ceAttributeObject)
+        batchClass = result['Value']
+        batchModuleFile = result['ModuleFile']
+        self.batchSystem = batchClass()
+        self.log.info("Batch system class from module: ", batchModuleFile)
+        return S_OK()
 
     def setParameters(self, ceOptions):
         """Add parameters from the given dictionary overriding the previous values

src/DIRAC/Resources/Computing/LocalComputingElement.py

@@ -15,10 +15,6 @@
    Underlying batch system that is going to be used to orchestrate executable files. The Batch System has to be
    accessible from the LocalCE. By default, the LocalComputingElement submits directly on the host via the Host class.
 
-ParallelLibrary:
-   Underlying parallel library used to generate a wrapper around the executable files to run them in parallel on
-   multiple nodes.
-
 SharedArea:
    Area used to store executable/output/error files if they are not aready defined via BatchOutput, BatchError,
    InfoArea, ExecutableArea and/or WorkArea. The path should be absolute.
@@ -80,7 +76,7 @@ def _reset(self):
         batchSystemName = self.ceParameters.get("BatchSystem", "Host")
         result = self.loadBatchSystem(batchSystemName)
         if not result["OK"]:
-            self.log.error("Failed to load the batch system plugin %s", self.batchSystem)
+            self.log.error("Failed to load the batch system plugin %s", batchSystemName)
             return result
 
         self.queue = self.ceParameters["Queue"]
@@ -106,13 +102,6 @@ def _reset(self):
         if not self.workArea.startswith("/"):
             self.workArea = os.path.join(self.sharedArea, self.workArea)
 
-        parallelLibraryName = self.ceParameters.get("ParallelLibrary")
-        if parallelLibraryName:
-            result = self.loadParallelLibrary(parallelLibraryName, self.executableArea)
-            if not result["OK"]:
-                self.log.error("Failed to load the parallel library plugin %s", parallelLibraryName)
-                return result
-
         result = self._prepareHost()
         if not result["OK"]:
             self.log.error("Failed to initialize CE", self.ceName)
@@ -181,14 +170,6 @@ def _prepareHost(self):
         return S_OK()
 
     def submitJob(self, executableFile, proxy=None, numberOfJobs=1):
-        copyExecutable = os.path.join(self.executableArea, os.path.basename(executableFile))
-        if self.parallelLibrary and executableFile != copyExecutable:
-            # Because we use a parallel library, the executable will become a dependency of the parallel library script
-            # Thus, it has to be defined in a specific area (executableArea) to be found and executed properly
-            # For this reason, we copy the executable from its location to executableArea
-            shutil.copy(executableFile, copyExecutable)
-            executableFile = copyExecutable
-
         if not os.access(executableFile, 5):
             os.chmod(executableFile, stat.S_IRWXU | stat.S_IRGRP | stat.S_IXGRP | stat.S_IROTH | stat.S_IXOTH)
 
@@ -206,10 +187,6 @@ def submitJob(self, executableFile, proxy=None, numberOfJobs=1):
         else:  # no proxy
             submitFile = executableFile
 
-        if self.parallelLibrary:
-            # Wrap the executable to be executed multiple times in parallel via a parallel library
-            submitFile = self.parallelLibrary.generateWrapper(submitFile)
-
         jobStamps = []
         for _i in range(numberOfJobs):
             jobStamps.append(makeGuid()[:8])
@@ -229,7 +206,7 @@ def submitJob(self, executableFile, proxy=None, numberOfJobs=1):
             "NumberOfGPUs": self.numberOfGPUs,
         }
         resultSubmit = self.batchSystem.submitJob(**batchDict)
-        if proxy or self.parallelLibrary:
+        if proxy:
             os.remove(submitFile)
 
         if resultSubmit["Status"] == 0:
@@ -311,9 +288,6 @@ def getJobOutput(self, jobID, localDir=None):
             return result
 
         jobStamp, _host, outputFile, errorFile = result["Value"]
-        if self.parallelLibrary:
-            # outputFile and errorFile are directly modified by parallelLib
-            self.parallelLibrary.processOutput(outputFile, errorFile)
 
         if not localDir:
             tempDir = tempfile.mkdtemp()

src/DIRAC/Resources/Computing/ParallelLibraries/ParallelLibrary.py

@@ -15,11 +15,6 @@
 
 The Parallel Libraries have to be defined in a CE section via the configuration system. See the page about
 configuring :ref:`resourcesComputing` for where the options can be placed.
-
-WorkingDirectory:
-   Generally defined in the ComputingElement source code depending on how they deal with the executables.
-   Indicates where the parallel library wrapper script has to be defined and retrieved.
-
 """
 from __future__ import absolute_import
 from __future__ import division

src/DIRAC/Resources/Computing/ParallelLibraries/Srun.py

@@ -14,8 +14,8 @@
 
 
 class Srun(ParallelLibrary):
-    def __init__(self, workingDirectory):
-        super(Srun, self).__init__("srun", workingDirectory)
+    def __init__(self):
+        super(Srun, self).__init__("srun")
 
     def generateWrapper(self, executableFile):
         """
@@ -25,21 +25,25 @@ def generateWrapper(self, executableFile):
         :param str executableFile: name of the executable file to wrap
         :return str: name of the wrapper that runs the executable via srun
         """
+        wrapper = 'srunExec.sh'
+        with open(executableFile, 'r') as f:
+            content = f.read()
 
         # Build the script to run the executable in parallel multiple times
-        # srun is the command to execute a task multiple time in parallel
-        # -l option: add the task ID to the output
-        # -k option: do not kill the slurm job if one of the nodes is broken
-        cmd = "#!/bin/bash\n"
-        cmd += "srun -l -k %s" % shlex_quote(executableFile)
-        self.log.debug("Command generated by Srun:", "%s" % cmd)
-
-        srunWrapper = os.path.join(self.workingDirectory, "srunExec.sh")
-        with open(srunWrapper, "w") as exe:
-            exe.write(cmd)
-        if not os.access(srunWrapper, os.R_OK | os.X_OK):
-            os.chmod(srunWrapper, stat.S_IRWXU | stat.S_IRGRP | stat.S_IXGRP | stat.S_IROTH + stat.S_IXOTH)
-        return srunWrapper
+        # - Embed the content of executableFile inside the parallel library wrapper script
+        # - srun is the command to execute a task multiple time in parallel
+        #   -l option: add the task ID to the output
+        #   -k option: do not kill the slurm job if one of the nodes is broken
+        cmd = """#!/bin/bash
+cat > %(wrapper)s << EOFEXEC
+%(content)s
+EOFEXEC
+chmod 755 %(wrapper)s
+srun -l -k %(wrapper)s
+""" % dict(wrapper=wrapper,
+           content=content
+           )
+    return cmd
 
     def processOutput(self, output, error, isFile=True):
         """

src/DIRAC/Resources/Computing/ParallelLibraries/test/Test_ParallelLibraries.py

@@ -18,6 +18,19 @@
 echo "hello world"
 """
 
+expectedContent = """#!/bin/bash
+
+cat > srunExec.sh << EOFEXEC
+
+#!/bin/bash
+
+echo "hello world"
+
+EOFEXEC
+chmod 755 srunExec.sh
+srun -l -k srunExec.sh
+"""
+
 srunOutput = """
 1: line1
 1: line2
@@ -61,14 +74,14 @@
 
 
 @pytest.fixture
-def generateParallelLibrary(parallelLibrary, parameters):
+def generateParallelLibrary(parallelLibrary):
     """Instantiate the requested Parallel Library"""
     # Instantiate an object from parallelLibrary class
     parallelLibraryPath = "DIRAC.Resources.Computing.ParallelLibraries.%s" % parallelLibrary
     plugin = __import__(parallelLibraryPath, globals(), locals(), [parallelLibrary])  # pylint: disable=unused-variable
     # Need to be reloaded to update the mock within the module, else, it will reuse the one when loaded the first time
     reload_module(plugin)
-    parallelLibraryStr = "plugin.%s(%s)" % (parallelLibrary, parameters)
+    parallelLibraryStr = "plugin.%s()" % (parallelLibrary)
     return eval(parallelLibraryStr)
 
 
@@ -100,6 +113,26 @@ def test_generateWrapper(generateParallelLibrary, parallelLibrary, parameters, e
     os.remove(executableFile)
     os.remove(res)
 
+@pytest.mark.parametrize(
+    "parallelLibrary, expectedContent", 
+    [
+        ('Srun', expectedContent),
+    ],
+)
+def test_generateWrapper(generateParallelLibrary, parallelLibrary, expectedContent):
+    """ Test generateWrapper()"""
+    parallelLibraryInstance = generateParallelLibrary
+
+    executableFile = 'executableFile.sh'
+    with open(executableFile, 'w') as f:
+        f.write(executableContent)
+
+    res = parallelLibraryInstance.generateWrapper(executableFile)
+    # Make sure a wrapper file has been generated and is executable
+    assert res == expectedContent
+
+    os.remove(executableFile)
+
 
 @pytest.mark.parametrize(
     "parallelLibrary, parameters, outputContent, expectedContent",

src/DIRAC/Resources/Computing/SSHComputingElement.py

@@ -10,10 +10,6 @@
    Underlying batch system that is going to be used to orchestrate executable files. The Batch System has to be
    accessible from the LocalCE. By default, the LocalComputingElement submits directly on the host via the Host class.
 
-ParallelLibrary:
-   Underlying parallel library used to generate a wrapper around the executable files to run them in parallel on
-   multiple nodes.
-
 SharedArea:
    Area used to store executable/output/error files if they are not aready defined via BatchOutput, BatchError,
    InfoArea, ExecutableArea and/or WorkArea. The path should be absolute.
@@ -412,13 +408,6 @@ def _reset(self):
         if not self.workArea.startswith("/"):
             self.workArea = os.path.join(self.sharedArea, self.workArea)
 
-        parallelLibraryName = self.ceParameters.get("ParallelLibrary")
-        if parallelLibraryName:
-            result = self.loadParallelLibrary(parallelLibraryName)
-            if not result["OK"]:
-                self.log.error("Failed to load the parallel library plugin %s", parallelLibraryName)
-                return result
-
         self.submitOptions = self.ceParameters.get("SubmitOptions", "")
         self.removeOutput = True
         if "RemoveOutput" in self.ceParameters:
@@ -599,21 +588,13 @@ def submitJob(self, executableFile, proxy, numberOfJobs=1):
         else:  # no proxy
             submitFile = executableFile
 
-        inputs = []
-        if self.parallelLibrary:
-            # In this case, the executable becomes an input of a parallel library script.
-            # It needs to be submitted along with the submitFile, which is a parallel library wrapper.
-            inputFile = os.path.join(self.executableArea, os.path.basename(submitFile))
-            inputs.append(inputFile)
-            submitFile = self.parallelLibrary.generateWrapper(inputFile)
-
-        result = self._submitJobToHost(submitFile, numberOfJobs, inputs=inputs)
-        if proxy or self.parallelLibrary:
+        result = self._submitJobToHost(submitFile, numberOfJobs)
+        if proxy:
             os.remove(submitFile)
 
         return result
 
-    def _submitJobToHost(self, executableFile, numberOfJobs, host=None, inputs=None):
+    def _submitJobToHost(self, executableFile, numberOfJobs, host=None):
         """Submit prepared executable to the given host"""
         ssh = SSH(host=host, parameters=self.ceParameters)
         # Copy the executable
@@ -622,14 +603,6 @@ def _submitJobToHost(self, executableFile, numberOfJobs, host=None, inputs=None)
         if not result["OK"]:
             return result
 
-        # Copy the executable dependencies if any
-        if inputs:
-            for localInput in inputs:
-                remoteInput = os.path.join(self.executableArea, os.path.basename(localInput))
-                result = ssh.scpCall(30, localInput, remoteInput, postUploadCommand="chmod +x %s" % remoteInput)
-                if not result["OK"]:
-                    return result
-
         jobStamps = []
         for _i in range(numberOfJobs):
             jobStamps.append(makeGuid()[:8])
@@ -849,7 +822,4 @@ def getJobOutput(self, jobID, localDir=None):
             output = result["Value"][1]
             error = result["Value"][1]
 
-        if self.parallelLibrary:
-            output, error = self.parallelLibrary.processOutput(output, error, isFile=bool(localDir))
-
         return S_OK((output, error))