diff --git a/CHANGELOG.rst b/CHANGELOG.rst index b67fd7cc..42f66180 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -9,6 +9,10 @@ Changelog version 1.10.0-dev ------------------ ++ Lower the memory use of ``igzip_threaded`` writers on CPython. Threads that + wait for their next block no longer keep their last blocks in memory. When + data is written slower than the threads compress it, writing with 8 threads + and the default 1 MiB blocks uses about 7 MiB less memory at its peak. + Fix ``python -m isal.igzip`` replacing the input file with empty output and exiting with status 0 when the output file is the input file, as in ``-f -o data data``. It exits with an error now. diff --git a/src/isal/igzip_threaded.py b/src/isal/igzip_threaded.py index b4fbacdf..9f2143c1 100644 --- a/src/isal/igzip_threaded.py +++ b/src/isal/igzip_threaded.py @@ -405,7 +405,10 @@ def _compress(self, index: int): # blocks from the output queues in order. It drops the block. compressed, crc = b"", 0 data_length = len(data) + # Do not hold blocks while waiting; zdict views a whole block. + del item, data, zdict out_queue.put((compressed, crc, data_length)) + del compressed in_queue.task_done() def _write(self): @@ -427,6 +430,7 @@ def _write(self): self._crc = isal_zlib.crc32_combine(self._crc, crc, data_length) self._size += data_length self._write_compressed(compressed) + del item, compressed output_queue.task_done() index += 1 @@ -454,9 +458,11 @@ def _compress_and_write(self): # _write_compressed() drops the block. compressed, crc = b"", 0 data_length = len(data) + del item, data, zdict self._crc = isal_zlib.crc32_combine(self._crc, crc, data_length) self._size += data_length self._write_compressed(compressed) + del compressed in_queue.task_done() def _write_compressed(self, compressed: bytes): diff --git a/tests/test_igzip_threaded.py b/tests/test_igzip_threaded.py index d9b7570c..232592e4 100644 --- a/tests/test_igzip_threaded.py +++ b/tests/test_igzip_threaded.py @@ -6,6 +6,7 @@ # PYTHON SOFTWARE FOUNDATION LICENSE VERSION 2. import errno +import gc import gzip import io import itertools @@ -17,6 +18,7 @@ import threading import time import types +import weakref import zlib from pathlib import Path @@ -538,3 +540,39 @@ def compress_and_crc(self, data, zdict): # raw holds only the gzip header. The writer thread dropped all blocks, # and close() did not end the stream. assert len(raw.getvalue()) == 10 + + +@pytest.mark.skipif(not hasattr(sys, "getrefcount"), + reason="Needs reference counting") +@pytest.mark.parametrize("threads", [1, 3]) +def test_threaded_writer_drops_blocks(monkeypatch, threads): + isal_zlib = igzip_threaded.isal_zlib + ParallelCompress = isal_zlib._ParallelCompress + compressed_refs = [] + + class TrackingCompressor: + def __init__(self, **kwargs): + self.compressor = ParallelCompress(**kwargs) + + def compress_and_crc(self, data, zdict): + compressed, crc = self.compressor.compress_and_crc(data, zdict) + # bytes cannot be weakly referenced; a memoryview can. + compressed = memoryview(compressed) + compressed_refs.append(weakref.ref(compressed)) + return compressed, crc + + monkeypatch.setattr(isal_zlib, "_ParallelCompress", TrackingCompressor) + # write() does not copy bytes objects, so the threads get these blocks. + blocks = [os.urandom(1024) for _ in range(threads + 1)] + refcounts = [sys.getrefcount(block) for block in blocks] + with igzip_threaded._ThreadedGzipWriter(io.BytesIO(), + threads=threads) as f: + for block in blocks: + f.write(block) + f.flush() + # Free-threaded builds may free objects from other threads late. + gc.collect() + # write() keeps the last block as the zdict for the next one. + assert ([sys.getrefcount(block) for block in blocks][:-1] == + refcounts[:-1]) + assert [ref() for ref in compressed_refs] == [None] * len(blocks)