From 137c4fc9b5c2fbee849b1f64e7db5370e617eec9 Mon Sep 17 00:00:00 2001 From: Chris Burr Date: Thu, 16 Jul 2026 10:12:47 +0200 Subject: [PATCH 1/2] feat: allow fractional MatchingDelay for sub-second matcher throttling MatchingDelay values were cast with int(), so the minimum spacing between matched jobs of a given type was 1 second (>=1 job/s). Sites needing a higher steady start rate (e.g. to smooth a spiky batch of a few hundred jobs into a couple of hundred per two minutes) could not express the required sub-second spacing. Parse the MatchingDelay path as float via a cast parameter on __extractCSData, so fractional seconds (e.g. 0.5) are honoured; DictCache already builds the TTL with timedelta(seconds=...), which accepts floats. RunningLimit continues to parse as int. --- .../ConfReference/Operations/JobScheduling/index.rst | 2 ++ src/DIRAC/WorkloadManagementSystem/Client/Limiter.py | 10 +++++++--- 2 files changed, 9 insertions(+), 3 deletions(-) diff --git a/docs/source/AdministratorGuide/Configuration/ConfReference/Operations/JobScheduling/index.rst b/docs/source/AdministratorGuide/Configuration/ConfReference/Operations/JobScheduling/index.rst index 101f119460f..d9f860f7db5 100644 --- a/docs/source/AdministratorGuide/Configuration/ConfReference/Operations/JobScheduling/index.rst +++ b/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 ======== diff --git a/src/DIRAC/WorkloadManagementSystem/Client/Limiter.py b/src/DIRAC/WorkloadManagementSystem/Client/Limiter.py index 226ebeb9693..016acd06b27 100644 --- a/src/DIRAC/WorkloadManagementSystem/Client/Limiter.py +++ b/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) @@ -200,7 +204,7 @@ def __getRunningCondition(self, siteName, gridCE=None): def updateDelayCounters(self, siteName, jid): # 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"] From 0ff974caf19489e5d2c2c9bc176fa36c577a1e56 Mon Sep 17 00:00:00 2001 From: Chris Burr Date: Thu, 16 Jul 2026 13:32:08 +0200 Subject: [PATCH 2/2] fix: arm MatchingDelay counter right after matching The matching-delay counter was armed at the very end of Matcher.selectJob, ~7 DB round-trips after the negative condition was read. Under the Matcher's concurrent request handling, every thread in that window reads the type as un-delayed and matches it, so bursts of jobs slip past the delay (observed ~15-25% of matches landing <0.1s apart at a site with a 1s delay). Move the arming to immediately after the match is confirmed Waiting, and reuse the JobType already fetched for the status check so no extra query is added: updateDelayCounters now accepts pre-fetched attributes and only queries the DB for any configured attribute that is missing. This shrinks the read->arm window from ~7 round-trips to the match itself, cutting the leak, and removes a query from the match path. --- .../WorkloadManagementSystem/Client/Limiter.py | 18 ++++++++++++------ .../WorkloadManagementSystem/Client/Matcher.py | 11 +++++++---- 2 files changed, 19 insertions(+), 10 deletions(-) diff --git a/src/DIRAC/WorkloadManagementSystem/Client/Limiter.py b/src/DIRAC/WorkloadManagementSystem/Client/Limiter.py index 016acd06b27..3d70386dfe1 100644 --- a/src/DIRAC/WorkloadManagementSystem/Client/Limiter.py +++ b/src/DIRAC/WorkloadManagementSystem/Client/Limiter.py @@ -201,7 +201,7 @@ 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, cast=float) @@ -217,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() diff --git a/src/DIRAC/WorkloadManagementSystem/Client/Matcher.py b/src/DIRAC/WorkloadManagementSystem/Client/Matcher.py index 6a81b8f85d2..d80b4adb6c9 100644 --- a/src/DIRAC/WorkloadManagementSystem/Client/Matcher.py +++ b/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)