[8.0] feat: PoolCE takes care of RAM requirements

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

docs/source/UserGuide/Tutorials/JobManagementAdvanced/index.rst

@@ -335,17 +335,14 @@ 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.
 
-.. versionadded:: v6r20p5
+``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.::
 
@@ -358,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)
 @@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@@
 

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"):

src/DIRAC/Interfaces/API/Job.py

@@ -522,6 +522,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. 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,
+                "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.
 
@@ -740,7 +784,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

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/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:
@@ -279,17 +282,14 @@ 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"]
         waitingJobs = result["WaitingJobs"]
         submittedJobs = result["SubmittedJobs"]
         availableProcessors = result.get("AvailableProcessors")
+
         ceInfoDict = dict(result)
 
         maxTotalJobs = int(self.ceParameters.get("MaxTotalJobs", 0))
@@ -493,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)"

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