feat (FileStorage): allow to specify block size

src/DIRAC/Resources/Catalog/FCConditionParser.py

@@ -1,6 +1,8 @@
 """
    Contains the mechanism to evaluate whether to use or not a catalog
 """
+from functools import lru_cache
+
 from pyparsing import infix_notation, opAssoc, Word, printables, Literal, Suppress
 
 from DIRAC import S_OK, gLogger
@@ -199,6 +201,25 @@ def __init__(self, vo=None, ro_methods=None):
 
         self.log = gLogger.getSubLogger(self.__class__.__name__)
 
+        # Bound per-instance cache (as opposed to a class-level one) so it is
+        # garbage collected along with the instance instead of growing forever
+        self.__parseCondition = lru_cache(maxsize=None)(self.__parseConditionUncached)
+
+    def __parseConditionUncached(self, conditionString):
+        """Parse a condition string into its evaluation tree.
+
+        Condition strings are static (they come from the CS or are fixed at
+        call time), so parsing them - which instantiates the (possibly costly)
+        plugins via :class:`ObjectLoader` - is cached per-instance in
+        :attr:`__parseCondition` and reused for every lfn/call.
+
+        :param str conditionString: the condition to parse
+        :returns: the root node (bool operator or PluginOperand) of the parsed expression
+        """
+        # res is a tuple whose first and only element is either
+        # one of the bool operator defined above, or a PluginOperand
+        return self.__boolExpr.parseString(conditionString)[0]
+
     def __evaluateCondition(self, conditionString, **kwargs):
         """Evaluate a condition against attributes, typically lfn.
         CAUTION: lfns are here given one by one
@@ -207,11 +228,8 @@ def __evaluateCondition(self, conditionString, **kwargs):
 
         self.log.debug(f"Testing {conditionString} against {kwargs}")
 
-        # Parse all the condition and evaluate it
-        # res is a tuple whose first and only element is either
-        # one of the bool operator defined above, or a PluginOperand
-        res = self.__boolExpr.parseString(conditionString)
-        res = res[0].eval(**kwargs)
+        parsedCondition = self.__parseCondition(conditionString)
+        res = parsedCondition.eval(**kwargs)
 
         self.log.debug(f"Evaluated to {res}")
 

src/DIRAC/Resources/Storage/FileStorage.py

@@ -148,6 +148,20 @@ def __init__(self, storageName, parameters):
 
         self.pluginName = "File"
         self.protocol = self.protocolParameters["Protocol"]
+        self.block_size = int(self._allProtocolParameters.get("BlockSize", 0))
+
+    def _do_copy(self, src, dst):
+        """
+        Either use shutil.copy2 or copyfileobj if the site is
+        strongly opinionated on the block size
+        """
+        if not self.block_size:
+            return shutil.copy2(src, dst)
+        else:
+            with open(src, "rb") as fd_r:
+                with open(dst, "wb") as fd_w:
+                    shutil.copyfileobj(fd_r, fd_w, self.block_size)
+            shutil.copystat(src, dst)
 
     def getURLBase(self, withWSUrl=False):
         return S_OK(self.basePath)
@@ -243,7 +257,7 @@ def getFile(self, path, localPath=False):
             try:
                 fileName = os.path.basename(src_url)
                 dest_url = os.path.join(localPath, fileName)
-                shutil.copy2(src_url, dest_url)
+                self._do_copy(src_url, dest_url)
 
                 fileSize = os.path.getsize(dest_url)
                 successful[src_url] = fileSize
@@ -277,7 +291,7 @@ def putFile(self, path, sourceSize=0):
                 dirname = os.path.dirname(dest_url)
                 if not os.path.exists(dirname):
                     os.makedirs(dirname)
-                shutil.copy2(src_file, dest_url)
+                self._do_copy(src_file, dest_url)
                 fileSize = os.path.getsize(dest_url)
                 try:
                     src_cks = fileAdler(src_file)