resolve and display causes of error as is best possible

currently we just get out cleanly, and make sure the causation is preserved.

Later (v4.0) we may introduce a TransferInterrupted or similar which can be
more useful than RuntimeError for indicating what happened.  (in place of
the RuntimeError("xxx failed.") in irods/manager/data_object_manager.)

Author: d-w-moore

irods/manager/data_object_manager.py

@@ -246,15 +246,19 @@ def _download(self, obj, local_path, num_threads, updatables=(), **options):
             if self.should_parallelize_transfer(
                 num_threads, o, open_options=options.items()
             ):
-                if not self.parallel_get(
-                    (obj, o),
-                    local_file,
-                    num_threads=num_threads,
-                    target_resource_name=options.get(kw.RESC_NAME_KW, ""),
-                    data_open_returned_values=data_open_returned_values_,
-                    updatables=updatables,
-                ):
-                    raise RuntimeError("parallel get failed")
+                error = RuntimeError("parallel get failed")
+                try:
+                    if not self.parallel_get(
+                        (obj, o),
+                        local_file,
+                        num_threads=num_threads,
+                        target_resource_name=options.get(kw.RESC_NAME_KW, ""),
+                        data_open_returned_values=data_open_returned_values_,
+                        updatables=updatables,
+                    ):
+                        raise error
+                except BaseException as e:
+                    raise error from e
             else:
                 with open(local_file, "wb") as f:
                     for chunk in chunks(o, self.READ_BUFFER_SIZE):
@@ -354,17 +358,21 @@ def put(
             ):
                 o = deferred_call(self.open, (obj, "w"), options)
                 f.close()
-                if not self.parallel_put(
-                    local_path,
-                    (obj, o),
-                    total_bytes=sizelist[0],
-                    num_threads=num_threads,
-                    target_resource_name=options.get(kw.RESC_NAME_KW, "")
-                    or options.get(kw.DEST_RESC_NAME_KW, ""),
-                    open_options=options,
-                    updatables=updatables,
-                ):
-                    raise RuntimeError("parallel put failed")
+                error = RuntimeError("parallel put failed")
+                try:
+                    if not self.parallel_put(
+                        local_path,
+                        (obj, o),
+                        total_bytes=sizelist[0],
+                        num_threads=num_threads,
+                        target_resource_name=options.get(kw.RESC_NAME_KW, "")
+                        or options.get(kw.DEST_RESC_NAME_KW, ""),
+                        open_options=options,
+                        updatables=updatables,
+                    ):
+                        raise error
+                except BaseException as e:
+                    raise error from e
             else:
                 with self.open(obj, "w", **options) as o:
                     # Set operation type to trigger acPostProcForPut

irods/parallel.py

@@ -473,7 +473,7 @@ def bytes_range_for_thread(i, num_threads, total_bytes, chunk):
     transfer_managers[mgr] = (_quit_current_transfer, [id(mgr)])
 
     try:
-        transfer_aborted = False
+        thread_setup_error = None
 
         for byte_range in ranges:
             if Io is None:
@@ -510,22 +510,23 @@ def bytes_range_for_thread(i, num_threads, total_bytes, chunk):
                 )
             except RuntimeError as error: 
                 # Executor was probably shut down before parallel transfer could be initiated.
-                transfer_aborted = True
+                thread_setup_error = error
                 break
             else:
                 mgr.add_future(f)
 
             counter += 1
             Io = File = None
 
-        if transfer_aborted:
-            return ((bytes_transferred:=0), total_size)
+        if thread_setup_error:
+            raise thread_setup_error
+        
+        bytes_transferred = 0
 
         if Operation.isNonBlocking():
             transfer_managers[mgr] = None
             return (futures, mgr, queueObject)
         else:
-            bytes_transferred = 0
             # Enable user attempts to cancel the current synchronous transfer.
             # At any given time, only one transfer manager key should map to a tuple object T.
             # You should be able to quit all threads of the current transfer by calling T[0](*T[1]).
@@ -538,7 +539,7 @@ def bytes_range_for_thread(i, num_threads, total_bytes, chunk):
         return (bytes_transferred, total_size)
 
     except BaseException as e:
-        if isinstance(e, (SystemExit, KeyboardInterrupt)):
+        if isinstance(e, (SystemExit, KeyboardInterrupt, RuntimeError)):
             mgr.quit()
         raise