Parallel Processing With multiprocessing: Conclusion

Thanks Dan for exposing this functional programming approach. In this particular context I guess it is worth mentioning that if “lambda” function is used inside the “map” function then we cannot re-use the same code inside Pool.map as it will fail to unpickle. So it is always good to use regular function and pass it to the map() as we exercise this functional programming approach.

Any ideas, please, why this would happen:

>>> def transform(x):
...     result = reduce(reducer,x,collections.defaultdict(list))
...     return result
... 
>>> pool = multiprocessing.Pool()
>>> result = pool.map(transform, scientists)
Traceback (most recent call last):
  File "<input>", line 1, in <module>
    result = pool.map(transform, scientists)
  File "/Users/dv/anaconda3/lib/python3.6/multiprocessing/pool.py", line 266, in map
    return self._map_async(func, iterable, mapstar, chunksize).get()
  File "/Users/dv/anaconda3/lib/python3.6/multiprocessing/pool.py", line 644, in get
    raise self._value
AttributeError: 'str' object has no attribute 'field'

@darth88vader88 I think that your transform function would try to reduce a sequence, but since you are using a map you are passing one item (a scientist) each time.

So I guess transform(scientists) would work for you, but transform(scientists[0]) would not.

Thanks, sir. Will certainly check out your suggestion.

I have a question about parallel processing. My computer is also a dual core with multi-threading, so I get how 4 processes run in parallel. What I don’t get is when I give the Pool object a parameter process=len(scientists), it is able to go beyond for processes at a time. Is it using parallelism along with concurrency in the case of going beyond the number of physical cores?

@norcal618: When you tell the pool to spawn N worker processes with the process parameter it follows that blindly and creates that number of processes. It doesn’t check how many physical or logical cores your machine has. It just creates the number of processes you tell it to, and then the operating system decides how those are run across the available cores. Does that clear it up a bit?

Parallel processing is great but if I need to keep track of the order of “task” and need them to come back in the order of for-loop I set up. How do I accomplish this without sacrificing the performance?

In addition, you have one function and one iteration up there. How do I handle two iterables? (I know we can use partial but I can only handle on iterable and one static variable. I have two parameters to the function and each parameter is a list of variables)

@philipwong37 There are many ways to keep track of the tasks’ order. One thing that comes to mind is using a queue, but you can also take advantage of the executor class from the concurrent.futures module.

If you want to process the results as soon as possible, then use the .map() method:

from concurrent.futures import ProcessPoolExecutor
import time
import random

def download(url):
    print(f"Starting the download of {url}")
    time.sleep(random.randint(500, 3000) / 1000)
    print(f"Downloaded {url}")
    return url

files = ["file1", "file2", "file3"]

with ProcessPoolExecutor() as executor:
    for result in executor.map(download, files):
        print(f"Main process got: {result}")

You’ll still process the files in the right order on the parent process even when they’re downloaded out of order by the individual subprocesses:

$ python example.py 
Starting the download of file1
Starting the download of file2
Starting the download of file3
Downloaded file2
Downloaded file1
Main process got: file1
Main process got: file2
Downloaded file3
Main process got: file3

On the other hand, you can wait until all files are downloaded before processing them, too:

from concurrent.futures import wait
# ...

with ProcessPoolExecutor() as executor:
    futures = [executor.submit(download, file) for file in files]
    wait(futures)
    for future in futures:
        print(f"Main process got: {future.result()}")

This won’t resume the main process until all subtasks are done:

$ python example.py 
Starting the download of file1
Starting the download of file2
Starting the download of file3
Downloaded file2
Downloaded file3
Downloaded file1
Main process got: file1
Main process got: file2
Main process got: file3

To answer your second question, try a Cartesian product of your input parameters:

from itertools import product
# ...

servers = ["locahost", "remote"]
files = ["file1", "file2", "file3"]

with ProcessPoolExecutor() as executor:
    for result in executor.map(download, *zip(*product(servers, files))):
        print(f"Main process got: {result}")

Output:

Starting the download of locahost/file1
Starting the download of locahost/file2
Starting the download of locahost/file3
Starting the download of remote/file1
Starting the download of remote/file2
Starting the download of remote/file3
Main process got: locahost/file1
Main process got: locahost/file2
Main process got: locahost/file3
Main process got: remote/file1
Main process got: remote/file2
Main process got: remote/file3