Merge pull request #6794 from chaen/v8.0_FIX_speadupOperations

[8.0] Speed up ReplicateAndRegister operation

Author: chrisburr

pyproject.toml

@@ -28,6 +28,31 @@ exclude = [
     '/tests/'
 ]
 
+[tool.pylint.basic]
+
+# We mostly have CamelCase, with a few differences.
+# In tests we have quite some snake_case, mostly due to pytest
+# We can instruct pylint to understand both, but the problem is that it
+# will stick to one style per file (i.e if the first variable is snake,
+# all the following must be snake)
+# It's not quite the case yet...
+# For the time being, I wrote the regex that matches best our code.
+# (except for the services with their export_ and types_ ...)
+# We will see about tests later...
+# See https://pylint.readthedocs.io/en/latest/user_guide/messages/convention/invalid-name.html#multiple-naming-styles-for-custom-regular-expressions
+
+# Camel case with capital letter first
+class-rgx = '([A-Z][a-z]*)+$'
+module-rgx = '([A-Z][a-z]*)+$'
+
+# Attributes, variables, functions and methods
+# are camelCase, but can start with one or two understcore
+attr-rgx = '(?:_*[a-z]+([A-Z][a-z]*)*)$'
+variable-rgx = '(?:_*[a-z]+([A-Z][a-z]*)*)$'
+function-rgx = '(?:_*[a-z]+([A-Z][a-z]*)*)$'
+method-rgx = '(?:_*[a-z]+([A-Z][a-z]*)*)$'
+
+argument-naming-style = "camelCase"
 
 [tool.pylint.main]
 # Files or directories to be skipped. They should be base names, not paths.

src/DIRAC/DataManagementSystem/Agent/RequestOperations/RegisterReplica.py

@@ -55,14 +55,17 @@ def __call__(self):
         # # loop over files
         registerOperations = {}
         successReplicas = 0
+
+        targetSE = self.operation.targetSEList[0]
+        replicaTuples = [(opFile.LFN, opFile.PFN, targetSE) for opFile in waitingFiles]
+
+        registerReplica = self.dm.registerReplica(replicaTuples, catalogs)
+
         for opFile in waitingFiles:
             # # get LFN
             lfn = opFile.LFN
             # # and others
-            targetSE = self.operation.targetSEList[0]
-            replicaTuple = (lfn, opFile.PFN, targetSE)
-            # # call ReplicaManager
-            registerReplica = self.dm.registerReplica(replicaTuple, catalogs)
+
             # # check results
             if not registerReplica["OK"] or lfn in registerReplica["Value"]["Failed"]:
                 # There have been some errors

src/DIRAC/DataManagementSystem/Agent/RequestOperations/ReplicateAndRegister.py

@@ -46,8 +46,15 @@
 from DIRAC.MonitoringSystem.Client.MonitoringReporter import MonitoringReporter
 
 
-def filterReplicas(opFile, logger=None, dataManager=None, opSources=None):
-    """filter out banned/invalid source SEs"""
+def filterReplicas(opFile, logger=None, dataManager=None, opSources=None, activeReplicas=None):
+    """filter out banned/invalid source SEs
+
+    :param list opSources: list of SE names to which limit the possible sources
+    :param dict activeReplicas: the result of dm.getActiveReplicas(*)["Value"]. Used as a cache
+
+    :returns: Valid list of SEs valid as source
+
+    """
 
     if logger is None:
         logger = gLogger
@@ -57,19 +64,21 @@ def filterReplicas(opFile, logger=None, dataManager=None, opSources=None):
     log = logger.getSubLogger("filterReplicas")
     result = defaultdict(list)
 
-    replicas = dataManager.getActiveReplicas(opFile.LFN, getUrl=False, preferDisk=True)
-    if not replicas["OK"]:
-        log.error("Failed to get active replicas", replicas["Message"])
-        return replicas
+    if not activeReplicas:
+        res = dataManager.getActiveReplicas(opFile.LFN, getUrl=False, preferDisk=True)
+        if not res["OK"]:
+            log.error("Failed to get active replicas", res["Message"])
+            return res
+        activeReplicas = res["Value"]
+
     reNotExists = re.compile(r".*such file.*")
-    replicas = replicas["Value"]
-    failed = replicas["Failed"].get(opFile.LFN, "")
+    failed = activeReplicas["Failed"].get(opFile.LFN, "")
     if reNotExists.match(failed.lower()):
         opFile.Status = "Failed"
         opFile.Error = failed
         return S_ERROR(failed)
 
-    replicas = replicas["Successful"].get(opFile.LFN, {})
+    replicas = activeReplicas["Successful"].get(opFile.LFN, {})
 
     # If user set sourceSEs, only consider those replicas
     if opSources:
@@ -253,9 +262,15 @@ def _addMetadataToFiles(self, toSchedule):
 
         return S_OK(filesToSchedule)
 
-    def _filterReplicas(self, opFile):
+    def _filterReplicas(self, opFile, activeReplicas):
         """filter out banned/invalid source SEs"""
-        return filterReplicas(opFile, logger=self.log, dataManager=self.dm, opSources=self.operation.sourceSEList)
+        return filterReplicas(
+            opFile,
+            logger=self.log,
+            dataManager=self.dm,
+            opSources=self.operation.sourceSEList,
+            activeReplicas=activeReplicas,
+        )
 
     def _checkExistingFTS3Operations(self):
         """
@@ -342,13 +357,22 @@ def fts3Transfer(self):
         if self.rmsMonitoring:
             self.rmsMonitoringReporter.addRecord(self.createRMSRecord("Attempted", len(self.getWaitingFilesList())))
 
-        for opFile in self.getWaitingFilesList():
+        waitingFiles = self.getWaitingFilesList()
+
+        allLFNs = [opFile.LFN for opFile in waitingFiles]
+        res = self.dm.getActiveReplicas(allLFNs, getUrl=False, preferDisk=True)
+        if not res["OK"]:
+            self.log.error("Failed to get active replicas", res["Message"])
+            return res
+        allActiveReplicas = res["Value"]
+
+        for opFile in waitingFiles:
             rmsFilesIds[opFile.FileID] = opFile
 
             opFile.Error = ""
 
             # # check replicas
-            replicas = self._filterReplicas(opFile)
+            replicas = self._filterReplicas(opFile, allActiveReplicas)
             if not replicas["OK"]:
                 continue
             replicas = replicas["Value"]
@@ -504,6 +528,13 @@ def dmTransfer(self, fromFTS=False):
         if self.rmsMonitoring:
             self.rmsMonitoringReporter.addRecord(self.createRMSRecord("Attempted", len(waitingFiles)))
 
+        allLFNs = [opFile.LFN for opFile in waitingFiles]
+        res = self.dm.getActiveReplicas(allLFNs, getUrl=False, preferDisk=True)
+        if not res["OK"]:
+            self.log.error("Failed to get active replicas", res["Message"])
+            return res
+        allActiveReplicas = res["Value"]
+
         for opFile in waitingFiles:
             if opFile.Error in (
                 "Couldn't get metadata",
@@ -518,7 +549,7 @@ def dmTransfer(self, fromFTS=False):
             lfn = opFile.LFN
 
             # Check if replica is at the specified source
-            replicas = self._filterReplicas(opFile)
+            replicas = self._filterReplicas(opFile, allActiveReplicas)
             if not replicas["OK"]:
                 self.log.error("Failed to check replicas", replicas["Message"])
                 continue