Enabling Multi-Threading in Kanzi with the `--jobs` Option for Parallel Compression

The --jobs flag enables parallel compression and decompression in Kanzi by storing the thread count in the Context object, which CompressedOutputStream and CompressedInputStream use to spawn worker threads and distribute block processing across CPU cores.

The kanzi-cpp library provides high-performance data compression with support for parallel processing via the --jobs command-line option. Enabling multi-threading in Kanzi with the --jobs option allows users to utilize multiple CPU cores for concurrent block compression and decompression, significantly improving throughput on large files. This article examines the internal architecture that makes parallel execution possible, from argument parsing in Kanzi.cpp to thread management in the I/O stream classes.

How the --jobs Flag Is Parsed

Command-line handling in Kanzi.cpp

The entry point for the Kanzi application resides in src/app/Kanzi.cpp. When the parser encounters the -j short form or the --jobs= long form, it extracts the integer value and stores it in the global Context under the key "jobs":

// src/app/Kanzi.cpp – handling of –jobs
if ((ctx == ARG_IDX_JOBS) || (arg.compare(0, 7, "--jobs=") == 0)) {
    // …
    context.putInt("jobs", value);
}

The help text, printed from the same file at lines 63–71, describes the default behavior:

log.println("   -j, --jobs=<jobs>", true);
#ifdef CONCURRENCY_ENABLED
   int cores = min(max(int(thread::hardware_concurrency()) / 2, 1), MAX_CONCURRENCY);
   // default = half the cores, 0 ⇒ all cores (capped at 64)
#endif

If the user omits the flag, the application defaults to half the available logical cores, capped at 64. Specifying --jobs=0 instructs Kanzi to use all available cores (subject to the same cap).

Propagating Job Counts Through the Context

Context storage and retrieval

All components that require concurrency information retrieve it from the Context object using getInt. The default fallback is 1 (single-threaded) if the key is absent:

int jobs = ctx.getInt("jobs", 1);   // default = 1 if not supplied

This pattern appears throughout the codebase:

• src/transform/BWT.cpp (lines 67–79) validates the job count for the Burrows-Wheeler transform.
• src/app/BlockCompressor.cpp (line 365) uses the value to parallelize compression across multiple input files.
• src/app/BlockDecompressor.cpp (line 284) performs the analogous operation for decompression.

Global job distribution logic

Kanzi distributes work by dividing the total job count among a set of tasks, where each task processes a contiguous range of blocks. The helper function Global::computeJobsPerTask, declared in src/Global.hpp and implemented in src/Global.cpp, performs an even distribution:

// src/Global.cpp – computeJobsPerTask
void Global::computeJobsPerTask(int jobsPerTask[], int jobs, int tasks) {
    int q = (jobs <= tasks) ? 1 : jobs / tasks;
    int r = (jobs <= tasks) ? 0 : jobs - q * tasks;
    for (int i = 0; i < tasks; i++) jobsPerTask[i] = q;
    int n = 0;
    while (r != 0) {
        jobsPerTask[n]++; r--; n++;
    }
}

The algorithm guarantees that each task receives either ⌊jobs/tasks⌋ or ⌈jobs/tasks⌉ worker threads, ensuring balanced load across CPU cores.

Parallel Execution in I/O Streams

CompressedOutputStream implementation

The CompressedOutputStream class, defined in src/io/CompressedOutputStream.hpp, manages parallel compression. Its constructor accepts a jobs parameter (defaulting to 1) and stores it in the member variable _jobs. After determining the number of input blocks, the stream invokes Global::computeJobsPerTask to populate the _jobsPerTask vector:

// src/io/CompressedOutputStream.cpp – constructor fragment
if (tasks > 1) {
    int nbTasks = (_nbInputBlocks != 0) ? min(_nbInputBlocks, _jobs) : _jobs;
    Global::computeJobsPerTask(&_jobsPerTask[0], _jobs, nbTasks);
}

During the compression loop, the stream iterates over _jobs and launches a std::future for each task, using the per-task job count to determine how many blocks each thread processes.

CompressedInputStream implementation

Decompression follows an identical pattern. The CompressedInputStream constructor, located in src/io/CompressedInputStream.cpp around line 616, retrieves the job count from the context, validates it, and calls computeJobsPerTask to obtain _jobsPerTask. The processing loop then spawns up to _jobs parallel decoding tasks to reconstruct the original data.

Transform-Level Threading Constraints

BWT single-threading limitation

While most Kanzi transforms are fully thread-safe, the Burrows-Wheeler Transform (BWT) currently enforces single-threaded execution. In src/transform/BWT.cpp (lines 67–79), the constructor validates the job parameter and throws an exception if parallelism is requested:

// src/transform/BWT.cpp – validation
if (jobs != 1)
    throw invalid_argument("The number of jobs is limited to 1 in this version");

This restriction means that even if the user specifies --jobs=8, any pipeline containing BWT will process that specific transform sequentially, though other stages (such as entropy coding) still benefit from parallelism.

Practical Usage Examples

Command-line usage

The --jobs option (short form -j) accepts an integer specifying the number of worker threads:

# Use half the cores (default behavior)

kanzi -c -i bigfile.bin -o bigfile.knz

# Explicitly request 8 parallel jobs

kanzi -c -i bigfile.bin -o bigfile.knz --jobs=8

# Use all available cores (capped at 64)

kanzi -c -i bigfile.bin -o bigfile.knz --jobs=0

C++ API integration

When embedding Kanzi as a library, set the job count via the Context object before constructing the compressor:

#include "kanzi/BlockCompressor.hpp"
#include "kanzi/Context.hpp"

int main() {
    kanzi::Context ctx;
    ctx.putInt("jobs", 4);               // request 4 concurrent jobs
    ctx.putString("entropy", "ANS0");     // choose entropy codec
    ctx.putString("transform", "BWT+ZRLT"); // choose transforms

    // compress a file
    kanzi::BlockCompressor compressor("input.txt", "output.knz", ctx);
    compressor.compress();

    // decompress the result
    kanzi::BlockDecompressor decompressor("output.knz", "recovered.txt", ctx);
    decompressor.decompress();
}

Key implementation details:

• ctx.putInt("jobs", 4); propagates the desired thread count.
• BlockCompressor and BlockDecompressor forward the context to CompressedOutputStream and CompressedInputStream, which spawn the worker threads.

Inspecting job distribution

To verify how Kanzi distributes jobs across tasks, you can invoke the same helper function used internally:

#include "kanzi/Global.hpp"
#include <iostream>

int main() {
    int jobs = 7;          // user requested 7 jobs
    int tasks = 3;         // suppose we have 3 blocks to process
    int perTask[3] = {0};

    kanzi::Global::computeJobsPerTask(perTask, jobs, tasks);
    std::cout << "Jobs per task: ";
    for (int i = 0; i < tasks; ++i) std::cout << perTask[i] << ' ';
    std::cout << std::endl;
}

Executing this program outputs Jobs per task: 3 2 2, illustrating the even distribution algorithm where the first r tasks receive one extra job when jobs is not evenly divisible by tasks.

Summary

• The --jobs (or -j) flag controls parallel compression and decompression in Kanzi by specifying the number of worker threads.
• The value is parsed in src/app/Kanzi.cpp and stored in the Context object under the key "jobs".
• Global::computeJobsPerTask in src/Global.cpp evenly distributes the requested jobs across processing tasks.
• CompressedOutputStream and CompressedInputStream spawn std::future worker threads based on the computed distribution.
• The BWT transform in src/transform/BWT.cpp currently limits jobs to 1, processing that stage sequentially even when multi-threading is enabled globally.
• Default behavior uses half the available logical cores (capped at 64), while --jobs=0 utilizes all available cores.

Frequently Asked Questions

What is the default value for --jobs in Kanzi?

If you do not specify the --jobs flag, Kanzi defaults to using half of the available logical CPU cores, capped at 64 (the MAX_CONCURRENCY constant). This logic is calculated in src/app/Kanzi.cpp when printing help text and applied when the Context is initialized. You can override this by providing an explicit value or by setting --jobs=0 to utilize all available cores.

Can I use --jobs with the BWT transform?

While you can enable multi-threading globally with --jobs, the Burrows-Wheeler Transform (BWT) currently enforces single-threaded execution. In src/transform/BWT.cpp, the constructor validates the job count and throws an invalid_argument if jobs != 1. Other transforms in the pipeline will still run in parallel, but the BWT stage will process blocks sequentially.

How does Kanzi distribute jobs across CPU cores?

Kanzi uses the Global::computeJobsPerTask function defined in src/Global.cpp to partition the requested job count among available tasks. The algorithm calculates a base quotient of jobs / tasks and distributes the remainder r by assigning one extra job to the first r tasks. This ensures balanced load distribution before CompressedOutputStream or CompressedInputStream spawn the actual worker threads via std::future.

Is there a maximum limit for the --jobs value?

Yes, Kanzi caps the maximum concurrency at 64 threads, defined by the MAX_CONCURRENCY constant. Even if you specify --jobs=128 or use --jobs=0 on a machine with more than 64 logical cores, the implementation in src/app/Kanzi.cpp and the I/O stream classes will limit the actual thread count to 64 to prevent excessive context switching and resource contention.

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