Allow sharing a loaded index across processes to avoid per-worker memory duplication

[leanhdung1994]

I have a Python file:

from concurrent.futures import ProcessPoolExecutor
import tarfile, rapidgzip

def processNdjson(ndjsonName):
    with rapidgzip.open(inTarDir) as myZip:
        myZip.import_index(rapidgzipDir)
        with tarfile.open(fileobj=myZip, mode="r:*") as f:
            member = f.getmember(ndjsonName)
            dataFile = f.extractfile(member)
            for oneLine in dataFile:
                # process oneLine here

if __name__ == "__main__":
    rapidgzipDir = ...
    inTarDir = ...
    nCore = 5
    ndjsonNames = ["name1.ndjson", "name2.ndjson"]

    with ProcessPoolExecutor(nCore) as pool:
        results = pool.map(worker, ndjsonNames)

Above,

• inTarDir is the directory to a .tar.gz file that contains multiple .ndjson files.
• rapidgzipDir is the pre-index file to be used by rapidgzip. This allows fast random access and is a drop-in replacement for the built-in Python gzip.GzipFile.
• Each process will

with rapidgzip.open(inTarDir) as myZip:
    myZip.import_index(rapidgzipDir)
    with tarfile.open(fileobj=myZip, mode="r:*") as f:

My concern: each command myZip.import_index(rapidgzipDir) will take up a certain amount of RAM (for example, 500MB for a 20GB .tar.gz file). This will grow linearly with nCore.

It would be great of we can load rapidgzipDir once into some index_map object in the memory. Then every with rapidgzip.open(inTarDir) as myZip in each worker can use this index_map object.

Thank you for your consideration.

[mxmlnkn]

That could be a concern and could be done, maybe by using mmap if it is a real file, but I really hope that this is not the main memory problem.

• If the index is really 500 MB for a 20 GB file, then please,
  • either convert the index to the compressed index format. This can be done really fast with: rapidgzip --import-index old.index --export-index new.index --index-format=gztool archive.gz. This should already reduce the size by 10x or so.
    • Unfortunately, this does not seem to work without the original archive, even though it probably could. I might have to check why. But, it still should be faster than full index recreation.
  • or better recreate the index with sparsity (--sparse-windows is on by default, but not used by the default index format :( ) and compression with: rapidgzip --export-index new.index --index-format=gztool archive.gz.
    • I really should change the default index export format to the compressed and sparse gztool format! Ratarmount is already using that as the default. That gave me the wrong impression that it also were the default in rapidgzip.
• Did you try with only the import_index call and without any read calls to check the memory usage for only that?
  • I would guess that the chunk buffer, required for parallel decompression, would take up more space. But it depends. It will definitely not grow with the archive size, so for sufficiently large archives, the index really should be the bigger problem.

[leanhdung1994]

@mxmlnkn

• The file enwiktionary_namespace_0.tar is 20GB.
• Its index file enwiktionary_namespace_0.gzindex is 1.3GB.
• When loaded into memory by myZip.import_index(rapidgzipDir), it takes approximately 500MB of RAM.

The size of the index file is usually not a problem. My concern is the RAM that it takes when we run myZip.import_index(rapidgzipDir).

I have

import rapidgzip, time

inTarDir = r"E:\Personal Projects\tmp\tarFiles\enwiktionary_namespace_0.tar.gz"
inRapidgzipDir = r'E:\Personal Projects\tmp\tarFiles\enwiktionary_namespace_0.gzindex'

myfile = rapidgzip.open(inTarDir)
myfile.import_index(inRapidgzipDir)

Before I run above script,

After I run it:

Hence you are probably write that "the chunk buffer, required for parallel decompression, would take up more space". In my case, I process a ndjson inside each process line-by-line, so I guess parallel decompression may not be as much beneficial (please correct me if I'm wrong on this).

• Can we configure this chunk buffer?
• I wonder if setting parallelization=1 helps:

with rapidgzip.open(inTarDir, parallelization=1) as myZip:
    myZip.import_index(rapidgzipDir)
    with tarfile.open(fileobj=myZip, mode="r:*") as f:

Also, does the parallelization by rapidgzip.open(inTarDir, parallelization=3) (inside each process) conflict with that by pool.map(worker, ndjsonNames)?

[mxmlnkn]

The size of the index file is usually not a problem. My concern is the RAM that it takes when we run myZip.import_index(rapidgzipDir).

My thought was that a smaller index size would also lead to less memory usage because it is fully loaded into memory.
But you might be right, I think I am compressing non-compressed indexes during import to avoid that memory usage.
You might reduce the index import time when using the compressed index format, though.

my case, I process a ndjson inside each process line-by-line, so I guess parallel decompression may not be as much beneficial (please correct me if I'm wrong on this).

I can't really say. It depends on how fast your parser is. You should measure it. If you are using simdjson or something like that, it might benefit, but otherwise, yes, you are probably much better off to use parallelization=1. This is also more efficient because parallelization=2 is roughly as fast as P=1. Only from P=3 and onwards, will it actually be faster. This happens because the parallel decompression has to do some more extraneous processing to make it work.

Edit: Forgot that you are importing indexes. In that case, even parallelization=2 should be twice as fast.

* Can we configure this chunk buffer?
* I wonder if setting `parallelization=1` helps:

Unfortunately, the chunk cache entry size is automatically derived from parallelization. A lot of finetuning went into it. A wrong value would degrade performance a lot:

https://github.com/mxmlnkn/indexed_bzip2/blob/b314b9b0c59b12233787b7535b305867622de164/src/core/BlockFetcher.hpp#L177-L183

Also, does the parallelization by rapidgzip.open(inTarDir, parallelization=3) (inside each process) conflict with that by pool.map(worker, ndjsonNames)?

If both are actually used at the same time, then, yes, of course. The actual parallelization will be limited by your number of CPU cores. Anything larger than that is probably only wasting memory. You might argue, however, that only one of the parallelized rapidgzip processes is used at any time. Then it might make sense.

[leanhdung1994]

Indeed, setting parallelization=1 reduces RAM usage.

The running time is almost the same:

• Without specifying this option, it runs for 1508 seconds.
• With parallelization=1, it runs for 1560 seconds.

Thank you very much for your elaboration.

[leanhdung1994]

@mxmlnkn When I set parallelization=0, then the RAM usage is the same as when I don't specify the option ``parallelization. Could you elaborate on the behavior of parallelization=0`?

[mxmlnkn]

If 0 is given, then the parallelism will be determined automatically. (default: 0)

https://github.com/mxmlnkn/indexed_bzip2/blob/b314b9b0c59b12233787b7535b305867622de164/src/rapidgzip/ParallelGzipReader.hpp#L283C15-L283C95

https://github.com/mxmlnkn/indexed_bzip2/blob/b314b9b0c59b12233787b7535b305867622de164/src/core/AffinityHelpers.hpp#L105-L124

It will try to use the number of available (logical) cores. If the process is limited to a subset of the cores by setting the affinity, e.g., with taskset, it will also try to detect that. But this is only implemented on Linux.

[leanhdung1994]

That could be a concern and could be done, maybe by using mmap if it is a real file.

I think it is in the same spirit as mxmlnkn/ratarmount#12.

[leanhdung1994]

This may be in the same spirit as #10.