From 4697dac865fae612948e8236fe7689cbfa608e03 Mon Sep 17 00:00:00 2001
From: Parker Loflin <parker@cronstable.dev>
Date: Fri, 6 Nov 2026 17:23:13 -0500
Subject: [PATCH] Do not keep blocks alive in idle igzip_threaded writer
 threads

Each compression thread of an igzip_threaded writer kept its last
block, its zdict and its compressed block in local variables while it
waited on its queues. The zdict is a memoryview of the previous block,
so it kept all of that block alive too. The writer thread kept its last
compressed block the same way. The threads now delete these names
before they wait, so a block is freed once it is compressed and
written.

On Linux, writing 64 MiB of text in 64 KiB chunks, 0.4 ms apart, with
the default 1 MiB blocks and 8 threads, the peak memory that
tracemalloc traced fell from 26.4 to 18.6 MiB. Peak RSS fell from 22.3
to 14.2 MiB on Python 3.11 and from 23.0 to 14.3 MiB on Python 3.14.
With 4 threads, peak RSS fell from 13.6 to 9.4 MiB. When the data comes
as fast as the threads compress it, the difference is within noise.
The speed does not change. PyPy frees the blocks only when its garbage
collector runs, so there this changes nothing.
---
 CHANGELOG.rst                |  4 ++++
 src/isal/igzip_threaded.py   |  6 ++++++
 tests/test_igzip_threaded.py | 38 ++++++++++++++++++++++++++++++++++++
 3 files changed, 48 insertions(+)

diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index 481b7c5f..f0214dbe 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 the gzip header that ``igzip_threaded.open`` writes when ``threads`` is
   not 0. It had the XFL and OS bytes swapped, so the ``file`` command reported
   the output as coming from VM/CMS or a FAT file system.
diff --git a/src/isal/igzip_threaded.py b/src/isal/igzip_threaded.py
index 570605b5..768acfcf 100644
--- a/src/isal/igzip_threaded.py
+++ b/src/isal/igzip_threaded.py
@@ -406,7 +406,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):
@@ -428,6 +431,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
 
@@ -455,9 +459,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)