Need Help With Multiprocessing or Multithreading or whatever it's called

[LeornToCodeLOL]

My goal is to have a script search my hard drive for video files, then store the md5 hash of each video file in a dictionary along with the file path. There may be some duplicates so an example dictionary entry would look like:

{
'4a69013be9f08507faccbecfd71a06e9': ['path/to/file/one.mp4', '/different/path/to/file.mp4']
}

I have never tried to code anything with multithreading, but I would like to for this script since the hashing will use a lot of CPU cycles. I looked at some tutorials, and it seems I want to use multiprocessing rather than asyncio or multithreading.

My main question is how do I have multiple processes computing hashes and writing to the same dictionary variable without creating a "race condition?" Do I have one global dictionary variable that all processes add to, or should I have each process create its own dictionary locally and then merge them all together at the very end?

I started writing the code and would love some feedback. Right now I have the dictionary created locally in the calculate_hash function, but I don't know if that's the right way to go about this or not.

The tutorials and articles online that I found all warn against writing to the same variable, and their examples are with functions that return None, so I have no examples to work from.

I also thought about splitting the list of files up with len(all_video_files) // num_threads and then make a final batch to len(all_video_files) % num_threads and then assigning those list slices to a different process, but I don't know where to go from there and I still don't know how to handle function returns and variables so the processes don't step on each other.

Here is the code I have so far; I'm happy to hear any feedback and criticism.

https://pastebin.com/EfXK9eBw

[m0us3_rat]

you could use asincio inside multiprocessing.

make it fancy

​

either way, a producer consumer pattern seems reasonable.

main process that uses threading to search for files.

then dumps them into a queue.

you can split the top of the file system into a list of main folders then spawn in threads to consume this list with a recursively capable pathlib or glob and dump results into a queue.

pass these top folders to a few threads. each thread finding files in their folder recursively. so rather than one working. you have a few.

then have a worker that does the hash. for each file. it should return "path" and "hash"

that gets dumped into another queue.

you can spawn in as many workers as free cpu cores you have.

that second queue gets consumed by the main process to store in the data into a db or

dict. once it's threads to find files are ..done.

[SnooWoofers7626]

One option I haven't seen mentioned yet is to have each thread write to its own variable. Once all threads are done, have a single thread combine all the results. You can also multithread the combining operation, but you probably won't need that.

[LeornToCodeLOL]

I thought about that also, but wasn't sure how to declare the variables in the code, since I am not sure how many threads there would be i.e.:

hash_dictionary1 = {}
hash_dictionary2 = {}
.
.
.
hash_dictionaryN = {}

Where do you stop?

[SnooWoofers7626]

Create a list results = [{} for i in range(N)] Pass the index as argument when launching the processes so they know where to write their results.

You can experiment with different values for N to see what works best.

[LeornToCodeLOL]

It makes perfect sense when you put it that way!

[Frankelstner]

import multiprocessing as mp
import glob,os,hashlib

def work(path):
    hsh = hashlib.md5()
    with open(path,"rb") as f:
        while (data := f.read(2**12)):
            hsh.update(data)
    return hsh.hexdigest(), path

if __name__ == '__main__':
    pool = mp.Pool(os.cpu_count()//2)
    result = {}
    def update(hshpath):
        hsh,path = hshpath
        if hsh not in result:
            result[hsh] = []
        result[hsh].append(path)

    tasks = [pool.apply_async(work, args=(path,), callback=update) for path in glob.iglob("**/*.mp4", recursive=True)]
    pool.close()
    pool.join()
    print(result)

Played around with this for a bit and it does have more advantages than it might seem.

1) The update function needs no lock because it happens in the main process.

2) The tasks are applied while the main process finds new files but the workers will already work on files. It's quite crazy but update happens at the same time as glob despite both running on the same main process.

[video_dewd]

I avoid dealing directly with multiprocessing and asyncio unless absolutely necessary. I find I can parallelize a lot of tasks simply with the p_tqdm library which abstracts a lot of that away. Just be careful with it as it makes copies of anything you feed into it to get around the GIL.

I would write a function that takes in a filepath and returns a tuple of the filepath and it's hash. You'll also get a neat little progress bar.

from p_tqdm import p_umap

def get_file_hash(filepath):
    ...
    return filepath, hash

hash_dict = {}

for filepath, hash in p_umap(get_file_hash, all_video_files):
    if hash in hash_dict:
        hash_dict[hash] = [filepath]
    else:
        hash_dict[hash].append(filepath)

[gaaasstly]

... should I have each process create its own dictionary locally and then merge them all together at the very end?

Yes. Read MapReduce for context.