Merge pull request #8402 from chaen/v9.0_feat_speedupGetReplicas

feat (DataManager): batch getReplicas at the FC level for speedup

Author: fstagni

src/DIRAC/DataManagementSystem/Client/DataManager.py

@@ -22,7 +22,7 @@
 from DIRAC.Core.Utilities import DErrno
 from DIRAC.Core.Utilities.Adler import fileAdler, compareAdler
 from DIRAC.Core.Utilities.File import makeGuid, getSize
-from DIRAC.Core.Utilities.List import randomize, breakListIntoChunks
+from DIRAC.Core.Utilities.List import randomize
 from DIRAC.Core.Utilities.ReturnValues import returnSingleResult
 from DIRAC.Core.Security.ProxyInfo import getProxyInfo
 from DIRAC.Core.Security.ProxyInfo import getVOfromProxyGroup
@@ -1677,18 +1677,15 @@ def getReplicas(
         """get replicas from catalogue and filter if requested
         Warning: all filters are independent, hence active and preferDisk should be set if using forJobs
         """
-        catalogReplicas = {}
-        failed = {}
+
         if not protocol:
             protocol = self.registrationProtocol
 
-        for lfnChunk in breakListIntoChunks(lfns, 1000):
-            res = self.fileCatalog.getReplicas(lfnChunk, allStatus=allStatus)
-            if res["OK"]:
-                catalogReplicas.update(res["Value"]["Successful"])
-                failed.update(res["Value"]["Failed"])
-            else:
-                return res
+        res = self.fileCatalog.getReplicas(lfns, allStatus=allStatus)
+        if not res["OK"]:
+            return res
+        catalogReplicas = res["Value"]["Successful"]
+        failed = res["Value"]["Failed"]
         if not getUrl:
             for lfn in catalogReplicas:
                 catalogReplicas[lfn] = dict.fromkeys(catalogReplicas[lfn], True)

src/DIRAC/Resources/Catalog/FileCatalogClient.py

@@ -1,15 +1,18 @@
-""" The FileCatalogClient is a class representing the client of the DIRAC File Catalog
-"""
+"""The FileCatalogClient is a class representing the client of the DIRAC File Catalog"""
+
 import json
 import os
 
 from DIRAC import S_OK, S_ERROR
+from DIRAC.Core.Utilities.List import breakListIntoChunks
 from DIRAC.Core.Tornado.Client.ClientSelector import TransferClientSelector as TransferClient
 
 from DIRAC.ConfigurationSystem.Client.Helpers.Registry import getVOMSAttributeForGroup, getDNForUsername
 from DIRAC.Resources.Catalog.Utilities import checkCatalogArguments
 from DIRAC.Resources.Catalog.FileCatalogClientBase import FileCatalogClientBase
 
+GET_REPLICAS_CHUNK_SIZE = 10_000
+
 
 class FileCatalogClient(FileCatalogClientBase):
     """Client code to the DIRAC File Catalogue"""
@@ -135,14 +138,20 @@ def __init__(self, url=None, **kwargs):
     @checkCatalogArguments
     def getReplicas(self, lfns, allStatus=False, timeout=120):
         """Get the replicas of the given files"""
-        rpcClient = self._getRPC(timeout=timeout)
-        result = rpcClient.getReplicas(lfns, allStatus)
+        successful = {}
+        failed = {}
 
-        if not result["OK"]:
-            return result
+        for chunk in breakListIntoChunks(lfns, GET_REPLICAS_CHUNK_SIZE):
+            rpcClient = self._getRPC(timeout=timeout)
+            result = rpcClient.getReplicas(chunk, allStatus)
+
+            if not result["OK"]:
+                return result
+            successful.update(result["Value"]["Successful"])
+            failed.update(result["Value"]["Failed"])
 
         # If there is no PFN returned, just set the LFN instead
-        lfnDict = result["Value"]
+        lfnDict = {"Successful": successful, "Failed": failed}
         for lfn in lfnDict["Successful"]:
             for se in lfnDict["Successful"][lfn]:
                 if not lfnDict["Successful"][lfn][se]: