{"id":258,"date":"2019-02-13T10:16:46","date_gmt":"2019-02-13T09:16:46","guid":{"rendered":"https:\/\/www.kth.se\/blogs\/pdc\/?p=258"},"modified":"2019-03-28T12:32:53","modified_gmt":"2019-03-28T11:32:53","slug":"parallel-programming-in-python-multiprocessing-part-1","status":"publish","type":"post","link":"https:\/\/www.kth.se\/blogs\/pdc\/2019\/02\/parallel-programming-in-python-multiprocessing-part-1\/","title":{"rendered":"Parallel programming in Python: multiprocessing (part 1)"},"content":{"rendered":"

Parallel programming solves big numerical problems by dividing them into smaller sub-tasks, and hence reduces the overall computational time on multi-processor and\/or multi-core machines. Parallel programming is well supported in traditional programming languages like C and FORTRAN, which are suitable for “heavy-duty” computational tasks. Traditionally, Python is considered to not support parallel programming very well, partly because of the global interpreter lock (GIL)<\/a>. However, things have changed over time. Thanks to the development of a rich variety of libraries and packages, the support for parallel programming in Python is now much better.<\/p>\n

This post (and the following part) will briefly introduce the\u00a0multiprocessing<\/code>\u00a0module<\/a> in Python, which effectively side-steps the GIL by using subprocesses instead of threads. The\u00a0multiprocessing<\/code>\u00a0module provides many useful features and is very suitable for symmetric multiprocessing (SMP)<\/a> and shared memory systems. In this post we focus on the\u00a0Pool<\/code>\u00a0class<\/a>\u00a0of the multiprocessing<\/code>\u00a0module, which controls a pool of worker processes and supports both synchronous and asynchronous\u00a0parallel execution.<\/p>\n

<\/p>\n

The Pool class<\/h2>\n

Creating a Pool object<\/h3>\n

To use the multiprocessing<\/code>\u00a0module, you need to import it first.<\/p>\n

\n
\n
import<\/span> multiprocessing as<\/span> mp\r\n<\/pre>\n<\/div>\n<\/div>\n
Documentation for the module can be displayed with the help<\/code>\u00a0method.<\/p>\n
\n
\n
help<\/span>(<\/span>mp)<\/span>\r\n<\/pre>\n<\/div>\n<\/div>\n
The module can detect the number of available CPU cores via the cpu_count<\/code>\u00a0method. (Note that we use the Python3 syntax for printing the resulting number.)<\/p>\n
\n
\n
nprocs =<\/span> mp.<\/span>cpu_count(<\/span>)<\/span>\r\nprint<\/span>(<\/span>f\"Number of CPU cores: {nprocs}\"<\/span>)<\/span>\r\n<\/pre>\n<\/div>\n<\/div>\n
In practice it is desirable to have one process per CPU core, so it is a good idea to set nprocs<\/code>\u00a0to be the number of available CPU cores. A Pool<\/code>\u00a0object can be created by passing the desired number of processes to the constructor.<\/p>\n
\n
\n
pool =<\/span> mp.<\/span>Pool(<\/span>processes=<\/span>nprocs)<\/span>\r\n<\/pre>\n<\/div>\n<\/div>\n
The map method<\/h3>\n
To demonstrate the usage of the Pool<\/code>\u00a0class, let’s define a simple function that calculates the square of a number.<\/p>\n
\n
\n
def<\/span> square(<\/span>x)<\/span>:<\/span>\r\n    return<\/span> x *<\/span> x\r\n<\/pre>\n<\/div>\n<\/div>\n
Suppose we want to use the square<\/code>\u00a0method to calculate the squares of a list of integers. In serial programming we can use the following code to compute and print the result, via\u00a0list comprehension<\/a>.<\/p>\n
\n
\n
result =<\/span> [<\/span>square(<\/span>x)<\/span> for<\/span> x in<\/span> range<\/span>(<\/span>20<\/span>)<\/span>]<\/span>\r\nprint<\/span>(<\/span>result)<\/span>\r\n<\/pre>\n<\/div>\n<\/div>\n
To execute the computation in parallel, we can use the map<\/code>\u00a0method of the Pool<\/code>\u00a0class, which is similar to the built-in\u00a0map<\/code>\u00a0function<\/a> in Python.<\/p>\n
\n
\n
result =<\/span> pool.<\/span>map<\/span>(<\/span>square,<\/span> range<\/span>(<\/span>20<\/span>)<\/span>)<\/span>\r\nprint<\/span>(<\/span>result)<\/span>\r\n<\/pre>\n<\/div>\n<\/div>\n
The above parallel code will print exactly the same result as the serial code, but the computations are actually distributed and executed in parallel on the worker processes. The map<\/code>\u00a0method will guarantee that the order of the output is correct.<\/p>\n
The starmap method<\/h3>\n
You may have noticed that the map<\/code>\u00a0method is only applicable to computational routines that accept a single argument (e.g. the previously defined square<\/code>\u00a0function). For routines that accept multiple arguments, the Pool<\/code>\u00a0class also provides the starmap<\/code>\u00a0method. For example, we can define a more general routine that computes a power of arbitrary order<\/p>\n
\n
\n
def<\/span> power_n(<\/span>x,<\/span> n)<\/span>:<\/span>\r\n    return<\/span> x **<\/span> n\r\n<\/pre>\n<\/div>\n<\/div>\n
and pass this power_n<\/code>\u00a0routine and a list of input arguments\u00a0to the starmap<\/code>\u00a0method.<\/p>\n
\n
\n
result =<\/span> pool.<\/span>starmap(<\/span>power_n,<\/span> [<\/span>(<\/span>x,<\/span> 2<\/span>)<\/span> for<\/span> x in<\/span> range<\/span>(<\/span>20<\/span>)<\/span>]<\/span>)<\/span>\r\nprint<\/span>(<\/span>result)<\/span>\r\n<\/pre>\n<\/div>\n<\/div>\n
Note that both map<\/code>\u00a0and starmap<\/code>\u00a0are synchronous methods. In other words, if a worker process finishes its sub-task very early, it will wait for the other worker processes to finish. This may lead to performance degradation if the workload is not well balanced among the worker processes.<\/p>\n
The apply_async method<\/h3>\n
The Pool<\/code>\u00a0class also provides the apply_async<\/code>\u00a0method that makes asynchronous execution of the worker processes possible. Unlike the map<\/code>\u00a0method, which executes a computational routine over a list of inputs, the apply_async<\/code>\u00a0method executes the routine only once. Therefore, in the previous example, we would need to define another routine, power_n_list<\/code>, that computes the values of a list of numbers raised to a particular power.<\/p>\n
\n
\n
def<\/span> power_n_list(<\/span>x_list,<\/span> n)<\/span>:<\/span>\r\n    return<\/span> [<\/span>x **<\/span> n for<\/span> x in<\/span> x_list]<\/span>\r\n<\/pre>\n<\/div>\n<\/div>\n
To use the apply_async<\/code>\u00a0method, we also need to divide the whole input list range(20)<\/code>\u00a0into sub-lists (which are known as slices) and distribute them to the worker processes. The slices can be prepared by the following slice_data<\/code>\u00a0method.<\/p>\n
\n
\n
def<\/span> slice_data(<\/span>data,<\/span> nprocs)<\/span>:<\/span>\r\n    aver,<\/span> res =<\/span> divmod<\/span>(<\/span>len<\/span>(<\/span>data)<\/span>,<\/span> nprocs)<\/span>\r\n    nums =<\/span> [<\/span>]<\/span>\r\n    for<\/span> proc in<\/span> range<\/span>(<\/span>nprocs)<\/span>:<\/span>\r\n        if<\/span> proc <<\/span> res:<\/span>\r\n            nums.<\/span>append(<\/span>aver +<\/span> 1<\/span>)<\/span>\r\n        else<\/span>:<\/span>\r\n            nums.<\/span>append(<\/span>aver)<\/span>\r\n    count =<\/span> 0<\/span>\r\n    slices =<\/span> [<\/span>]<\/span>\r\n    for<\/span> proc in<\/span> range<\/span>(<\/span>nprocs)<\/span>:<\/span>\r\n        slices.<\/span>append(<\/span>data[<\/span>count:<\/span> count+<\/span>nums[<\/span>proc]<\/span>]<\/span>)<\/span>\r\n        count +<\/span>=<\/span> nums[<\/span>proc]<\/span>\r\n    return<\/span> slices\r\n<\/pre>\n<\/div>\n<\/div>\n
Then we can pass the power_n_list<\/code>\u00a0routine and the sliced input lists to the apply_async<\/code>\u00a0method.<\/p>\n
\n
\n
inp_lists =<\/span> slice_data(<\/span>range<\/span>(<\/span>20<\/span>)<\/span>,<\/span> nprocs)<\/span>\r\nmulti_result =<\/span> [<\/span>pool.<\/span>apply_async(<\/span>power_n_list,<\/span> (<\/span>inp,<\/span> 2<\/span>)<\/span>)<\/span> for<\/span> inp in<\/span> inp_lists]<\/span>\r\n<\/pre>\n<\/div>\n<\/div>\n
The actual result can be obtained using the get<\/code>\u00a0method and nested\u00a0list comprehension<\/a>.<\/p>\n
\n
\n
result =<\/span> [<\/span>x for<\/span> p in<\/span> multi_result for<\/span> x in<\/span> p.<\/span>get(<\/span>)<\/span>]<\/span>\r\nprint<\/span>(<\/span>result)<\/span>\r\n<\/pre>\n<\/div>\n<\/div>\n
Note that the apply_async<\/code>\u00a0method itself does not guarantee the correct order of the output. In the above example, apply_async<\/code>\u00a0was used with\u00a0list comprehension<\/a>\u00a0so that the result remained ordered (see also the examples<\/a>).<\/p>\n
Example: computing \u03c0<\/h2>\n
After that brief introduction, we can use the Pool<\/code>\u00a0class to do useful things. Here we use the calculation of \u03c0 as a simple example to demonstrate the parallelization of Python code. The formula for computing \u03c0 is given below.<\/p>\n
\"\"<\/p>\n
Serial code<\/h3>\n
With the above formula we can compute the value of\u00a0\u03c0 via numerical integration over a large number of points. For example we can choose to use 10 million points. The serial code is shown below.<\/p>\n
\n
\n
nsteps =<\/span> 10000000<\/span>\r\ndx =<\/span> 1.0<\/span> \/<\/span> nsteps\r\npi =<\/span> 0.0<\/span>\r\nfor<\/span> i in<\/span> range<\/span>(<\/span>nsteps)<\/span>:<\/span>\r\n    x =<\/span> (<\/span>i +<\/span> 0.5<\/span>)<\/span> *<\/span> dx\r\n    pi +<\/span>=<\/span> 4.0<\/span> \/<\/span> (<\/span>1.0<\/span> +<\/span> x *<\/span> x)<\/span>\r\npi *<\/span>=<\/span> dx\r\n<\/pre>\n<\/div>\n<\/div>\n
Parallel code<\/h3>\n
To parallelize the serial code for computing \u03c0, we need to divide the\u00a0for<\/span>\u00a0loop into sub-tasks and distribute them to the worker processes. In other words, we need to evenly distribute the task of evaluating the integrand at 10 million points. This can be conveniently done by providing the start, stop and step arguments to the built-in\u00a0range<\/code>\u00a0function<\/a>. The first integer in the range<\/code>\u00a0function is the start of the sequence, and should be set as the index or rank of the process. The second integer is the number of integration points, namely the end of the sequence. The third integer is the step between the adjacent elements in the sequence, and is set as the number of processes to avoid double counting. For example, the following calc_partial_pi<\/code>\u00a0function uses the range<\/code>\u00a0function for the sub-task on a worker process.<\/p>\n
\n
\n
def<\/span> calc_partial_pi(<\/span>rank,<\/span> nprocs,<\/span> nsteps,<\/span> dx)<\/span>:<\/span>\r\n    partial_pi =<\/span> 0.0<\/span>\r\n    for<\/span> i in<\/span> range<\/span>(<\/span>rank,<\/span> nsteps,<\/span> nprocs)<\/span>:<\/span>\r\n        x =<\/span> (<\/span>i +<\/span> 0.5<\/span>)<\/span> *<\/span> dx\r\n        partial_pi +<\/span>=<\/span> 4.0<\/span> \/<\/span> (<\/span>1.0<\/span> +<\/span> x *<\/span> x)<\/span>\r\n    partial_pi *<\/span>=<\/span> dx\r\n    return<\/span> partial_pi\r\n<\/pre>\n<\/div>\n<\/div>\n
With the calc_partial_pi<\/code>\u00a0function we can prepare the input arguments for the sub-tasks and compute the value of \u03c0 using the starmap<\/code>\u00a0method of the Pool<\/code>\u00a0class, as shown below.<\/p>\n
\n
\n
nprocs =<\/span> mp.<\/span>cpu_count(<\/span>)<\/span>\r\ninputs =<\/span> [<\/span>(<\/span>rank,<\/span> nprocs,<\/span> nsteps,<\/span> dx)<\/span> for<\/span> rank in<\/span> range<\/span>(<\/span>nprocs)<\/span>]<\/span>\r\n\r\npool =<\/span> mp.<\/span>Pool(<\/span>processes=<\/span>nprocs)<\/span>\r\nresult =<\/span> pool.<\/span>starmap(<\/span>calc_partial_pi,<\/span> inputs)<\/span>\r\npi =<\/span> sum<\/span>(<\/span>result)<\/span>\r\n<\/pre>\n<\/div>\n<\/div>\n
Asynchronous parallel calculation can be carried out with the apply_async<\/code>\u00a0method of the Pool<\/code>\u00a0class. We can make use of the calc_partial_pi<\/code>\u00a0function and the inputs<\/code>\u00a0list since both starmap<\/code>\u00a0and apply_async<\/code>\u00a0support multiple arguments. The difference is that starmap<\/code>\u00a0returns the results from all processes, while apply_async<\/code>\u00a0returns the result from a single process. The code using the apply_async<\/code>\u00a0method is shown below.<\/p>\n
\n
\n
multi_result =<\/span> [<\/span>pool.<\/span>apply_async(<\/span>calc_partial_pi,<\/span> inp)<\/span> for<\/span> inp in<\/span> inputs]<\/span>\r\nresult =<\/span> [<\/span>p.<\/span>get(<\/span>)<\/span> for<\/span> p in<\/span> multi_result]<\/span>\r\npi =<\/span> sum<\/span>(<\/span>result)<\/span>\r\n<\/pre>\n<\/div>\n<\/div>\n
In a previous post we have discussed the scaling of parallel programs<\/a>. We can also run a scaling test for the parallel Python code based on the starmap<\/code>\u00a0and apply_async<\/code>\u00a0methods of the Pool<\/code>\u00a0class. From the figure below we can see that the two methods provide very similar scaling for computing the value of \u03c0.<\/p>\n
\"\"<\/p>\n
Summary<\/h2>\n
We have briefly shown the basics of the map<\/code>, starmap<\/code>\u00a0and apply_async<\/code>\u00a0methods from the Pool<\/code>\u00a0class.<\/p>\n