fix: adding basic capabilities to InProcessCE

Author: fstagni

src/DIRAC/Resources/Computing/InProcessComputingElement.py

@@ -1,15 +1,15 @@
-""" The simplest of the "inner" CEs (meaning it's used by a jobAgent inside a pilot)
+"""The simplest of the "inner" CEs (meaning it's used by a jobAgent inside a pilot)
 
-    A "InProcess" CE instance submits jobs in the current process.
-    This is the standard "inner CE" invoked from the JobAgent, main alternative being the PoolCE
+A "InProcess" CE instance submits jobs in the current process.
+This is the standard "inner CE" invoked from the JobAgent, main alternative being the PoolCE
 """
+
 import os
 import stat
 
-from DIRAC import S_OK, S_ERROR
-from DIRAC.Core.Utilities.ThreadScheduler import gThreadScheduler
+from DIRAC import S_ERROR, S_OK
 from DIRAC.Core.Utilities.CGroups2 import CG2Manager
-
+from DIRAC.Core.Utilities.ThreadScheduler import gThreadScheduler
 from DIRAC.Resources.Computing.ComputingElement import ComputingElement
 
 
@@ -21,8 +21,8 @@ def __init__(self, ceUniqueID):
         self.submittedJobs = 0
         self.runningJobs = 0
 
-        self.processors = int(self.ceParameters.get("NumberOfProcessors", 1))
-        self.maxRAM = int(self.ceParameters.get("MaxRAM", 0))
+        self.processors = 1
+        self.maxRAM = 0
         self.ceParameters["MaxTotalJobs"] = 1
 
     def submitJob(self, executableFile, proxy=None, inputs=None, **kwargs):
@@ -34,6 +34,16 @@ def submitJob(self, executableFile, proxy=None, inputs=None, **kwargs):
         :param list inputs: dependencies of executableFile
         :return: S_OK(payload exit code) / S_ERROR() if submission issue
         """
+        self.processors = int(self.ceParameters.get("NumberOfProcessors", self.processors))
+        self.maxRAM = int(self.ceParameters.get("MaxRAM", self.maxRAM))
+
+        if "numberOfProcessors" in kwargs:
+            if self.processors < int(kwargs["numberOfProcessors"]):
+                return S_ERROR("Requesting processors not available")
+        if "MaxRAM" in kwargs:
+            if self.maxRAM < int(kwargs["MaxRAM"]):
+                return S_ERROR("Requesting RAM not available")
+
         payloadEnv = dict(os.environ)
         payloadProxy = ""
         renewTask = None

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

@@ -8,25 +8,41 @@
 
 import pytest
 
-from DIRAC.Resources.Computing.test.Test_PoolComputingElement import jobScript, _stopJob
-from DIRAC.WorkloadManagementSystem.Utilities.Utils import createJobWrapper
-
 # sut
 from DIRAC.Resources.Computing.InProcessComputingElement import InProcessComputingElement
+from DIRAC.Resources.Computing.test.Test_PoolComputingElement import _stopJob, jobScript
+from DIRAC.WorkloadManagementSystem.Utilities.Utils import createJobWrapper
 
 
 @pytest.mark.slow
-def test_submitJob():
+@pytest.mark.parametrize(
+    "ce_parameters, available_processors, ram",
+    [
+        ({}, 1, 0),
+        ({"NumberOfProcessors": 8}, 8, 0),
+        ({"MaxRAM": 2048}, 1, 2048),
+        ({"NumberOfProcessors": 8, "MaxRAM": 2048}, 8, 2048),
+    ],
+)
+def test_submitJob(ce_parameters, available_processors, ram):
+    # initialization
+    ce = InProcessComputingElement("InProcessCE")
+    ce.ceParameters = ce_parameters
+
+    # simple
     with open("testJob.py", "w") as execFile:
         execFile.write(jobScript % "1")
     os.chmod("testJob.py", 0o755)
 
-    ce = InProcessComputingElement("InProcessCE")
     res = ce.submitJob("testJob.py", None)
     assert res["OK"] is True
     res = ce.getCEStatus()
     assert res["OK"] is True
     assert res["SubmittedJobs"] == 1
+    assert res["RunningJobs"] == 0
+    assert res["WaitingJobs"] == 0
+    assert res["AvailableProcessors"] == available_processors
+    assert res["AvailableRAM"] == ram
     _stopJob(1)
     for ff in ["testJob.py", "stop_job_2", "job.info", "std.out"]:
         if os.path.isfile(ff):
@@ -44,29 +60,68 @@ def test_submitJob():
 
     wrapperFile = createJobWrapper(
         jobID=2, jobParams=jobParams, resourceParams=resourceParams, optimizerParams=optimizerParams, logLevel="DEBUG"
-    )["Value"][
-        "JobExecutablePath"
-    ]  # This is not under test, assuming it works fine
+    )["Value"]["JobExecutablePath"]  # This is not under test, assuming it works fine
     res = ce.submitJob(
         wrapperFile,
         proxy=None,
-        numberOfProcessors=4,
-        maxNumberOfProcessors=8,
+        numberOfProcessors=available_processors,
+        maxNumberOfProcessors=available_processors,
         wholeNode=False,
         mpTag=True,
-        MinRAM=2500,
-        MaxRAM=4000,
+        MinRAM=ram,
+        MaxRAM=ram,
         jobDesc={"jobParams": jobParams, "resourceParams": resourceParams, "optimizerParams": optimizerParams},
     )
     assert res["OK"] is True
+    _stopJob(2)
 
     res = ce.getCEStatus()
     assert res["OK"] is True
     assert res["SubmittedJobs"] == 2
+    assert res["RunningJobs"] == 0
+    assert res["WaitingJobs"] == 0
+    assert res["AvailableProcessors"] == available_processors
+    assert res["AvailableRAM"] == ram
 
-    _stopJob(2)
     for ff in ["testJob.py", "stop_job_2", "job.info", "std.out"]:
         if os.path.isfile(ff):
             os.remove(ff)
     if os.path.isdir("job"):
         shutil.rmtree("job")
+
+    # failing
+    with open("testJob.py", "w") as execFile:
+        execFile.write(jobScript % "3")
+    os.chmod("testJob.py", 0o755)
+
+    jobParams = {"JobType": "User", "Executable": "testJob.py"}
+    resourceParams = {"GridCE": "some_CE"}
+    optimizerParams = {}
+
+    wrapperFile = createJobWrapper(
+        jobID=3, jobParams=jobParams, resourceParams=resourceParams, optimizerParams=optimizerParams, logLevel="DEBUG"
+    )["Value"]["JobExecutablePath"]  # This is not under test, assuming it works fine
+
+    res = ce.submitJob(
+        wrapperFile,
+        proxy=None,
+        numberOfProcessors=4 + available_processors,
+        maxNumberOfProcessors=8 + available_processors,
+        wholeNode=False,
+        mpTag=True,
+        MinRAM=2500,
+        MaxRAM=4000,
+        jobDesc={"jobParams": jobParams, "resourceParams": resourceParams, "optimizerParams": optimizerParams},
+    )
+    assert res["OK"] is False
+    res = ce.getCEStatus()
+    assert res["OK"] is True
+    assert res["SubmittedJobs"] == 2
+    assert res["RunningJobs"] == 0
+    assert res["WaitingJobs"] == 0
+    assert res["AvailableProcessors"] == available_processors
+    assert res["AvailableRAM"] == ram
+    _stopJob(1)
+    for ff in ["testJob.py", "stop_job_3", "job.info", "std.out"]:
+        if os.path.isfile(ff):
+            os.remove(ff)