Implementing Streaming Compression and Decompression with Kanzi: A Complete C++ Guide

Kanzi provides a fully-streaming API through CompressedOutputStream and CompressedInputStream classes that process data incrementally in configurable blocks, automatically distributing work across CPU cores via task-based parallelism.

The flanglet/kanzi-cpp repository delivers a high-performance compression engine designed for scenarios where data arrives incrementally or total size is unknown beforehand. Unlike batch-oriented compressors, Kanzi’s I/O layer in src/io/ enables real-time, block-by-block processing while maintaining random-access capabilities and optional integrity checking.

Core Streaming Architecture

Kanzi’s streaming implementation centers on two primary I/O classes that manage the full lifecycle of compression and decompression pipelines.

CompressedOutputStream (Compression Pipeline)

Located in src/io/CompressedOutputStream.hpp and src/io/CompressedOutputStream.cpp, this class manages block buffering, transform application, entropy encoding, and bit-stream output. Key methods include writeHeader() for emitting stream metadata, processBuffer() for accumulating input data, and submitBlock() for dispatching work units. The class uses internal _buffers to accumulate raw data until reaching the configured block size or receiving an explicit flush() call.

CompressedInputStream (Decompression Pipeline)

Defined in src/io/CompressedInputStream.hpp and src/io/CompressedInputStream.cpp, this component handles header parsing via readHeader(), on-demand block decoding through submitBlock(), and seekable random access via seek(bitPos). The stream maintains an _available byte counter to track ready-to-read data and resets the pipeline when seeking to block boundaries.

Task-Based Concurrency

When CONCURRENCY_ENABLED is defined (the default for modern compilers), Kanzi utilizes a ThreadPool from src/concurrent.hpp to parallelize block processing. The headers define template classes EncodingTask (in CompressedOutputStream.hpp) and DecodingTask (in CompressedInputStream.hpp) that encapsulate the transform-plus-entropy work for individual blocks. These tasks execute XXHash32/64 checksum computation (implemented in src/util/XXHash.hpp) alongside codec operations.

How Streaming Compression Works

The compression workflow follows a strict pipeline from context configuration to bit-stream finalization.

1. Context Configuration – Create a Context object (defined in src/Context.hpp) specifying the entropy codec (e.g., "ANS0"), transform pipeline (e.g., "BWT+LZ"), block size, and job count. This context is reusable across multiple streams.

2. Stream Construction – Instantiate CompressedOutputStream with an underlying OutputStream (file, socket, or memory buffer). The constructor immediately calls writeHeader() to emit a bit-stream header recording these parameters.

3. Block Buffering – Incoming data accumulates in internal _buffers. When the buffer reaches the configured block size or upon flush(), processBuffer() copies the data to an EncodingTask.

4. Parallel Processing – submitBlock() hands tasks to the thread pool. Each EncodingTask applies transforms via TransformFactory (from src/transform/TransformFactory.hpp), encodes entropy via EntropyEncoderFactory (from src/entropy/EntropyEncoderFactory.hpp), and writes results into a shared DefaultOutputBitStream (_obs), maintaining byte alignment on block boundaries.

5. Finalization – Calling close() flushes any pending blocks, finalizes the underlying stream, and optionally writes a footer containing the total uncompressed size.

#include <fstream>
#include "kanzi/CompressedOutputStream.hpp"
#include "kanzi/Context.hpp"

int main() {
    std::ofstream fout("data.knz", std::ios::binary);
    
    kanzi::Context ctx;
    ctx.putString("entropy", "ANS0");
    ctx.putString("transform", "BWT+LZ");
    ctx.putInt("blockSize", 4*1024*1024);  // 4 MiB blocks
    ctx.putInt("jobs", 4);                 // 4 threads

    kanzi::CompressedOutputStream cos(fout, ctx);
    
    std::string data = "Streaming data chunk...";
    cos.write(data.c_str(), data.size());
    cos.close();  // Flushes automatically
}

How Streaming Decompression Works

Decompression reverses the pipeline using lazy evaluation and supports random access at block granularity.

1. Header Parsing – The CompressedInputStream constructor calls readHeader() to populate a Context with the original compression parameters, unless operating in headerless mode.

2. On-Demand Decoding – When read() requests data, the stream checks _available bytes. If insufficient, submitBlock() creates a DecodingTask for the next compressed block.

3. Block Processing – The task reads per-block metadata (size, checksum, transform identifiers), decodes the entropy layer, applies inverse transforms via TransformFactory, and verifies checksums.

4. Data Delivery – Decoded bytes populate internal _buffers. The read() method copies from these buffers to the caller’s buffer, updating _available.

5. Random Access – The seek(bitPos) method resets the pipeline, cancels outstanding futures, and re-initializes the block queue, enabling seeks only to block boundaries for large-file archival.

#include <fstream>
#include "kanzi/CompressedInputStream.hpp"

int main() {
    std::ifstream fin("data.knz", std::ios::binary);
    kanzi::CompressedInputStream cis(fin);
    
    std::vector<char> buffer(8192);
    std::size_t total = 0;
    
    while (true) {
        std::size_t got = cis.read(buffer.data(), buffer.size());
        total += got;
        if (got < buffer.size()) break;  // EOF
    }
    // buffer[0..total) holds original data
}

C API for Cross-Language Integration

The repository provides a thin C wrapper in src/api/Compressor.cpp and src/api/Decompressor.cpp exposing the streaming model through opaque handles. This API enables integration with Python ctypes, Java JNI, and other foreign function interfaces.

• initCompressor() creates a CompressedOutputStream bound to a FILE*.
• compress() writes raw buffers incrementally, returning compressed byte counts.
• disposeCompressor() finalizes and releases resources.

The decompressor provides mirrored functions: initDecompressor(), decompress(), and disposeDecompressor().

#include "kanzi/kanzi.h"
#include <stdio.h>

int main() {
    FILE *src = fopen("bigfile.bin", "rb");
    FILE *dst = fopen("bigfile.knz", "wb");
    
    struct cData cfg = {0};
    strcpy(cfg.entropy, "ANS0");
    strcpy(cfg.transform, "BWT+LZ");
    cfg.blockSize = 4*1024*1024;
    cfg.jobs = 4;

    struct cContext *ctx = NULL;
    if (initCompressor(&cfg, dst, &ctx) != 0) return -1;

    unsigned char buf[65536];
    size_t read, written;
    
    while ((read = fread(buf, 1, sizeof(buf), src)) > 0) {
        if (compress(ctx, buf, read, &written) != 0) break;
    }
    
    disposeCompressor(&ctx, &written);
    fclose(src); 
    fclose(dst);
}

Summary

• Block-based streaming – Kanzi processes data through CompressedOutputStream and CompressedInputStream without requiring total input size upfront.
• Parallel execution – EncodingTask and DecodingTask templates distribute work across a ThreadPool when CONCURRENCY_ENABLED is active.
• Configurable pipelines – The Context class and factory methods in src/transform/TransformFactory.hpp and src/entropy/EntropyEncoderFactory.hpp allow runtime selection of transforms and entropy codecs.
• Seekable archives – Block-aligned output enables seek(bitPos) for random access decompression, suitable for video archives and large datasets.
• C interoperability – The C API in src/api/ provides opaque handles for embedding in other languages while preserving the full streaming workflow.

Frequently Asked Questions

What block size should I use for streaming compression with Kanzi?

Select a block size between 1 MiB and 8 MiB via ctx.putInt("blockSize", 4*1024*1024). Larger blocks improve compression ratio but increase latency before data is written to the output stream, while smaller blocks enable finer-grained parallelism and faster flushing.

Can I seek to arbitrary byte positions in a Kanzi-compressed stream?

No, seeking is supported only on block boundaries. The seek(bitPos) method in CompressedInputStream resets the decoding pipeline to specific block starts, making Kanzi suitable for applications requiring random access to large files but not byte-level seeking within compressed data.

How does Kanzi verify data integrity during streaming?

Each block optionally includes a checksum computed via XXHash32 or XXHash64 (implemented in src/util/XXHash.hpp). When enabled in the Context, EncodingTask computes checksums during compression and DecodingTask verifies them during decompression, ensuring block-level integrity without buffering the entire stream.

Is the Kanzi streaming API thread-safe for concurrent stream processing?

Individual CompressedOutputStream and CompressedInputStream instances are not thread-safe for concurrent calls, but the internal ThreadPool safely handles parallel block processing across multiple tasks. For concurrent compression of multiple independent streams, create separate stream instances each with their own Context configuration.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}