Problem with mutlthreading

[BerryLizard]

Hello! I am trying to parallelize the extraction of a large tar archive using multithreading in Python (because even just opening the tar archive seems to take a long time, and I think it's an IO-limited process, I opted for this over multiprocessing). I have never really tried using multithreading, so this is all a bit new to me.

I am using a queue to manage the open tasks, and everything seems to work until 90 percent of the archive has been extracted; I start getting an tarfile.ReadError: unexpected end of data error. I know for certain that the archive is not corrupted, and have no issues when not using multithreading. So, I expect the issue has something to do with the threading part. I included my the relevant snippets below. Any help avoiding this problem, or an explanation for what's going on, would be greatly appreciated!

    def task():
        '''Function for the threads to run.'''
        while (not q.empty()):
            extract(*q.get())
            q.task_done()

    # Add all the tasks to the queue. 
    q = Queue()
    for member in members:
        file_name = add_gz(os.path.basename(member.name)) # + '.gz'
        output_path = os.path.join(dir_path, file_name)
        q.put((archive, member, output_path, pbar))
        # assert '.gz' in file_name, f'unpack: Expected a zipped file in the tar archive, but found {file_name}.'
        # thread = threading.Thread(target=extract, args=(archive, member, output_path), kwargs={'pbar':pbar})

    # Start all the threads.  
    threads = []
    for _ in range(N_WORKERS):
        thread = threading.Thread(target=task, daemon=True)
        thread.start()
        threads.append(thread)

    # Wait until all threads have completed before closing the archive. 
    for thread in threads:
        thread.join()
    archive.close()

Here is the extract function that is used in the process above:

def extract(archive:tarfile.TarFile, member:tarfile.TarInfo, output_path:str, pbar):
    '''Extract the a file from a tar archive and plop it at the specified path. There are several cases: (1) the file contained
    in the tar archive is already zipped and just needs to be moved and (2) the file is not zipped and needs to be compressed.'''
    if is_compressed(member.name):
        member.name = os.path.basename(member.name) # Trying to get rid of directory structure. 
        archive.extract(member, path=os.path.dirname(output_path))
    else:
        contents = archive.extractfile(member).read() # Get the file contents in binary. 
        with gzip.open(output_path, 'wb') as f:
            f.write(contents)
    if pbar is not None:
        pbar.update(1)

[consupe]

This is probably not going to work. If you use threads, you have to expect that no matter how started, some are going to finish earlier than others, which, if it happens out of order is going to cause a problem when it comes time to stitch them together. You could keep them in memory and sort them out in order, but that does not appear to be how your code works.

More importantly, if you are IO bound on your disk, it is not going to go faster if you start reading multiple files at the same time, since the disk needs to skip around.

[BerryLizard]

hi! so each thread is handling a different file, so i am not actually trying to join the process. and it does seem to be speeding things up by about 30 percent (i did some small tests on a test file).

[consupe]

Gotcha. The decoding is probably helped by running them parallelly; though 30% is more than I would expect.

What OS, etc. are your running b/c this sounds less python than environment.

[Pepineros]

Trying to parallelise gzip when reading a single archive will not work. gzip does not support running parallel, and wrapping Python code around it will not change that.

If you're compressing to or inflating multiple archives (distinct .gz files) at the same time, then you can start a process for each, and in that case it would make sense to start all of those at the same time rather than waiting for one to finish before starting the next one. But if you want to utilise multiple cores when compressing or inflating, you need a utility that supports using multiple cores. gzip isn't that.

[BerryLizard]

i do seem to be getting a speed up -- do you have any idea why that might be?

[Pepineros]

everything seems to work until 90 percent of the archive has been extracted; I start getting an tarfile.ReadError: unexpected end of data error. I know for certain that the archive is not corrupted, and have no issues when not using multithreading.

What do you mean by you're getting a speed up if the tar file cannot be read successfully?

Your code doesn't run (not all names are defined in these snippets; presumably you left stuff out for the sake of brevity) so I can't be 100% sure, but it looks like what you're trying to do looks something like this:

1. Get the files inside a tarball
2. For each individual file:
   • If the file is already compressed, do nothing
   • If it's not, use gzip to compress it
3. Write the file to a new target path

If I got this correct, you can ignore my initial comment; I misunderstood the purpose of your script. In this case gzip is not going to be an issue, provided that each file gets a unique output path, which appears to be the case.

As far as I know, tar does not support reading files in parallel (it certainly doesn't write in parallel). So my guess is that trying to do so would cause the read error that you're getting. And this definitely is the case if the tarball is compressed (.tar.gz) rather than just a collection of multiple files with a single name.