[9.1] feat: allow fractional MatchingDelay for sub-second matcher throttling

docs/source/AdministratorGuide/Configuration/ConfReference/Operations/JobScheduling/index.rst

@@ -42,6 +42,8 @@ running`_. But instead of defining the maximum amount of jobs that can run at a
 For instance *JobScheduling/MatchingDelay/DIRAC.Somewhere.co/JobType/MonteCarlo=10* won't allow jobs with *JobType=MonteCarlo* to start at
 site *DIRAC.Somewhere.co* with less than 10 seconds between them.
 
+The value may be fractional (e.g. ``0.5``) to allow more than one job to start per second.
+
 Example
 ========
 

src/DIRAC/WorkloadManagementSystem/Client/Limiter.py

@@ -132,9 +132,13 @@ def __mergeCond(self, negCond, addCond):
                     negCond[attr].append(value)
         return negCond
 
-    def __extractCSData(self, section):
+    def __extractCSData(self, section, cast=int):
         """Extract limiting information from the CS in the form:
         { 'JobType' : { 'Merge' : 20, 'MCGen' : 1000 } }
+
+        :param cast: callable used to convert each value. ``int`` for job-count
+            limits (RunningLimit) and ``float`` for delays in seconds (MatchingDelay),
+            which may be sub-second.
         """
         stuffDict = self.csDictCache.get(section)
         if stuffDict:
@@ -153,7 +157,7 @@ def __extractCSData(self, section):
                 return result
             attLimits = result["Value"]
             try:
-                attLimits = {k: int(attLimits[k]) for k in attLimits}
+                attLimits = {k: cast(attLimits[k]) for k in attLimits}
             except Exception as excp:
                 errMsg = f"{section}/{attName} has to contain numbers: {str(excp)}"
                 self.log.error(errMsg)
@@ -197,10 +201,10 @@ def __getRunningCondition(self, siteName, gridCE=None):
         # negCond is something like : {'JobType': ['Merge']}
         return S_OK(negCond)
 
-    def updateDelayCounters(self, siteName, jid):
+    def updateDelayCounters(self, siteName, jid, knownAtts=None):
         # Get the info from the CS
         siteSection = f"{self.__matchingDelaySection}/{siteName}"
-        result = self.__extractCSData(siteSection)
+        result = self.__extractCSData(siteSection, cast=float)
         if not result["OK"]:
             return result
         delayDict = result["Value"]
@@ -213,11 +217,17 @@ def updateDelayCounters(self, siteName, jid):
                 self.log.error("Attribute does not exist in the JobDB. Please fix it!", f"({attName})")
             else:
                 attNames.append(attName)
-        result = self.jobDB.getJobAttributes(jid, attNames)
-        if not result["OK"]:
-            self.log.error("Error while retrieving attributes", f"coming from {siteSection}: {result['Message']}")
-            return result
-        atts = result["Value"]
+        # Reuse any attributes the caller already fetched; only query the DB for the missing ones.
+        # This lets the caller arm the counter right after matching without an extra round trip.
+        knownAtts = knownAtts or {}
+        atts = {attName: knownAtts[attName] for attName in attNames if attName in knownAtts}
+        missing = [attName for attName in attNames if attName not in atts]
+        if missing:
+            result = self.jobDB.getJobAttributes(jid, missing)
+            if not result["OK"]:
+                self.log.error("Error while retrieving attributes", f"coming from {siteSection}: {result['Message']}")
+                return result
+            atts.update(result["Value"])
         # Create the DictCache if not there
         if siteName not in self.delayMem:
             self.delayMem[siteName] = DictCache()

src/DIRAC/WorkloadManagementSystem/Client/Matcher.py

@@ -97,7 +97,7 @@ def selectJob(self, resourceDescription, credDict):
                 return {}
 
             jobID = result["jobId"]
-            resAtt = self.jobDB.getJobAttributes(jobID, ["Status"])
+            resAtt = self.jobDB.getJobAttributes(jobID, ["Status", "JobType"])
             if not resAtt["OK"]:
                 raise RuntimeError("Could not retrieve job attributes")
             if not resAtt["Value"]:
@@ -109,6 +109,12 @@ def selectJob(self, resourceDescription, credDict):
                     raise RuntimeError(result["Message"])
                 raise RuntimeError(f"Job {str(jobID)} is not in Waiting state")
 
+            # Arm the matching-delay counter right after the match is confirmed, to keep the window
+            # in which concurrent requests can slip past the delay as small as possible. Reuse the
+            # attributes just fetched (JobType) so this does not add a DB query.
+            if self.opsHelper.getValue("JobScheduling/CheckMatchingDelay", True):
+                self.limiter.updateDelayCounters(resourceDict["Site"], jobID, knownAtts=resAtt["Value"])
+
             self._reportStatus(resourceDict, jobID)
 
             result = self.jobDB.getJobJDL(jobID)
@@ -133,9 +139,6 @@ def selectJob(self, resourceDescription, credDict):
             if not resAtt["Value"]:
                 raise RuntimeError("No attributes returned for job")
 
-            if self.opsHelper.getValue("JobScheduling/CheckMatchingDelay", True):
-                self.limiter.updateDelayCounters(resourceDict["Site"], jobID)
-
             pilotInfoReportedFlag = resourceDict.get("PilotInfoReportedFlag", False)
             if not pilotInfoReportedFlag:
                 self._updatePilotInfo(resourceDict)