feat: Computing Elements use MaxRAM

Author: fstagni

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,18 +522,26 @@ def setDestination(self, destination):
         return S_OK()
 
     #############################################################################
-    def setRAMRequirements(self, ramRequired: int = 0):
+    def setRAMRequirements(self, ramRequired: int = 0, maxRAMRequested: 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",
+                "MinRAM",
                 "JDL",
                 ramRequired,
                 "GBs of RAM requested",
             )
+        if maxRAMRequested:
+            self._addParameter(
+                self.workflow,
+                "MaxRAM",
+                "JDL",
+                maxRAMRequested,
+                "GBs of RAM requested (Max to be used)",
+            )
 
     def setNumberOfProcessors(self, numberOfProcessors=None, minNumberOfProcessors=None, maxNumberOfProcessors=None):
         """Helper function.

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:
             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()
+        if self.ceParameters.get("MaxRAM", 0) == 0:
+            self.ceParameters["MaxRAM"] = getAvailableRAM()
 
         for key in ceOptions:
             if key in INTEGER_PARAMETERS:

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["MaxRAM"] = self.maxRAM
         return result

src/DIRAC/Resources/Computing/PoolComputingElement.py

@@ -61,7 +61,9 @@ 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.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
@@ -77,9 +79,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 max_ram:
+            self.ram = self.ceParameters["MaxRAM"]
         # Indicates that the submission is done asynchronously
         # The result is not immediately available
         self.ceParameters["AsyncSubmission"] = True
@@ -190,14 +191,18 @@ 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?
         availableMemory = self.ram - sum(self.ramPerTask.values())
         if availableMemory < requestedMemory:
             return None
 
+        # If there's a maximum amount of RAM allowed for the job, use that as maximum,
+        if "MaxRAM" in kwargs and kwargs["MaxRAM"]:
+            requestedMemory = int(kwargs["MaxRAM"])
+
         return requestedMemory
 
     def finalizeJob(self, taskID, future):

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["MaxRAM"] = self.maxRAM
         return result