Illustrates (de)compression of multiple lz4framed files/streams without overhead of using the data

threaded_lz4.py

#!/usr/bin/env python3

"""Illustrates (de)compression of multiple lz4framed files/streams without overhead of using the data."""

from time import monotonic
from sys import argv
from os import cpu_count
from concurrent.futures import ThreadPoolExecutor

from lz4framed import Decompressor, Compressor, Lz4FramedNoDataError


# Maximum read size in bytes during compression
READ_SIZE = 4 * 1024 * 1024


def decompress_file(filename):
    total = 0
    with open(filename, 'rb') as raw:
        try:
            for chunk in Decompressor(raw):
                total += len(chunk)
        except Lz4FramedNoDataError:
            # Data incomplete error case. In this example we simply ignore the data
            total = 0
    # Returning filename also so produce mapping
    return (filename, total)


def compress_file(filename):
    total = 0
    with open(filename, 'rb') as raw:
        compressor = Compressor()
        try:
            while True:
                total += len(compressor.update(raw.read1(READ_SIZE)))
        except Lz4FramedNoDataError:
            # End of input file reached
            pass
    return (filename, total)


def main():
    # Try setting to 1 to see difference
    workers = cpu_count() or 1
    # workers = 1

    # Whether to test compression or decompression
    method = compress_file
    # method = decompress_file

    print('Will (de)compress with %d workers' % workers)

    pool = ThreadPoolExecutor(max_workers=workers)
    start = monotonic()

    # (De)compress all given files and produce of mapping file name to processed size in bytes
    processed = dict(pool.map(method, argv[1:]))

    print('Completed in %.2fs' % (monotonic() - start))
    for name, size in processed.items():
        print('Size of %s: %d' % (name, size))


if __name__ == '__main__':
    main()