From 8160d53d70b0a1feddec187b3276fd9d9d60e45f Mon Sep 17 00:00:00 2001 From: Parker Loflin Date: Thu, 5 Nov 2026 17:17:44 -0500 Subject: [PATCH] Keep igzip_threaded workers running after the opening thread ends The workers stopped when the thread that called open() ended. Reads then returned truncated data and writes waited forever. Now they stop only when the main thread has ended too, which still lets Python exit with an unclosed file. --- CHANGELOG.rst | 8 +++++ src/isal/igzip_threaded.py | 23 ++++++++++--- tests/test_igzip_threaded.py | 62 ++++++++++++++++++++++++++++++++++++ 3 files changed, 88 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.rst b/CHANGELOG.rst index d65c0609..3d643c0c 100644 --- a/CHANGELOG.rst +++ b/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``, diff --git a/src/isal/igzip_threaded.py b/src/isal/igzip_threaded.py index 8968e5eb..2e805829 100644 --- a/src/isal/igzip_threaded.py +++ b/src/isal/igzip_threaded.py @@ -97,6 +97,7 @@ 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() @@ -104,17 +105,24 @@ 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() diff --git a/tests/test_igzip_threaded.py b/tests/test_igzip_threaded.py index b4f43e04..b4998e76 100644 --- a/tests/test_igzip_threaded.py +++ b/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():