From 7dffc72d65971610309b98c309bda81e34b9f15b Mon Sep 17 00:00:00 2001 From: Federico Stagni Date: Wed, 25 Jun 2025 11:51:13 +0200 Subject: [PATCH 1/8] fix: minor typo fixes --- src/DIRAC/Interfaces/API/Job.py | 2 +- .../WorkloadManagementSystem/Utilities/JobParameters.py | 8 ++++---- src/DIRAC/tests/Utilities/testJobDefinitions.py | 4 ++-- 3 files changed, 7 insertions(+), 7 deletions(-) diff --git a/src/DIRAC/Interfaces/API/Job.py b/src/DIRAC/Interfaces/API/Job.py index 870c49b62e6..433e1ef7e86 100755 --- a/src/DIRAC/Interfaces/API/Job.py +++ b/src/DIRAC/Interfaces/API/Job.py @@ -740,7 +740,7 @@ def setTag(self, tags): Example usage: >>> job = Job() - >>> job.setTag( ['WholeNode','8GBMemory'] ) + >>> job.setTag( ['WholeNode','8GB'] ) :param tags: single tag string or a list of tags :type tags: str or python:list diff --git a/src/DIRAC/WorkloadManagementSystem/Utilities/JobParameters.py b/src/DIRAC/WorkloadManagementSystem/Utilities/JobParameters.py index 774456f8ba9..1f260ee066a 100644 --- a/src/DIRAC/WorkloadManagementSystem/Utilities/JobParameters.py +++ b/src/DIRAC/WorkloadManagementSystem/Utilities/JobParameters.py @@ -1,10 +1,10 @@ """ DIRAC Workload Management System utility module to get available memory and processors from mjf """ -import os import multiprocessing +import os from urllib.request import urlopen -from DIRAC import gLogger, gConfig +from DIRAC import gConfig, gLogger from DIRAC.Core.Utilities.List import fromChar @@ -116,8 +116,8 @@ def getNumberOfProcessors(siteName=None, gridCE=None, queue=None): if numberOfProcessors: return numberOfProcessors - # 4) looks in CS for tags - gLogger.info(f"Getting tagsfor {siteName}: {gridCE}: {queue}") + # 3) looks in CS for tags + 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", "") diff --git a/src/DIRAC/tests/Utilities/testJobDefinitions.py b/src/DIRAC/tests/Utilities/testJobDefinitions.py index a4df06c23a6..59995bbe397 100644 --- a/src/DIRAC/tests/Utilities/testJobDefinitions.py +++ b/src/DIRAC/tests/Utilities/testJobDefinitions.py @@ -232,9 +232,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: From c009ebea6172773f8c0cf55cfdbadd2717762910 Mon Sep 17 00:00:00 2001 From: Federico Stagni Date: Wed, 25 Jun 2025 15:03:51 +0200 Subject: [PATCH 2/8] fix: remove deprecated SudoComputingElement --- .../Computing/PoolComputingElement.py | 21 ++++--------------- 1 file changed, 4 insertions(+), 17 deletions(-) diff --git a/src/DIRAC/Resources/Computing/PoolComputingElement.py b/src/DIRAC/Resources/Computing/PoolComputingElement.py index 231136d995b..ec3ce078219 100644 --- a/src/DIRAC/Resources/Computing/PoolComputingElement.py +++ b/src/DIRAC/Resources/Computing/PoolComputingElement.py @@ -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,18 @@ **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) + """wrapper around ce.submitJob: decides which 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 @@ -134,13 +128,6 @@ def submitJob(self, executableFile, proxy=None, inputs=None, **kwargs): # 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) From b0dfab6687ef3c53ecf50deeb4502bcd03f19a1e Mon Sep 17 00:00:00 2001 From: Federico Stagni Date: Wed, 25 Jun 2025 16:40:16 +0200 Subject: [PATCH 3/8] feat: PoolCE can subdivide RAM --- src/DIRAC/Interfaces/API/Job.py | 13 +++++ .../Computing/PoolComputingElement.py | 50 +++++++++++++------ .../test/Test_PoolComputingElement.py | 43 +++++++++------- 3 files changed, 75 insertions(+), 31 deletions(-) diff --git a/src/DIRAC/Interfaces/API/Job.py b/src/DIRAC/Interfaces/API/Job.py index 433e1ef7e86..17d8af0550e 100755 --- a/src/DIRAC/Interfaces/API/Job.py +++ b/src/DIRAC/Interfaces/API/Job.py @@ -522,6 +522,19 @@ def setDestination(self, destination): return S_OK() ############################################################################# + def setRAMRequirements(self, ramRequired: int = 0): + """Helper function. + Specify the RAM requirements for the job in GB. 0 (default) means no specific requirements. + """ + if ramRequired: + self._addParameter( + self.workflow, + "MaxRAM", + "JDL", + ramRequired, + "GBs of RAM requested", + ) + def setNumberOfProcessors(self, numberOfProcessors=None, minNumberOfProcessors=None, maxNumberOfProcessors=None): """Helper function. diff --git a/src/DIRAC/Resources/Computing/PoolComputingElement.py b/src/DIRAC/Resources/Computing/PoolComputingElement.py index ec3ce078219..b3f4fb26c39 100644 --- a/src/DIRAC/Resources/Computing/PoolComputingElement.py +++ b/src/DIRAC/Resources/Computing/PoolComputingElement.py @@ -61,6 +61,8 @@ def __init__(self, ceUniqueID): self.taskID = 0 self.processorsPerTask = {} self.userNumberPerTask = {} + self.ram = 1024 # Default RAM in GB (this 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) @@ -74,22 +76,16 @@ def _reset(self): self.processors = int(self.ceParameters.get("NumberOfProcessors", self.processors)) self.ceParameters["MaxTotalJobs"] = self.processors + max_ram = int(self.ceParameters.get("MaxRAM", 0)) + if max_ram > 0: + self.ram = max_ram // 1024 # Convert from MB to GB + self.ceParameters["MaxRAM"] = self.ram # 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. @@ -112,15 +108,22 @@ 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"]) @@ -132,6 +135,7 @@ def submitJob(self, executableFile, proxy=None, inputs=None, **kwargs): # Submission future = self.pPool.submit(executeJob, executableFile, proxy, self.taskID, inputs, **taskKwargs) self.processorsPerTask[future] = processorsForJob + self.ramPerTask[future] = memoryForJob future.add_done_callback(functools.partial(self.finalizeJob, self.taskID)) taskID = self.taskID @@ -141,7 +145,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( @@ -178,6 +182,24 @@ 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 GB or None if not enough memory + """ + + # # job requirements + requestedMemory = kwargs.get("MaxRAM", 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 + + return requestedMemory + def finalizeJob(self, taskID, future): """Finalize the job by updating the process utilisation counters @@ -209,7 +231,7 @@ 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 return result diff --git a/src/DIRAC/Resources/Computing/test/Test_PoolComputingElement.py b/src/DIRAC/Resources/Computing/test/Test_PoolComputingElement.py index 57aab6b4a86..ef40888a02e 100644 --- a/src/DIRAC/Resources/Computing/test/Test_PoolComputingElement.py +++ b/src/DIRAC/Resources/Computing/test/Test_PoolComputingElement.py @@ -83,7 +83,7 @@ def createAndDelete(): def test_submit_and_shutdown(createAndDelete): time.sleep(0.5) - ceParameters = {"WholeNode": True, "NumberOfProcessors": 4} + ceParameters = {"WholeNode": True, "NumberOfProcessors": 4, "MaxRAM": 4} ce = PoolComputingElement("TestPoolCE") ce.setParameters(ceParameters) @@ -371,28 +371,37 @@ 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": 8}, 1, 8), + (None, None, {"mpTag": True, "wholeNode": True}, 16, 0), + (None, None, {"mpTag": True, "wholeNode": False}, 1, 0), + (None, None, {"mpTag": True, "numberOfProcessors": 4, "MaxRAM": 4}, 4, 4), + (None, None, {"mpTag": True, "numberOfProcessors": 4, "maxNumberOfProcessors": 8}, 8, 0), + (None, None, {"mpTag": True, "numberOfProcessors": 4, "maxNumberOfProcessors": 32}, 16, 0), + ({1: 4}, {1: 4}, {"mpTag": True, "wholeNode": True}, 0, 0), + ({1: 4}, {1: 4}, {"mpTag": True, "wholeNode": False}, 1, 0), + ({1: 4}, {1: 4}, {"mpTag": True, "numberOfProcessors": 2, "MaxRAM": 8}, 2, 8), + ({1: 4}, {1: 4}, {"mpTag": True, "numberOfProcessors": 16, "MaxRAM": 12}, 0, 12), + ({1: 4}, {1: 4}, {"mpTag": True, "maxNumberOfProcessors": 2, "MaxRAM": 16}, 2, 16), + ({1: 4}, {1: 4}, {"mpTag": True, "maxNumberOfProcessors": 16, "MaxRAM": 32}, 12, None), + ({1: 4, 2: 8}, {1: 4}, {"mpTag": True, "numberOfProcessors": 2}, 2, 0), + ({1: 4, 2: 8}, {1: 4}, {"mpTag": True, "numberOfProcessors": 4}, 4, 0), + ({1: 4, 2: 8, 3: 8}, {1: 4}, {"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 = 32 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 From cfb94edb491a2f255e892a65a8e9473835871a9b Mon Sep 17 00:00:00 2001 From: Federico Stagni Date: Wed, 15 Oct 2025 13:49:39 +0200 Subject: [PATCH 4/8] fix: remove switch for CREAM CE --- src/DIRAC/Resources/Computing/ComputingElement.py | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/src/DIRAC/Resources/Computing/ComputingElement.py b/src/DIRAC/Resources/Computing/ComputingElement.py index 4b73626eba9..93b89787bae 100755 --- a/src/DIRAC/Resources/Computing/ComputingElement.py +++ b/src/DIRAC/Resources/Computing/ComputingElement.py @@ -279,11 +279,7 @@ def available(self, jobIDList=None): result["WaitingJobs"] = 0 result["SubmittedJobs"] = 0 else: - result = self.ceParameters.get("CEType") - if result and result == "CREAM": - result = self.getCEStatus(jobIDList) # pylint: disable=too-many-function-args - else: - result = self.getCEStatus() + result = self.getCEStatus() if not result["OK"]: return result runningJobs = result["RunningJobs"] From 6d4d439b8ca546fda5c414a4c89fbb81a5670db4 Mon Sep 17 00:00:00 2001 From: Federico Stagni Date: Wed, 25 Jun 2025 17:01:27 +0200 Subject: [PATCH 5/8] test: added tests for RAM limits --- .../tests/Utilities/testJobDefinitions.py | 36 +++++++++++++++++++ tests/System/unitTestUserJobs.py | 8 +++++ 2 files changed, 44 insertions(+) diff --git a/src/DIRAC/tests/Utilities/testJobDefinitions.py b/src/DIRAC/tests/Utilities/testJobDefinitions.py index 59995bbe397..e31be9fbd04 100644 --- a/src/DIRAC/tests/Utilities/testJobDefinitions.py +++ b/src/DIRAC/tests/Utilities/testJobDefinitions.py @@ -282,6 +282,42 @@ 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.setTag("4GB") + 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.setTag(["2GB", "4GB_MAX"]) + return endOfAllJobs(J) + + def parametricJob(): """Creates a parametric job with 3 subjobs which are simple hello world jobs""" diff --git a/tests/System/unitTestUserJobs.py b/tests/System/unitTestUserJobs.py index e529e441c95..f67708559be 100644 --- a/tests/System/unitTestUserJobs.py +++ b/tests/System/unitTestUserJobs.py @@ -106,6 +106,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"]) From c17c2f5ac6d8f41ed47f3ac97fe3d9cc82ec3203 Mon Sep 17 00:00:00 2001 From: Federico Stagni Date: Tue, 1 Jul 2025 15:32:38 +0200 Subject: [PATCH 6/8] docs: ComputingElement docs update --- .../Resources/computingelements.rst | 21 ++++++------------- .../Tutorials/JobManagementAdvanced/index.rst | 2 -- 2 files changed, 6 insertions(+), 17 deletions(-) diff --git a/docs/source/AdministratorGuide/Resources/computingelements.rst b/docs/source/AdministratorGuide/Resources/computingelements.rst index 76d2874c901..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`. @@ -114,7 +110,7 @@ configuration. Interacting with Grid Sites @@@@@@@@@@@@@@@@@@@@@@@@@@@ -The :mod:`~DIRAC.Resources.Computing.HTCondorCEComputingElement` and the :mod:`~DIRAC.Resources.Computing.ARCComputingElement` eases +The :mod:`~DIRAC.Resources.Computing.HTCondorCEComputingElement` and the :mod:`~DIRAC.Resources.Computing.AREXComputingElement` eases the interactions with grid sites, by managing pilots using the underlying batch systems. Instances of such CEs are generally setup by the site administrators. @@ -132,11 +128,6 @@ The :mod:`~DIRAC.Resources.Computing.CloudComputingElement` allows submission to (via the standard SiteDirector agent). The instances are contextualised using cloud-init. -Delegating to BOINC (Volunteering Computing) -@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@ -There exists a :mod:`~DIRAC.Resources.Computing.BOINCComputingElement` to submit pilots to a BOINC server. - - Computing Elements within allocated computing resources @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@ The :mod:`~DIRAC.Resources.Computing.InProcessComputingElement` is usually invoked by a Pilot-Job (JobAgent agent) to execute user diff --git a/docs/source/UserGuide/Tutorials/JobManagementAdvanced/index.rst b/docs/source/UserGuide/Tutorials/JobManagementAdvanced/index.rst index 0b70d79c050..a939114b72a 100644 --- a/docs/source/UserGuide/Tutorials/JobManagementAdvanced/index.rst +++ b/docs/source/UserGuide/Tutorials/JobManagementAdvanced/index.rst @@ -345,8 +345,6 @@ using the "setNumberOfProcessors" method of the API:: Calling ``Job().setNumberOfProcessors()``, with a value bigger than 1, will translate into adding also the "MultiProcessor" tag to the job description. -.. versionadded:: v6r20p5 - Users can specify in the job descriptions NumberOfProcessors and WholeNode parameters, e.g.:: NumberOfProcessors = 16; From 92382a177062a3962156a4c480f7a0333a1c6776 Mon Sep 17 00:00:00 2001 From: Federico Stagni Date: Mon, 14 Jul 2025 17:17:47 +0200 Subject: [PATCH 7/8] feat: created a utility function getAvailableRAM --- .../Resources/Computing/ComputingElement.py | 5 +- .../Utilities/JobParameters.py | 98 ++++++++++++------- .../scripts/dirac_wms_get_wn_parameters.py | 9 +- 3 files changed, 71 insertions(+), 41 deletions(-) diff --git a/src/DIRAC/Resources/Computing/ComputingElement.py b/src/DIRAC/Resources/Computing/ComputingElement.py index 93b89787bae..3130e4948b5 100755 --- a/src/DIRAC/Resources/Computing/ComputingElement.py +++ b/src/DIRAC/Resources/Computing/ComputingElement.py @@ -57,6 +57,7 @@ from DIRAC.WorkloadManagementSystem.Utilities.JobParameters import ( getNumberOfProcessors, getNumberOfGPUs, + getAvailableRAM, ) INTEGER_PARAMETERS = ["CPUTime", "NumberOfProcessors", "NumberOfPayloadProcessors", "MaxRAM"] @@ -235,12 +236,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/RAM is present in the description but is equal to zero # interpret it as needing local evaluation if self.ceParameters.get("NumberOfProcessors", -1) == 0: self.ceParameters["NumberOfProcessors"] = getNumberOfProcessors() if self.ceParameters.get("NumberOfGPUs", -1) == 0: self.ceParameters["NumberOfGPUs"] = getNumberOfGPUs() + if self.ceParameters.get("RAM", -1) == 0: + self.ceParameters["RAM"] = getAvailableRAM() for key in ceOptions: if key in INTEGER_PARAMETERS: diff --git a/src/DIRAC/WorkloadManagementSystem/Utilities/JobParameters.py b/src/DIRAC/WorkloadManagementSystem/Utilities/JobParameters.py index 1f260ee066a..f5dbd2626e4 100644 --- a/src/DIRAC/WorkloadManagementSystem/Utilities/JobParameters.py +++ b/src/DIRAC/WorkloadManagementSystem/Utilities/JobParameters.py @@ -34,31 +34,7 @@ def getJobFeatures(): return features -def getProcessorFromMJF(): - jobFeatures = getJobFeatures() - if jobFeatures: - try: - return int(jobFeatures["allocated_cpu"]) - except KeyError: - gLogger.error( - "MJF is available but allocated_cpu is not an integer", repr(jobFeatures.get("allocated_cpu")) - ) - return None - - -def getMemoryFromMJF(): - jobFeatures = getJobFeatures() - if jobFeatures: - try: - return int(jobFeatures["max_rss_bytes"]) - except KeyError: - gLogger.error( - "MJF is available but max_rss_bytes is not an integer", repr(jobFeatures.get("max_rss_bytes")) - ) - return None - - -def getMemoryFromProc(): +def _getMemoryFromProc(): meminfo = {i.split()[0].rstrip(":"): int(i.split()[1]) for i in open("/proc/meminfo").readlines()} maxRAM = meminfo["MemTotal"] if maxRAM: @@ -72,7 +48,6 @@ def getNumberOfProcessors(siteName=None, gridCE=None, queue=None): Tries to find it in this order: 1) from the /Resources/Computing/CEDefaults/NumberOfProcessors (which is what the pilot fills up) - 2) if not present from JobFeatures 3) if not present looks in CS for "NumberOfProcessors" Queue or CE option 4) if not present but there's WholeNode tag, look what the WN provides using multiprocessing.cpu_count() 5) return 1 @@ -84,14 +59,7 @@ def getNumberOfProcessors(siteName=None, gridCE=None, queue=None): if numberOfProcessors: return numberOfProcessors - # 2) from MJF - gLogger.info("Getting numberOfProcessors from MJF") - numberOfProcessors = getProcessorFromMJF() - if numberOfProcessors: - return numberOfProcessors - gLogger.info("NumberOfProcessors could not be found in MJF") - - # 3) looks in CS for "NumberOfProcessors" Queue or CE or site option + # 2) looks in CS for "NumberOfProcessors" Queue or CE or site option if not siteName: siteName = gConfig.getValue("/LocalSite/Site", "") if not gridCE: @@ -237,3 +205,65 @@ def getNumberOfGPUs(siteName=None, gridCE=None, queue=None): # 3) return 0 gLogger.info("NumberOfGPUs could not be found in CS") return 0 + + +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/AvailableRAM (which is what the pilot might fill up) + 2) if not present looks in CS for "AvailableRAM" Queue or CE 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}/MaxRAM", + f"/Resources/Sites/{grid}/{siteName}/CEs/{gridCE}/MaxRAM", + f"/Resources/Sites/{grid}/{siteName}/MaxRAM", + ] + 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("AvailableRAM could not be found in CS, and WholeNode tag not found") + return 0 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 07a1042f88a..5b3b2497b6c 100755 --- a/src/DIRAC/WorkloadManagementSystem/scripts/dirac_wms_get_wn_parameters.py +++ b/src/DIRAC/WorkloadManagementSystem/scripts/dirac_wms_get_wn_parameters.py @@ -39,15 +39,12 @@ def main(): gLogger.info("Getting number of processors") numberOfProcessor = JobParameters.getNumberOfProcessors(Site, ceName, Queue) - gLogger.info("Getting memory (RAM) from MJF") - maxRAM = JobParameters.getMemoryFromMJF() - if not maxRAM: - gLogger.info("maxRAM could not be found in MJF, using JobParameters.getMemoryFromProc()") - 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])) From 8dd57efd7d56b9d74c1988685ce38c22f0e579c0 Mon Sep 17 00:00:00 2001 From: Federico Stagni Date: Tue, 28 Oct 2025 11:51:58 +0100 Subject: [PATCH 8/8] feat: Computing Elements use MaxRAM --- .../Tutorials/JobManagementAdvanced/index.rst | 23 ++++-- src/DIRAC/Core/Utilities/CGroups2.py | 1 + src/DIRAC/Interfaces/API/Job.py | 37 ++++++++- .../Resources/Computing/ComputingElement.py | 12 +-- .../Computing/InProcessComputingElement.py | 3 + .../Computing/PoolComputingElement.py | 48 +++++++++--- .../Computing/SingularityComputingElement.py | 9 +++ .../test/Test_InProcessComputingElement.py | 2 + .../test/Test_PoolComputingElement.py | 77 ++++++++++++++----- .../Utilities/JobParameters.py | 29 +++++-- .../Utilities/RemoteRunner.py | 5 +- .../Utilities/test/Test_RemoteRunner.py | 10 ++- .../tests/Utilities/testJobDefinitions.py | 5 +- .../Resources/Computing/Test_SingularityCE.py | 2 + 14 files changed, 208 insertions(+), 55 deletions(-) diff --git a/docs/source/UserGuide/Tutorials/JobManagementAdvanced/index.rst b/docs/source/UserGuide/Tutorials/JobManagementAdvanced/index.rst index a939114b72a..dd2842f14c0 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(2, 4) + +Calling ``Job().setRAMRequirements()`` takes 2 values, where the first is the minimum required amount of RAM (in GB) 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 17d8af0550e..9c63fe98422 100755 --- a/src/DIRAC/Interfaces/API/Job.py +++ b/src/DIRAC/Interfaces/API/Job.py @@ -522,18 +522,49 @@ def setDestination(self, destination): return S_OK() ############################################################################# - def setRAMRequirements(self, ramRequired: int = 0): + def setRAMRequirements(self, ramRequired: int = 0, maxRAM: int = 0): """Helper function. - Specify the RAM requirements for the job in GB. 0 (default) means no specific requirements. + Specify the RAM requirements for the job. 0 (default) means no specific requirements. + + Example usage: + + >>> job = Job() + >>> job.setRAMRequirements(ramRequired=2) + 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=2, maxRAM=4) + means that the job needs 2 GBs of RAM to work. 4 GBs will then be the upper limit for CG2 limits. + + >>> job.setRAMRequirements(ramRequired=4, maxRAM=4) + means that we should match this job if there is at least 4 available GBs of run. At the same time, CG2 will not allow to use more than that. + + >>> job.setRAMRequirements(maxRAM=4) + 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=8, maxRAM=4) + 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, - "MaxRAM", + "MinRAM", "JDL", ramRequired, "GBs of RAM requested", ) + if maxRAM: + self._addParameter( + self.workflow, + "MaxRAM", + "JDL", + maxRAM, + "Max GBs of RAM to be used", + ) def setNumberOfProcessors(self, numberOfProcessors=None, minNumberOfProcessors=None, maxNumberOfProcessors=None): """Helper function. diff --git a/src/DIRAC/Resources/Computing/ComputingElement.py b/src/DIRAC/Resources/Computing/ComputingElement.py index 3130e4948b5..edb9ae41ceb 100755 --- a/src/DIRAC/Resources/Computing/ComputingElement.py +++ b/src/DIRAC/Resources/Computing/ComputingElement.py @@ -236,14 +236,14 @@ def setParameters(self, ceOptions): generalCEDict.update(self.ceParameters) self.ceParameters = generalCEDict - # If NumberOfProcessors/GPUs/RAM 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 self.ceParameters.get("RAM", -1) == 0: - self.ceParameters["RAM"] = getAvailableRAM() + if int(self.ceParameters.get("MaxRAM", 0)) == 0: + self.ceParameters["MaxRAM"] = getAvailableRAM() for key in ceOptions: if key in INTEGER_PARAMETERS: @@ -289,6 +289,7 @@ def available(self, jobIDList=None): waitingJobs = result["WaitingJobs"] submittedJobs = result["SubmittedJobs"] availableProcessors = result.get("AvailableProcessors") + ceInfoDict = dict(result) maxTotalJobs = int(self.ceParameters.get("MaxTotalJobs", 0)) @@ -492,6 +493,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", 1)) 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 b3f4fb26c39..87eac16a28b 100644 --- a/src/DIRAC/Resources/Computing/PoolComputingElement.py +++ b/src/DIRAC/Resources/Computing/PoolComputingElement.py @@ -29,8 +29,8 @@ from DIRAC.Resources.Computing.SingularityComputingElement import SingularityComputingElement -def executeJob(executableFile, proxy, taskID, inputs, **kwargs): - """wrapper around ce.submitJob: decides which CE to use (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 @@ -46,6 +46,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) @@ -60,8 +63,9 @@ def __init__(self, ceUniqueID): self.pPool = None self.taskID = 0 self.processorsPerTask = {} - self.userNumberPerTask = {} - self.ram = 1024 # Default RAM in GB (this is an arbitrary large value in case of no limit) + self.ram = ( + 1024 # Available RAM for the node, in GB. The default value is an arbitrary large value in case of no limit + ) self.ramPerTask = {} # This CE will effectively submit to another "Inner"CE @@ -76,10 +80,8 @@ def _reset(self): self.processors = int(self.ceParameters.get("NumberOfProcessors", self.processors)) self.ceParameters["MaxTotalJobs"] = self.processors - max_ram = int(self.ceParameters.get("MaxRAM", 0)) - if max_ram > 0: - self.ram = max_ram // 1024 # Convert from MB to GB - self.ceParameters["MaxRAM"] = self.ram + 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 @@ -128,12 +130,19 @@ def submitJob(self, executableFile, proxy=None, inputs=None, **kwargs): 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", {}) # 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)) @@ -190,7 +199,7 @@ def _getMemoryForJobs(self, kwargs): """ # # job requirements - requestedMemory = kwargs.get("MaxRAM", 0) + requestedMemory = kwargs.get("MinRAM", 0) # # now check what the slot can provide # Do we have enough memory? @@ -198,6 +207,14 @@ def _getMemoryForJobs(self, kwargs): 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): @@ -206,23 +223,26 @@ def finalizeJob(self, taskID, future): :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}GB 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 @@ -234,6 +254,10 @@ def getCEStatus(self): 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 05546c7c428..1d86e6012cb 100644 --- a/src/DIRAC/Resources/Computing/SingularityComputingElement.py +++ b/src/DIRAC/Resources/Computing/SingularityComputingElement.py @@ -115,6 +115,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(): @@ -413,6 +414,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 * 1024, 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 ) @@ -449,4 +456,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 c7c023dda93..7201f7cd521 100644 --- a/src/DIRAC/Resources/Computing/test/Test_InProcessComputingElement.py +++ b/src/DIRAC/Resources/Computing/test/Test_InProcessComputingElement.py @@ -52,6 +52,8 @@ def test_submitJob(): maxNumberOfProcessors=8, wholeNode=False, mpTag=True, + MinRAM=2, + MaxRAM=4, 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 ef40888a02e..3c107072e85 100644 --- a/src/DIRAC/Resources/Computing/test/Test_PoolComputingElement.py +++ b/src/DIRAC/Resources/Computing/test/Test_PoolComputingElement.py @@ -50,7 +50,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,18 +66,22 @@ 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): @@ -145,7 +149,7 @@ def test_executeJob_wholeNode4(createAndDelete): time.sleep(0.5) taskIDs = {} - ceParameters = {"WholeNode": True, "NumberOfProcessors": 4} + ceParameters = {"WholeNode": True, "NumberOfProcessors": 4, "MaxRAM": 16} ce = PoolComputingElement("TestPoolCE") ce.setParameters(ceParameters) @@ -159,9 +163,11 @@ def test_executeJob_wholeNode4(createAndDelete): result = ce.getCEStatus() assert result["UsedProcessors"] == 1 assert result["AvailableProcessors"] == 3 + assert result["UsedRAM"] == 0 + assert result["AvailableRAM"] == 16 assert result["RunningJobs"] == 1 - jobParams = {"mpTag": True, "numberOfProcessors": 2} + jobParams = {"mpTag": True, "numberOfProcessors": 2, "MaxRAM": 4} result = ce.submitJob("testPoolCEJob_1.py", None, **jobParams) assert result["OK"] is True taskID = result["Value"] @@ -171,6 +177,9 @@ def test_executeJob_wholeNode4(createAndDelete): result = ce.getCEStatus() assert result["UsedProcessors"] == 3 assert result["AvailableProcessors"] == 1 + assert result["UsedRAM"] == 4 + assert result["AvailableRAM"] == 12 + assert result["RunningJobs"] == 2 # now trying again would fail @@ -194,7 +203,7 @@ def test_executeJob_wholeNode8(createAndDelete): time.sleep(0.5) taskIDs = {} - ceParameters = {"WholeNode": True, "NumberOfProcessors": 8} + ceParameters = {"WholeNode": True, "NumberOfProcessors": 8, "MaxRAM": 32} ce = PoolComputingElement("TestPoolCE") ce.setParameters(ceParameters) @@ -217,8 +226,10 @@ def test_executeJob_wholeNode8(createAndDelete): result = ce.getCEStatus() assert result["UsedProcessors"] == 5 + assert result["UsedRAM"] == 0 + assert result["AvailableRAM"] == 32 - jobParams = {"numberOfProcessors": 2} # This is same as asking for SP + jobParams = {"numberOfProcessors": 2, "MinRAM": 4, "MaxRAM": 8} # 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 +238,65 @@ def test_executeJob_wholeNode8(createAndDelete): result = ce.getCEStatus() assert result["UsedProcessors"] == 6 + assert result["UsedRAM"] == 8 + assert result["AvailableRAM"] == 24 - # now trying again would fail - jobParams = {"mpTag": True, "numberOfProcessors": 3} + jobParams = {"MinRAM": 8, "MaxRAM": 8} # 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"] == 16 + assert result["AvailableRAM"] == 16 + + jobParams = {"MaxRAM": 24} # 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"] == 16 + assert result["AvailableRAM"] == 16 + + # 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"] def test_executeJob_submitAndStop(createAndDelete): @@ -378,14 +415,18 @@ def test_executeJob_WholeNodeJobs(createAndDelete): (None, None, {"mpTag": True, "MaxRAM": 8}, 1, 8), (None, None, {"mpTag": True, "wholeNode": True}, 16, 0), (None, None, {"mpTag": True, "wholeNode": False}, 1, 0), + (None, None, {"mpTag": True, "numberOfProcessors": 4, "MinRAM": 2}, 4, 2), (None, None, {"mpTag": True, "numberOfProcessors": 4, "MaxRAM": 4}, 4, 4), + (None, None, {"mpTag": True, "numberOfProcessors": 4, "MaxRAM": 36}, 4, None), + (None, None, {"mpTag": True, "numberOfProcessors": 4, "MinRAM": 2, "MaxRAM": 4}, 4, 4), (None, None, {"mpTag": True, "numberOfProcessors": 4, "maxNumberOfProcessors": 8}, 8, 0), (None, None, {"mpTag": True, "numberOfProcessors": 4, "maxNumberOfProcessors": 32}, 16, 0), ({1: 4}, {1: 4}, {"mpTag": True, "wholeNode": True}, 0, 0), ({1: 4}, {1: 4}, {"mpTag": True, "wholeNode": False}, 1, 0), - ({1: 4}, {1: 4}, {"mpTag": True, "numberOfProcessors": 2, "MaxRAM": 8}, 2, 8), - ({1: 4}, {1: 4}, {"mpTag": True, "numberOfProcessors": 16, "MaxRAM": 12}, 0, 12), + ({1: 4}, {1: 4}, {"mpTag": True, "numberOfProcessors": 2, "MinRAM": 8}, 2, 8), + ({1: 4}, {1: 4}, {"mpTag": True, "numberOfProcessors": 16, "MinRAM": 8, "MaxRAM": 12}, 0, 12), ({1: 4}, {1: 4}, {"mpTag": True, "maxNumberOfProcessors": 2, "MaxRAM": 16}, 2, 16), + ({1: 4}, {1: 4}, {"mpTag": True, "numberOfProcessors": 2, "MaxRAM": 8}, 2, 8), ({1: 4}, {1: 4}, {"mpTag": True, "maxNumberOfProcessors": 16, "MaxRAM": 32}, 12, None), ({1: 4, 2: 8}, {1: 4}, {"mpTag": True, "numberOfProcessors": 2}, 2, 0), ({1: 4, 2: 8}, {1: 4}, {"mpTag": True, "numberOfProcessors": 4}, 4, 0), diff --git a/src/DIRAC/WorkloadManagementSystem/Utilities/JobParameters.py b/src/DIRAC/WorkloadManagementSystem/Utilities/JobParameters.py index f5dbd2626e4..7932a8268d3 100644 --- a/src/DIRAC/WorkloadManagementSystem/Utilities/JobParameters.py +++ b/src/DIRAC/WorkloadManagementSystem/Utilities/JobParameters.py @@ -213,8 +213,8 @@ def getAvailableRAM(siteName=None, gridCE=None, queue=None): The siteName/gridCE/queue parameters are normally not necessary. Tries to find it in this order: - 1) from the /Resources/Computing/CEDefaults/AvailableRAM (which is what the pilot might fill up) - 2) if not present looks in CS for "AvailableRAM" Queue or CE option + 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 """ @@ -238,9 +238,9 @@ def getAvailableRAM(siteName=None, gridCE=None, queue=None): grid = siteName.split(".")[0] csPaths = [ - f"/Resources/Sites/{grid}/{siteName}/CEs/{gridCE}/Queues/{queue}/MaxRAM", - f"/Resources/Sites/{grid}/{siteName}/CEs/{gridCE}/MaxRAM", - f"/Resources/Sites/{grid}/{siteName}/MaxRAM", + 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) @@ -265,5 +265,22 @@ def getAvailableRAM(siteName=None, gridCE=None, queue=None): return _getMemoryFromProc() # 4) return 0 - gLogger.info("AvailableRAM could not be found in CS, and WholeNode tag not found") + 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/RemoteRunner.py b/src/DIRAC/WorkloadManagementSystem/Utilities/RemoteRunner.py index 3bcf20f78a4..bb544f143a4 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 @@ -197,6 +198,7 @@ def _setUpWorkloadCE(self, numberOfProcessorsPayload=1): return result ceType = result["Value"]["CEType"] ceParams = result["Value"] + print(ceParams) # Build CE ceFactory = ComputingElementFactory() @@ -206,6 +208,7 @@ def _setUpWorkloadCE(self, numberOfProcessorsPayload=1): workloadCE = result["Value"] # Set the number of processors available according to the need of the payload + print(workloadCE.ceParameters) numberOfProcessorsCE = workloadCE.ceParameters.get("NumberOfProcessors", 1) if numberOfProcessorsCE < 1 or numberOfProcessorsPayload < 1: self.log.warn( 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/tests/Utilities/testJobDefinitions.py b/src/DIRAC/tests/Utilities/testJobDefinitions.py index e31be9fbd04..76c1584d912 100644 --- a/src/DIRAC/tests/Utilities/testJobDefinitions.py +++ b/src/DIRAC/tests/Utilities/testJobDefinitions.py @@ -296,7 +296,8 @@ def memory_4GB(): J.setExecutable("mpTest.py") J.setNumberOfProcessors(numberOfProcessors=2) - J.setTag("4GB") + J.setRAMRequirements(ramRequired=2, maxRAM=4) + return endOfAllJobs(J) @@ -314,7 +315,7 @@ def memory_2_to4GB(): J.setExecutable("mpTest.py") J.setNumberOfProcessors(numberOfProcessors=2) - J.setTag(["2GB", "4GB_MAX"]) + J.setRAMRequirements(ramRequired=4, maxRAM=4) return endOfAllJobs(J) diff --git a/tests/Integration/Resources/Computing/Test_SingularityCE.py b/tests/Integration/Resources/Computing/Test_SingularityCE.py index 90f271fa450..896b898c255 100644 --- a/tests/Integration/Resources/Computing/Test_SingularityCE.py +++ b/tests/Integration/Resources/Computing/Test_SingularityCE.py @@ -64,6 +64,8 @@ def test_submitJobWrapper(): maxNumberOfProcessors=8, wholeNode=False, mpTag=True, + MinRAM=2, + MaxRAM=4, jobDesc={"jobParams": jobParams, "resourceParams": resourceParams, "optimizerParams": optimizerParams}, )