feat: moved memory limits from GB to MB

Author: fstagni

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

@@ -363,9 +363,9 @@ using the "setRAMRequirements" method of the API::
 
       j = Job()
       ...
-      j.setRAMRequirements(2, 4)
+      j.setRAMRequirements(2500, 4000)
 
-Calling ``Job().setRAMRequirements()`` takes 2 values, where the first is the minimum required amount of RAM (in GB) that the job requests.
+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.
 
 

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
@@ -519,26 +520,26 @@ def setDestination(self, destination):
     #############################################################################
     def setRAMRequirements(self, ramRequired: int = 0, maxRAM: int = 0):
         """Helper function.
-        Specify the RAM requirements for the job. 0 (default) means no specific requirements.
+        Specify the RAM requirements for the job, in MB. 0 (default) means no specific requirements.
 
         Example usage:
 
         >>> job = Job()
-        >>> job.setRAMRequirements(ramRequired=2)
+        >>> 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=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=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=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(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=4)
+        >>> 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=8, maxRAM=4)
+        >>> job.setRAMRequirements(ramRequired=8000, maxRAM=4000)
         Makes no sense, an error will be raised
         """
         if ramRequired and maxRAM and ramRequired > maxRAM:
@@ -550,15 +551,15 @@ def setRAMRequirements(self, ramRequired: int = 0, maxRAM: int = 0):
                 "MinRAM",
                 "JDL",
                 ramRequired,
-                "GBs of RAM requested",
+                "MBs of RAM requested",
             )
         if maxRAM:
             self._addParameter(
                 self.workflow,
                 "MaxRAM",
                 "JDL",
                 maxRAM,
-                "Max GBs of RAM to be used",
+                "Max MBs of RAM to be used",
             )
 
     def setNumberOfProcessors(self, numberOfProcessors=None, minNumberOfProcessors=None, maxNumberOfProcessors=None):
@@ -753,7 +754,7 @@ def setTag(self, tags):
         Example usage:
 
         >>> job = Job()
-        >>> job.setTag( ['WholeNode','8GB'] )
+        >>> job.setTag( ['WholeNode'] )
 
         :param tags: single tag string or a list of tags
         :type tags: str or python:list

src/DIRAC/Resources/Computing/ComputingElement.py

@@ -408,7 +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", 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)"

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.
 
@@ -19,6 +19,7 @@
 
 **Code Documentation**
 """
+
 import concurrent.futures
 import functools
 
@@ -64,7 +65,8 @@ def __init__(self, ceUniqueID):
         self.taskID = 0
         self.processorsPerTask = {}
         self.ram = (
-            1024  # Available RAM for the node, in GB. The default value is an arbitrary large value in case of no limit
+            1024
+            * 1024  # Available RAM for the node, in MB. The default value is an arbitrary large value in case of no limit
         )
         self.ramPerTask = {}
 
@@ -195,7 +197,7 @@ 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
+        :return: memory in MB or None if not enough memory
         """
 
         # # job requirements
@@ -227,7 +229,7 @@ def finalizeJob(self, taskID, future):
 
         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) and {ram}GB 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

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
@@ -420,7 +421,7 @@ def submitJob(self, executableFile, proxy=None, **kwargs):
         # 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)
+                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

src/DIRAC/Resources/Computing/test/Test_InProcessComputingElement.py

@@ -54,8 +54,8 @@ def test_submitJob():
         maxNumberOfProcessors=8,
         wholeNode=False,
         mpTag=True,
-        MinRAM=2,
-        MaxRAM=4,
+        MinRAM=2500,
+        MaxRAM=4000,
         jobDesc={"jobParams": jobParams, "resourceParams": resourceParams, "optimizerParams": optimizerParams},
     )
     assert res["OK"] is True

src/DIRAC/Resources/Computing/test/Test_PoolComputingElement.py

@@ -2,6 +2,7 @@
 """
 tests for PoolComputingElement module
 """
+
 import os
 import time
 
@@ -87,7 +88,7 @@ def createAndDelete():
 def test_submit_and_shutdown(createAndDelete):
     time.sleep(0.5)
 
-    ceParameters = {"WholeNode": True, "NumberOfProcessors": 4, "MaxRAM": 4}
+    ceParameters = {"WholeNode": True, "NumberOfProcessors": 4, "MaxRAM": 3800}
     ce = PoolComputingElement("TestPoolCE")
     ce.setParameters(ceParameters)
 
@@ -149,7 +150,7 @@ def test_executeJob_wholeNode4(createAndDelete):
     time.sleep(0.5)
     taskIDs = {}
 
-    ceParameters = {"WholeNode": True, "NumberOfProcessors": 4, "MaxRAM": 16}
+    ceParameters = {"WholeNode": True, "NumberOfProcessors": 4, "MaxRAM": 16000}
     ce = PoolComputingElement("TestPoolCE")
     ce.setParameters(ceParameters)
 
@@ -164,10 +165,10 @@ def test_executeJob_wholeNode4(createAndDelete):
     assert result["UsedProcessors"] == 1
     assert result["AvailableProcessors"] == 3
     assert result["UsedRAM"] == 0
-    assert result["AvailableRAM"] == 16
+    assert result["AvailableRAM"] == 16000
     assert result["RunningJobs"] == 1
 
-    jobParams = {"mpTag": True, "numberOfProcessors": 2, "MaxRAM": 4}
+    jobParams = {"mpTag": True, "numberOfProcessors": 2, "MaxRAM": 4000}
     result = ce.submitJob("testPoolCEJob_1.py", None, **jobParams)
     assert result["OK"] is True
     taskID = result["Value"]
@@ -177,8 +178,8 @@ 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["UsedRAM"] == 4000
+    assert result["AvailableRAM"] == 12000
 
     assert result["RunningJobs"] == 2
 
@@ -203,7 +204,7 @@ def test_executeJob_wholeNode8(createAndDelete):
     time.sleep(0.5)
     taskIDs = {}
 
-    ceParameters = {"WholeNode": True, "NumberOfProcessors": 8, "MaxRAM": 32}
+    ceParameters = {"WholeNode": True, "NumberOfProcessors": 8, "MaxRAM": 32000}
     ce = PoolComputingElement("TestPoolCE")
     ce.setParameters(ceParameters)
 
@@ -227,9 +228,9 @@ def test_executeJob_wholeNode8(createAndDelete):
     result = ce.getCEStatus()
     assert result["UsedProcessors"] == 5
     assert result["UsedRAM"] == 0
-    assert result["AvailableRAM"] == 32
+    assert result["AvailableRAM"] == 32000
 
-    jobParams = {"numberOfProcessors": 2, "MinRAM": 4, "MaxRAM": 8}  # 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"]
@@ -238,10 +239,10 @@ def test_executeJob_wholeNode8(createAndDelete):
 
     result = ce.getCEStatus()
     assert result["UsedProcessors"] == 6
-    assert result["UsedRAM"] == 8
-    assert result["AvailableRAM"] == 24
+    assert result["UsedRAM"] == 8000
+    assert result["AvailableRAM"] == 24000
 
-    jobParams = {"MinRAM": 8, "MaxRAM": 8}  # This is same as asking for SP
+    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"]
@@ -250,10 +251,10 @@ def test_executeJob_wholeNode8(createAndDelete):
 
     result = ce.getCEStatus()
     assert result["UsedProcessors"] == 7
-    assert result["UsedRAM"] == 16
-    assert result["AvailableRAM"] == 16
+    assert result["UsedRAM"] == 16000
+    assert result["AvailableRAM"] == 16000
 
-    jobParams = {"MaxRAM": 24}  # This will fail
+    jobParams = {"MaxRAM": 24000}  # This will fail
     result = ce.submitJob("testPoolCEJob_6.py", None, **jobParams)
     assert result["OK"] is True
     taskID = result["Value"]
@@ -262,8 +263,8 @@ def test_executeJob_wholeNode8(createAndDelete):
 
     result = ce.getCEStatus()
     assert result["UsedProcessors"] == 7
-    assert result["UsedRAM"] == 16
-    assert result["AvailableRAM"] == 16
+    assert result["UsedRAM"] == 16000
+    assert result["AvailableRAM"] == 16000
 
     # now trying again would fail
     jobParams = {"mpTag": True, "numberOfProcessors": 3}
@@ -413,31 +414,31 @@ def test_executeJob_WholeNodeJobs(createAndDelete):
     [
         (None, None, {}, 1, 0),
         (None, None, {"mpTag": False}, 1, 0),
-        (None, None, {"mpTag": True, "MaxRAM": 8}, 1, 8),
+        (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": 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, "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: 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, "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),
-        ({1: 4, 2: 8, 3: 8}, {1: 4}, {"mpTag": True, "numberOfProcessors": 4}, 0, 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__getLimitsForJobs(processorsPerTask, ramPerTask, kwargs, expected_processors, expected_memory):
     ce = PoolComputingElement("TestPoolCE")
     ce.processors = 16
-    ce.ram = 32
+    ce.ram = 32000
 
     if processorsPerTask:
         ce.processorsPerTask = processorsPerTask

src/DIRAC/WorkloadManagementSystem/Client/DownloadInputData.py

@@ -1,6 +1,7 @@
-""" The Download Input Data module wraps around the Replica Management
-    components to provide access to datasets by downloading locally
+"""The Download Input Data module wraps around the Replica Management
+components to provide access to datasets by downloading locally
 """
+
 import os
 import random
 import tempfile