diff --git a/docs/source/AdministratorGuide/Resources/computingelements.rst b/docs/source/AdministratorGuide/Resources/computingelements.rst index 27850cb7cc0..f33497dc7d2 100644 --- a/docs/source/AdministratorGuide/Resources/computingelements.rst +++ b/docs/source/AdministratorGuide/Resources/computingelements.rst @@ -57,25 +57,22 @@ of the *ComputingElement* is located inside the corresponding site section in th # Site administrative domain LCG { - # Site section + # Site section. This is the DIRAC's site name. LCG.CNAF.it { - # Site name + # Alternative site name (e.g. site name in GOC DB) Name = CNAF - # List of valid CEs on the site - CE = ce01.infn.it, ce02.infn.it - # Section describing each CE CEs { - # Specific CE description section + # Specific CE description section. This site name is unique. ce01.infn.it { - # Type of the CE + # Type of the CE. "HTCondorCE" and "AREX" and "SSH" are the most common types. CEType = HTCondorCE - # Section to describe various queue in the CE + # Section to describe various (logical) queues in the CE. Queues { long @@ -93,7 +90,6 @@ of the *ComputingElement* is located inside the corresponding site section in th This is the general structure in which specific CE descriptions are inserted. The CE configuration is part of the general DIRAC configuration -It can be placed in the general Configuration Service or in the local configuration of the DIRAC installation. Examples of the configuration can be found in the :ref:`full_configuration_example`, in the *Resources/Computing* section. You can find the options of a specific CE in the code documentation: :mod:`DIRAC.Resources.Computing`. diff --git a/docs/source/UserGuide/Tutorials/JobManagementAdvanced/index.rst b/docs/source/UserGuide/Tutorials/JobManagementAdvanced/index.rst index a939114b72a..b3f72e7eb24 100644 --- a/docs/source/UserGuide/Tutorials/JobManagementAdvanced/index.rst +++ b/docs/source/UserGuide/Tutorials/JobManagementAdvanced/index.rst @@ -335,16 +335,15 @@ Jobs that can (or should) run using more than 1 processor should be described as using the "setNumberOfProcessors" method of the API:: j = Job() - j.setCPUTime(500) - j.setExecutable('echo',arguments='hello') - j.setExecutable('ls',arguments='-l') - j.setExecutable('echo', arguments='hello again') - j.setName('MP test') + ... j.setNumberOfProcessors(16) Calling ``Job().setNumberOfProcessors()``, with a value bigger than 1, will translate into adding also the "MultiProcessor" tag to the job description. +``Job().setNumberOfProcessors()`` takes at most 3 arguments, in this order: +``numberOfProcessors`` is the exact number of requested processors, ``minNumberOfProcessors`` is the minimum allowed, ``maxNumberOfProcessors`` the maximum. + Users can specify in the job descriptions NumberOfProcessors and WholeNode parameters, e.g.:: NumberOfProcessors = 16; @@ -356,6 +355,20 @@ This will be translated internally into 16Processors and WholeNode tags. This would allow resources (WN's) to put flexibly requirements on jobs to be taken, for example, avoiding single-core jobs on a multi-core nodes. +Setting memory limits +@@@@@@@@@@@@@@@@@@@@@ + +Jobs can (and probably should) set RAM limits. +using the "setRAMRequirements" method of the API:: + + j = Job() + ... + j.setRAMRequirements(2500, 4000) + +Calling ``Job().setRAMRequirements()`` takes 2 values, where the first is the minimum required amount of RAM (in MB) that the job requests. +The second value instead specifies the limit that should not be surpassed. + + Submitting jobs with specifc requirements (e.g. GPU) @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@ diff --git a/src/DIRAC/Core/Utilities/CGroups2.py b/src/DIRAC/Core/Utilities/CGroups2.py index f874f85483f..f7cb460f572 100644 --- a/src/DIRAC/Core/Utilities/CGroups2.py +++ b/src/DIRAC/Core/Utilities/CGroups2.py @@ -300,6 +300,7 @@ def systemCall(self, *args, **kwargs): if "ceParameters" in kwargs: if cpuLimit := kwargs["ceParameters"].get("CPULimit", None): cores = float(cpuLimit) + # MemoryLimitMB should be the job upper limit if memoryMB := int(kwargs["ceParameters"].get("MemoryLimitMB", 0)): memory = memoryMB * 1024 * 1024 if kwargs["ceParameters"].get("MemoryNoSwap", "no").lower() in ("yes", "true"): diff --git a/src/DIRAC/Interfaces/API/Job.py b/src/DIRAC/Interfaces/API/Job.py index c1d95160bf2..1baba96000d 100755 --- a/src/DIRAC/Interfaces/API/Job.py +++ b/src/DIRAC/Interfaces/API/Job.py @@ -1,27 +1,28 @@ """ - Job Base Class +Job Base Class - This class provides generic job definition functionality suitable for any VO. +This class provides generic job definition functionality suitable for any VO. - Helper functions are documented with example usage for the DIRAC API. An example - script (for a simple executable) would be:: +Helper functions are documented with example usage for the DIRAC API. An example +script (for a simple executable) would be:: - from DIRAC.Interfaces.API.Dirac import Dirac - from DIRAC.Interfaces.API.Job import Job + from DIRAC.Interfaces.API.Dirac import Dirac + from DIRAC.Interfaces.API.Job import Job - j = Job() - j.setCPUTime(500) - j.setExecutable('/bin/echo hello') - j.setExecutable('yourPythonScript.py') - j.setExecutable('/bin/echo hello again') - j.setName('MyJobName') + j = Job() + j.setCPUTime(500) + j.setExecutable('/bin/echo hello') + j.setExecutable('yourPythonScript.py') + j.setExecutable('/bin/echo hello again') + j.setName('MyJobName') - dirac = Dirac() - jobID = dirac.submitJob(j) - print 'Submission Result: ',jobID + dirac = Dirac() + jobID = dirac.submitJob(j) + print 'Submission Result: ',jobID - Note that several executables can be provided and wil be executed sequentially. +Note that several executables can be provided and wil be executed sequentially. """ + import os import re import shlex @@ -517,6 +518,50 @@ def setDestination(self, destination): return S_OK() ############################################################################# + def setRAMRequirements(self, ramRequired: int = 0, maxRAM: int = 0): + """Helper function. + Specify the RAM requirements for the job, in MB. 0 (default) means no specific requirements. + + Example usage: + + >>> job = Job() + >>> job.setRAMRequirements(ramRequired=2000) + means that the job needs at least 2 GBs of RAM to work. This is taken into consideration at job's matching time. + The job definition does not specify an upper limit. + From a user's point of view this is fine (normally, not for admins). + + >>> job.setRAMRequirements(ramRequired=500, maxRAM=3800) + means that the job needs 500 MBs of RAM to work. 3.8 GBs will then be the upper limit for CG2 limits. + + >>> job.setRAMRequirements(ramRequired=3200, maxRAM=3200) + means that we should match this job if there is at least 3.2 available GBs of run. At the same time, CG2 will not allow to use more than that. + + >>> job.setRAMRequirements(maxRAM=4000) + means that the job does not set a min amount of RAM (so can match--run "everywhere"), but the 4 GBs will then be the upper limit for CG2 limits. + + >>> job.setRAMRequirements(ramRequired=8000, maxRAM=4000) + Makes no sense, an error will be raised + """ + if ramRequired and maxRAM and ramRequired > maxRAM: + return self._reportError("Invalid settings, ramRequired is higher than maxRAM") + + if ramRequired: + self._addParameter( + self.workflow, + "MinRAM", + "JDL", + ramRequired, + "MBs of RAM requested", + ) + if maxRAM: + self._addParameter( + self.workflow, + "MaxRAM", + "JDL", + maxRAM, + "Max MBs of RAM to be used", + ) + def setNumberOfProcessors(self, numberOfProcessors=None, minNumberOfProcessors=None, maxNumberOfProcessors=None): """Helper function to set the number of processors required by the job. @@ -709,7 +754,7 @@ def setTag(self, tags): Example usage: >>> job = Job() - >>> job.setTag( ['WholeNode','8GBMemory'] ) + >>> job.setTag( ['WholeNode'] ) :param tags: single tag string or a list of tags :type tags: str or python:list diff --git a/src/DIRAC/Resources/Computing/ComputingElement.py b/src/DIRAC/Resources/Computing/ComputingElement.py index 39d1a1fe645..0033ffa3e65 100755 --- a/src/DIRAC/Resources/Computing/ComputingElement.py +++ b/src/DIRAC/Resources/Computing/ComputingElement.py @@ -1,40 +1,40 @@ -""" The Computing Element class is a base class for all the various - types CEs. It serves several purposes: +"""The Computing Element class is a base class for all the various +types CEs. It serves several purposes: - - collects general CE related parameters to generate CE description - for the job matching - - provides logic for evaluation of the number of available CPU slots - - provides logic for the proxy renewal while executing jobs + - collects general CE related parameters to generate CE description + for the job matching + - provides logic for evaluation of the number of available CPU slots + - provides logic for the proxy renewal while executing jobs - The CE parameters are collected from the following sources, in hierarchy - descending order: +The CE parameters are collected from the following sources, in hierarchy +descending order: - - parameters provided through setParameters() method of the class - - parameters in /LocalSite configuration section - - parameters in /LocalSite//ResourceDict configuration section - - parameters in /LocalSite/ResourceDict configuration section - - parameters in /LocalSite/ configuration section - - parameters in /Resources/Computing/ configuration section - - parameters in /Resources/Computing/CEDefaults configuration section + - parameters provided through setParameters() method of the class + - parameters in /LocalSite configuration section + - parameters in /LocalSite//ResourceDict configuration section + - parameters in /LocalSite/ResourceDict configuration section + - parameters in /LocalSite/ configuration section + - parameters in /Resources/Computing/ configuration section + - parameters in /Resources/Computing/CEDefaults configuration section - The ComputingElement objects are usually instantiated with the help of - ComputingElementFactory. +The ComputingElement objects are usually instantiated with the help of +ComputingElementFactory. - The ComputingElement class can be considered abstract. 3 kinds of abstract ComputingElements - can be distinguished from it: +The ComputingElement class can be considered abstract. 3 kinds of abstract ComputingElements +can be distinguished from it: - - Remote ComputingElement: includes methods to interact with a remote ComputingElement - (e.g. HtCondorCEComputingElement, AREXComputingElement). - - Inner ComputingElement: includes methods to locally interact with an underlying worker node. - It is worth noting that an Inner ComputingElement provides synchronous submission - (the submission of a job is blocking the execution until its completion). It deals with one job at a time. - - Inner Pool ComputingElement: includes methods to locally interact with Inner ComputingElements asynchronously. - It can manage a pool of jobs running simultaneously. + - Remote ComputingElement: includes methods to interact with a remote ComputingElement + (e.g. HtCondorCEComputingElement, AREXComputingElement). + - Inner ComputingElement: includes methods to locally interact with an underlying worker node. + It is worth noting that an Inner ComputingElement provides synchronous submission + (the submission of a job is blocking the execution until its completion). It deals with one job at a time. + - Inner Pool ComputingElement: includes methods to locally interact with Inner ComputingElements asynchronously. + It can manage a pool of jobs running simultaneously. - To configure the use of Tokens for CEs: +To configure the use of Tokens for CEs: - * the CE is able to receive any token. Validation: 'Tag = Token' should be included in the CE parameters. - * the CE is able to receive VO-specifc tokens. Validation: 'Tag = Token:' should be included in the CE parameters. +* the CE is able to receive any token. Validation: 'Tag = Token' should be included in the CE parameters. +* the CE is able to receive VO-specifc tokens. Validation: 'Tag = Token:' should be included in the CE parameters. """ @@ -50,6 +50,7 @@ from DIRAC.WorkloadManagementSystem.Utilities.JobParameters import ( getNumberOfGPUs, getNumberOfProcessors, + getAvailableRAM, ) INTEGER_PARAMETERS = ["CPUTime", "NumberOfProcessors", "NumberOfPayloadProcessors", "MaxRAM"] @@ -211,12 +212,14 @@ def setParameters(self, ceOptions): generalCEDict.update(self.ceParameters) self.ceParameters = generalCEDict - # If NumberOfProcessors/GPUs is present in the description but is equal to zero + # If NumberOfProcessors/GPUs/MaxRAM is present in the description but is equal to zero # interpret it as needing local evaluation - if self.ceParameters.get("NumberOfProcessors", -1) == 0: + if int(self.ceParameters.get("NumberOfProcessors", -1)) == 0: self.ceParameters["NumberOfProcessors"] = getNumberOfProcessors() - if self.ceParameters.get("NumberOfGPUs", -1) == 0: + if int(self.ceParameters.get("NumberOfGPUs", -1)) == 0: self.ceParameters["NumberOfGPUs"] = getNumberOfGPUs() + if int(self.ceParameters.get("MaxRAM", 0)) == 0: + self.ceParameters["MaxRAM"] = getAvailableRAM() for key in ceOptions: if key in INTEGER_PARAMETERS: @@ -252,6 +255,7 @@ def available(self): runningJobs = result["RunningJobs"] waitingJobs = result["WaitingJobs"] availableProcessors = result.get("AvailableProcessors") + ceInfoDict = dict(result) maxTotalJobs = int(self.ceParameters.get("MaxTotalJobs", 0)) @@ -404,6 +408,7 @@ def getDescription(self): result = self.getCEStatus() if result["OK"]: ceDict["NumberOfProcessors"] = result.get("AvailableProcessors", result.get("NumberOfProcessors", 1)) + ceDict["MaxRAM"] = result.get("AvailableRAM", result.get("MaxRAM", 1024)) else: self.log.error( "Failure getting CE status", "(we keep going without the number of waiting and running pilots/jobs)" diff --git a/src/DIRAC/Resources/Computing/InProcessComputingElement.py b/src/DIRAC/Resources/Computing/InProcessComputingElement.py index 4a5f91ab23f..96969335c30 100755 --- a/src/DIRAC/Resources/Computing/InProcessComputingElement.py +++ b/src/DIRAC/Resources/Computing/InProcessComputingElement.py @@ -22,6 +22,7 @@ def __init__(self, ceUniqueID): self.runningJobs = 0 self.processors = int(self.ceParameters.get("NumberOfProcessors", 1)) + self.maxRAM = int(self.ceParameters.get("MaxRAM", 0)) self.ceParameters["MaxTotalJobs"] = 1 def submitJob(self, executableFile, proxy=None, inputs=None, **kwargs): @@ -118,4 +119,6 @@ def getCEStatus(self): result["WaitingJobs"] = 0 # processors result["AvailableProcessors"] = self.processors + # RAM + result["AvailableRAM"] = self.maxRAM return result diff --git a/src/DIRAC/Resources/Computing/PoolComputingElement.py b/src/DIRAC/Resources/Computing/PoolComputingElement.py index 231136d995b..fed710d4433 100644 --- a/src/DIRAC/Resources/Computing/PoolComputingElement.py +++ b/src/DIRAC/Resources/Computing/PoolComputingElement.py @@ -1,4 +1,4 @@ -""" The Pool Computing Element is an "inner" CE (meaning it's used by a jobAgent inside a pilot) +"""The Pool Computing Element is an "inner" CE (meaning it's used by a jobAgent inside a pilot) It's used running several jobs simultaneously in separate processes, managed by a ProcessPool. @@ -10,7 +10,7 @@ LocalCEType = Pool The Pool Computing Element is specific: it embeds an additional "inner" CE - (`InProcess` by default, `Sudo`, `Singularity`). The "inner" CE can be specified such as:: + (`InProcess` by default, or `Singularity`). The "inner" CE can be specified such as:: LocalCEType = Pool/Singularity @@ -19,24 +19,19 @@ **Code Documentation** """ -import functools -import os + import concurrent.futures +import functools -from DIRAC import S_OK, S_ERROR +from DIRAC import S_ERROR, S_OK from DIRAC.ConfigurationSystem.private.ConfigurationData import ConfigurationData - from DIRAC.Resources.Computing.ComputingElement import ComputingElement - from DIRAC.Resources.Computing.InProcessComputingElement import InProcessComputingElement from DIRAC.Resources.Computing.SingularityComputingElement import SingularityComputingElement -# Number of unix users to run job payloads with sudo -MAX_NUMBER_OF_SUDO_UNIX_USERS = 32 - -def executeJob(executableFile, proxy, taskID, inputs, **kwargs): - """wrapper around ce.submitJob: decides which CE to use (Sudo or InProcess or Singularity) +def executeJob(executableFile, proxy, taskID, inputs, innerCEParameters, **kwargs): + """wrapper around ce.submitJob: decides which inner CE to use (InProcess or Singularity) :param str executableFile: location of the executable file :param str proxy: proxy file location to be used for job submission @@ -52,6 +47,9 @@ def executeJob(executableFile, proxy, taskID, inputs, **kwargs): else: ce = InProcessComputingElement("Task-" + str(taskID)) + # adding the number of processors to use and the RAM + ce.ceParameters["NumberOfProcessors"] = innerCEParameters["NumberOfProcessors"] + ce.ceParameters["MaxRAM"] = innerCEParameters["MaxRAM"] return ce.submitJob(executableFile, proxy, inputs=inputs, **kwargs) @@ -66,7 +64,11 @@ def __init__(self, ceUniqueID): self.pPool = None self.taskID = 0 self.processorsPerTask = {} - self.userNumberPerTask = {} + self.ram = ( + 1024 + * 1024 # Available RAM for the node, in MB. The default value is an arbitrary large value in case of no limit + ) + self.ramPerTask = {} # This CE will effectively submit to another "Inner"CE # (by default to the InProcess CE) @@ -80,22 +82,14 @@ def _reset(self): self.processors = int(self.ceParameters.get("NumberOfProcessors", self.processors)) self.ceParameters["MaxTotalJobs"] = self.processors + if self.ceParameters.get("MaxRAM", 0): + self.ram = int(self.ceParameters["MaxRAM"]) # Indicates that the submission is done asynchronously # The result is not immediately available self.ceParameters["AsyncSubmission"] = True self.innerCESubmissionType = self.ceParameters.get("InnerCESubmissionType", self.innerCESubmissionType) return S_OK() - def getProcessorsInUse(self): - """Get the number of currently allocated processor cores - - :return: number of processors in use - """ - processorsInUse = 0 - for future in self.processorsPerTask: - processorsInUse += self.processorsPerTask[future] - return processorsInUse - ############################################################################# def submitJob(self, executableFile, proxy=None, inputs=None, **kwargs): """Method to submit job. @@ -118,33 +112,41 @@ def submitJob(self, executableFile, proxy=None, inputs=None, **kwargs): self.taskID += 1 return S_OK(taskID) - # Now persisting the job limits for later use in pilot.cfg file (pilot 3 default) + memoryForJob = self._getMemoryForJobs(kwargs) + if memoryForJob is None: + self.taskResults[self.taskID] = S_ERROR("Not enough memory for the job") + taskID = self.taskID + self.taskID += 1 + return S_OK(taskID) + + # Now persisting the job limits for later use in pilot.cfg file cd = ConfigurationData(loadDefaultCFG=False) res = cd.loadFile("pilot.cfg") if not res["OK"]: self.log.error("Could not load pilot.cfg", res["Message"]) else: - # only NumberOfProcessors for now, but RAM (or other stuff) can also be added jobID = int(kwargs.get("jobDesc", {}).get("jobID", 0)) cd.setOptionInCFG("/Resources/Computing/JobLimits/%d/NumberOfProcessors" % jobID, processorsForJob) + cd.setOptionInCFG("/Resources/Computing/JobLimits/%d/MaxRAM" % jobID, memoryForJob) res = cd.dumpLocalCFGToFile("pilot.cfg") if not res["OK"]: self.log.error("Could not dump cfg to pilot.cfg", res["Message"]) + # Define the innerCEParameters + innerCEParameters = {} + innerCEParameters["NumberOfProcessors"] = processorsForJob + innerCEParameters["MaxRAM"] = memoryForJob + # Here we define task kwargs: adding complex objects like thread.Lock can trigger errors in the task taskKwargs = {"InnerCESubmissionType": self.innerCESubmissionType} taskKwargs["jobDesc"] = kwargs.get("jobDesc", {}) - if self.innerCESubmissionType == "Sudo": - for nUser in range(MAX_NUMBER_OF_SUDO_UNIX_USERS): - if nUser not in self.userNumberPerTask.values(): - break - taskKwargs["NUser"] = nUser - if "USER" in os.environ: - taskKwargs["PayloadUser"] = os.environ["USER"] + f"p{str(nUser).zfill(2)}" # Submission - future = self.pPool.submit(executeJob, executableFile, proxy, self.taskID, inputs, **taskKwargs) + future = self.pPool.submit( + executeJob, executableFile, proxy, self.taskID, inputs, innerCEParameters, **taskKwargs + ) self.processorsPerTask[future] = processorsForJob + self.ramPerTask[future] = memoryForJob future.add_done_callback(functools.partial(self.finalizeJob, self.taskID)) taskID = self.taskID @@ -154,7 +156,7 @@ def submitJob(self, executableFile, proxy=None, inputs=None, **kwargs): def _getProcessorsForJobs(self, kwargs): """helper function""" - processorsInUse = self.getProcessorsInUse() + processorsInUse = sum(self.processorsPerTask.values()) availableProcessors = self.processors - processorsInUse self.log.verbose( @@ -191,29 +193,58 @@ def _getProcessorsForJobs(self, kwargs): return requestedProcessors + def _getMemoryForJobs(self, kwargs): + """helper function to get the memory that will be allocated for the job + + :param kwargs: job parameters + :return: memory in MB or None if not enough memory + """ + + # # job requirements + requestedMemory = kwargs.get("MinRAM", 0) + + # # now check what the slot can provide + # Do we have enough memory? + availableMemory = self.ram - sum(self.ramPerTask.values()) + if availableMemory < requestedMemory: + return None + + # if there's a MaxRAM requested, we allocate it all (if it fits), + # and if it doesn't, we stop here instead of using MinRAM + if kwargs.get("MaxRAM", 0): + if availableMemory >= kwargs.get("MaxRAM"): + requestedMemory = kwargs.get("MaxRAM") + else: + return None + + return requestedMemory + def finalizeJob(self, taskID, future): """Finalize the job by updating the process utilisation counters :param future: evaluating the future result """ nProc = self.processorsPerTask.pop(future) + ram = self.ramPerTask.pop(future, None) result = future.result() # This would be the result of the e.g. InProcess.submitJob() if result["OK"]: - self.log.info("Task finished successfully:", f"{taskID}; {nProc} processor(s) freed") + self.log.info("Task finished successfully:", f"{taskID}; {nProc} processor(s) and {ram}MB freed") else: self.log.error("Task failed submission:", f"{taskID}; message: {result['Message']}") self.taskResults[taskID] = result def getCEStatus(self): """Method to return information on running and waiting jobs, - as well as the number of processors (used, and available). + as well as the number of processors (used, and available), + and the RAM (used, and available) :return: dictionary of numbers of jobs per status and processors (used, and available) """ result = S_OK() nJobs = 0 + for _j, value in self.processorsPerTask.items(): if value > 0: nJobs += 1 @@ -222,9 +253,13 @@ def getCEStatus(self): result["WaitingJobs"] = 0 # dealing with processors - processorsInUse = self.getProcessorsInUse() + processorsInUse = sum(self.processorsPerTask.values()) result["UsedProcessors"] = processorsInUse result["AvailableProcessors"] = self.processors - processorsInUse + # dealing with RAM + result["UsedRAM"] = sum(self.ramPerTask.values()) + result["AvailableRAM"] = self.ram - sum(self.ramPerTask.values()) + return result def getDescription(self): diff --git a/src/DIRAC/Resources/Computing/SingularityComputingElement.py b/src/DIRAC/Resources/Computing/SingularityComputingElement.py index 12c66badca5..d8bb1cf4b99 100644 --- a/src/DIRAC/Resources/Computing/SingularityComputingElement.py +++ b/src/DIRAC/Resources/Computing/SingularityComputingElement.py @@ -1,16 +1,17 @@ -""" SingularityCE is a type of "inner" CEs - (meaning it's used by a jobAgent inside a pilot). - A computing element class using singularity containers, - where Singularity is supposed to be found on the WN. +"""SingularityCE is a type of "inner" CEs +(meaning it's used by a jobAgent inside a pilot). +A computing element class using singularity containers, +where Singularity is supposed to be found on the WN. - The goal of this CE is to start the job in the container set by - the "ContainerRoot" config option. +The goal of this CE is to start the job in the container set by +the "ContainerRoot" config option. - DIRAC can be re-installed within the container. +DIRAC can be re-installed within the container. - See the Configuration/Resources/Computing documention for details on - where to set the option parameters. +See the Configuration/Resources/Computing documention for details on +where to set the option parameters. """ + import json import os import re @@ -115,6 +116,7 @@ def __init__(self, ceUniqueID): self.__installDIRACInContainer = False self.processors = int(self.ceParameters.get("NumberOfProcessors", 1)) + self.maxRAM = int(self.ceParameters.get("MaxRAM", 0)) @staticmethod def __findInstallBaseDir(): @@ -415,6 +417,12 @@ def submitJob(self, executableFile, proxy=None, **kwargs): self.log.debug(f"Execute singularity command: {cmd}") self.log.debug(f"Execute singularity env: {self.__getEnv()}") + # systemCall below uses ceParameters["MemoryLimitMB"] as CG2 upper memory limit + # if there's a max RAM available to the job, use that + if self.maxRAM: + self.ceParameters["MemoryLimitMB"] = min( + self.maxRAM, self.ceParameters.get("MemoryLimitMB", 1024 * 1024) + ) # 1024 * 1024 is an arbitrary large number result = CG2Manager().systemCall( 0, cmd, callbackFunction=self.sendOutput, env=self.__getEnv(), ceParameters=self.ceParameters ) @@ -451,4 +459,6 @@ def getCEStatus(self): result["WaitingJobs"] = 0 # processors result["AvailableProcessors"] = self.processors + # RAM + result["AvailableRAM"] = self.maxRAM return result diff --git a/src/DIRAC/Resources/Computing/test/Test_InProcessComputingElement.py b/src/DIRAC/Resources/Computing/test/Test_InProcessComputingElement.py index 282b9197dfc..3b8be05700a 100644 --- a/src/DIRAC/Resources/Computing/test/Test_InProcessComputingElement.py +++ b/src/DIRAC/Resources/Computing/test/Test_InProcessComputingElement.py @@ -54,6 +54,8 @@ def test_submitJob(): maxNumberOfProcessors=8, wholeNode=False, mpTag=True, + MinRAM=2500, + MaxRAM=4000, jobDesc={"jobParams": jobParams, "resourceParams": resourceParams, "optimizerParams": optimizerParams}, ) assert res["OK"] is True diff --git a/src/DIRAC/Resources/Computing/test/Test_PoolComputingElement.py b/src/DIRAC/Resources/Computing/test/Test_PoolComputingElement.py index a630b9bcdeb..fc18fcd497d 100644 --- a/src/DIRAC/Resources/Computing/test/Test_PoolComputingElement.py +++ b/src/DIRAC/Resources/Computing/test/Test_PoolComputingElement.py @@ -2,6 +2,7 @@ """ tests for PoolComputingElement module """ + import os import time @@ -50,7 +51,7 @@ def _stopJob(nJob): @pytest.fixture def createAndDelete(): - for i in range(6): + for i in range(9): with open(f"testPoolCEJob_{i}.py", "w") as execFile: execFile.write(jobScript % i) os.chmod(f"testPoolCEJob_{i}.py", 0o755) @@ -66,24 +67,28 @@ def createAndDelete(): time.sleep(0.5) # stopping the jobs - for i in range(6): + for i in range(9): _stopJob(i) # removing testPoolCEJob files # this will also stop the futures unless they are already stopped! - for i in range(6): + for i in range(9): try: os.remove(f"testPoolCEJob_{i}.py") - os.remove("testBadPoolCEJob.py") except OSError: pass + try: + os.remove("testBadPoolCEJob.py") + except OSError: + pass + @pytest.mark.slow def test_submit_and_shutdown(createAndDelete): time.sleep(0.5) - ceParameters = {"WholeNode": True, "NumberOfProcessors": 4} + ceParameters = {"WholeNode": True, "NumberOfProcessors": 4, "MaxRAM": 3800} ce = PoolComputingElement("TestPoolCE") ce.setParameters(ceParameters) @@ -145,7 +150,7 @@ def test_executeJob_wholeNode4(createAndDelete): time.sleep(0.5) taskIDs = {} - ceParameters = {"WholeNode": True, "NumberOfProcessors": 4} + ceParameters = {"WholeNode": True, "NumberOfProcessors": 4, "MaxRAM": 16000} ce = PoolComputingElement("TestPoolCE") ce.setParameters(ceParameters) @@ -159,9 +164,11 @@ def test_executeJob_wholeNode4(createAndDelete): result = ce.getCEStatus() assert result["UsedProcessors"] == 1 assert result["AvailableProcessors"] == 3 + assert result["UsedRAM"] == 0 + assert result["AvailableRAM"] == 16000 assert result["RunningJobs"] == 1 - jobParams = {"mpTag": True, "numberOfProcessors": 2} + jobParams = {"mpTag": True, "numberOfProcessors": 2, "MaxRAM": 4000} result = ce.submitJob("testPoolCEJob_1.py", None, **jobParams) assert result["OK"] is True taskID = result["Value"] @@ -171,6 +178,9 @@ def test_executeJob_wholeNode4(createAndDelete): result = ce.getCEStatus() assert result["UsedProcessors"] == 3 assert result["AvailableProcessors"] == 1 + assert result["UsedRAM"] == 4000 + assert result["AvailableRAM"] == 12000 + assert result["RunningJobs"] == 2 # now trying again would fail @@ -194,7 +204,7 @@ def test_executeJob_wholeNode8(createAndDelete): time.sleep(0.5) taskIDs = {} - ceParameters = {"WholeNode": True, "NumberOfProcessors": 8} + ceParameters = {"WholeNode": True, "NumberOfProcessors": 8, "MaxRAM": 32000} ce = PoolComputingElement("TestPoolCE") ce.setParameters(ceParameters) @@ -217,8 +227,10 @@ def test_executeJob_wholeNode8(createAndDelete): result = ce.getCEStatus() assert result["UsedProcessors"] == 5 + assert result["UsedRAM"] == 0 + assert result["AvailableRAM"] == 32000 - jobParams = {"numberOfProcessors": 2} # This is same as asking for SP + jobParams = {"numberOfProcessors": 2, "MinRAM": 4000, "MaxRAM": 8000} # This is same as asking for SP result = ce.submitJob("testPoolCEJob_4.py", None, **jobParams) assert result["OK"] is True taskID = result["Value"] @@ -227,39 +239,65 @@ def test_executeJob_wholeNode8(createAndDelete): result = ce.getCEStatus() assert result["UsedProcessors"] == 6 + assert result["UsedRAM"] == 8000 + assert result["AvailableRAM"] == 24000 - # now trying again would fail - jobParams = {"mpTag": True, "numberOfProcessors": 3} + jobParams = {"MinRAM": 8000, "MaxRAM": 8000} # This is same as asking for SP result = ce.submitJob("testPoolCEJob_5.py", None, **jobParams) assert result["OK"] is True taskID = result["Value"] assert taskID == 3 + taskIDs[taskID] = True + + result = ce.getCEStatus() + assert result["UsedProcessors"] == 7 + assert result["UsedRAM"] == 16000 + assert result["AvailableRAM"] == 16000 + + jobParams = {"MaxRAM": 24000} # This will fail + result = ce.submitJob("testPoolCEJob_6.py", None, **jobParams) + assert result["OK"] is True + taskID = result["Value"] + assert taskID == 4 + taskIDs[taskID] = False + + result = ce.getCEStatus() + assert result["UsedProcessors"] == 7 + assert result["UsedRAM"] == 16000 + assert result["AvailableRAM"] == 16000 + + # now trying again would fail + jobParams = {"mpTag": True, "numberOfProcessors": 3} + result = ce.submitJob("testPoolCEJob_7.py", None, **jobParams) + assert result["OK"] is True + taskID = result["Value"] + assert taskID == 5 taskIDs[taskID] = False # waiting and submit again while len(ce.taskResults) < 2: time.sleep(0.1) - jobParams = {"mpTag": True, "numberOfProcessors": 3} - result = ce.submitJob("testPoolCEJob_5.py", None, **jobParams) + jobParams = {"mpTag": True, "numberOfProcessors": 1} + result = ce.submitJob("testPoolCEJob_8.py", None, **jobParams) assert result["OK"] is True taskID = result["Value"] - assert taskID == 4 + assert taskID == 6 taskIDs[taskID] = True result = ce.shutdown() assert result["OK"] is True assert isinstance(result["Value"], dict) - assert len(result["Value"]) == 5 + assert len(result["Value"]) == 7 - while len(ce.taskResults) < 5: + while len(ce.taskResults) < 7: time.sleep(0.1) for taskID, expectedResult in taskIDs.items(): submissionResult = ce.taskResults[taskID] assert submissionResult["OK"] is expectedResult if not submissionResult["OK"]: - assert "Not enough processors" in submissionResult["Message"] + assert submissionResult["Message"] in ["Not enough processors for the job", "Not enough memory for the job"] @pytest.mark.slow @@ -372,28 +410,41 @@ def test_executeJob_WholeNodeJobs(createAndDelete): @pytest.mark.parametrize( - "processorsPerTask, kwargs, expected", + "processorsPerTask, ramPerTask, kwargs, expected_processors, expected_memory", [ - (None, {}, 1), - (None, {"mpTag": False}, 1), - (None, {"mpTag": True}, 1), - (None, {"mpTag": True, "wholeNode": True}, 16), - (None, {"mpTag": True, "wholeNode": False}, 1), - (None, {"mpTag": True, "numberOfProcessors": 4}, 4), - (None, {"mpTag": True, "numberOfProcessors": 4, "maxNumberOfProcessors": 8}, 8), - (None, {"mpTag": True, "numberOfProcessors": 4, "maxNumberOfProcessors": 32}, 16), - ({1: 4}, {"mpTag": True, "wholeNode": True}, 0), - ({1: 4}, {"mpTag": True, "wholeNode": False}, 1), - ({1: 4}, {"mpTag": True, "numberOfProcessors": 2}, 2), - ({1: 4}, {"mpTag": True, "maxNumberOfProcessors": 2}, 2), - ({1: 4}, {"mpTag": True, "maxNumberOfProcessors": 16}, 12), + (None, None, {}, 1, 0), + (None, None, {"mpTag": False}, 1, 0), + (None, None, {"mpTag": True, "MaxRAM": 8000}, 1, 8000), + (None, None, {"mpTag": True, "wholeNode": True}, 16, 0), + (None, None, {"mpTag": True, "wholeNode": False}, 1, 0), + (None, None, {"mpTag": True, "numberOfProcessors": 4, "MinRAM": 2000}, 4, 2000), + (None, None, {"mpTag": True, "numberOfProcessors": 4, "MaxRAM": 4000}, 4, 4000), + (None, None, {"mpTag": True, "numberOfProcessors": 4, "MaxRAM": 36000}, 4, None), + (None, None, {"mpTag": True, "numberOfProcessors": 4, "MinRAM": 2000, "MaxRAM": 4000}, 4, 4000), + (None, None, {"mpTag": True, "numberOfProcessors": 4, "maxNumberOfProcessors": 8}, 8, 0), + (None, None, {"mpTag": True, "numberOfProcessors": 4, "maxNumberOfProcessors": 32}, 16, 0), + ({1: 4}, {1: 4000}, {"mpTag": True, "wholeNode": True}, 0, 0), + ({1: 4}, {1: 4000}, {"mpTag": True, "wholeNode": False}, 1, 0), + ({1: 4}, {1: 4000}, {"mpTag": True, "numberOfProcessors": 2, "MinRAM": 8000}, 2, 8000), + ({1: 4}, {1: 4000}, {"mpTag": True, "numberOfProcessors": 16, "MinRAM": 8000, "MaxRAM": 12000}, 0, 12000), + ({1: 4}, {1: 4000}, {"mpTag": True, "maxNumberOfProcessors": 2, "MaxRAM": 16000}, 2, 16000), + ({1: 4}, {1: 4000}, {"mpTag": True, "numberOfProcessors": 2, "MaxRAM": 8000}, 2, 8000), + ({1: 4}, {1: 4000}, {"mpTag": True, "maxNumberOfProcessors": 16, "MaxRAM": 32000}, 12, None), + ({1: 4, 2: 8}, {1: 4000}, {"mpTag": True, "numberOfProcessors": 2}, 2, 0), + ({1: 4, 2: 8}, {1: 4000}, {"mpTag": True, "numberOfProcessors": 4}, 4, 0), + ({1: 4, 2: 8, 3: 8}, {1: 4000}, {"mpTag": True, "numberOfProcessors": 4}, 0, 0), ], ) -def test__getProcessorsForJobs(processorsPerTask, kwargs, expected): +def test__getLimitsForJobs(processorsPerTask, ramPerTask, kwargs, expected_processors, expected_memory): ce = PoolComputingElement("TestPoolCE") ce.processors = 16 + ce.ram = 32000 if processorsPerTask: ce.processorsPerTask = processorsPerTask + if ramPerTask: + ce.ramPerTask = ramPerTask res = ce._getProcessorsForJobs(kwargs) - assert res == expected + assert res == expected_processors + res = ce._getMemoryForJobs(kwargs) + assert res == expected_memory diff --git a/src/DIRAC/WorkloadManagementSystem/Client/Matcher.py b/src/DIRAC/WorkloadManagementSystem/Client/Matcher.py index 3eb1c19bf89..ec01894a7ad 100644 --- a/src/DIRAC/WorkloadManagementSystem/Client/Matcher.py +++ b/src/DIRAC/WorkloadManagementSystem/Client/Matcher.py @@ -1,7 +1,8 @@ -""" Encapsulate here the logic for matching jobs +"""Encapsulate here the logic for matching jobs - Utilities and classes here are used by MatcherHandler +Utilities and classes here are used by MatcherHandler """ + import time from DIRAC import convertToPy3VersionNumber, gLogger @@ -75,7 +76,7 @@ def selectJob(self, resourceDescription, credDict): toPrintDict["Tag"] = [] if "Tag" in resourceDict: for tag in resourceDict["Tag"]: - if not tag.endswith("GB") and not tag.endswith("Processors"): + if not tag.endswith("MB") and not tag.endswith("Processors"): toPrintDict["Tag"].append(tag) if not toPrintDict["Tag"]: toPrintDict.pop("Tag") @@ -195,7 +196,7 @@ def _processResourceDescription(self, resourceDescription): maxRAM = resourceDescription.get("MaxRAM") if maxRAM: try: - maxRAM = int(maxRAM / 1000) + maxRAM = int(maxRAM) except ValueError: maxRAM = None nProcessors = resourceDescription.get("NumberOfProcessors") @@ -204,9 +205,9 @@ def _processResourceDescription(self, resourceDescription): nProcessors = int(nProcessors) except ValueError: nProcessors = None - for param, key in [(maxRAM, "GB"), (nProcessors, "Processors")]: - if param and param <= 1024: - paramList = list(range(2, param + 1)) + for param, key, limit, increment in [(maxRAM, "MB", 1024 * 1024, 256), (nProcessors, "Processors", 1024, 1)]: + if param and param <= limit: + paramList = list(range(increment, param + increment, increment)) paramTags = ["%d%s" % (par, key) for par in paramList] if paramTags: resourceDict.setdefault("Tag", []).extend(paramTags) diff --git a/src/DIRAC/WorkloadManagementSystem/Executor/JobScheduling.py b/src/DIRAC/WorkloadManagementSystem/Executor/JobScheduling.py index 6e711add055..ea05dbb4a9a 100755 --- a/src/DIRAC/WorkloadManagementSystem/Executor/JobScheduling.py +++ b/src/DIRAC/WorkloadManagementSystem/Executor/JobScheduling.py @@ -1,12 +1,12 @@ -""" The Job Scheduling Executor takes the information gained from all previous - optimizers and makes a scheduling decision for the jobs. +"""The Job Scheduling Executor takes the information gained from all previous +optimizers and makes a scheduling decision for the jobs. - Subsequent to this jobs are added into a Task Queue and pilot agents can be submitted. +Subsequent to this jobs are added into a Task Queue and pilot agents can be submitted. - All issues preventing the successful resolution of a site candidate are discovered - here where all information is available. +All issues preventing the successful resolution of a site candidate are discovered +here where all information is available. - This Executor will fail affected jobs meaningfully. +This Executor will fail affected jobs meaningfully. """ import random @@ -355,7 +355,7 @@ def _getTagsFromManifest(self, jobManifest): if "MaxRAM" in jobManifest: maxRAM = jobManifest.getOption("MaxRAM", 0) if maxRAM: - tagList.append("%dGB" % maxRAM) + tagList.append(f"{maxRAM}MB") # other tags? Just add them if "Tags" in jobManifest: diff --git a/src/DIRAC/WorkloadManagementSystem/Executor/test/Test_Executor.py b/src/DIRAC/WorkloadManagementSystem/Executor/test/Test_Executor.py index 2f5d56501fb..732cbffc6ff 100644 --- a/src/DIRAC/WorkloadManagementSystem/Executor/test/Test_Executor.py +++ b/src/DIRAC/WorkloadManagementSystem/Executor/test/Test_Executor.py @@ -1,5 +1,5 @@ -""" pytest(s) for Executors -""" +"""pytest(s) for Executors""" + # pylint: disable=protected-access, missing-docstring from unittest.mock import MagicMock @@ -54,9 +54,9 @@ def test__applySiteFilter(sites, banned, expected): ({}, []), ({"Tag": "bof"}, ["bof"]), ({"Tags": "bof, bif"}, ["bof", "bif"]), - ({"MaxRAM": 2}, ["2GB"]), - ({"Tags": "bof, bif", "MaxRAM": 2}, ["bof", "bif", "2GB"]), - ({"WholeNode": "yes", "MaxRAM": 2}, ["WholeNode", "MultiProcessor", "2GB"]), + ({"MaxRAM": 2500}, ["2500MB"]), + ({"Tags": "bof, bif", "MaxRAM": 2048}, ["bof", "bif", "2048MB"]), + ({"WholeNode": "yes", "MaxRAM": 2048}, ["WholeNode", "MultiProcessor", "2048MB"]), ({"NumberOfProcessors": 1}, []), ({"NumberOfProcessors": 4}, ["MultiProcessor", "4Processors"]), ({"NumberOfProcessors": 4, "MinNumberOfProcessors": 2}, ["MultiProcessor", "4Processors"]), diff --git a/src/DIRAC/WorkloadManagementSystem/Utilities/JobParameters.py b/src/DIRAC/WorkloadManagementSystem/Utilities/JobParameters.py index eeaf985523e..e86e429c3e6 100644 --- a/src/DIRAC/WorkloadManagementSystem/Utilities/JobParameters.py +++ b/src/DIRAC/WorkloadManagementSystem/Utilities/JobParameters.py @@ -10,7 +10,7 @@ def getMemoryFromProc(): meminfo = {i.split()[0].rstrip(":"): int(i.split()[1]) for i in open("/proc/meminfo").readlines()} maxRAM = meminfo["MemTotal"] if maxRAM: - return int(maxRAM / 1024) + return int(maxRAM / 1024) # from KB to MB def getNumberOfProcessors(siteName=None, gridCE=None, queue=None): @@ -57,7 +57,7 @@ def getNumberOfProcessors(siteName=None, gridCE=None, queue=None): return numberOfProcessors # 3) looks in CS for tags - gLogger.info(f"Getting tagsfor {siteName}: {gridCE}: {queue}") + gLogger.info(f"Getting tags for {siteName}: {gridCE}: {queue}") # Tags of the CE tags = fromChar( gConfig.getValue(f"/Resources/Sites/{siteName.split('.')[0]}/{siteName}/CEs/{gridCE}/Tag", "") @@ -201,3 +201,82 @@ def getJobParameters(jobIDs: list[int], parName: str | None, vo: str = "") -> di if jobID not in final: final[jobID] = parameters[jobID] return S_OK(final) + + +def getAvailableRAM(siteName=None, gridCE=None, queue=None): + """Gets the available RAM on a certain CE/queue/node (what the pilot administers) + + The siteName/gridCE/queue parameters are normally not necessary. + + Tries to find it in this order: + 1) from the /Resources/Computing/CEDefaults/MaxRAM (which is what the pilot might fill up) + 2) if not present looks in CS for "MemoryLimitMB" Queue or CE or site option + 3) if not present but there's WholeNode tag, look what the WN provides using _getMemoryFromProc() + 4) return 0 + """ + + # 1) from /Resources/Computing/CEDefaults/MaxRAM + gLogger.info("Getting MaxRAM from /Resources/Computing/CEDefaults/MaxRAM") + availableRAM = gConfig.getValue("/Resources/Computing/CEDefaults/MaxRAM", None) + if availableRAM: + return availableRAM + + # 2) looks in CS for "MaxRAM" Queue or CE or site option + if not siteName: + siteName = gConfig.getValue("/LocalSite/Site", "") + if not gridCE: + gridCE = gConfig.getValue("/LocalSite/GridCE", "") + if not queue: + queue = gConfig.getValue("/LocalSite/CEQueue", "") + if not (siteName and gridCE and queue): + gLogger.error("Could not find AvailableRAM: missing siteName or gridCE or queue. Returning 0") + return 0 + + grid = siteName.split(".")[0] + csPaths = [ + f"/Resources/Sites/{grid}/{siteName}/CEs/{gridCE}/Queues/{queue}/MemoryLimitMB", + f"/Resources/Sites/{grid}/{siteName}/CEs/{gridCE}/MemoryLimitMB", + f"/Resources/Sites/{grid}/{siteName}/MemoryLimitMB", + ] + for csPath in csPaths: + gLogger.info("Looking in", csPath) + availableRAM = gConfig.getValue(csPath, None) + if availableRAM: + return availableRAM + + # 3) checks if 'WholeNode' is one of the used tags + # Tags of the CE + tags = fromChar( + gConfig.getValue(f"/Resources/Sites/{siteName.split('.')[0]}/{siteName}/CEs/{gridCE}/Tag", "") + ) + fromChar(gConfig.getValue(f"/Resources/Sites/{siteName.split('.')[0]}/{siteName}/Cloud/{gridCE}/Tag", "")) + # Tags of the Queue + tags += fromChar( + gConfig.getValue(f"/Resources/Sites/{siteName.split('.')[0]}/{siteName}/CEs/{gridCE}/Queues/{queue}/Tag", "") + ) + fromChar( + gConfig.getValue(f"/Resources/Sites/{siteName.split('.')[0]}/{siteName}/Cloud/{gridCE}/VMTypes/{queue}/Tag", "") + ) + + if "WholeNode" in tags: + gLogger.info("Found WholeNode tag, using getMemoryFromProc()") + return getMemoryFromProc() + + # 4) return 0 + gLogger.info("RAM limits could not be found in CS, and WholeNode tag not found") + return 0 + + +def getRAMForJob(jobID): + """Gets the RAM allowed for the job. + This can be used to communicate to your job payload the RAM it's allowed to use, + so this function should be called from your extension. + + If the JobAgent is using "InProcess" CE (which is the default), + then what's returned will basically be the same of what's returned by the getAvailableRAM() function above + """ + + # from /Resources/Computing/JobLimits/jobID/MaxRAM (set by PoolComputingElement) + ram = gConfig.getValue(f"Resources/Computing/JobLimits/{jobID}/MaxRAM") + if ram: + return ram + + return getAvailableRAM() diff --git a/src/DIRAC/WorkloadManagementSystem/Utilities/QueueUtilities.py b/src/DIRAC/WorkloadManagementSystem/Utilities/QueueUtilities.py index 0aa1c34b5ea..58311f3d3b3 100644 --- a/src/DIRAC/WorkloadManagementSystem/Utilities/QueueUtilities.py +++ b/src/DIRAC/WorkloadManagementSystem/Utilities/QueueUtilities.py @@ -1,5 +1,5 @@ -"""Utilities to help Computing Element Queues manipulation -""" +"""Utilities to help Computing Element Queues manipulation""" + import hashlib from DIRAC import S_OK, S_ERROR @@ -222,10 +222,10 @@ def matchQueue(jobJDL, queueDict, fullMatch=False): return S_OK({"Match": False, "Reason": noMatchReasons[0]}) # 5. RAM - ram = job.getAttributeInt("RAM") + ram = job.getAttributeInt("MaxRAM") # If MaxRAM is not specified in the queue description, assume 2GB - if ram and ram > int(queueDict.get("MaxRAM", 2048) / 1024): - noMatchReasons.append("Job RAM %d requirement not satisfied" % ram) + if ram and ram > int(queueDict.get("MaxRAM", 2048)): + noMatchReasons.append(f"Job RAM {ram} requirement not satisfied") if not fullMatch: return S_OK({"Match": False, "Reason": noMatchReasons[0]}) diff --git a/src/DIRAC/WorkloadManagementSystem/Utilities/RemoteRunner.py b/src/DIRAC/WorkloadManagementSystem/Utilities/RemoteRunner.py index 39d5f2d4a26..e76f474509d 100644 --- a/src/DIRAC/WorkloadManagementSystem/Utilities/RemoteRunner.py +++ b/src/DIRAC/WorkloadManagementSystem/Utilities/RemoteRunner.py @@ -1,4 +1,4 @@ -""" RemoteRunner +"""RemoteRunner RemoteRunner has been designed to send scripts/applications and input files on remote worker nodes having no outbound connectivity (e.g. supercomputers) @@ -6,6 +6,7 @@ Mostly called by workflow modules, RemoteRunner is generally the last component to get through before the script/application execution on a remote machine. """ + import hashlib import os import shlex diff --git a/src/DIRAC/WorkloadManagementSystem/Utilities/test/Test_RemoteRunner.py b/src/DIRAC/WorkloadManagementSystem/Utilities/test/Test_RemoteRunner.py index 2ba26fa1c40..18e334f968a 100644 --- a/src/DIRAC/WorkloadManagementSystem/Utilities/test/Test_RemoteRunner.py +++ b/src/DIRAC/WorkloadManagementSystem/Utilities/test/Test_RemoteRunner.py @@ -1,5 +1,4 @@ -""" Test class for Job Agent -""" +"""Test class for Job Agent""" # imports import pytest @@ -60,11 +59,16 @@ def test__wrapCommand(command, workingDirectory, expectedContent): (1, 1, True, 1), (2, 2, True, 2), (1, 2, True, 1), + ( + 1, + 0, + True, + 1, + ), # if ceNumberOfProcessors is 0, it will be interpreted as needing local evaluation. That will return 1. # CE has less processors than the payload requests (2, 1, False, "Not enough processors to execute the command"), # Specific case: we should not have 0 (0, 1, False, "Inappropriate NumberOfProcessors value"), - (1, 0, False, "Inappropriate NumberOfProcessors value"), (-4, 1, False, "Inappropriate NumberOfProcessors value"), (1, -4, False, "Inappropriate NumberOfProcessors value"), (0, 0, False, "Inappropriate NumberOfProcessors value"), diff --git a/src/DIRAC/WorkloadManagementSystem/scripts/dirac_wms_get_wn_parameters.py b/src/DIRAC/WorkloadManagementSystem/scripts/dirac_wms_get_wn_parameters.py index 890ee15b218..7badcf96c92 100755 --- a/src/DIRAC/WorkloadManagementSystem/scripts/dirac_wms_get_wn_parameters.py +++ b/src/DIRAC/WorkloadManagementSystem/scripts/dirac_wms_get_wn_parameters.py @@ -39,12 +39,12 @@ def main(): gLogger.info("Getting number of processors") numberOfProcessor = JobParameters.getNumberOfProcessors(Site, ceName, Queue) - gLogger.info("Getting memory (RAM)") - maxRAM = JobParameters.getMemoryFromProc() - gLogger.info("Getting number of GPUs") numberOfGPUs = JobParameters.getNumberOfGPUs(Site, ceName, Queue) + gLogger.info("Getting maximum RAM") + maxRAM = JobParameters.getAvailableRAM(Site, ceName, Queue) + # just communicating it back gLogger.notice(" ".join(str(wnPar) for wnPar in [numberOfProcessor, maxRAM, numberOfGPUs])) diff --git a/src/DIRAC/tests/Utilities/testJobDefinitions.py b/src/DIRAC/tests/Utilities/testJobDefinitions.py index 65c5e3ad7c2..61064d4b9e3 100644 --- a/src/DIRAC/tests/Utilities/testJobDefinitions.py +++ b/src/DIRAC/tests/Utilities/testJobDefinitions.py @@ -1,5 +1,4 @@ -""" Collection of user jobs for testing purposes -""" +"""Collection of user jobs for testing purposes""" # pylint: disable=invalid-name @@ -290,9 +289,9 @@ def mpJob(): def mp3Job(): - """simple hello world job, with 2 to 4 processors""" + """simple hello world job, with 3 processors""" - J = baseToAllJobs("min2max4Job") + J = baseToAllJobs("min3Job") try: J.setInputSandbox([find_all("mpTest.py", rootPath, "DIRAC/tests/Utilities")[0]]) except IndexError: @@ -340,6 +339,43 @@ def wholeNodeJob(): return endOfAllJobs(J) +def memory_4GB(): + """simple hello world job, with a memory requirement of 4 GB and MultiProcessor tags""" + + J = baseToAllJobs("memory_4GB") + try: + J.setInputSandbox([find_all("mpTest.py", rootPath, "DIRAC/tests/Utilities")[0]]) + except IndexError: + try: + J.setInputSandbox([find_all("mpTest.py", ".", "DIRAC/tests/Utilities")[0]]) + except IndexError: # we are in Jenkins + J.setInputSandbox([find_all("mpTest.py", os.environ["WORKSPACE"], "DIRAC/tests/Utilities")[0]]) + + J.setExecutable("mpTest.py") + J.setNumberOfProcessors(numberOfProcessors=2) + J.setRAMRequirements(ramRequired=2500, maxRAM=4000) + + return endOfAllJobs(J) + + +def memory_2_to4GB(): + """simple hello world job, with a memory requirement of 2 to 4 GB and MultiProcessor tags""" + + J = baseToAllJobs("memory_2_to_4GB") + try: + J.setInputSandbox([find_all("mpTest.py", rootPath, "DIRAC/tests/Utilities")[0]]) + except IndexError: + try: + J.setInputSandbox([find_all("mpTest.py", ".", "DIRAC/tests/Utilities")[0]]) + except IndexError: # we are in Jenkins + J.setInputSandbox([find_all("mpTest.py", os.environ["WORKSPACE"], "DIRAC/tests/Utilities")[0]]) + + J.setExecutable("mpTest.py") + J.setNumberOfProcessors(numberOfProcessors=2) + J.setRAMRequirements(ramRequired=4000, maxRAM=4000) + return endOfAllJobs(J) + + def parametricJob(): """Creates a parametric job with 3 subjobs which are simple hello world jobs""" diff --git a/tests/Integration/Resources/Computing/Test_SingularityCE.py b/tests/Integration/Resources/Computing/Test_SingularityCE.py index 0bba47a5a8b..712ba8641a0 100644 --- a/tests/Integration/Resources/Computing/Test_SingularityCE.py +++ b/tests/Integration/Resources/Computing/Test_SingularityCE.py @@ -1,7 +1,7 @@ #!/bin/env python -""" This integration test is for "Inner" Computing Element SingularityComputingElement - This test is here and not in the unit tests because it requires singularity to be installed. +"""This integration test is for "Inner" Computing Element SingularityComputingElement +This test is here and not in the unit tests because it requires singularity to be installed. """ import os @@ -70,6 +70,8 @@ def test_submitJobWrapper(): maxNumberOfProcessors=8, wholeNode=False, mpTag=True, + MinRAM=2500, + MaxRAM=4000, jobDesc={"jobParams": jobParams, "resourceParams": resourceParams, "optimizerParams": optimizerParams}, ) diff --git a/tests/Integration/WorkloadManagementSystem/Test_TaskQueueDB.py b/tests/Integration/WorkloadManagementSystem/Test_TaskQueueDB.py index 367bf64a1d7..80c26faffd7 100644 --- a/tests/Integration/WorkloadManagementSystem/Test_TaskQueueDB.py +++ b/tests/Integration/WorkloadManagementSystem/Test_TaskQueueDB.py @@ -1,15 +1,16 @@ -""" This integration test only need the TaskQueueDB - (which should of course be properly defined also in the configuration), - and connects directly to it +"""This integration test only need the TaskQueueDB +(which should of course be properly defined also in the configuration), +and connects directly to it - Run this test with:: - "python -m pytest tests/Integration/WorkloadManagementSystem/Test_TaskQueueDB.py" +Run this test with:: + "python -m pytest tests/Integration/WorkloadManagementSystem/Test_TaskQueueDB.py" - Suggestion: for local testing, run this with:: - python -m pytest -c ../pytest.ini -vv tests/Integration/WorkloadManagementSystem/Test_TaskQueueDB.py +Suggestion: for local testing, run this with:: + python -m pytest -c ../pytest.ini -vv tests/Integration/WorkloadManagementSystem/Test_TaskQueueDB.py """ + import DIRAC from DIRAC import gLogger @@ -685,10 +686,6 @@ def test_chainWithTags(): # This is translated to "#Processors" by the SiteDirector result = tqDB.matchAndGetTaskQueue({"CPUTime": 50000, "Tag": "4Processors"}, numQueuesToGet=4) assert result["OK"] - # FIXME: this is not interpreted in any different way --- is it correct? - # I believe it should be instead interpreted in a way similar to CPUTime - # FIXME: the MaxRam parameter has a similar fate, and becomes "#GB", - # and then there's no specific matching about it. for jobId in range(1, 8): result = tqDB.deleteJob(jobId) diff --git a/tests/System/unitTestUserJobs.py b/tests/System/unitTestUserJobs.py index 6af9fa04dc5..b922062cd2a 100644 --- a/tests/System/unitTestUserJobs.py +++ b/tests/System/unitTestUserJobs.py @@ -122,6 +122,14 @@ def test_submit(self): self.assertTrue(res["OK"]) jobsSubmittedList.append(res["Value"]) + res = memory_4GB() + self.assertTrue(res["OK"]) + jobsSubmittedList.append(res["Value"]) + + res = memory_2_to4GB() + self.assertTrue(res["OK"]) + jobsSubmittedList.append(res["Value"]) + res = parametricJob() self.assertTrue(res["OK"]) jobsSubmittedList.append(res["Value"]) diff --git a/tests/Workflow/Integration/Test_UserJobs.py b/tests/Workflow/Integration/Test_UserJobs.py index e8df2eeeed8..6418073920b 100644 --- a/tests/Workflow/Integration/Test_UserJobs.py +++ b/tests/Workflow/Integration/Test_UserJobs.py @@ -1,10 +1,10 @@ -""" Testing the API and a bit more. - It will submit a number of test jobs locally (via runLocal), using the python unittest to assess the results. - Can be automatized. +"""Testing the API and a bit more. +It will submit a number of test jobs locally (via runLocal), using the python unittest to assess the results. +Can be automatized. """ + # pylint: disable=protected-access, wrong-import-position, invalid-name, missing-docstring import multiprocessing -import os import sys import unittest