check distributed files availability after transfer

Author: zhaoyuyoung

python/lsst/ctrl/bps/panda/utils.py

@@ -205,18 +205,19 @@ def copy_files_for_distribution(files_to_stage, file_distribution_uri, max_copy_
             folder_uri = file_distribution_uri.join(folder_name, forceDirectory=False)
             files_to_copy[ResourcePath(local_pfn, forceDirectory=False)] = folder_uri
 
-    copy_executor = concurrent.futures.ThreadPoolExecutor(max_workers=max_copy_workers)
-    future_file_copy = []
-    for src, trgt in files_to_copy.items():
-        _LOG.debug("Staging %s to %s", src, trgt)
-        # S3 clients explicitly instantiate here to overpass this
-        # https://stackoverflow.com/questions/52820971/is-boto3-client-thread-safe
-        trgt.exists()
-        future_file_copy.append(copy_executor.submit(trgt.transfer_from, src, transfer="copy"))
-
-    for future in concurrent.futures.as_completed(future_file_copy):
-        if future.result() is not None:
-            raise RuntimeError("Error of placing files to the distribution point")
+    with concurrent.futures.ThreadPoolExecutor(max_workers=max_copy_workers) as copy_executor:
+        future_file_copy = {}
+        for src, trgt in files_to_copy.items():
+            _LOG.debug("Staging %s to %s", src, trgt)
+            future = copy_executor.submit(trgt.transfer_from, src, transfer="copy")
+            future_file_copy[future] = trgt
+
+        for future in concurrent.futures.as_completed(future_file_copy):
+            trgt = future_file_copy[future]
+            if future.result() is not None:
+                raise RuntimeError(f"Error of placing file to the distribution point: {trgt}")
+            if not trgt.exists():
+                raise RuntimeError(f"File was not copied to the distribution point: {trgt}")
 
 
 def get_idds_client(config):