Keep igzip_threaded workers running after the opening thread ends

CHANGELOG.rst

@@ -7,6 +7,14 @@ Changelog
 .. This document is user facing. Please word the changes in such a way
 .. that users understand how the changes affect the new version.
 
+version 1.10.0-dev
+------------------
++ Fix ``igzip_threaded`` files that stopped working when the thread that
+  opened them ended, for example a ``ThreadPoolExecutor`` thread after
+  ``shutdown()``. Reads returned truncated data without an error, and writes
+  waited forever.
+
+
 version 1.9.1
 -----------------
 + Fix BrokenPipeError when piping ``python -m isal.igzip`` into ``head``,

src/isal/igzip_threaded.py

@@ -97,24 +97,32 @@ def __init__(self, filename, queue_size=2, block_size=1024 * 1024):
         self.worker = threading.Thread(target=self._decompress)
         self._closed = False
         self.running = True
+        self._main_thread = threading.main_thread()
         self._calling_thread = threading.current_thread()
         self.worker.start()
 
     def _check_closed(self, msg=None):
         if self._closed:
             raise ValueError("I/O operation on closed file")
 
+    def _calling_threads_alive(self) -> bool:
+        # Workers stop when both the thread that opened the file and the main
+        # thread have ended, so an unclosed file does not block exit. Check
+        # both: the opening thread may end first, and at exit Python marks
+        # the main thread stopped before it waits for the other threads.
+        return self._calling_thread.is_alive() or self._main_thread.is_alive()
+
     def _decompress(self):
         block_size = self.block_size
         block_queue = self.queue
-        while self.running and self._calling_thread.is_alive():
+        while self.running and self._calling_threads_alive():
             try:
                 data = self.fileobj.read(block_size)
             except Exception as e:
                 self.exception = e
                 data = b""
             # An empty block tells the reader about EOF or an error
-            while self.running and self._calling_thread.is_alive():
+            while self.running and self._calling_threads_alive():
                 try:
                     block_queue.put(data, timeout=0.05)
                     break
@@ -234,6 +242,7 @@ def __init__(self,
         if "b" not in mode:
             mode += "b"
         self.lock = threading.Lock()
+        self._main_thread = threading.main_thread()
         self._calling_thread = threading.current_thread()
         self.exception: Optional[Exception] = None
         self.level = level
@@ -289,6 +298,10 @@ def _check_closed(self, msg=None):
         if self._closed:
             raise ValueError("I/O operation on closed file")
 
+    def _calling_threads_alive(self) -> bool:
+        # See _ThreadedGzipReader._calling_threads_alive().
+        return self._calling_thread.is_alive() or self._main_thread.is_alive()
+
     def _write_gzip_header(self):
         """Simple gzip header. Only xfl flag is set according to level."""
         magic1 = 0x1f
@@ -390,7 +403,7 @@ def _compress(self, index: int):
             try:
                 item = in_queue.get(timeout=0.05)
             except queue.Empty:
-                if not (self.running and self._calling_thread.is_alive()):
+                if not (self.running and self._calling_threads_alive()):
                     return
                 continue
             if item is None:  # Sentinel from stop()
@@ -418,7 +431,7 @@ def _write(self):
             try:
                 item = output_queue.get(timeout=0.05)
             except queue.Empty:
-                if not (self.running and self._calling_thread.is_alive()):
+                if not (self.running and self._calling_threads_alive()):
                     return
                 continue
             if item is None:  # Sentinel from stop()
@@ -440,7 +453,7 @@ def _compress_and_write(self):
             try:
                 item = in_queue.get(timeout=0.05)
             except queue.Empty:
-                if not (self.running and self._calling_thread.is_alive()):
+                if not (self.running and self._calling_threads_alive()):
                     return
                 continue
             if item is None:  # Sentinel from stop()

tests/test_igzip_threaded.py

@@ -5,6 +5,7 @@
 # This file is part of python-isal which is distributed under the
 # PYTHON SOFTWARE FOUNDATION LICENSE VERSION 2.
 
+import concurrent.futures
 import errno
 import gzip
 import io
@@ -237,6 +238,67 @@ def test_close_writer(threads):
     assert_writer_closed(f)
 
 
+def open_in_finished_thread(*args, **kwargs):
+    # Leaving the with block shuts down the executor, which ends its thread.
+    with concurrent.futures.ThreadPoolExecutor(1) as executor:
+        return executor.submit(igzip_threaded.open, *args, **kwargs).result()
+
+
+@pytest.mark.timeout(10)
+def test_threaded_reader_opened_in_finished_thread():
+    data = os.urandom(256 * 1024)
+    # Small blocks fill the queue, so the worker reads most of the file after
+    # the thread that opened it has ended.
+    with open_in_finished_thread(io.BytesIO(gzip.compress(data)), "rb",
+                                 block_size=8 * 1024) as f:
+        assert f.read() == data
+
+
+@pytest.fixture
+def queue_timeouts(monkeypatch):
+    """Collects the threads whose get() on a queue timed out."""
+    threads = set()
+
+    class Queue(queue.Queue):
+        def get(self, block=True, timeout=None):
+            try:
+                return super().get(block, timeout)
+            except queue.Empty:
+                threads.add(threading.current_thread())
+                raise
+
+    monkeypatch.setattr(igzip_threaded, "queue", types.SimpleNamespace(
+        Queue=Queue, Empty=queue.Empty, Full=queue.Full))
+    return threads
+
+
+@pytest.mark.timeout(10)
+@pytest.mark.parametrize("threads", [1, 2])
+def test_threaded_writer_opened_in_finished_thread(queue_timeouts, threads):
+    data = os.urandom(64 * 1024)
+    output = io.BytesIO()
+    f = open_in_finished_thread(output, "wb", threads=threads,
+                                block_size=8 * 1024)
+    # The workers check whether to stop when a queue get() times out. Wait
+    # until each worker has checked, or stopped, after the thread that
+    # opened the file ended, so a failing test leaves no worker behind.
+    workers = [f.raw.output_worker] + f.raw.compression_workers
+    queue_timeouts.clear()
+    while not all(w in queue_timeouts or not w.is_alive() for w in workers):
+        time.sleep(0.001)
+
+    def write_and_close():
+        f.write(data)
+        f.close()
+
+    # write() and close() wait forever if the workers stopped.
+    writer = threading.Thread(target=write_and_close, daemon=True)
+    writer.start()
+    writer.join(timeout=5)
+    assert not writer.is_alive()
+    assert gzip.decompress(output.getvalue()) == data
+
+
 # io.BufferedReader.writable() and io.BufferedWriter.readable() are always
 # False, so check the raw streams.
 def test_reader_not_writable():