From 42384b132da4d6e160f24b390f6dd74a8548f79c Mon Sep 17 00:00:00 2001 From: ptweezy Date: Tue, 20 Oct 2026 09:57:09 -0400 Subject: [PATCH] Raise write errors in gzip_ng_threaded instead of waiting forever When raw.write() raised an error in the thread that writes the compressed blocks, that thread exited without calling task_done(). As a result, flush() waited forever, and so did closing the file object that gzip_ng_threaded.open() returns. write() blocked forever when an input queue was full. The thread now stores the error and drops later blocks until stop() ends it, so write(), flush(), and close() can raise the error in the calling thread. After an error, close() closes raw only if closefd is true, so it no longer closes a file object passed to gzip_ng_threaded.open(). close() now calls flush() before it writes the trailer, as python-isal's close() has done since pycompression/python-isal@5ef868b, so the trailer always comes after the queued blocks. This ports pycompression/python-isal#275. --- CHANGELOG.rst | 5 ++ src/zlib_ng/gzip_ng_threaded.py | 48 ++++++++++++----- tests/test_gzip_ng_threaded.py | 92 +++++++++++++++++++++++++++++++++ 3 files changed, 131 insertions(+), 14 deletions(-) diff --git a/CHANGELOG.rst b/CHANGELOG.rst index ca11abb..854ea04 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -9,6 +9,11 @@ Changelog version 1.0.1-dev ----------------- ++ Fix ``gzip_ng_threaded`` writers that waited forever when writing to the + underlying file failed, for example, because the disk was full. A later call + to ``write()``, ``flush()``, or ``close()`` raises the error. After an error, + ``close()`` no longer closes a file object that you passed to + ``gzip_ng_threaded.open``. + Wheels are now built for Windows arm64 architectures. + Fix a crash when calling ``copy()`` on a flushed compress object on Python 3.15. Python 3.15 changed diff --git a/src/zlib_ng/gzip_ng_threaded.py b/src/zlib_ng/gzip_ng_threaded.py index dee8be7..934a8fb 100644 --- a/src/zlib_ng/gzip_ng_threaded.py +++ b/src/zlib_ng/gzip_ng_threaded.py @@ -329,25 +329,32 @@ def flush(self): # Wait for all data to be written for out_q in self.output_queues: out_q.join() + if self.exception: + raise self.exception self.raw.flush() def close(self): if self._closed: return - self.raw.write(zlib_ng.compress(b"", wbits=-15)) - trailer = struct.pack(" bool: @@ -388,7 +395,7 @@ def _write(self): continue self._crc = zlib_ng.crc32_combine(self._crc, crc, data_length) self._size += data_length - self.raw.write(compressed) + self._write_compressed(compressed) output_queue.task_done() index += 1 @@ -413,9 +420,22 @@ def _compress_and_write(self): data_length = len(data) self._crc = zlib_ng.crc32_combine(self._crc, crc, data_length) self._size += data_length - self.raw.write(compressed) + self._write_compressed(compressed) in_queue.task_done() + def _write_compressed(self, compressed: bytes): + # After a write error, this thread keeps taking blocks from the queues + # and drops them until stop() ends it. That way, no thread waits + # forever on a full queue, and write(), flush(), and close() can raise + # the error in the calling thread. + if self.exception: + return + try: + self.raw.write(compressed) + except Exception as e: + with self.lock: + self.exception = e + def _set_error_and_empty_queue(self, error, q): with self.lock: self.exception = error diff --git a/tests/test_gzip_ng_threaded.py b/tests/test_gzip_ng_threaded.py index a587193..0ede0c0 100644 --- a/tests/test_gzip_ng_threaded.py +++ b/tests/test_gzip_ng_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 errno import gzip import io import itertools @@ -12,6 +13,8 @@ import subprocess import sys import tempfile +import time +import types import zlib from pathlib import Path @@ -112,6 +115,83 @@ def test_threaded_write_error(threads): error.match("Compressed output exceeds buffer size") +class FullDiskStream(io.BytesIO): + """Accepts the 10-byte gzip header, then fails like a full disk.""" + before_error = None + + def write(self, b): + if self.tell() == 0: + return super().write(b) + if self.before_error: + self.before_error() + raise OSError(errno.ENOSPC, os.strerror(errno.ENOSPC)) + + +def assert_writer_closed(f): + assert f.closed + workers = [f.output_worker] + f.compression_workers + assert not any(worker.is_alive() for worker in workers) + # Make sure double closing does not raise errors + f.close() + + +@pytest.mark.timeout(5) +@pytest.mark.parametrize("threads", [1, 3]) +def test_threaded_write_error_with_full_queues(threads): + raw = FullDiskStream() + f = gzip_ng_threaded._ThreadedGzipWriter(raw, threads=threads) + queues = f.input_queues + f.output_queues + + def wait_until_queues_are_full(): + while not all(q.full() for q in queues): + time.sleep(0.001) + + # Fail when all queues are full, so write() usually waits for room in a + # full input queue. + raw.before_error = wait_until_queues_are_full + with pytest.raises(OSError) as error: + while True: + f.write(b"x") + assert error.value.errno == errno.ENOSPC + with pytest.raises(OSError): + f.close() + assert_writer_closed(f) + + +@pytest.mark.timeout(5) +@pytest.mark.parametrize("threads", [1, 3]) +def test_threaded_flush_error(threads): + raw = FullDiskStream() + f = gzip_ng_threaded.open(raw, "wb", threads=threads) + f.write(b"x") + with pytest.raises(OSError) as error: + f.flush() + assert error.value.errno == errno.ENOSPC + # Later calls raise the error again. + with pytest.raises(OSError): + f.flush() + with pytest.raises(OSError): + f.close() + assert_writer_closed(f.raw) + # The test passed the stream to open(), so close() leaves it open. + assert not raw.closed + + +@pytest.mark.timeout(5) +@pytest.mark.parametrize("threads", [1, 3]) +def test_threaded_close_error(monkeypatch, threads): + raw = FullDiskStream() + # Make open() treat raw as a file that it opened by name. + monkeypatch.setattr(gzip_ng_threaded, "builtins", + types.SimpleNamespace(open=lambda name, mode: raw)) + with pytest.raises(OSError) as error: + with gzip_ng_threaded.open("full.gz", "wb", threads=threads) as f: + f.write(b"x") + assert error.value.errno == errno.ENOSPC + assert_writer_closed(f.raw) + assert raw.closed + + def test_close_reader(): tmp = io.BytesIO(Path(TEST_FILE).read_bytes()) f = gzip_ng_threaded._ThreadedGzipReader(tmp, "rb") @@ -131,6 +211,18 @@ def test_close_writer(threads): f.close() +@pytest.mark.parametrize("threads", [1, 3]) +def test_close_writer_writes_queued_blocks(threads): + raw = io.BytesIO() + f = gzip_ng_threaded._ThreadedGzipWriter( + raw, threads=threads, block_size=8 * 1024) + data = os.urandom(64 * 1024) + f.write(data) + # close() must write the queued blocks before the gzip trailer. + f.close() + assert gzip.decompress(raw.getvalue()) == data + + def test_reader_not_writable(): with gzip_ng_threaded.open(TEST_FILE, "rb") as f: assert not f.writable()