diff --git a/.github/release_checklist.md b/.github/release_checklist.md index 8ae899dc..4f98307c 100644 --- a/.github/release_checklist.md +++ b/.github/release_checklist.md @@ -1,5 +1,5 @@ Release checklist -- [ ] Check outstanding issues on JIRA and Github. +- [ ] Check outstanding issues on GitHub. - [ ] Check [latest documentation](https://python-isal.readthedocs.io/en/latest/) looks fine. - [ ] Create a release branch. - [ ] Change current development version in `CHANGELOG.rst` to stable version. @@ -7,9 +7,9 @@ Release checklist - [ ] Merge the release branch into `main`. - [ ] Created an annotated tag with the stable version number. Include changes from CHANGELOG.rst. -- [ ] Push tag to remote. This triggers the wheel/sdist build on github CI. +- [ ] Push tag to remote. This triggers the wheel/sdist build on GitHub CI. - [ ] merge `main` branch back into `develop`. - [ ] Build the new tag on readthedocs. Only build the last patch version of each minor version. So `1.1.1` and `1.2.0` but not `1.1.0`, `1.1.1` and `1.2.0`. -- [ ] Create a new release on github. +- [ ] Create a new release on GitHub. - [ ] Update the package on conda-forge. diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 955ca769..76a87acc 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -16,10 +16,10 @@ jobs: - uses: actions/checkout@v4 with: submodules: recursive - - name: Set up Python 3.10 - uses: actions/setup-python@v5 + - name: Set up Python 3.11 + uses: actions/setup-python@v7 with: - python-version: "3.10" + python-version: "3.11" - name: Install tox run: pip install tox - name: Lint @@ -31,15 +31,17 @@ jobs: tox_env: - docs - twine_check - runs-on: ubuntu-latest + # Use the same Ubuntu release as build.os in .readthedocs.yml, so the + # documentation is checked against the libisal that Read the Docs uses. + runs-on: ubuntu-22.04 steps: - uses: actions/checkout@v4 with: submodules: recursive - name: Set up Python 3.10 - uses: actions/setup-python@v5 + uses: actions/setup-python@v7 with: - python-version: "3.10" + python-version: "3.11" - name: Install isal run: sudo apt-get install libisal-dev - name: Install tox and upgrade setuptools and pip @@ -54,28 +56,29 @@ jobs: strategy: matrix: python-version: - - "3.10" - "3.11" - "3.12" - "3.13" - "3.14" - "3.14t" - - "pypy-3.10" + - "3.15" + - "3.15t" - "pypy-3.11" + - "pypy-3.12" os: ["ubuntu-latest"] include: - - os: "macos-13" - python-version: "3.10" + - os: "macos-15-intel" + python-version: "3.11" - os: "macos-latest" - python-version: "3.10" + python-version: "3.11" - os: "windows-latest" - python-version: "3.10" + python-version: "3.11" steps: - uses: actions/checkout@v4 with: submodules: recursive - name: Set up Python ${{ matrix.python-version }} - uses: actions/setup-python@v5 + uses: actions/setup-python@v7 with: python-version: ${{ matrix.python-version }} allow-prereleases: true @@ -106,7 +109,7 @@ jobs: strategy: matrix: python_version: - - "3.10" + - "3.11" steps: - uses: actions/checkout@v4 with: @@ -136,17 +139,14 @@ jobs: shell: bash -l {0} strategy: matrix: - os: ["ubuntu-latest", "macos-13", "windows-latest"] - python_version: [ "python" ] - include: - - os: "ubuntu-latest" - python_version: "pypy" + os: ["ubuntu-latest", "macos-15-intel", "windows-latest"] + python_version: [ "python==3.11" ] steps: - uses: actions/checkout@v4 with: submodules: recursive - name: Install miniconda. - uses: conda-incubator/setup-miniconda@v3 # https://github.com/conda-incubator/setup-miniconda. + uses: conda-incubator/setup-miniconda@v4 # https://github.com/conda-incubator/setup-miniconda. with: channels: conda-forge,defaults - name: Install requirements (universal) @@ -159,6 +159,49 @@ jobs: env: PYTHON_ISAL_LINK_DYNAMIC: True + # Run the tests on CPython builds with AddressSanitizer and ThreadSanitizer + # from https://github.com/nascheme/cpython_sanity. + test-sanitizers: + runs-on: ubuntu-latest + # A crash under ThreadSanitizer can hang instead of exiting. + timeout-minutes: 30 + strategy: + fail-fast: false + matrix: + include: + - sanitizer: address + image: ghcr.io/nascheme/cpython-asan:3.14 + # The free-threaded build runs Python threads in parallel. + - sanitizer: thread + image: ghcr.io/nascheme/cpython-tsan:3.14t + container: + image: ${{ matrix.image }} + steps: + - uses: actions/checkout@v4 + with: + submodules: recursive + - name: Install build dependencies + run: apt-get update && apt-get install -y nasm + - name: Install python-isal with ${{ matrix.sanitizer }} sanitizer + # setuptools-scm needs git, and the container user does not own the + # repository. + run: | + git config --global --add safe.directory "$GITHUB_WORKSPACE" + python -m pip install . pytest pytest-timeout + env: + # The extension modules get -fsanitize from CPython. CFLAGS adds it + # to the C code of ISA-L. + CFLAGS: -fsanitize=${{ matrix.sanitizer }} -fno-omit-frame-pointer -g + - name: Run tests + # Do not set PYTHONDEVMODE. Its debug allocator pads each allocation, + # so AddressSanitizer misses small overflows. --capture=sys prints + # sanitizer reports from C code even when they end the process, and + # -v prints the name of the test before its report. + run: python -m pytest -v tests --capture=sys + env: + # Exit at the first data race. A crash after a race can hang. + TSAN_OPTIONS: halt_on_error=1 + deploy: if: github.event_name == 'push' && startsWith(github.ref, 'refs/tags') runs-on: ${{ matrix.os }} @@ -168,12 +211,13 @@ jobs: - test-static - test-dynamic - test-arch + - test-sanitizers strategy: matrix: os: - ubuntu-latest - - macos-13 - - macos-14 + - macos-15-intel + - macos-latest - windows-latest cibw_archs_linux: ["x86_64"] cibw_before_all_linux: @@ -195,7 +239,7 @@ jobs: with: submodules: recursive fetch-depth: 0 # Fetch everything to get accurately versioned tag. - - uses: actions/setup-python@v2 # Some issues where caused by higher versions. + - uses: actions/setup-python@v7 name: Install Python - name: Install cibuildwheel twine wheel run: python -m pip install cibuildwheel twine wheel @@ -216,6 +260,7 @@ jobs: - name: Build wheels run: cibuildwheel --output-dir dist env: + CIBW_ENABLE: "pypy" CIBW_SKIP: "*-win32 *-manylinux_i686 cp38-macosx_*arm64 cp39-macosx_*arm64" # Skip 32 bit and problematic mac builds. CIBW_ARCHS_LINUX: ${{ matrix.cibw_archs_linux }} CIBW_BEFORE_ALL_LINUX: ${{ matrix.cibw_before_all_linux }} diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 36e20a2d..307b330b 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -8,6 +8,151 @@ Changelog .. that users understand how the changes affect the new version. +version 1.9.0 +----------------- + +By far most of the work on this release has been done by +`@ptweezy `_. Almost all bug fixes and improvements +are their work. + +Major changes: + ++ Updated included ISA-L library to 2.32.1. ++ Python 3.10 and PyPy 3.10 are no longer supported. Added support for PyPy + 3.12 and Python 3.15. ++ Restore PyPy wheel builds. ++ Fix builds on macOS with Apple Silicon, which used the portable C code of + ISA-L rather than its optimized aarch64 assembly. On an Apple M4, + decompression is 1.6 to 3 times faster, CRC-32 is 50 times faster, and + Adler-32 is 7 times faster. + +Optimizations and features: + ++ ``isal_zlib.compressobj`` supports ``strategy=Z_FIXED`` at compression + level 0. Level 0 then uses fixed Huffman codes instead of starting each + block with a Huffman table of about 110 bytes. This shrinks small flushed + writes, such as websocket messages: JSON messages of about 440 bytes, each + followed by ``flush(Z_SYNC_FLUSH)``, compress to an average of 75 instead + of 188 bytes. Blocks of a few KiB or more can be larger: about 5% for JSON + and 10% for Python source code at 64 KiB. Other strategies, and ``Z_FIXED`` + at levels 1 to 3, still give a warning and use the default strategy. ++ Speed up ``import isal.igzip`` and ``import isal.igzip_threaded`` by + importing ``argparse`` and ``shutil`` only when the command line interface + runs, and by dropping the ``multiprocessing`` import. ++ Speed up reading lines from binary ``igzip`` files through iteration, + ``readlines()``, and ``readline()``. On an Apple M4, reading FASTQ data is + 1.13 times faster, and reading lines of a few bytes is 1.5 times faster. ++ Speed up each ``write()`` call to binary ``igzip`` files on CPython. On an + AMD Ryzen 7 9800X3D with Python 3.14.8, writing FASTQ data or Python source + code one line at a time is 1.1 times faster. ++ Speed up many small writes to ``igzip.IGzipFile`` on Python 3.12 and later. + ``IGzipFile`` collects these writes in the write buffer of + ``gzip.GzipFile`` and compresses them in large chunks, which makes writing + lines 1.4 times faster. + +Bug fixes: + ++ Fix zlib streams compressed with a preset dictionary so they can be decoded + by standard zlib, including builds linked against older ISA-L releases. ++ Fix backward seeking in igzip files by discarding buffered compressed data + when rewinding the underlying file. ++ Fix a crash when ``igzip.decompress`` gets an object that is not + bytes-like, such as a ``str``. It raises a ``TypeError`` instead. ++ Fix a crash after creating ``igzip_lib.IgzipDecompressor`` objects with a + ``zdict``. ++ Fix error handling when more than one interpreter imports + ``isal.isal_zlib`` or ``isal.igzip_lib``, for example, the main interpreter + and a legacy subinterpreter that ``Py_NewInterpreter()`` creates. Each + import replaced the error classes that the C code raises, so + ``except isal_zlib.error``, ``except igzip_lib.error``, and + ``except gzip.BadGzipFile`` caught errors only in the interpreter that + imported the modules last. ++ Fix a hang in ``isal_zlib.Compress``: after ``flush()`` raised an error for + an unsupported mode, the next call on the same object never returned. ++ Fix a crash when you call ``flush(Z_FINISH)`` twice on a level 0 + ``isal_zlib.Compress`` object. ++ Fix ``unused_data`` of ``isal_zlib.Decompress`` and + ``igzip_lib.IgzipDecompressor``, which could contain bytes from outside the + input buffer when ``max_length`` split the end of a stream. At the end of a + stream, ``isal_zlib.Decompress.unconsumed_tail`` is empty, matching the + CPython fix for gh-158169. ++ Fix ``igzip_lib.IgzipDecompressor.needs_input``, which could be ``True`` + after ``decompress()`` returned ``max_length`` bytes while more output was + still available. Readers that stop at the end of the file when + ``needs_input`` is ``True``, such as CPython's ``DecompressReader``, raised + ``EOFError`` for valid data. ++ Fix ``isal_zlib.decompressobj`` with a ``wbits`` of 40 to 47, which detects + zlib or gzip data. It failed on gzip data when the first ``decompress()`` + call got fewer than 2 bytes, for example, when the data arrived byte by + byte. ++ Fix ``isal_zlib.decompressobj`` with a ``wbits`` of 40 to 47 after a + ``flush()`` call that came before any data. It then failed on both zlib and + gzip data. ++ ``igzip_threaded.open`` raises ``ValueError`` for a ``block_size`` of about + 3.6 GiB or more and ``TypeError`` for a ``compresslevel`` that is not an + integer. Both raised ``SystemError`` before. ++ Fix ``igzip_threaded`` writers at compression level 0, which raised + ``OverflowError`` for data that does not compress, such as random bytes or + already compressed files. At compression level 0, ``block_size`` must now be + less than about 2 GiB. ++ Extraordinarily large gzip headers with either a very long comment or + filename will now cause an ``OverflowError`` rather than use more and more + memory. The size is limited to the internal buffer size which is currently + set at 512 KiB. ++ Fix ``igzip_threaded`` readers and writers that waited for a polling timeout + of up to 50 ms at the end of the stream or when closed. Writing and closing + a small file takes less than a millisecond instead of 60 ms. ++ Fix ``igzip_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 + ``igzip_threaded.open``. ++ Fix ``igzip_threaded`` writers with more than one thread that waited + forever when compressing a block failed, for example, because there was not + enough memory. A later call to ``write()``, ``flush()``, or ``close()`` + raises the error. ++ Fix a crash on the free-threaded build when one thread reads + ``unused_data`` or ``unconsumed_tail`` of an ``isal_zlib`` decompressor + while another thread decompresses with it. Reading the attributes of + ``isal_zlib`` decompressors and ``igzip_lib.IgzipDecompressor`` from other + threads is now safe, and so is calling ``decompress()`` from several + threads at the same time when ``wbits`` detects the format. ++ ``isal_zlib.crc32_combine`` and the ``read()`` and ``seek()`` methods of the + private ``isal_zlib._GzipReader`` class raise ``TypeError`` instead of + ``SystemError`` for arguments of the wrong type or number. + ``igzip.IGzipFile`` was not affected. ++ Fix a crash on PyPy when you create an ``isal_zlib.Compress`` or + ``isal_zlib.Decompress`` object by calling the type, for example, + ``isal_zlib.Compress()``, and then use it. Calling the type raises + ``TypeError`` now, as it does on CPython. Use ``compressobj()`` and + ``decompressobj()`` instead. ++ Fix ``readinto()`` of the private ``isal_zlib._GzipReader`` class, which + wrote into read-only objects such as ``bytes``. It raises ``TypeError`` now. + ``igzip.IGzipFile`` was not affected. ++ Fix the ``mtime`` attribute of ``igzip.IGzipFile`` objects in read mode, + which was ``None`` for gzip files with a modification time of 0. It is 0 now, + as in the ``gzip`` module. ``gzip -n``, ``python -m isal.igzip -n``, and + ``igzip_threaded`` writers make such files, and so does ``gzip.compress`` by + default on Python 3.14. ++ Fix a crash when there is not enough memory to create an + ``igzip_lib.IgzipDecompressor``. It raises ``MemoryError`` now. ++ Fix a crash when there is not enough memory to raise ``BadGzipFile`` for + data that is not in the gzip format, for example, in ``igzip.decompress``. + It raises ``MemoryError`` now. ++ Fix a crash in ``igzip_lib.IgzipDecompressor.decompress()`` after a call on + the same object raised ``MemoryError``. The next call used the input of the + failed call after the caller had released it. This matches the CPython fix + for gh-148395. ++ Fix a crash on macOS and PyPy when there is not enough memory to create the + internal lock of an ``isal_zlib.Compress``, ``isal_zlib.Decompress``, or + ``igzip_lib.IgzipDecompressor`` object. It raises ``MemoryError`` now. ++ Fix the ``-0`` and ``--fast`` options of ``python -m isal.igzip``, which + compressed at level 2 instead of level 0. ++ Fix crashes and wrong data when one thread calls ``seek()`` of the + private ``isal_zlib._GzipReader`` class while another thread reads from the + same object. ``igzip.IGzipFile`` was not affected. + + version 1.8.0 ----------------- + Python 3.14 is supported. diff --git a/README.rst b/README.rst index 734b2f74..fd4ce61b 100644 --- a/README.rst +++ b/README.rst @@ -113,12 +113,12 @@ your project please list a python-isal dependency as follows. ``setup.cfg``:: install_requires = - isal; platform.machine == "x86_64" or platform.machine == "AMD64" or platform.machine == "aarch64" + isal; platform.machine == "x86_64" or platform.machine == "AMD64" or platform.machine == "aarch64" or platform.machine == "arm64" ``setup.py``:: extras_require={ - ":platform.machine == 'x86_64' or platform.machine == 'AMD64' or platform.machine == 'aarch64'": ['isal'] + ":platform.machine == 'x86_64' or platform.machine == 'AMD64' or platform.machine == 'aarch64' or platform.machine == 'arm64'": ['isal'] }, .. dependency end @@ -139,8 +139,9 @@ Differences with zlib and gzip modules ``FINISH_FLUSH``. Other flush modes are not supported and will raise errors. + ``zlib.Z_DEFAULT_STRATEGY``, ``zlib.Z_RLE`` etc. are exposed as ``isal_zlib.Z_DEFAULT_STRATEGY``, ``isal_zlib.Z_RLE`` etc. for compatibility - reasons. However, ``isal_zlib`` only supports a default strategy and will - give warnings when other strategies are used. + reasons. However, ``isal_zlib`` supports only the default strategy, and + ``Z_FIXED`` at compression level 0. It gives a warning and uses the default + strategy for other strategies, and for ``Z_FIXED`` at levels 1 to 3. + ``zlib`` supports different memory levels from 1 to 9 (with 8 default). ``isal_zlib`` supports memory levels smallest, small, medium, large and largest. These have been mapped to levels 1, 2-3, 4-6, 7-8 and 9. So @@ -223,6 +224,8 @@ This project builds upon the software and experience of many. Many thanks to: + The `github actions team `_ for creating the actions CI service that enables building and testing on all three major operating systems. ++ `@ptweezy `_ for discovering and fixing lots and + lots of bugs as well as suggesting and implementing quite a few improvements. + `@animalize `_ for explaining how to test and build python-isal for ARM 64-bit platforms. + And last but not least: everyone who submitted a bug report or a feature diff --git a/pyproject.toml b/pyproject.toml index c1cbce68..d1bbfb46 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -17,11 +17,11 @@ requires-python = ">=3.9" # Because of setuptools version classifiers = [ "Programming Language :: Python :: 3 :: Only", "Programming Language :: Python :: 3", - "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", "Programming Language :: Python :: 3.13", "Programming Language :: Python :: 3.14", + "Programming Language :: Python :: 3.15", "Programming Language :: Python :: Implementation :: CPython", "Programming Language :: Python :: Implementation :: PyPy", "Programming Language :: C", diff --git a/setup.py b/setup.py index 1ed64c2d..75c52987 100644 --- a/setup.py +++ b/setup.py @@ -113,15 +113,21 @@ def build_isa_l(): cpu_count = os.cpu_count() or 1 # os.cpu_count() can return None run_args = dict(cwd=build_dir, env=build_env) if SYSTEM_IS_UNIX: + make_args = [] if platform.machine() == "aarch64": cflags_param = "CFLAGS_aarch64" + elif sys.platform == "darwin" and platform.machine() == "arm64": + # Makefile.unx does not recognize "arm64" and builds the slow + # portable C code. ISA-L still checks CPU features at runtime. + cflags_param = "CFLAGS_aarch64" + make_args = ["host_cpu=aarch64", "arch=aarch64"] else: cflags_param = "CFLAGS_" make_cmd = "make" if SYSTEM_IS_BSD: make_cmd = "gmake" subprocess.run([make_cmd, "-j", str(cpu_count), "-f", "Makefile.unx", - "isa-l.h", "bin/isa-l.a", + "isa-l.h", "bin/isa-l.a", *make_args, f"{cflags_param}={build_env.get('CFLAGS', '')}"], **run_args) elif SYSTEM_IS_WINDOWS: diff --git a/src/isal/igzip.py b/src/isal/igzip.py index af1b4712..eb9b8042 100644 --- a/src/isal/igzip.py +++ b/src/isal/igzip.py @@ -24,12 +24,10 @@ """Similar to the stdlib gzip module. But using the Intel Storage Accelaration Library to speed up its methods.""" -import argparse import builtins import gzip import io import os -import shutil import struct import sys import time @@ -57,6 +55,10 @@ BadGzipFile = gzip.BadGzipFile # type: ignore +# Since Python 3.12 GzipFile.write buffers small writes and passes them to +# _write_raw in large chunks, which compress much faster. +_GZIP_BUFFERS_WRITES = hasattr(gzip.GzipFile, "_write_raw") + # The open method was copied from the CPython source with minor adjustments. def open(filename, mode="rb", compresslevel=_COMPRESS_LEVEL_TRADEOFF, @@ -166,6 +168,14 @@ def __init__(self, filename=None, mode=None, raw = _GzipReader(self.fileobj, READ_BUFFER_SIZE) self._buffer = io.BufferedReader(raw) + def readline(self, size=-1): + """Read a line of decompressed data, or at most size bytes of it.""" + # GzipFile.closed tests whether fileobj is None, so testing fileobj + # here skips the two Python calls GzipFile.readline makes per line. + if self.fileobj is None: + self._check_not_closed() + return self._buffer.readline(size) + def __repr__(self): s = repr(self.fileobj) return '' @@ -194,7 +204,10 @@ def _write_gzip_header(self, compresslevel=_COMPRESS_LEVEL_TRADEOFF): super()._write_gzip_header() def write(self, data): - self._check_not_closed() + # GzipFile.closed tests whether fileobj is None, so testing fileobj + # here skips two Python calls for each write. + if self.fileobj is None: + self._check_not_closed() if self.mode != WRITE: import errno raise OSError(errno.EBADF, "write() on read-only IGzipFile object") @@ -202,7 +215,12 @@ def write(self, data): if self.fileobj is None: raise ValueError("write() on closed IGzipFile object") - if isinstance(data, bytes): + if _GZIP_BUFFERS_WRITES: + return self._buffer.write(data) + return self._write_raw(data) + + def _write_raw(self, data): + if isinstance(data, (bytes, bytearray)): length = len(data) else: # accept any data that supports the buffer protocol @@ -216,6 +234,15 @@ def write(self, data): self.offset += length return length + def seek(self, offset, whence=io.SEEK_SET): + position = super().seek(offset, whence) + if self.mode == WRITE and _GZIP_BUFFERS_WRITES: + # GzipFile.seek returns the offset before the zeros it writes + # leave the write buffer. + self._buffer.flush() + return self.offset + return position + # Aliases for improved compatibility with CPython gzip module. GzipFile = IGzipFile @@ -255,6 +282,7 @@ def decompress(data): def _argument_parser(): + import argparse # Only the command line interface needs it parser = argparse.ArgumentParser() parser.description = ( "A simple command line interface for the igzip module. " @@ -277,7 +305,8 @@ def _argument_parser(): "-3", "--best", action="store_const", dest="compresslevel", const=_COMPRESS_LEVEL_BEST, help="use compression level 3 (best)") - compress_group.set_defaults(compress=True) + compress_group.set_defaults(compress=True, + compresslevel=_COMPRESS_LEVEL_TRADEOFF) compress_group.add_argument( "-d", "--decompress", action="store_const", dest="compress", @@ -302,9 +331,10 @@ def _argument_parser(): def main(): + import shutil # Only the command line interface needs it args = _argument_parser().parse_args() - compresslevel = args.compresslevel or _COMPRESS_LEVEL_TRADEOFF + compresslevel = args.compresslevel if args.output: out_filepath = args.output diff --git a/src/isal/igzip_libmodule.c b/src/isal/igzip_libmodule.c index d498145f..d2896bb3 100644 --- a/src/isal/igzip_libmodule.c +++ b/src/isal/igzip_libmodule.c @@ -32,8 +32,10 @@ typedef struct { separately. Conversion and looping is encapsulated in decompress_buf() */ Py_ssize_t avail_in_real; - char eof; /* T_BOOL expects a char */ + char eof; char needs_input; + uint8_t saved_bitbuffer[8]; + Py_ssize_t saved_bitbuffer_size; /* Struct inflate state contains a massive buffer at the end. Put it at the end of the IgzipDecompressor so members can be accessed easily. */ struct inflate_state state; @@ -42,7 +44,9 @@ typedef struct { static void IgzipDecompressor_dealloc(IgzipDecompressor *self) { - PyThread_free_lock(self->lock); + if (self->lock != NULL) { + PyThread_free_lock(self->lock); + } PyMem_Free(self->input_buffer); Py_CLEAR(self->unused_data); Py_CLEAR(self->zdict); @@ -161,8 +165,10 @@ decompress_buf(IgzipDecompressor *self, Py_ssize_t max_length) self->avail_in_real += self->state.avail_in; - if (_PyBytes_Resize(&RetVal, self->state.next_out - - (uint8_t *)PyBytes_AS_STRING(RetVal)) != 0) + /* Prevent empty string object from being resized. */ + if (PyBytes_GET_SIZE(RetVal) != 0 && + _PyBytes_Resize(&RetVal, self->state.next_out - + (uint8_t *)PyBytes_AS_STRING(RetVal)) != 0) goto error; return RetVal; @@ -224,6 +230,7 @@ decompress(IgzipDecompressor *self, uint8_t *data, size_t len, Py_ssize_t max_le input_buffer_in_use = 0; } + uint8_t *input_start = self->state.next_in; result = decompress_buf(self, max_length); if(result == NULL) { self->state.next_in = NULL; @@ -240,14 +247,24 @@ decompress(IgzipDecompressor *self, uint8_t *data, size_t len, Py_ssize_t max_le goto error; char *new_data_ptr = PyBytes_AS_STRING(new_data); bitbuffer_copy(&(self->state), new_data_ptr, bytes_in_bitbuffer); - memcpy(new_data_ptr + bytes_in_bitbuffer, self->state.next_in, - self->avail_in_real); + if (copy_from_next_in(&(self->state), input_start, + self->saved_bitbuffer, + self->saved_bitbuffer_size, + new_data_ptr + bytes_in_bitbuffer, + self->avail_in_real) < 0) { + Py_DECREF(new_data); + goto error; + } Py_XSETREF(self->unused_data, new_data); } } else if (self->avail_in_real == 0) { self->state.next_in = NULL; - self->needs_input = 1; + /* When the output is full, isal_inflate can still read all of the + input and keep the decompressed data in its own buffer. Ask for + more input only when the output was not full, like CPython's + LZMADecompressor does. */ + self->needs_input = (self->state.avail_out != 0); } else { self->needs_input = 0; @@ -280,9 +297,17 @@ decompress(IgzipDecompressor *self, uint8_t *data, size_t len, Py_ssize_t max_le self->state.next_in = self->input_buffer; } } + if (!self->eof) { + save_bitbuffer(&(self->state), self->saved_bitbuffer, + &self->saved_bitbuffer_size); + } return result; error: + /* next_in can point to the buffer of the caller, which is released after + this call. Reset it so the next call does not use it, as CPython does + since gh-148395. */ + self->state.next_in = NULL; Py_XDECREF(result); return NULL; } @@ -456,11 +481,17 @@ igzip_lib_IgzipDecompressor__new__(PyTypeObject *type, return NULL; } IgzipDecompressor *self = PyObject_New(IgzipDecompressor, type); + if (self == NULL) { + return NULL; + } self->eof = 0; self->needs_input = 1; + self->saved_bitbuffer_size = 0; self->avail_in_real = 0; self->input_buffer = NULL; self->input_buffer_size = 0; + self->lock = NULL; + Py_XINCREF(zdict); self->zdict = zdict; self->unused_data = PyBytes_FromStringAndSize(NULL, 0); if (self->unused_data == NULL) { @@ -502,16 +533,56 @@ PyDoc_STRVAR(IgzipDecompressor_needs_input_doc, PyDoc_STRVAR(IgzipDecompressor_crc_doc, "The checksum that is saved if DECOMP_ZLIB* or DECOMP_GZIP* flags are used."); -static PyMemberDef IgzipDecompressor_members[] = { - {"eof", T_BOOL, offsetof(IgzipDecompressor, eof), - READONLY, IgzipDecompressor_eof__doc__}, - {"unused_data", T_OBJECT_EX, offsetof(IgzipDecompressor, unused_data), - READONLY, IgzipDecompressor_unused_data__doc__}, - {"needs_input", T_BOOL, offsetof(IgzipDecompressor, needs_input), READONLY, - IgzipDecompressor_needs_input_doc}, - {"crc", T_UINT, - offsetof(IgzipDecompressor, state) + offsetof(struct inflate_state, crc), - READONLY, IgzipDecompressor_crc_doc}, +/* The getters take the lock, because without the GIL another thread can + change these attributes in decompress(). */ +static PyObject * +IgzipDecompressor_get_eof(IgzipDecompressor *self, void *Py_UNUSED(closure)) +{ + ENTER_ZLIB(self); + char eof = self->eof; + LEAVE_ZLIB(self); + return PyBool_FromLong(eof); +} + +static PyObject * +IgzipDecompressor_get_unused_data(IgzipDecompressor *self, + void *Py_UNUSED(closure)) +{ + ENTER_ZLIB(self); + PyObject *unused_data = self->unused_data; + Py_INCREF(unused_data); + LEAVE_ZLIB(self); + return unused_data; +} + +static PyObject * +IgzipDecompressor_get_needs_input(IgzipDecompressor *self, + void *Py_UNUSED(closure)) +{ + ENTER_ZLIB(self); + char needs_input = self->needs_input; + LEAVE_ZLIB(self); + return PyBool_FromLong(needs_input); +} + +static PyObject * +IgzipDecompressor_get_crc(IgzipDecompressor *self, void *Py_UNUSED(closure)) +{ + ENTER_ZLIB(self); + uint32_t crc = self->state.crc; + LEAVE_ZLIB(self); + return PyLong_FromUnsignedLong(crc); +} + +static PyGetSetDef IgzipDecompressor_properties[] = { + {"eof", (getter)IgzipDecompressor_get_eof, NULL, + IgzipDecompressor_eof__doc__, NULL}, + {"unused_data", (getter)IgzipDecompressor_get_unused_data, NULL, + IgzipDecompressor_unused_data__doc__, NULL}, + {"needs_input", (getter)IgzipDecompressor_get_needs_input, NULL, + IgzipDecompressor_needs_input_doc, NULL}, + {"crc", (getter)IgzipDecompressor_get_crc, NULL, + IgzipDecompressor_crc_doc, NULL}, {NULL} }; @@ -523,7 +594,7 @@ static PyTypeObject IgzipDecompressor_Type = { .tp_flags = Py_TPFLAGS_DEFAULT, .tp_doc = igzip_lib_IgzipDecompressor___init____doc__, .tp_methods = IgzipDecompressor_methods, - .tp_members = IgzipDecompressor_members, + .tp_getset = IgzipDecompressor_properties, .tp_new = igzip_lib_IgzipDecompressor__new__, }; @@ -605,7 +676,10 @@ static struct PyModuleDef igzip_lib_module = { PyModuleDef_HEAD_INIT, "igzip_lib", /* name of module */ igizp_lib_module_documentation, /* module documentation, may be NULL */ - 0, + /* -1 because IsalError is a C global. A subinterpreter then gets a copy + of this module instead of running PyInit_igzip_lib again, which would + replace IsalError with a new class. */ + -1, IgzipLibMethods }; diff --git a/src/isal/igzip_threaded.py b/src/isal/igzip_threaded.py index 355b7f72..8968e5eb 100644 --- a/src/isal/igzip_threaded.py +++ b/src/isal/igzip_threaded.py @@ -7,7 +7,6 @@ import builtins import io -import multiprocessing import os import queue import struct @@ -52,10 +51,7 @@ def open(filename, mode="rb", compresslevel=igzip._COMPRESS_LEVEL_TRADEOFF, try: threads = len(os.sched_getaffinity(0)) except: # noqa: E722 - try: - threads = multiprocessing.cpu_count() - except: # noqa: E722 - threads = 1 + threads = os.cpu_count() or 1 if "r" in mode: gzip_file = io.BufferedReader( _ThreadedGzipReader(filename, block_size=block_size)) @@ -116,20 +112,25 @@ def _decompress(self): data = self.fileobj.read(block_size) except Exception as e: self.exception = e - return - if not data: - return + data = b"" + # An empty block tells the reader about EOF or an error while self.running and self._calling_thread.is_alive(): try: block_queue.put(data, timeout=0.05) break except queue.Full: pass + if not data: + return def readinto(self, b): self._check_closed() result = self.buffer.readinto(b) if result == 0: + if self.eof: + if self.exception: + raise self.exception + return 0 while True: try: data_from_queue = self.queue.get(timeout=0.01) @@ -140,6 +141,11 @@ def readinto(self, b): raise self.exception # EOF reached return 0 + if not data_from_queue: + self.eof = True + if self.exception: + raise self.exception + return 0 self.buffer = io.BytesIO(data_from_queue) result = self.buffer.readinto(b) self.pos += result @@ -156,6 +162,12 @@ def close(self) -> None: if self._closed: return self.running = False + # Empty the queue so a worker blocked on put can stop + while True: + try: + self.queue.get_nowait() + except queue.Empty: + break self.worker.join() self.fileobj.close() if self.closefd: @@ -228,16 +240,27 @@ def __init__(self, self.previous_block = b"" # Deflating random data results in an output a little larger than the # input. Making the output buffer 10% larger is sufficient overkill. - compress_buffer_size = block_size + max(block_size // 10, 500) + # Level 0 is the exception: it codes all data with one precomputed + # Huffman table and never falls back to stored blocks, so random data + # grows by more than 20%. Deflate uses at most 15 bits per input byte, + # so twice the block size plus 500 bytes for the block header, the + # end-of-block code, and the sync flush marker always fits. + if level == 0: + compress_buffer_size = 2 * block_size + 500 + else: + compress_buffer_size = block_size + max(block_size // 10, 500) self.block_size = block_size self.compressors: List[isal_zlib._ParallelCompress] = [ isal_zlib._ParallelCompress(buffersize=compress_buffer_size, level=level) for _ in range(threads) ] if threads > 1: - self.input_queues: List[queue.Queue[Tuple[bytes, memoryview]]] = [ + # None is the sentinel from stop() + self.input_queues: List[ + queue.Queue[Optional[Tuple[bytes, memoryview]]]] = [ queue.Queue(queue_size) for _ in range(threads)] - self.output_queues: List[queue.Queue[Tuple[bytes, int, int]]] = [ + self.output_queues: List[ + queue.Queue[Optional[Tuple[bytes, int, int]]]] = [ queue.Queue(queue_size) for _ in range(threads)] self.output_worker = threading.Thread(target=self._write) self.compression_workers = [ @@ -287,6 +310,13 @@ def start(self): def stop(self): """Stop, but do not care for remaining work""" self.running = False + # Wake up threads that wait on a queue. A thread with a full queue + # is busy and checks self.running when it is done. + for q in self.input_queues + self.output_queues: + try: + q.put_nowait(None) + except queue.Full: + pass for worker in self.compression_workers: worker.join() self.output_worker.join() @@ -324,26 +354,29 @@ def flush(self): # Wait for all data to be written for out_q in self.output_queues: out_q.join() + # The threads store an error before they call task_done(), so after + # the joins, this check sees an error in any block written so far. + if self.exception: + raise self.exception self.raw.flush() def close(self) -> None: if self._closed: return - self.flush() - self.raw.write(isal_zlib.compress(b"", wbits=-15)) - trailer = struct.pack(" bool: @@ -355,17 +388,23 @@ def _compress(self, index: int): compressor: isal_zlib._ParallelCompress = self.compressors[index] while True: try: - data, zdict = in_queue.get(timeout=0.05) + item = in_queue.get(timeout=0.05) except queue.Empty: if not (self.running and self._calling_thread.is_alive()): return continue + if item is None: # Sentinel from stop() + in_queue.task_done() + return + data, zdict = item try: compressed, crc = compressor.compress_and_crc(data, zdict) except Exception as e: - in_queue.task_done() - self._set_error_and_empty_queue(e, in_queue) - return + with self.lock: + self.exception = e + # Pass on an empty block, so the writer thread keeps taking + # blocks from the output queues in order. It drops the block. + compressed, crc = b"", 0 data_length = len(data) out_queue.put((compressed, crc, data_length)) in_queue.task_done() @@ -377,14 +416,18 @@ def _write(self): out_index = index % self.threads output_queue = output_queues[out_index] try: - compressed, crc, data_length = output_queue.get(timeout=0.05) + item = output_queue.get(timeout=0.05) except queue.Empty: if not (self.running and self._calling_thread.is_alive()): return continue + if item is None: # Sentinel from stop() + output_queue.task_done() + return + compressed, crc, data_length = item self._crc = isal_zlib.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 @@ -395,34 +438,41 @@ def _compress_and_write(self): compressor = self.compressors[0] while True: try: - data, zdict = in_queue.get(timeout=0.05) + item = in_queue.get(timeout=0.05) except queue.Empty: if not (self.running and self._calling_thread.is_alive()): return continue + if item is None: # Sentinel from stop() + in_queue.task_done() + return + data, zdict = item try: compressed, crc = compressor.compress_and_crc(data, zdict) except Exception as e: - in_queue.task_done() - self._set_error_and_empty_queue(e, in_queue) - return + with self.lock: + self.exception = e + # _write_compressed() drops the block. + compressed, crc = b"", 0 data_length = len(data) self._crc = isal_zlib.crc32_combine(self._crc, crc, data_length) self._size += data_length - self.raw.write(compressed) + self._write_compressed(compressed) in_queue.task_done() - def _set_error_and_empty_queue(self, error, q): - with self.lock: - self.exception = error - # Abort everything and empty the queue - self.running = False - while True: - try: - _ = q.get(timeout=0.05) - q.task_done() - except queue.Empty: - return + def _write_compressed(self, compressed: bytes): + # After a compression or write error, the threads keep taking blocks + # from the queues, and this method drops them until stop() sends the + # None sentinel. 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 writable(self) -> bool: return True diff --git a/src/isal/isa-l b/src/isal/isa-l index c387163f..7c3479e0 160000 --- a/src/isal/isa-l +++ b/src/isal/isa-l @@ -1 +1 @@ -Subproject commit c387163fcbca62701d646149564c550c78a4f985 +Subproject commit 7c3479e0a9dac17f448603ec1ad64c7c625f530c diff --git a/src/isal/isal_shared.h b/src/isal/isal_shared.h index 61610276..f4f4519b 100644 --- a/src/isal/isal_shared.h +++ b/src/isal/isal_shared.h @@ -21,7 +21,6 @@ Changes compared to CPython: */ #define PY_SSIZE_T_CLEAN #include -#include "structmember.h" // PyMemberDef #include #include @@ -219,6 +218,40 @@ static int bitbuffer_copy(struct inflate_state *state, char *to, size_t n){ return 0; } +/* When a stream ends, isal_inflate moves next_in back over the unused bytes + in the bit buffer. If a previous call read those bytes, next_in points + before the current input. This keeps a copy of them after each call. */ +static inline void +save_bitbuffer(struct inflate_state *state, uint8_t *saved, + Py_ssize_t *saved_size) +{ + *saved_size = bitbuffer_size(state); + bitbuffer_copy(state, (char *)saved, *saved_size); +} + +/* Copy size bytes starting at state->next_in. Bytes before input_start come + from the saved bit buffer. */ +static int +copy_from_next_in(struct inflate_state *state, uint8_t *input_start, + uint8_t *saved, Py_ssize_t saved_size, + char *to, Py_ssize_t size) +{ + uint8_t *from = state->next_in; + Py_ssize_t before_input = 0; + if ((uintptr_t)from < (uintptr_t)input_start) { + before_input = (Py_ssize_t)((uintptr_t)input_start - (uintptr_t)from); + if (before_input > saved_size || before_input > size) { + PyErr_SetString(PyExc_RuntimeError, + "Leftover input before the buffer is unavailable"); + return -1; + } + memcpy(to, saved + saved_size - before_input, before_input); + from = input_start; + } + memcpy(to + before_input, from, size - before_input); + return 0; +} + static inline void arrange_input_buffer(uint32_t *avail_in, Py_ssize_t *remains) { diff --git a/src/isal/isal_zlibmodule.c b/src/isal/isal_zlibmodule.c index 3a280ec8..3ee12d83 100644 --- a/src/isal/isal_zlibmodule.c +++ b/src/isal/isal_zlibmodule.c @@ -60,6 +60,14 @@ Changes compared to CPython: #define MEMORYVIEW_READINTO_FLAGS PyBUF_WRITE #endif +/* Only compressobj() and decompressobj() create Compress and Decompress + objects. CPython does not let you call a static type without tp_new, but + PyPy does, and the uninitialized object crashes when you use it. This flag + makes PyPy raise TypeError too. Python 3.9 does not have the flag. */ +#ifndef Py_TPFLAGS_DISALLOW_INSTANTIATION +#define Py_TPFLAGS_DISALLOW_INSTANTIATION 0 +#endif + static PyTypeObject IsalZlibCompType; static PyTypeObject IsalZlibDecompType; @@ -141,10 +149,13 @@ static inline int zlib_mem_level_to_isal(int mem_level) { return ZLIB_MEM_LEVEL_TO_ISAL[mem_level]; } +/* A gzip stream starts with 31, 139. A zlib stream cannot start with 31, + because the low 4 bits of its first byte are the compression method 8. + So the first byte is enough, even when the data arrives byte by byte. */ static inline int data_is_gzip(Py_buffer *data){ uint8_t *buf = (uint8_t *)data->buf; - return (data->len > 1) && (buf[0] == 31) && (buf[1] == 139); + return (data->len > 0) && (buf[0] == 31); } @@ -280,7 +291,7 @@ isal_zlib_crc32_combine(PyObject *module, PyObject *args) { uint32_t crc2 = 0; Py_ssize_t crc2_length = 0; static char *format = "IIn:crc32combine"; - if (PyArg_ParseTuple(args, format, &crc1, &crc2, &crc2_length) < 0) { + if (!PyArg_ParseTuple(args, format, &crc1, &crc2, &crc2_length)) { return NULL; } return PyLong_FromUnsignedLong( @@ -310,8 +321,8 @@ ParallelCompress__new__(PyTypeObject *type, PyObject *args, PyObject *kwargs) int level = ISAL_DEFAULT_COMPRESSION; static char *format = "n|i:ParallelCompress__new__"; static char *kwarg_names[] = {"buffersize", "level", NULL}; - if (PyArg_ParseTupleAndKeywords(args, kwargs, format, kwarg_names, - &buffer_size, &level) < 0) { + if (!PyArg_ParseTupleAndKeywords(args, kwargs, format, kwarg_names, + &buffer_size, &level)) { return NULL; } uint32_t level_buf_size; @@ -323,6 +334,7 @@ ParallelCompress__new__(PyTypeObject *type, PyObject *args, PyObject *kwargs) PyErr_Format(PyExc_ValueError, "buffersize must be at most %zd, got %zd", (Py_ssize_t)UINT32_MAX, buffer_size); + return NULL; } ParallelCompress *self = PyObject_New(ParallelCompress, type); if (self == NULL) { @@ -570,6 +582,7 @@ typedef struct PyObject_HEAD uint8_t *level_buf; PyObject *zdict; + struct isal_zlib_header zlib_header; int is_initialised; PyThread_type_lock lock; // isal_zstream should be at the bottom as it contains buffers inside the struct. @@ -581,7 +594,9 @@ Comp_dealloc(compobject *self) { if (self->is_initialised && self->level_buf != NULL) PyMem_Free(self->level_buf); - PyThread_free_lock(self->lock); + if (self->lock != NULL) { + PyThread_free_lock(self->lock); + } Py_XDECREF(self->zdict); Py_TYPE(self)->tp_free((PyObject *)self); } @@ -595,6 +610,7 @@ newcompobject() return NULL; self->is_initialised = 0; self->zdict = NULL; + self->zlib_header.dict_flag = 0; self->level_buf = NULL; self->lock = PyThread_allocate_lock(); if (self->lock == NULL) { @@ -615,6 +631,8 @@ typedef struct int is_initialised; int method_set; char eof; + uint8_t saved_bitbuffer[8]; + Py_ssize_t saved_bitbuffer_size; PyThread_type_lock lock; // inflate_state should be at the bottom as it contains buffers inside the struct. struct inflate_state zst; @@ -623,7 +641,9 @@ typedef struct static void Decomp_dealloc(decompobject *self) { - PyThread_free_lock(self->lock); + if (self->lock != NULL) { + PyThread_free_lock(self->lock); + } Py_XDECREF(self->unused_data); Py_XDECREF(self->unconsumed_tail); Py_XDECREF(self->zdict); @@ -663,9 +683,13 @@ newdecompobject() if (self == NULL) return NULL; self->eof = 0; + self->saved_bitbuffer_size = 0; self->is_initialised = 0; self->method_set = 0; self->zdict = NULL; + // Decomp_dealloc uses these members when initialization fails + self->unconsumed_tail = NULL; + self->lock = NULL; self->unused_data = PyBytes_FromStringAndSize("", 0); if (self->unused_data == NULL) { Py_DECREF(self); @@ -676,7 +700,7 @@ newdecompobject() Py_DECREF(self); return NULL; } - self->lock = PyThread_allocate_lock(); + self->lock = PyThread_allocate_lock(); if (self->lock == NULL) { Py_DECREF(self); PyErr_SetString(PyExc_MemoryError, "Unable to allocate lock"); @@ -701,11 +725,11 @@ isal_zlib_compressobj_impl(PyObject *module, int level, int method, int wbits, method); goto error; } - if (strategy != Z_DEFAULT_STRATEGY){ + if (strategy != Z_DEFAULT_STRATEGY && !(strategy == Z_FIXED && level == 0)){ err = PyErr_WarnEx( PyExc_UserWarning, - "Only one strategy is supported when using isal_zlib. Using the " - "default strategy.", + "isal_zlib supports only Z_DEFAULT_STRATEGY, and Z_FIXED at " + "compression level 0. Using the default strategy.", 1); if (err == -1) // Warning was turned into an exception. @@ -748,6 +772,17 @@ isal_zlib_compressobj_impl(PyObject *module, int level, int method, int wbits, self->zst.level = level; self->zst.hist_bits = (uint16_t)hist_bits; self->zst.gzip_flag = (uint16_t)flag; + if (strategy == Z_FIXED && level == 0) { + /* Level 0 encodes every block with the stream's Huffman table, so the + static table gives fixed Huffman blocks only. Levels 1 to 3 choose + the codes for each block themselves and ignore this setting. */ + err = isal_deflate_set_hufftables(&(self->zst), NULL, + IGZIP_HUFFTABLE_STATIC); + if (err != COMP_OK) { + isal_deflate_error(err); + goto error; + } + } self->is_initialised = 1; if (zdict->buf == NULL) { @@ -755,8 +790,19 @@ isal_zlib_compressobj_impl(PyObject *module, int level, int method, int wbits, } else { err = isal_deflate_set_dict(&(self->zst), zdict->buf, (uint32_t)zdict->len); - if (err == COMP_OK) + if (err == COMP_OK) { + if (flag == IGZIP_ZLIB && zdict->len != 0) { + // Set every field instead of calling isal_zlib_header_init, + // which ISA-L 2.30 does not have. + self->zlib_header.info = hist_bits - 8; + self->zlib_header.level = level == 0 ? 0 : 1; + self->zlib_header.dict_flag = 1; + self->zlib_header.dict_id = isal_adler32( + 1, zdict->buf, (uint32_t)zdict->len); + self->zst.gzip_flag = IGZIP_ZLIB_NO_HDR; + } goto success; + } PyErr_SetString(PyExc_ValueError, "Invalid dictionary"); goto error; } @@ -817,6 +863,31 @@ isal_zlib_decompressobj_impl(PyObject *module, int wbits, PyObject *zdict) return (PyObject *)self; } +/* Called with the compressor locked and a newly arranged output buffer. + * The six-byte dictionary header is written once, including when flush() + * is the first operation. ISA-L still generates the Adler-32 trailer. */ +static int +write_pending_zlib_header(compobject *self) +{ + if (self->zlib_header.dict_flag) { + uint8_t *header = self->zst.next_out; + if (isal_write_zlib_header(&self->zst, &self->zlib_header) != 0) { + PyErr_SetString(PyExc_RuntimeError, + "Insufficient space for the zlib dictionary header"); + return -1; + } + /* Older ISA-L releases write DICTID in little-endian order. + * RFC 1950 requires network byte order, regardless of ISA-L version. */ + uint32_t dict_id = self->zlib_header.dict_id; + header[2] = (uint8_t)(dict_id >> 24); + header[3] = (uint8_t)(dict_id >> 16); + header[4] = (uint8_t)(dict_id >> 8); + header[5] = (uint8_t)dict_id; + self->zlib_header.dict_flag = 0; + } + return 0; +} + static PyObject * isal_zlib_Compress_compress_impl(compobject *self, Py_buffer *data) /*[clinic end generated code: output=5d5cd791cbc6a7f4 input=0d95908d6e64fab8]*/ @@ -838,6 +909,9 @@ isal_zlib_Compress_compress_impl(compobject *self, Py_buffer *data) if (obuflen < 0) goto error; + if (write_pending_zlib_header(self) < 0) + goto error; + Py_BEGIN_ALLOW_THREADS err = isal_deflate(&self->zst); Py_END_ALLOW_THREADS @@ -875,7 +949,8 @@ save_unconsumed_input(decompobject *self, Py_buffer *data, int err) Py_ssize_t new_size, left_size; PyObject *new_data; Py_ssize_t bytes_in_bitbuffer = bitbuffer_size(&(self->zst)); - left_size = (uint8_t *)data->buf + data->len - self->zst.next_in; + left_size = (Py_ssize_t)((uintptr_t)data->buf + data->len - + (uintptr_t)self->zst.next_in); if (left_size + bytes_in_bitbuffer > (PY_SSIZE_T_MAX - old_size)) { PyErr_NoMemory(); return -1; @@ -889,9 +964,16 @@ save_unconsumed_input(decompobject *self, Py_buffer *data, int err) memcpy(new_data_ptr, PyBytes_AS_STRING(self->unused_data), old_size); bitbuffer_copy(&(self->zst), new_data_ptr + old_size, bytes_in_bitbuffer); - memcpy(new_data_ptr + old_size + bytes_in_bitbuffer, - self->zst.next_in, left_size); + if (copy_from_next_in(&(self->zst), data->buf, + self->saved_bitbuffer, + self->saved_bitbuffer_size, + new_data_ptr + old_size + bytes_in_bitbuffer, + left_size) < 0) { + Py_DECREF(new_data); + return -1; + } Py_SETREF(self->unused_data, new_data); + self->zst.next_in = (uint8_t *)data->buf + data->len; self->zst.avail_in = 0; } } @@ -908,6 +990,10 @@ save_unconsumed_input(decompobject *self, Py_buffer *data, int err) Py_SETREF(self->unconsumed_tail, new_data); } + if (self->zst.block_state != ISAL_BLOCK_FINISH) { + save_bitbuffer(&(self->zst), self->saved_bitbuffer, + &self->saved_bitbuffer_size); + } return 0; } @@ -927,7 +1013,18 @@ isal_zlib_Decompress_decompress_impl(decompobject *self, Py_buffer *data, else hard_limit = max_length; + /* limit amount of data allocated to max_length */ + if (max_length && obuflen > max_length) + obuflen = max_length; + + ENTER_ZLIB(self); + if (!self->method_set) { + if (data->len == 0) { + /* Detect the format when the first byte arrives. */ + LEAVE_ZLIB(self); + return PyBytes_FromStringAndSize(NULL, 0); + } if (data_is_gzip(data)){ self->zst.crc_flag = ISAL_GZIP; } @@ -937,12 +1034,6 @@ isal_zlib_Decompress_decompress_impl(decompobject *self, Py_buffer *data, self->method_set = 1; } - /* limit amount of data allocated to max_length */ - if (max_length && obuflen > max_length) - obuflen = max_length; - - ENTER_ZLIB(self); - self->zst.next_in = data->buf; ibuflen = data->len; @@ -1036,6 +1127,7 @@ isal_zlib_Compress_flush_impl(compobject *self, int mode) } else { PyErr_Format(IsalError, "Unsupported flush mode: %d", mode); + LEAVE_ZLIB(self); return NULL; } @@ -1049,6 +1141,11 @@ isal_zlib_Compress_flush_impl(compobject *self, int mode) goto error; } + if (write_pending_zlib_header(self) < 0) { + Py_CLEAR(RetVal); + goto error; + } + Py_BEGIN_ALLOW_THREADS err = isal_deflate(&self->zst); Py_END_ALLOW_THREADS @@ -1066,6 +1163,7 @@ isal_zlib_Compress_flush_impl(compobject *self, int mode) mode is Z_FINISH, but checking both for safety*/ if (self->zst.internal_state.state == ZSTATE_END && mode == Z_FINISH) { PyMem_FREE(self->level_buf); + self->level_buf = NULL; self->zst.level_buf_size = 0; self->zst.level_buf = NULL; self->is_initialised = 0; @@ -1098,6 +1196,14 @@ isal_zlib_Decompress_flush_impl(decompobject *self, Py_ssize_t length) ENTER_ZLIB(self); + if (!self->method_set) { + /* No data yet, so the format is unknown. isal_inflate would start a + raw deflate block, and then decode the zlib or gzip header that + comes next as deflate data. */ + LEAVE_ZLIB(self); + return PyBytes_FromStringAndSize(NULL, 0); + } + if (PyObject_GetBuffer(self->unconsumed_tail, &data, PyBUF_SIMPLE) == -1) { LEAVE_ZLIB(self); return NULL; @@ -1175,8 +1281,10 @@ PyDoc_STRVAR(isal_zlib_compressobj__doc__, " Valid values range from 1 to 9. Higher values result in higher memory\n" " usage, faster compression, and smaller output.\n" " strategy\n" -" Used to tune the compression algorithm. Not supported by ISA-L.\n" -" Only a default strategy is used.\n" +" Used to tune the compression algorithm. At level 0, Z_FIXED uses fixed\n" +" Huffman codes, which gives smaller output when small amounts of data are\n" +" flushed. Other strategies, and Z_FIXED at levels 1 to 3, give a warning\n" +" and use the default strategy.\n" " zdict\n" " The predefined compression dictionary - a sequence of bytes\n" " containing subsequences that are likely to occur in the input data."); @@ -1422,13 +1530,11 @@ static PyTypeObject IsalZlibCompType = { .tp_name = "isal_zlib.Compress", .tp_doc = "Object returned by isal_zlib.compressobj", .tp_basicsize = sizeof(compobject), - .tp_flags = Py_TPFLAGS_DEFAULT, + .tp_flags = Py_TPFLAGS_DEFAULT | Py_TPFLAGS_DISALLOW_INSTANTIATION, .tp_dealloc = (destructor)Comp_dealloc, .tp_methods = comp_methods, }; -#define COMP_OFF(x) offsetof(decompobject, x) - PyDoc_STRVAR(Decomp_unconsumed_tail, "A bytes object that contains any data that was not consumed by the last\n" "decompress() call because it exceeded the limit for the uncompressed data\n" @@ -1436,24 +1542,55 @@ PyDoc_STRVAR(Decomp_unconsumed_tail, "feed it (possibly with further data concatenated to it) back to a \n" "subsequent decompress() method call in order to get correct output."); -static PyMemberDef Decomp_members[] = { - {"unused_data", T_OBJECT, COMP_OFF(unused_data), READONLY, - "Data found after the end of the compressed stream."}, - {"unconsumed_tail", T_OBJECT, COMP_OFF(unconsumed_tail), READONLY, - Decomp_unconsumed_tail}, - {"eof", T_BOOL, COMP_OFF(eof), READONLY, - "True if the end-of-stream marker has been reached."}, +/* The getters take the lock, because without the GIL another thread can + change these attributes in decompress() or flush(). */ +static PyObject * +Decomp_get_unused_data(decompobject *self, void *Py_UNUSED(closure)) +{ + ENTER_ZLIB(self); + PyObject *unused_data = self->unused_data; + Py_INCREF(unused_data); + LEAVE_ZLIB(self); + return unused_data; +} + +static PyObject * +Decomp_get_unconsumed_tail(decompobject *self, void *Py_UNUSED(closure)) +{ + ENTER_ZLIB(self); + PyObject *unconsumed_tail = self->unconsumed_tail; + Py_INCREF(unconsumed_tail); + LEAVE_ZLIB(self); + return unconsumed_tail; +} + +static PyObject * +Decomp_get_eof(decompobject *self, void *Py_UNUSED(closure)) +{ + ENTER_ZLIB(self); + char eof = self->eof; + LEAVE_ZLIB(self); + return PyBool_FromLong(eof); +} + +static PyGetSetDef Decomp_properties[] = { + {"unused_data", (getter)Decomp_get_unused_data, NULL, + "Data found after the end of the compressed stream.", NULL}, + {"unconsumed_tail", (getter)Decomp_get_unconsumed_tail, NULL, + Decomp_unconsumed_tail, NULL}, + {"eof", (getter)Decomp_get_eof, NULL, + "True if the end-of-stream marker has been reached.", NULL}, {NULL}, }; static PyTypeObject IsalZlibDecompType = { .tp_name = "isal_zlib.Decompress", - .tp_doc = "Object returned by isal_zlib.compressobj.", + .tp_doc = "Object returned by isal_zlib.decompressobj.", .tp_basicsize = sizeof(decompobject), - .tp_flags = Py_TPFLAGS_DEFAULT, + .tp_flags = Py_TPFLAGS_DEFAULT | Py_TPFLAGS_DISALLOW_INSTANTIATION, .tp_dealloc = (destructor)Decomp_dealloc, .tp_methods = Decomp_methods, - .tp_members = Decomp_members, + .tp_getset = Decomp_properties, }; #define GzipReader_HEADER 1 @@ -1474,6 +1611,9 @@ typedef struct _GzipReaderStruct { char stream_phase; char all_bytes_read; char closed; + /* Set when the first gzip header is read. A modification time of 0 is + valid, so _last_mtime cannot tell whether a header was read. */ + char has_last_mtime; uint32_t _last_mtime; PyThread_type_lock lock; struct inflate_state state; @@ -1488,7 +1628,9 @@ static void GzipReader_dealloc(GzipReader *self) PyMem_Free(self->memview); } Py_XDECREF(self->fp); - PyThread_free_lock(self->lock); + if (self->lock != NULL) { + PyThread_free_lock(self->lock); + } Py_TYPE(self)->tp_free(self); } @@ -1529,8 +1671,15 @@ GzipReader__new__(PyTypeObject *type, PyObject *args, PyObject *kwargs) return NULL; } GzipReader *self = PyObject_New(GzipReader, type); + if (self == NULL) { + return NULL; + } + // GzipReader_dealloc uses these members when initialization fails + self->input_buffer = NULL; + self->memview = NULL; + self->fp = NULL; + self->lock = NULL; if (PyObject_HasAttrString(fp, "read")) { - self->memview = NULL; self->buffer_size = buffer_size; self->input_buffer = PyMem_Malloc(self->buffer_size); if (self->input_buffer == NULL) { @@ -1540,14 +1689,17 @@ GzipReader__new__(PyTypeObject *type, PyObject *args, PyObject *kwargs) self->buffer_end = self->input_buffer; self->all_bytes_read = 0; } else { - self->memview = PyMem_Malloc(sizeof(Py_buffer)); - if (self->memview == NULL) { + Py_buffer *memview = PyMem_Malloc(sizeof(Py_buffer)); + if (memview == NULL) { + Py_DECREF(self); return PyErr_NoMemory(); } - if (PyObject_GetBuffer(fp, self->memview, PyBUF_SIMPLE) < 0) { + if (PyObject_GetBuffer(fp, memview, PyBUF_SIMPLE) < 0) { + PyMem_Free(memview); Py_DECREF(self); return NULL; } + self->memview = memview; self->buffer_size = self->memview->len; self->input_buffer = self->memview->buf; self->buffer_end = self->input_buffer + self->buffer_size; @@ -1560,6 +1712,7 @@ GzipReader__new__(PyTypeObject *type, PyObject *args, PyObject *kwargs) self->fp = fp; self->stream_phase = GzipReader_HEADER; self->closed = 0; + self->has_last_mtime = 0; self->_last_mtime = 0; self->lock = PyThread_allocate_lock(); if (self->lock == NULL) { @@ -1581,17 +1734,15 @@ GzipReader_read_from_file(GzipReader *self) uint8_t *buffer_end = self->buffer_end; size_t remaining = buffer_end - current_pos; if (remaining == self->buffer_size) { - /* Buffer is full but a new read request was issued. This will be due - to the header being bigger than the header. Enlarge the buffer - to accommodate the hzip header. */ - size_t new_buffer_size = self->buffer_size * 2; - uint8_t *tmp_buffer = PyMem_Realloc(self->input_buffer, new_buffer_size); - if (tmp_buffer == NULL) { - PyErr_NoMemory(); - return -1; - } - self->input_buffer = tmp_buffer; - self->buffer_size = new_buffer_size; + /* Buffer is full but a new read request was issued. This will be due + to the header being bigger than the buffer. To prevent adverserial + inputs with infinite header sizes an error is thrown here. */ + PyErr_Format( + PyExc_OverflowError, + "Gzip header exceeds input buffer size of %lu", + self->buffer_size + ); + return -1; } else if (remaining > 0) { memmove(self->input_buffer, current_pos, remaining); } @@ -1629,7 +1780,28 @@ GzipReader_read_from_file(GzipReader *self) #define FNAME 8 #define FCOMMENT 16 -static PyObject *BadGzipFile; // Import BadGzipFile error for consistency +/* Raise gzip.BadGzipFile. Look up the class every time, because each + interpreter has its own gzip module, but all interpreters share the C + globals of this module. */ +static void +raise_bad_gzip_file(const char *format, ...) +{ + PyObject *gzip_module = PyImport_ImportModule("gzip"); + if (gzip_module == NULL) { + return; + } + PyObject *bad_gzip_file = PyObject_GetAttrString(gzip_module, + "BadGzipFile"); + Py_DECREF(gzip_module); + if (bad_gzip_file == NULL) { + return; + } + va_list vargs; + va_start(vargs, format); + PyErr_FormatV(bad_gzip_file, format, vargs); + va_end(vargs); + Py_DECREF(bad_gzip_file); +} static inline uint32_t load_u32_le(void *address) { #if PY_BIG_ENDIAN @@ -1684,7 +1856,10 @@ GzipReader_read_into_buffer(GzipReader *self, uint8_t *out_buffer, size_t out_bu if (!(magic1 == 0x1f && magic2 == 0x8b)) { Py_BLOCK_THREADS; PyObject *magic_obj = PyBytes_FromStringAndSize((char *)current_pos, 2); - PyErr_Format(BadGzipFile, + if (magic_obj == NULL) { + return -1; + } + raise_bad_gzip_file( "Not a gzipped file (%R)", magic_obj); Py_DECREF(magic_obj); @@ -1693,11 +1868,12 @@ GzipReader_read_into_buffer(GzipReader *self, uint8_t *out_buffer, size_t out_bu uint8_t method = current_pos[2]; if (method != 8) { Py_BLOCK_THREADS; - PyErr_SetString(BadGzipFile, "Unknown compression method"); + raise_bad_gzip_file("Unknown compression method"); return -1; } uint8_t flags = current_pos[3]; self->_last_mtime = load_u32_le(current_pos + 4); + self->has_last_mtime = 1; // Skip XFL and header flag uint8_t *header_cursor = current_pos + 10; if (flags & FEXTRA) { @@ -1737,8 +1913,7 @@ GzipReader_read_into_buffer(GzipReader *self, uint8_t *out_buffer, size_t out_bu 0, current_pos, header_cursor - current_pos) & 0xFFFF; if (header_crc != crc) { Py_BLOCK_THREADS; - PyErr_Format( - BadGzipFile, + raise_bad_gzip_file( "Corrupted gzip header. Checksums do not " "match: %04x != %04x", crc, header_crc @@ -1792,8 +1967,7 @@ GzipReader_read_into_buffer(GzipReader *self, uint8_t *out_buffer, size_t out_bu current_pos += 4; if (crc != self->state.crc) { Py_BLOCK_THREADS; - PyErr_Format( - BadGzipFile, + raise_bad_gzip_file( "CRC check failed %u != %u", crc, self->state.crc ); @@ -1803,7 +1977,7 @@ GzipReader_read_into_buffer(GzipReader *self, uint8_t *out_buffer, size_t out_bu current_pos += 4; if (length != self->state.total_out) { Py_BLOCK_THREADS; - PyErr_SetString(BadGzipFile, "Incorrect length of data produced"); + raise_bad_gzip_file("Incorrect length of data produced"); return -1; } self->stream_phase = GzipReader_NULL_BYTES; @@ -1849,7 +2023,11 @@ static PyObject * GzipReader_readinto(GzipReader *self, PyObject *buffer_obj) { Py_buffer view; - if (PyObject_GetBuffer(buffer_obj, &view, PyBUF_SIMPLE) < 0) { + if (PyObject_GetBuffer(buffer_obj, &view, PyBUF_WRITABLE) < 0) { + /* Raise TypeError, like the readinto() methods of io. */ + PyErr_Format(PyExc_TypeError, + "readinto() argument must be read-write bytes-like " + "object, not %.200s", Py_TYPE(buffer_obj)->tp_name); return NULL; } uint8_t *buffer = view.buf; @@ -1864,16 +2042,10 @@ GzipReader_readinto(GzipReader *self, PyObject *buffer_obj) return PyLong_FromSsize_t((Py_ssize_t)written_size); } +/* Called with self->lock held */ static PyObject * -GzipReader_seek(GzipReader *self, PyObject *args, PyObject *kwargs) +GzipReader_seek_impl(GzipReader *self, Py_ssize_t offset, Py_ssize_t whence) { - Py_ssize_t offset; - Py_ssize_t whence = SEEK_SET; - static char *keywords[] = {"offset", "whence", NULL}; - static char format[] = {"n|n:GzipReader.seek"}; - if (PyArg_ParseTupleAndKeywords(args, kwargs, format, keywords, &offset, &whence) < 0) { - return NULL; - } // Recalculate offset as an absolute file position. if (whence == SEEK_SET) { ; @@ -1918,6 +2090,10 @@ GzipReader_seek(GzipReader *self, PyObject *args, PyObject *kwargs) if (seek_result == NULL) { return NULL; } + Py_DECREF(seek_result); + /* Discard compressed data buffered before rewinding the file. */ + self->current_pos = self->input_buffer; + self->buffer_end = self->input_buffer; self->stream_phase = GzipReader_HEADER; self->_pos = 0; self->all_bytes_read = 0; @@ -1950,6 +2126,24 @@ GzipReader_seek(GzipReader *self, PyObject *args, PyObject *kwargs) return PyLong_FromLongLong(self->_pos); } +static PyObject * +GzipReader_seek(GzipReader *self, PyObject *args, PyObject *kwargs) +{ + Py_ssize_t offset; + Py_ssize_t whence = SEEK_SET; + static char *keywords[] = {"offset", "whence", NULL}; + static char format[] = {"n|n:GzipReader.seek"}; + if (!PyArg_ParseTupleAndKeywords(args, kwargs, format, keywords, &offset, &whence)) { + return NULL; + } + /* Take the lock like read() does. Decompression releases the GIL, so + another thread could otherwise read while seek() changes the state. */ + ENTER_ZLIB(self); + PyObject *result = GzipReader_seek_impl(self, offset, whence); + LEAVE_ZLIB(self); + return result; +} + static PyObject * GzipReader_readall(GzipReader *self, PyObject *Py_UNUSED(ignore)) { @@ -2026,7 +2220,7 @@ static PyObject * GzipReader_read(GzipReader *self, PyObject *args) { Py_ssize_t size = -1; - if (PyArg_ParseTuple(args, "|n:GzipReader.read", &size) < 0) { + if (!PyArg_ParseTuple(args, "|n:GzipReader.read", &size)) { return NULL; } if (size < 0) { @@ -2091,7 +2285,7 @@ GzipReader_flush(GzipReader *self, PyObject *Py_UNUSED(ignore)) { static PyObject * GzipReader_get_last_mtime(GzipReader *self, void *Py_UNUSED(closure)) { - if (self->_last_mtime) { + if (self->has_last_mtime) { return PyLong_FromUnsignedLong(self->_last_mtime); } Py_RETURN_NONE; @@ -2171,7 +2365,10 @@ static struct PyModuleDef isal_zlib_module = { PyModuleDef_HEAD_INIT, "isal_zlib", /* name of module */ isal_zlib_module_documentation, /* module documentation, may be NULL */ - 0, + /* -1, as in igzip_lib, because IsalError is a C global. A subinterpreter + then gets a copy of this module instead of running PyInit_isal_zlib + again. */ + -1, IsalZlibMethods, }; @@ -2197,18 +2394,6 @@ PyInit_isal_zlib(void) return NULL; } - PyObject *gzip_module = PyImport_ImportModule("gzip"); - if (gzip_module == NULL) { - return NULL; - } - - BadGzipFile = PyObject_GetAttrString(gzip_module, "BadGzipFile"); - if (BadGzipFile == NULL) { - return NULL; - } - Py_INCREF(BadGzipFile); - - Py_INCREF(IsalError); if (PyModule_AddObject(m, "IsalError", IsalError) < 0) { return NULL; diff --git a/tests/test_compat.py b/tests/test_compat.py index 0ea54f70..3f339851 100644 --- a/tests/test_compat.py +++ b/tests/test_compat.py @@ -11,10 +11,16 @@ import gzip import itertools +import random +import subprocess +import sys +import textwrap +import threading +import warnings import zlib from pathlib import Path -from isal import igzip, isal_zlib +from isal import igzip, igzip_lib, isal_zlib import pytest @@ -36,21 +42,57 @@ @pytest.mark.parametrize(["data_size", "value"], - itertools.product(DATA_SIZES, SEEDS)) + list(itertools.product(DATA_SIZES, SEEDS))) def test_crc32(data_size, value): data = DATA[:data_size] assert zlib.crc32(data, value) == isal_zlib.crc32(data, value) @pytest.mark.parametrize(["data_size", "value"], - itertools.product(DATA_SIZES, SEEDS)) + list(itertools.product(DATA_SIZES, SEEDS))) def test_adler32(data_size, value): data = DATA[:data_size] assert zlib.adler32(data, value) == isal_zlib.adler32(data, value) +@pytest.mark.parametrize("args", [("a", 1, 2), (1, "a", 2), (1, 2, "a"), + (1, 2)]) +def test_crc32_combine_wrong_arguments(args): + with pytest.raises(TypeError): + isal_zlib.crc32_combine(*args) + + +@pytest.mark.parametrize("factory", + [isal_zlib.compressobj, isal_zlib.decompressobj]) +def test_compress_and_decompress_types_cannot_be_called(factory): + with pytest.raises(TypeError): + type(factory())() + + +@pytest.mark.skipif(sys.implementation.name != "cpython", + reason="Needs _testcapi.set_nomemory") +@pytest.mark.parametrize("factory", ["compressobj", "decompressobj"]) +def test_compressobj_decompressobj_no_memory(factory): + pytest.importorskip("_testcapi") + # Fail each memory allocation in turn, such as the one for the lock. A + # subprocess keeps the failing allocations away from the test runner. + code = textwrap.dedent(f""" + import _testcapi + from isal import isal_zlib + for start in range(20): + _testcapi.set_nomemory(start, start + 1) + try: + isal_zlib.{factory}() + except MemoryError: + pass + finally: + _testcapi.remove_mem_hooks() + """) + subprocess.run([sys.executable, "-c", code], check=True) + + @pytest.mark.parametrize(["data_size", "level"], - itertools.product(DATA_SIZES, range(4))) + list(itertools.product(DATA_SIZES, range(4)))) def test_compress(data_size, level): data = DATA[:data_size] compressed = isal_zlib.compress(data, level=level) @@ -58,7 +100,7 @@ def test_compress(data_size, level): @pytest.mark.parametrize(["data_size", "level"], - itertools.product(DATA_SIZES, range(10))) + list(itertools.product(DATA_SIZES, range(10)))) def test_decompress_zlib(data_size, level): data = DATA[:data_size] compressed = zlib.compress(data, level=level) @@ -67,8 +109,8 @@ def test_decompress_zlib(data_size, level): @pytest.mark.parametrize(["data_size", "level", "wbits", "memLevel"], - itertools.product([128 * 1024], range(4), - WBITS_RANGE, range(1, 10))) + list(itertools.product([128 * 1024], range(4), + WBITS_RANGE, range(1, 10)))) def test_decompress_wbits(data_size, level, wbits, memLevel): data = DATA[:data_size] compressobj = zlib.compressobj(level=level, wbits=wbits, memLevel=memLevel) @@ -78,7 +120,7 @@ def test_decompress_wbits(data_size, level, wbits, memLevel): @pytest.mark.parametrize(["data_size", "level"], - itertools.product(DATA_SIZES, range(4))) + list(itertools.product(DATA_SIZES, range(4)))) def test_decompress_isal_zlib(data_size, level): data = DATA[:data_size] compressed = isal_zlib.compress(data, level=level) @@ -87,9 +129,19 @@ def test_decompress_isal_zlib(data_size, level): assert decompressed == data +@pytest.mark.parametrize("module", [isal_zlib, igzip_lib]) +def test_decompress_bufsize(module): + data = DATA[:10_000] + compressed = module.compress(data) + # Like zlib.decompress, decompress accepts a bufsize of 0. + assert module.decompress(compressed, bufsize=0) == data + with pytest.raises(ValueError, match="bufsize must be non-negative"): + module.decompress(compressed, bufsize=-1) + + @pytest.mark.parametrize(["data_size", "level", "wbits", "memLevel"], - itertools.product([128 * 1024], range(4), - WBITS_RANGE, range(1, 10))) + list(itertools.product([128 * 1024], range(4), + WBITS_RANGE, range(1, 10)))) def test_compress_compressobj(data_size, level, wbits, memLevel): data = DATA[:data_size] compressobj = isal_zlib.compressobj(level=level, @@ -100,9 +152,93 @@ def test_compress_compressobj(data_size, level, wbits, memLevel): assert data == decompressed +def test_compressobj_flush_unsupported_mode(): + compressor = isal_zlib.compressobj() + with pytest.raises(isal_zlib.error) as error: + compressor.flush(isal_zlib.Z_PARTIAL_FLUSH) + error.match("Unsupported flush mode") + # The next call hangs if the failed flush kept the lock + result = [] + thread = threading.Thread(target=lambda: result.append( + compressor.compress(b"data") + compressor.flush()), daemon=True) + thread.start() + thread.join(timeout=10) + assert result + assert zlib.decompress(result[0]) == b"data" + + +def test_compressobj_level_0_flush_after_finish(): + compressor = isal_zlib.compressobj(0) + compressed = compressor.compress(b"data") + compressor.flush() + assert compressor.flush() == b"" + assert zlib.decompress(compressed) == b"data" + + +@pytest.mark.parametrize("flush_mode", [isal_zlib.Z_SYNC_FLUSH, + isal_zlib.Z_FULL_FLUSH]) +@pytest.mark.parametrize(["strategy", "btype"], + [(isal_zlib.Z_DEFAULT_STRATEGY, 2), + (isal_zlib.Z_FIXED, 1)]) +def test_compressobj_level_0_block_type(strategy, btype, flush_mode): + with warnings.catch_warnings(): + warnings.simplefilter("error") + compressor = isal_zlib.compressobj(0, wbits=-15, strategy=strategy) + decompressor = zlib.decompressobj(wbits=-15) + for i in range(0, 64 * 1024, 4096): + data = DATA[i:i + 4096] + compressed = compressor.compress(data) + compressor.flush(flush_mode) + # Sync and full flushes end on a byte boundary, so each output starts + # with a new block header. Bits 1 and 2 of the first byte hold BTYPE: + # 1 for fixed Huffman codes, 2 for dynamic Huffman codes. + assert compressed[0] >> 1 & 3 == btype + assert decompressor.decompress(compressed) == data + + +# Gzip has no preset dictionary, so zdict is only tested with zlib and raw. +@pytest.mark.parametrize(["wbits", "use_zdict"], + [(wbits, False) for wbits in WBITS_RANGE] + + [(wbits, True) for wbits in WBITS_RANGE + if not 25 <= wbits <= 31]) +def test_compressobj_level_0_z_fixed_round_trip(wbits, use_zdict): + kwargs = {"zdict": DATA[:1024]} if use_zdict else {} + with warnings.catch_warnings(): + warnings.simplefilter("error") + compressor = isal_zlib.compressobj( + 0, wbits=wbits, strategy=isal_zlib.Z_FIXED, **kwargs) + # Random bytes use all literal codes, including the 9-bit ones. + data = DATA[:64 * 1024] + random.Random(0).randbytes(64 * 1024) + flush_modes = itertools.cycle([isal_zlib.Z_SYNC_FLUSH, + isal_zlib.Z_FULL_FLUSH, + isal_zlib.Z_NO_FLUSH]) + compressed = b"" + for i, mode in zip(range(0, len(data), 1000), flush_modes): + compressed += (compressor.compress(data[i:i + 1000]) + + compressor.flush(mode)) + compressed += compressor.flush() + decompressor = zlib.decompressobj(wbits=wbits, **kwargs) + assert decompressor.decompress(compressed) == data + assert decompressor.eof + + +@pytest.mark.parametrize(["level", "strategy"], + list(itertools.product(range(4), + [isal_zlib.Z_FILTERED, + isal_zlib.Z_HUFFMAN_ONLY, + isal_zlib.Z_RLE])) + + [(level, isal_zlib.Z_FIXED) for level in range(1, 4)]) +def test_compressobj_unsupported_strategy(level, strategy): + data = DATA[:64 * 1024] + with pytest.warns(UserWarning, match="Using the default strategy"): + compressor = isal_zlib.compressobj(level, strategy=strategy) + default_compressor = isal_zlib.compressobj(level) + compressed = compressor.compress(data) + compressor.flush() + assert compressed == (default_compressor.compress(data) + + default_compressor.flush()) + + @pytest.mark.parametrize(["data_size", "level", "wbits", "memLevel"], - itertools.product([128 * 1024], range(4), - WBITS_RANGE, range(1, 10))) + list(itertools.product([128 * 1024], range(4), + WBITS_RANGE, range(1, 10)))) def test_decompress_decompressobj(data_size, level, wbits, memLevel): data = DATA[:data_size] compressobj = zlib.compressobj(level=level, wbits=wbits, memLevel=memLevel) @@ -123,7 +259,7 @@ def test_decompressobj_unconsumed_tail(): @pytest.mark.parametrize(["data_size", "level"], - itertools.product(DATA_SIZES, range(4))) + list(itertools.product(DATA_SIZES, range(4)))) def test_igzip_compress(data_size, level): data = DATA[:data_size] compressed = igzip.compress(data, compresslevel=level) @@ -131,7 +267,7 @@ def test_igzip_compress(data_size, level): @pytest.mark.parametrize(["data_size", "level"], - itertools.product(DATA_SIZES, range(10))) + list(itertools.product(DATA_SIZES, range(10)))) def test_decompress_gzip(data_size, level): data = DATA[:data_size] compressed = gzip.compress(data, compresslevel=level) @@ -140,7 +276,7 @@ def test_decompress_gzip(data_size, level): @pytest.mark.parametrize(["data_size", "level"], - itertools.product(DATA_SIZES, range(4))) + list(itertools.product(DATA_SIZES, range(4)))) def test_decompress_igzip(data_size, level): data = DATA[:data_size] compressed = igzip.compress(data, compresslevel=level) @@ -150,7 +286,7 @@ def test_decompress_igzip(data_size, level): @pytest.mark.parametrize(["unused_size", "wbits"], - itertools.product([26], [-15, 15, 31])) + list(itertools.product([26], [-15, 15, 31]))) def test_unused_data(unused_size, wbits): unused_data = b"abcdefghijklmnopqrstuvwxyz"[:unused_size] compressor = zlib.compressobj(wbits=wbits) @@ -162,6 +298,67 @@ def test_unused_data(unused_size, wbits): assert decompressor.unused_data == unused_data +@pytest.mark.parametrize("wbits", [15, 31]) +def test_decompressobj_detects_format_byte_by_byte(wbits): + data = DATA[:1000] + compressor = zlib.compressobj(wbits=wbits) + compressed = compressor.compress(data) + compressor.flush() + decompressor = isal_zlib.decompressobj(wbits=47) + result = b"".join(decompressor.decompress(compressed[i:i + 1]) + for i in range(len(compressed))) + assert result == data + assert decompressor.eof + + +@pytest.mark.parametrize("wbits", [15, 31]) +def test_decompressobj_detects_format_after_empty_input(wbits): + data = DATA[:1000] + compressor = zlib.compressobj(wbits=wbits) + compressed = compressor.compress(data) + compressor.flush() + for module in (zlib, isal_zlib): + decompressor = module.decompressobj(wbits=47) + assert decompressor.decompress(b"") == b"" + assert decompressor.decompress(compressed[:1]) == b"" + assert decompressor.decompress(compressed[1:] + b"unused") == data + assert decompressor.eof + assert decompressor.unused_data == b"unused" + + +@pytest.mark.parametrize("wbits", [15, 31]) +def test_decompressobj_detects_format_after_flush(wbits): + data = DATA[:1000] + compressor = zlib.compressobj(wbits=wbits) + compressed = compressor.compress(data) + compressor.flush() + for module in (zlib, isal_zlib): + decompressor = module.decompressobj(wbits=47) + assert decompressor.flush() == b"" + assert decompressor.decompress(compressed) == data + assert decompressor.eof + + +@pytest.mark.parametrize("data", [b"", b"\x1f", b"\x78"]) +def test_decompress_detect_format_incomplete(data): + with pytest.raises(zlib.error): + zlib.decompress(data, wbits=47) + with pytest.raises(isal_zlib.error): + isal_zlib.decompress(data, wbits=47) + + +@pytest.mark.parametrize("wbits", [-15, 15, 31]) +def test_decompressobj_unconsumed_tail_after_eof(wbits): + source = b"abcdefghijklmnopqrstuvwxyz" + remainder = b"0123456789" + compressor = zlib.compressobj(wbits=wbits) + compressed = compressor.compress(source) + compressor.flush() + decompressor = isal_zlib.decompressobj(wbits=wbits) + data = decompressor.decompress(compressed + remainder, 1) + data += decompressor.decompress(decompressor.unconsumed_tail) + assert decompressor.eof + assert data == source + assert decompressor.unconsumed_tail == b"" + assert decompressor.unused_data == remainder + + def test_zlib_dictionary_decompress(): dictionary = b"bla" data = b"bladiebla" diff --git a/tests/test_dictionary_header.py b/tests/test_dictionary_header.py new file mode 100644 index 00000000..028b88c7 --- /dev/null +++ b/tests/test_dictionary_header.py @@ -0,0 +1,70 @@ +"""Cross-library checks for dictionary headers and compressor lifecycle.""" +import random +import zlib + +from isal import isal_zlib + +import pytest + + +@pytest.mark.parametrize('level', range(4)) +@pytest.mark.parametrize('wbits', [9, 15, -15]) +@pytest.mark.parametrize('dictionary_size', [11, 32768, 40000]) +@pytest.mark.parametrize( + 'first_operation', ['compress', 'empty', 'sync', 'finish']) +def test_dictionary_cross_decode(level, wbits, dictionary_size, + first_operation): + dictionary = random.Random(21).randbytes(dictionary_size) + data = dictionary[-256:] * 5 + random.Random(22).randbytes(4096) + compressor = isal_zlib.compressobj(level, wbits=wbits, zdict=dictionary) + packed = b'' + if first_operation == 'empty': + packed += compressor.compress(b'') + elif first_operation == 'sync': + packed += compressor.flush(isal_zlib.Z_SYNC_FLUSH) + elif first_operation == 'finish': + data = b'' + for pos in range(0, len(data), 17): + packed += compressor.compress(data[pos:pos+17]) + packed += compressor.flush() + if wbits > 0: + assert packed[0] >> 4 == wbits - 8 + assert packed[1] & 32 + assert int.from_bytes(packed[2:6], 'big') == zlib.adler32(dictionary) + assert int.from_bytes(packed[:2], 'big') % 31 == 0 + decompressor = zlib.decompressobj(wbits, zdict=dictionary) + assert decompressor.decompress(packed) + decompressor.flush() == data + assert decompressor.eof + assert decompressor.unused_data == b'' + + +@pytest.mark.parametrize('level', range(4)) +@pytest.mark.parametrize( + 'flush_mode', [isal_zlib.Z_SYNC_FLUSH, isal_zlib.Z_FULL_FLUSH]) +def test_dictionary_multiple_flushes(level, flush_mode): + dictionary = b'hello world' + compressor = isal_zlib.compressobj(level, zdict=dictionary) + data = dictionary * 1000 + packed = compressor.compress(data[:5500]) + compressor.flush(flush_mode) + packed += compressor.compress(data[5500:]) + compressor.flush() + for module in (zlib, isal_zlib): + decompressor = module.decompressobj(zdict=dictionary) + assert decompressor.decompress(packed) + decompressor.flush() == data + + +@pytest.mark.parametrize('dictionary', [None, b'']) +def test_no_dictionary_header(dictionary): + kwargs = {} if dictionary is None else {'zdict': dictionary} + compressor = isal_zlib.compressobj(**kwargs) + data = b'hello world' * 100 + packed = compressor.compress(data) + compressor.flush() + assert not packed[1] & 32 + assert zlib.decompress(packed) == data + + +def test_wrong_dictionary_rejected_by_zlib(): + compressor = isal_zlib.compressobj(zdict=b'hello world') + packed = compressor.compress(b'hello world' * 10) + compressor.flush() + decompressor = zlib.decompressobj(zdict=b'wrong dictionary') + with pytest.raises(zlib.error): + decompressor.decompress(packed) diff --git a/tests/test_freethreading.py b/tests/test_freethreading.py index 10d1d53b..adac1b6a 100644 --- a/tests/test_freethreading.py +++ b/tests/test_freethreading.py @@ -1,4 +1,5 @@ import concurrent.futures +import io import random import string import threading @@ -75,10 +76,9 @@ NUM_THREADS = 10 NUM_ITERATIONS = 20 NUM_JOBS = 50 # To simulate 50 jobs running in 10 threads -barrier = threading.Barrier(parties=NUM_THREADS) -def isal_compress_decompress(compress, decompress): +def isal_compress_decompress(compress, decompress, barrier): for _ in range(NUM_ITERATIONS): barrier.wait() x = compress(HAMLET_SCENE) @@ -101,10 +101,169 @@ def isal_compress_decompress(compress, decompress): ] ) def test_isal_compress_decompress_threaded(compress, decompress): + # A failed job makes fewer barrier waits than the others, so when the + # jobs run out, the other threads would wait at the barrier forever. + # The timeout makes them fail instead. + barrier = threading.Barrier(parties=NUM_THREADS, timeout=10) with concurrent.futures.ThreadPoolExecutor(NUM_THREADS) as executor: futures = [ - executor.submit(isal_compress_decompress, compress, decompress) + executor.submit(isal_compress_decompress, compress, decompress, + barrier) for _ in range(NUM_JOBS) ] for future in concurrent.futures.as_completed(futures): future.result() # To fire assertion error if there is one + + +def run_in_threads(function): + """Call function(index) in NUM_THREADS threads at the same time.""" + start = threading.Barrier(parties=NUM_THREADS) + + def run(index): + start.wait() + return function(index) + + with concurrent.futures.ThreadPoolExecutor(NUM_THREADS) as executor: + return list(executor.map(run, range(NUM_THREADS))) + + +def test_compressobj_shared_between_threads(): + compressobj = isal_zlib.compressobj() + + def compress(index): + for _ in range(NUM_ITERATIONS): + compressobj.compress(HAMLET_SCENE) + if index == 0: + compressobj.flush(isal_zlib.Z_SYNC_FLUSH) + + # The order of the calls is unknown, so the output cannot be checked. + # This test is for ThreadSanitizer. + run_in_threads(compress) + compressobj.flush() + + +def test_decompressobj_shared_between_threads(): + decompressobj = isal_zlib.decompressobj() + decompressobj.decompress(isal_zlib.compress(HAMLET_SCENE)) + assert decompressobj.eof + + def decompress(index): + # Data after the end of the stream is added to unused_data. + for _ in range(NUM_ITERATIONS): + assert decompressobj.decompress(bytes([index]) * 10) == b"" + + run_in_threads(decompress) + unused_data = decompressobj.unused_data + blocks = [unused_data[i:i + 10] for i in range(0, len(unused_data), 10)] + assert sorted(blocks) == sorted( + bytes([index]) * 10 + for index in range(NUM_THREADS) + for _ in range(NUM_ITERATIONS) + ) + + +def test_igzip_decompressor_shared_between_threads(): + data = HAMLET_SCENE * 100 + decompressor = igzip_lib.IgzipDecompressor() + first = decompressor.decompress(igzip_lib.compress(data), 100) + + def decompress(index): + size = 0 + while True: + try: + size += len(decompressor.decompress(b"", 100)) + except EOFError: + return size + + assert len(first) + sum(run_in_threads(decompress)) == len(data) + + +def test_decompressobj_shared_between_threads_detects_format(): + compressed = isal_zlib.compress(HAMLET_SCENE, wbits=31) + for _ in range(NUM_ITERATIONS): + # wbits=47 detects gzip or zlib in the first decompress() call. + decompressobj = isal_zlib.decompressobj(wbits=47) + results = run_in_threads( + lambda index: decompressobj.decompress(compressed)) + assert sorted(results) == [b""] * (NUM_THREADS - 1) + [HAMLET_SCENE] + assert decompressobj.unused_data == compressed * (NUM_THREADS - 1) + + +def test_decompressobj_attributes_while_decompressing(): + data = HAMLET_SCENE * 100 + compressed = isal_zlib.compress(data) + b"unused" + decompressobj = isal_zlib.decompressobj() + done = threading.Event() + + def decompress(): + output = [] + unconsumed = compressed + while not decompressobj.eof: + output.append(decompressobj.decompress(unconsumed, 100)) + unconsumed = decompressobj.unconsumed_tail + # Each call replaces unused_data with a longer bytes object. + for _ in range(NUM_ITERATIONS): + decompressobj.decompress(b"!") + return b"".join(output) + + def run(index): + if index == 0: + try: + return decompress() + finally: + done.set() + while not done.is_set(): + decompressobj.eof + decompressobj.unused_data + decompressobj.unconsumed_tail + + assert run_in_threads(run)[0] == data + assert decompressobj.unused_data == b"unused" + b"!" * NUM_ITERATIONS + + +def test_igzip_decompressor_attributes_while_decompressing(): + data = HAMLET_SCENE * 100 + compressed = igzip_lib.compress(data, flag=igzip_lib.COMP_GZIP) + decompressor = igzip_lib.IgzipDecompressor(flag=igzip_lib.DECOMP_GZIP) + done = threading.Event() + + def decompress(): + output = [decompressor.decompress(compressed + b"unused", 100)] + while not decompressor.eof: + output.append(decompressor.decompress(b"", 100)) + return b"".join(output) + + def run(index): + if index == 0: + try: + return decompress() + finally: + done.set() + while not done.is_set(): + decompressor.eof + decompressor.unused_data + decompressor.needs_input + decompressor.crc + + assert run_in_threads(run)[0] == data + assert decompressor.unused_data == b"unused" + assert decompressor.crc == isal_zlib.crc32(data) + + +def test_gzip_reader_seek_while_reading(): + # Each number appears once, so a chunk that mixes up parts is not in it. + data = b"".join(b"%08d" % i for i in range(128 * 1024)) + compressed = isal_zlib.compress(data, wbits=31) + reader = isal_zlib._GzipReader(io.BytesIO(compressed)) + + def run(index): + for i in range(NUM_ITERATIONS): + if index % 2: + reader.seek(len(data) // 2 if i % 2 else 0) + else: + # seek() can come between two reads, but not during one. + assert reader.read(64 * 1024) in data + + run_in_threads(run) + reader.seek(0) + assert reader.read() == data diff --git a/tests/test_gzip_compliance.py b/tests/test_gzip_compliance.py index 5d71577c..ac1ebe7c 100644 --- a/tests/test_gzip_compliance.py +++ b/tests/test_gzip_compliance.py @@ -34,6 +34,7 @@ from subprocess import PIPE, Popen from test.support import _4G, bigmemtest # type: ignore from test.support.script_helper import assert_python_failure, assert_python_ok # type: ignore # noqa: E501 +from unittest import mock from isal import igzip @@ -65,14 +66,11 @@ def seek(self, *args): class BaseTest(unittest.TestCase): - def __init__(self, methodName): + def setUp(self): + # Make a unique file name. The tests create the file themselves. fileno, self.filename = tempfile.mkstemp() os.close(fileno) - super().__init__(methodName) - - def setUp(self): - if os.path.exists(self.filename): - os.unlink(self.filename) + os.unlink(self.filename) def tearDown(self): if os.path.exists(self.filename): @@ -837,8 +835,8 @@ def test_decompress_infile_outfile(self): with igzip.open(igzipname, mode='wb') as fp: fp.write(self.data) - sys.argv = ['', '-d', igzipname] - igzip.main() + with mock.patch.object(sys, 'argv', ['', '-d', igzipname]): + igzip.main() with open(os.path.join(TEMPDIR, "testigzip"), "rb") as gunziped: self.assertEqual(gunziped.read(), self.data) diff --git a/tests/test_igzip.py b/tests/test_igzip.py index 7b32b363..6d9cbf5d 100644 --- a/tests/test_igzip.py +++ b/tests/test_igzip.py @@ -11,12 +11,14 @@ import gzip import io +import itertools import os import re -import shutil import struct +import subprocess import sys import tempfile +import textwrap import zlib from gzip import FCOMMENT, FEXTRA, FHCRC, FNAME, FTEXT # type: ignore from pathlib import Path @@ -36,11 +38,26 @@ def test_wrong_compresslevel_igzipfile(): error.match("Compression level should be between 0 and 3") -def test_repr(): - tempdir = tempfile.mkdtemp() - with igzip.IGzipFile(os.path.join(tempdir, "test.gz"), "wb") as test: +def test_repr(tmp_path): + with igzip.IGzipFile(str(tmp_path / "test.gz"), "wb") as test: assert "