feat: executables are extracted from inputs in ARC CEs

Author: aldbr

src/DIRAC/Resources/Computing/ARC6ComputingElement.py

@@ -144,7 +144,7 @@ def submitJob(self, executableFile, proxy, numberOfJobs=1):
                 jobdescs = arc.JobDescriptionList()
 
                 # Get the job into the ARC way
-                xrslString, diracStamp = self._writeXRSL(executableFile)
+                xrslString, diracStamp = self._writeXRSL(executableFile, [], [])
                 self.log.debug("XRSL string submitted : %s" % xrslString)
                 self.log.debug("DIRAC stamp for job : %s" % diracStamp)
 

src/DIRAC/Resources/Computing/ARCComputingElement.py

@@ -167,13 +167,12 @@ def _addCEConfigDefaults(self):
         ComputingElement._addCEConfigDefaults(self)
 
     #############################################################################
-    def _writeXRSL(self, executableFile, inputs=None, outputs=None, executables=None):
+    def _writeXRSL(self, executableFile, inputs, outputs):
         """Create the JDL for submission
 
         :param str executableFile: executable to wrap in a XRSL file
-        :param str/list inputs: path of the dependencies to include along with the executable
-        :param str/list outputs: path of the outputs that we want to get at the end of the execution
-        :param str/list executables: path to inputs that should have execution mode on the remote worker node
+        :param list inputs: path of the dependencies to include along with the executable
+        :param list outputs: path of the outputs that we want to get at the end of the execution
         """
         diracStamp = makeGuid()[:8]
         # Evaluate the number of processors to allocate
@@ -191,34 +190,25 @@ def _writeXRSL(self, executableFile, inputs=None, outputs=None, executables=None
                 "xrslMPExtraString": self.xrslMPExtraString,
             }
 
-        # Files that would need execution rights on the remote worker node
-        xrslExecutables = ""
-        if executables:
-            if not isinstance(executables, list):
-                executables = [executables]
-            xrslExecutables = "(executables=%s)" % " ".join(map(os.path.basename, executables))
-            # Add them to the inputFiles
-            if not inputs:
-                inputs = []
-            if not isinstance(inputs, list):
-                inputs = [inputs]
-            inputs += executables
-
         # Dependencies that have to be embedded along with the executable
         xrslInputs = ""
-        if inputs:
-            if not isinstance(inputs, list):
-                inputs = [inputs]
-            for inputFile in inputs:
-                xrslInputs += '(%s "%s")' % (os.path.basename(inputFile), inputFile)
+        executables = []
+        for inputFile in inputs:
+            inputFileBaseName = os.path.basename(inputFile)
+            if os.access(inputFile, os.X_OK):
+                # Files that would need execution rights on the remote worker node
+                executables.append(inputFileBaseName)
+            xrslInputs += '(%s "%s")' % (inputFileBaseName, inputFile)
+
+        # Executables are added to the XRSL
+        xrslExecutables = ""
+        if executables:
+            xrslExecutables = "(executables=%s)" % " ".join(executables)
 
         # Output files to retrieve once the execution is complete
         xrslOutputs = '("%s.out" "") ("%s.err" "")' % (diracStamp, diracStamp)
-        if outputs:
-            if not isinstance(outputs, list):
-                outputs = [outputs]
-            for outputFile in outputs:
-                xrslOutputs += '(%s "")' % (outputFile)
+        for outputFile in outputs:
+            xrslOutputs += '(%s "")' % (outputFile)
 
         xrsl = """
 &(executable="%(executable)s")
@@ -247,6 +237,13 @@ def _writeXRSL(self, executableFile, inputs=None, outputs=None, executables=None
     def _bundlePreamble(self, executableFile):
         """Bundle the preamble with the executable file"""
         wrapperContent = "%s\n./%s" % (self.preamble, executableFile)
+
+        # We need to make sure the executable file can be executed by the wrapper
+        # By adding the execution mode to the file, the file will be processed as an "executable" in the XRSL
+        # This is done in _writeXRSL()
+        if not os.access(executableFile, os.X_OK):
+            os.chmod(executableFile, stat.S_IRWXU | stat.S_IRGRP | stat.S_IXGRP | stat.S_IROTH + stat.S_IXOTH)
+
         return writeScript(wrapperContent, os.getcwd())
 
     #############################################################################
@@ -299,13 +296,14 @@ def submitJob(self, executableFile, proxy, numberOfJobs=1, inputs=None, outputs=
             return result
         self.usercfg.ProxyPath(os.environ["X509_USER_PROXY"])
 
-        self.log.verbose("Executable file path: %s" % executableFile)
-        if not os.access(executableFile, 5):
-            os.chmod(executableFile, stat.S_IRWXU | stat.S_IRGRP | stat.S_IXGRP | stat.S_IROTH + stat.S_IXOTH)
+        if not inputs:
+            inputs = []
+        if not outputs:
+            outputs = []
 
-        executables = None
+        self.log.verbose("Executable file path: %s" % executableFile)
         if self.preamble:
-            executables = [executableFile]
+            inputs.append(executableFile)
             executableFile = self._bundlePreamble(executableFile)
 
         batchIDList = []
@@ -325,7 +323,7 @@ def submitJob(self, executableFile, proxy, numberOfJobs=1, inputs=None, outputs=
             # The basic job description
             jobdescs = arc.JobDescriptionList()
             # Get the job into the ARC way
-            xrslString, diracStamp = self._writeXRSL(executableFile, inputs, outputs, executables)
+            xrslString, diracStamp = self._writeXRSL(executableFile, inputs, outputs)
             self.log.debug("XRSL string submitted : %s" % xrslString)
             self.log.debug("DIRAC stamp for job : %s" % diracStamp)
             # The arc bindings don't accept unicode objects in Python 2 so xrslString must be explicitly cast

src/DIRAC/Resources/Computing/AREXComputingElement.py

@@ -304,13 +304,12 @@ def _getDelegationID(self, arcJobID):
 
     #############################################################################
 
-    def _getArcJobID(self, executableFile, inputs, outputs, executables, delegation):
+    def _getArcJobID(self, executableFile, inputs, outputs, delegation):
         """Get an ARC JobID endpoint to upload executables and inputs.
 
         :param str executableFile: executable to submit
         :param list inputs: list of input files
         :param list outputs: list of expected output files
-        :param list executables: list of secondary executables (will be uploaded with the executable mode)
         :param str delegation: delegation ID
 
         :return: tuple containing a job ID and a stamp
@@ -320,7 +319,7 @@ def _getArcJobID(self, executableFile, inputs, outputs, executables, delegation)
         query = self._urlJoin("jobs")
 
         # Get the job into the ARC way
-        xrslString, diracStamp = self._writeXRSL(executableFile, inputs, outputs, executables)
+        xrslString, diracStamp = self._writeXRSL(executableFile, inputs, outputs)
         xrslString += delegation
         self.log.debug("XRSL string submitted", "is %s" % xrslString)
         self.log.debug("DIRAC stamp for job", "is %s" % diracStamp)
@@ -344,21 +343,16 @@ def _getArcJobID(self, executableFile, inputs, outputs, executables, delegation)
         arcJobID = responseJob["id"]
         return S_OK((arcJobID, diracStamp))
 
-    def _uploadJobDependencies(self, arcJobID, executableFile, inputs, executables):
+    def _uploadJobDependencies(self, arcJobID, executableFile, inputs):
         """Upload job dependencies so that the job can start.
         This includes the executables and the inputs.
 
         :param str arcJobID: ARC job ID
         :param str executableFile: executable file
         :param list inputs: inputs required by the executable file
-        :param list executables: executables require by the executable file
         """
         filesToSubmit = [executableFile]
-        filesToSubmit += executables
-        if inputs:
-            if not isinstance(inputs, list):
-                inputs = [inputs]
-            filesToSubmit += inputs
+        filesToSubmit += inputs
 
         for fileToSubmit in filesToSubmit:
             queryExecutable = self._urlJoin(os.path.join("jobs", arcJobID, "session", os.path.basename(fileToSubmit)))
@@ -397,10 +391,14 @@ def submitJob(self, executableFile, proxy, numberOfJobs=1, inputs=None, outputs=
         else:
             delegation = "\n(delegationid=%s)" % result["Value"]
 
+        if not inputs:
+            inputs = []
+        if not outputs:
+            outputs = []
+
         # If there is a preamble, then we bundle it in an executable file
-        executables = []
         if self.preamble:
-            executables = [executableFile]
+            inputs.append(executableFile)
             executableFile = self._bundlePreamble(executableFile)
 
         # Submit multiple jobs sequentially.
@@ -410,14 +408,14 @@ def submitJob(self, executableFile, proxy, numberOfJobs=1, inputs=None, outputs=
         batchIDList = []
         stampDict = {}
         for _ in range(numberOfJobs):
-            result = self._getArcJobID(executableFile, inputs, outputs, executables, delegation)
+            result = self._getArcJobID(executableFile, inputs, outputs, delegation)
             if not result["OK"]:
                 break
             arcJobID, diracStamp = result["Value"]
 
             # At this point, only the XRSL job has been submitted to AREX services
             # Here we also upload the executable, other executable files and inputs.
-            result = self._uploadJobDependencies(arcJobID, executableFile, inputs, executables)
+            result = self._uploadJobDependencies(arcJobID, executableFile, inputs)
             if not result["OK"]:
                 break