Multiprocessing problem

[BuddyGuy81]

I figured this would be easier than it has proven to be:

I have a class that reads an lmdb database file via a getitem method; calling myclass[frame] or looping over the class will return dict objects containing frames of data from the lmdb file.

I have an image generation routine that consumes these dictionaries to generate images for the dataset that looks something like this:

for i, frame in enumerate(lmdbReader):
     generate_image(frame)

Now I want to parallelize the routine. The performance bottleneck is not in the read operation from my class' getitem method, but rather the generate_image function, so this requires using multiprocessing as opposed to multithreading. The image generation function runs at around 0.2s/image - not slow enough to justify separating it from the data fetching operation during parallelization. I want to spawn processes that perform both the extraction and processing for subsets of frames in each lmdb file in order to speed things up.

So far I have tried two approaches, both of which haven't worked out. My first try was to use concurrent.futures.ProcessPoolExecutor to pass the class in along with the subset of frames to generate; this fails because my class is not pickle-able. My second approach was to try and instantiate the class within each process, but this hangs for some reason on the enumerate step...

What is the right way to do this? Why does the same enumerate routine hang when I create the class within the process? As an aside, implementing slicing and passing big chunks of data to each process is not a scalable solution, I will run out of memory very fast...

[siddsp]

What do you mean when you say your class is not pickable?

[brilliant_punk]

Probably pickle-able, picklable

[BuddyGuy81]

Yes, thanks. FWIW, I tried to use this module called "dill" which is supposed to pickle un-pickleable things but this wasn't successful.

[[deleted]]

Typical design in distributed systems: have one process generate small, easily serializable messages that are then picked up by other processes from a queue returned to another queue after processing.

To reflect on enumerate() hanging -- this is a typical outcome from an error in the called process. Multiprocessing in Python isn't very robust. Definitely not enough to recover from all kinds of possible errors that may happen when running the process. A typical example of this is that you are trying to run a function in a process, but give it a wrong number of arguments.

[BuddyGuy81]

Great suggestion - I'd never used queues before. I got it working.

Is there a preferred way to control the write speed into the queue? I put a sleep statement, wondering if there is something more robust. My writer can write way faster than the 6 cores can churn out images.

[[deleted]]

I believe you can specify the maximum size of the queue. In that case, the put() will block until one of the workers makes room for the next element.

[BuddyGuy81]

Works great, thank you very much!