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():