[9.1] Improve DFC speed

src/DIRAC/DataManagementSystem/DB/FileCatalogComponents/FileManager/FileManagerBase.py

@@ -4,7 +4,10 @@
 
 import os
 import stat
+import threading
 
+from cachetools import LRUCache, cachedmethod
+from cachetools.keys import hashkey
 from DIRAC import S_OK, S_ERROR, gLogger
 from DIRAC.Core.Utilities.List import intListToString
 from DIRAC.Core.Utilities.Pfn import pfnunparse
@@ -16,6 +19,8 @@ class FileManagerBase:
     def __init__(self, database=None):
         self.db = database
         self.statusDict = {}
+        self.statusIntCache = LRUCache(maxsize=8)
+        self.statusIntCacheLock = threading.RLock()
 
     def _getConnection(self, connection):
         if connection:
@@ -1063,19 +1068,28 @@ def getReplicaStatus(self, lfns, connection=False):
     # General usage methods
     #
 
+    @cachedmethod(
+        lambda self: self.statusIntCache,
+        key=lambda _, status, *args, **kwargs: hashkey(status),
+        lock=lambda self: self.statusIntCacheLock,
+    )
     def _getStatusInt(self, status, connection=False):
         connection = self._getConnection(connection)
         req = "SELECT StatusID FROM FC_Statuses WHERE Status = %s"
         res = self.db._query(req, args=(status,), conn=connection)
         if not res["OK"]:
             return res
         if res["Value"]:
-            return S_OK(res["Value"][0][0])
+            statusID = res["Value"][0][0]
+            self.statusDict[statusID] = status
+            return S_OK(statusID)
         req = "INSERT INTO FC_Statuses (Status) VALUES (%s)"
         res = self.db._update(req, args=(status,), conn=connection)
         if not res["OK"]:
             return res
-        return S_OK(res["lastRowId"])
+        statusID = res["lastRowId"]
+        self.statusDict[statusID] = status
+        return S_OK(statusID)
 
     def _getIntStatus(self, statusID, connection=False):
         if statusID in self.statusDict:

src/DIRAC/DataManagementSystem/DB/FileCatalogComponents/FileManager/FileManagerPs.py

@@ -1,5 +1,5 @@
-""" FileManager for ... ?
-"""
+"""FileManager for ... ?"""
+
 import os
 import datetime
 
@@ -294,7 +294,11 @@ def _insertFiles(self, lfns, uid, gid, connection=False):
             fileDesc[(dirID, fileName)] = lfn
 
         chunkSize = 200
-        allChunks = list(self.__chunks(list(lfns), chunkSize))
+        if len(lfns) == 1:
+            allChunks = []
+            lfnsToRetry = lfns
+        else:
+            allChunks = list(self.__chunks(list(lfns), chunkSize))
 
         for lfnChunk in allChunks:
             result = self.__insertMultipleFiles(fileValues, lfnChunk)
@@ -529,7 +533,11 @@ def _insertReplicas(self, lfns, master=False, connection=False):
                 repValues[lfn] = (fileID, seID, statusID, replicaType, pfn)
                 repDesc[(fileID, seID)] = lfn
 
-        allChunks = list(self.__chunks(list(lfns), chunkSize))
+        if len(lfns) == 1:
+            allChunks = []
+            lfnsToRetry = lfns
+        else:
+            allChunks = list(self.__chunks(list(lfns), chunkSize))
 
         for lfnChunk in allChunks:
             result = self.__insertMultipleReplicas(repValues, lfnChunk)

src/DIRAC/DataManagementSystem/DB/FileCatalogWithFkAndPsDB.sql

@@ -828,8 +828,13 @@ BEGIN
   DEALLOCATE PREPARE stmt;
 
 
-  SET @sql = CONCAT('INSERT INTO FC_DirectoryUsage (DirID, SEID, SESize, SEFiles) SELECT SQL_NO_CACHE f.DirID, 1, f.Size, 1 FROM FC_Files f WHERE ', fileDesc);
-  SET @sql = CONCAT(@sql, ' ON DUPLICATE KEY UPDATE SESize = SESize + f.Size, SEFiles = SEFiles + 1');
+  SET @sql = CONCAT(
+    'INSERT INTO FC_DirectoryUsage (DirID, SEID, SESize, SEFiles) ',
+    'SELECT SQL_NO_CACHE f.DirID, 1, SUM(f.Size), COUNT(*) ',
+    'FROM FC_Files f WHERE ', fileDesc,
+    ' GROUP BY f.DirID '
+  );
+  SET @sql = CONCAT(@sql, ' ON DUPLICATE KEY UPDATE SESize = SESize + VALUES(SESize), SEFiles = SEFiles + VALUES(SEFiles)');
 
   PREPARE stmt FROM @sql;
   EXECUTE stmt;
@@ -1068,8 +1073,13 @@ BEGIN
   DEALLOCATE PREPARE stmt;
 
 
-  SET @sql = CONCAT('INSERT INTO FC_DirectoryUsage (DirID, SEID, SESize, SEFiles) SELECT SQL_NO_CACHE f.DirID, SEID, f.Size, 1 FROM FC_Files f, FC_Replicas r WHERE f.FileID = r.FileID AND (', replicaDesc);
-  SET @sql = CONCAT(@sql, ') ON DUPLICATE KEY UPDATE SESize = SESize + f.Size, SEFiles = SEFiles + 1');
+  SET @sql = CONCAT(
+    'INSERT INTO FC_DirectoryUsage (DirID, SEID, SESize, SEFiles) ',
+    'SELECT SQL_NO_CACHE f.DirID, r.SEID, SUM(f.Size), COUNT(*) ',
+    'FROM FC_Files f JOIN FC_Replicas r ON f.FileID = r.FileID WHERE (', replicaDesc,
+    ') GROUP BY f.DirID, r.SEID'
+  );
+  SET @sql = CONCAT(@sql, ' ON DUPLICATE KEY UPDATE SESize = SESize + VALUES(SESize), SEFiles = SEFiles + VALUES(SEFiles)');
   -- insert into FC_DirectoryUsage (DirID, SEID, SESize, SEFiles) select f.DirID, 1, f.Size as size_diff, 1 as file_diff from FC_Files f where (DirID = 1 and FileName = 'a.txt') OR (DirID = 1 and FileName = '1.txt') on duplicate key update SESize = SESize + f.Size, SEFiles = SEFiles + 1;
   PREPARE stmt FROM @sql;
   EXECUTE stmt;

tests/Integration/DataManagementSystem/Test_DataManager.py

@@ -179,7 +179,7 @@ def test_registerFile(dm, tempFile):
     fileSize = 10000
     storageElementName = "SE-1"
     fileGuid = makeGuid()
-    checkSum = None
+    checkSum = "fakeCks"
     fileTuple = (lfn, physicalFile, fileSize, storageElementName, fileGuid, checkSum)
     registerRes = dm.registerFile(fileTuple)
     removeFileRes = dm.removeFile(lfn)
@@ -198,7 +198,7 @@ def test_registerReplica(dm, tempFile):
     fileSize = 10000
     storageElementName = "SE-1"
     fileGuid = makeGuid()
-    checkSum = None
+    checkSum = "fakeCks"
     fileTuple = (lfn, physicalFile, fileSize, storageElementName, fileGuid, checkSum)
     registerRes = dm.registerFile(fileTuple)
     seName = "SE-1"