Try 12

Author: forsyth2

zstash/extract.py

@@ -548,7 +548,7 @@ def extractFiles(  # noqa: C901
         raise
 
 
-# FIXME: C901 '_extractFiles_impl' is too complex (42)
+# FIXME: C901 '_extractFiles_impl' is too complex (51)
 def _extractFiles_impl(  # noqa: C901
     files: List[FilesRow],
     keep_files: bool,
@@ -575,16 +575,12 @@ def _extractFiles_impl(  # noqa: C901
     tfname: str
     newtar: bool = True
     nfiles: int = len(files)
+
+    # For multiprocess workers, we'll set up logging to queue later
+    # Don't capture logs immediately to avoid blocking on initial setup
+    setup_logging_later: bool = False
     if multiprocess_worker:
-        # All messages to the logger will now be sent to
-        # this queue, instead of sys.stdout.
-        sh = logging.StreamHandler(multiprocess_worker.print_queue)
-        sh.setLevel(logging.DEBUG)
-        formatter: logging.Formatter = logging.Formatter("%(levelname)s: %(message)s")
-        sh.setFormatter(formatter)
-        logger.addHandler(sh)
-        # Don't have the logger print to the console as the message come in.
-        logger.propagate = False
+        setup_logging_later = True
 
     for i in range(nfiles):
         files_row: FilesRow = files[i]
@@ -593,11 +589,31 @@ def _extractFiles_impl(  # noqa: C901
         if newtar:
             newtar = False
             tfname = os.path.join(cache, files_row.tar)
+
+            # Set up print queue synchronization NOW, after we know which tar we're processing
+            if setup_logging_later and multiprocess_worker:
+                # All messages to the logger will now be sent to
+                # this queue, instead of sys.stdout.
+                sh = logging.StreamHandler(multiprocess_worker.print_queue)
+                sh.setLevel(logging.DEBUG)
+                formatter: logging.Formatter = logging.Formatter(
+                    "%(levelname)s: %(message)s"
+                )
+                sh.setFormatter(formatter)
+                logger.addHandler(sh)
+                # Don't have the logger print to the console as the message come in.
+                logger.propagate = False
+                setup_logging_later = False
+
             # Everytime we're extracting a new tar, if running in parallel,
             # let the process know.
             # This is to synchronize the print statements.
             if multiprocess_worker:
-                multiprocess_worker.set_curr_tar(files_row.tar)
+                try:
+                    multiprocess_worker.set_curr_tar(files_row.tar)
+                except Exception:
+                    # If setting current tar fails, continue anyway
+                    pass
 
             # Use args.hpss directly - it's always set correctly
             if args.hpss is not None:
@@ -751,7 +767,11 @@ def _extractFiles_impl(  # noqa: C901
             failures.append(files_row)
 
         if multiprocess_worker:
-            multiprocess_worker.print_contents()
+            try:
+                multiprocess_worker.print_contents()
+            except (TimeoutError, Exception):
+                # If printing fails, continue - work is still done
+                pass
 
         # Close current archive?
         if i == nfiles - 1 or files[i].tar != files[i + 1].tar:
@@ -762,7 +782,12 @@ def _extractFiles_impl(  # noqa: C901
             tar.close()
 
             if multiprocess_worker:
-                multiprocess_worker.done_enqueuing_output_for_tar(files_row.tar)
+                try:
+                    multiprocess_worker.done_enqueuing_output_for_tar(files_row.tar)
+                except TimeoutError:
+                    # If we timeout trying to signal completion, just continue
+                    # The monitor synchronization failed but the work is done
+                    pass
 
             # Open new archive next time
             newtar = True
@@ -781,7 +806,11 @@ def _extractFiles_impl(  # noqa: C901
 
     if multiprocess_worker:
         # If there are things left to print, print them.
-        multiprocess_worker.print_all_contents()
+        try:
+            multiprocess_worker.print_all_contents()
+        except (TimeoutError, Exception):
+            # If final printing fails, at least report failures
+            pass
 
         # Add the failures to the queue.
         # When running with multiprocessing, the function multiprocess_extract()

zstash/parallel.py

@@ -73,7 +73,7 @@ def wait_turn(
         with self._cv:
             attempted: bool = False
             workers_curr_tar_bytes = workers_curr_tar.encode("utf-8")
-            max_wait_time = 120.0  # Maximum 2 minutes total wait
+            max_wait_time = 10.0  # Maximum 10 seconds total wait - fail fast
             start_time = time.time()
 
             while self._current_tar.value != workers_curr_tar_bytes: