Do not keep blocks alive in idle igzip_threaded writer threads

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.

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

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)