{"id":306,"date":"2019-03-08T14:45:06","date_gmt":"2019-03-08T13:45:06","guid":{"rendered":"https:\/\/www.kth.se\/blogs\/pdc\/?p=306"},"modified":"2019-03-28T10:52:40","modified_gmt":"2019-03-28T09:52:40","slug":"parallel-programming-in-python-multiprocessing-part-2","status":"publish","type":"post","link":"https:\/\/www.kth.se\/blogs\/pdc\/2019\/03\/parallel-programming-in-python-multiprocessing-part-2\/","title":{"rendered":"Parallel programming in Python: multiprocessing (part 2)"},"content":{"rendered":"

In the previous post<\/a> we introduced the Pool<\/code>\u00a0class of the multiprocessing<\/code>\u00a0module. In this post we continue on and introduce the\u00a0Process<\/code>\u00a0class<\/a>, which makes it possible to have direct control over individual processes.<\/p>\n

A process can be created by providing a target function and its input arguments to the Process<\/code>\u00a0constructor. The process can then be started with the start<\/code>\u00a0method and ended using the join<\/code>\u00a0method. Below is a very simple example that prints the square of a number.<\/p>\n

\n
\n
import<\/span> multiprocessing as<\/span> mp\r\n\r\ndef<\/span> square(<\/span>x)<\/span>:<\/span>\r\n\u00a0 \u00a0 print<\/span>(<\/span>x *<\/span> x)<\/span>\r\n\r\np =<\/span> mp.<\/span>Process(<\/span>target=<\/span>square,<\/span> args=<\/span>(<\/span>5<\/span>,<\/span>)<\/span>)<\/span>\r\np.<\/span>start(<\/span>)<\/span>\r\np.<\/span>join(<\/span>)<\/span>\r\n<\/pre>\n<\/div>\n<\/div>\n
<\/p>\n
Process and Queue<\/h2>\n
In practice, we often want the process and the function to return the computed result, rather than just printing the result as in the previous example. This can be achieved with help from the Queue<\/code> class<\/a> in the multiprocessing<\/code> module. The Queue<\/code> class includes the put<\/code>\u00a0method for depositing data and the get<\/code>\u00a0method for retrieving data. The code in the earlier example can be changed as follows so that it returns the result (rather than printing it). In this example, a Queue<\/code>\u00a0object named qout<\/code>\u00a0is used to save the result.<\/p>\n
\n
\n
import<\/span> multiprocessing as<\/span> mp\r\n\r\ndef<\/span> square(<\/span>x,<\/span> q)<\/span>:<\/span>\r\n\u00a0 \u00a0 q.<\/span>put(<\/span>x *<\/span> x)<\/span>\r\n\r\nqout =<\/span> mp.<\/span>Queue(<\/span>)<\/span>\r\np =<\/span> mp.<\/span>Process(<\/span>target=<\/span>square,<\/span> args=<\/span>(<\/span>5<\/span>,<\/span> qout)<\/span>)<\/span>\r\np.<\/span>start(<\/span>)<\/span>\r\np.<\/span>join(<\/span>)<\/span>\r\n\r\nresult =<\/span> qout.<\/span>get(<\/span>)<\/span>\r\nprint<\/span>(<\/span>result)<\/span>\r\n<\/pre>\n<\/div>\n<\/div>\n
Now we can parallelize the code by creating multiple processes and running them simultaneously.<\/p>\n
\n
\n
import<\/span> multiprocessing as<\/span> mp\r\n\r\ndef<\/span> square(<\/span>x,<\/span> q)<\/span>:<\/span>\r\n\u00a0 \u00a0 q.<\/span>put(<\/span>x *<\/span> x)<\/span>\r\n\r\nqout =<\/span> mp.<\/span>Queue(<\/span>)<\/span>\r\nprocesses =<\/span> [<\/span>mp.<\/span>Process(<\/span>target=<\/span>square,<\/span> args=<\/span>(<\/span>i,<\/span> qout)<\/span>)<\/span>\r\n\u00a0\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 for<\/span> i in<\/span> range<\/span>(<\/span>2<\/span>,<\/span> 10<\/span>)<\/span>]<\/span>\r\n\r\nfor<\/span> p in<\/span> processes:<\/span>\r\n\u00a0 \u00a0 p.<\/span>start(<\/span>)<\/span>\r\n\r\nfor<\/span> p in<\/span> processes:<\/span>\r\n\u00a0 \u00a0 p.<\/span>join(<\/span>)<\/span>\r\n\r\nresult =<\/span> [<\/span>qout.<\/span>get(<\/span>)<\/span> for<\/span> p in<\/span> processes]<\/span>\r\nprint<\/span>(<\/span>result)<\/span>\r\n<\/pre>\n<\/div>\n<\/div>\n
We would expect the output to be as follows.<\/p>\n
\n
\n
[4, 9, 16, 25, 36, 49, 64, 81]\r\n<\/pre>\n<\/div>\n<\/div>\n
However the values in the Queue<\/code>\u00a0object will be stored in the order in which the individual processes finish. This means that the order of the squares in qout<\/code>\u00a0will not necessarily be in accordance with the ascending order of the input values to be squared. To demonstrate this, we can introduce some randomness in the execution time of the square function. This mimics what happens in realistic calculations where the computational load is not perfectly balanced amongst the processes.<\/p>\n
\n
\n
from<\/span> time import<\/span> sleep\r\nfrom<\/span> random import<\/span> randint\r\n\r\ndef<\/span> square(<\/span>x,<\/span> q)<\/span>:<\/span>\r\n\u00a0 \u00a0 sleep(<\/span>0.01<\/span> *<\/span> randint(<\/span>0<\/span>,<\/span> 100<\/span>)<\/span>)<\/span>\r\n  \u00a0 q.<\/span>put(<\/span>x *<\/span> x)<\/span>\r\n<\/pre>\n<\/div>\n<\/div>\n
Then, for example, we might obtain the following results, instead of the list of square numbers in ascending order.<\/p>\n
\n
\n
[36, 9, 16, 81, 25, 64, 4, 49]\r\n<\/pre>\n<\/div>\n<\/div>\n
Bear in mind that the randomising function is only introduced in this short example to show the effect of asynchronous execution of processes. In realistic code that would perform far more calculations than in the short example, the order of the output values is almost always different from the order of the corresponding input values.<\/p>\n
If we want the results to be output in the same order as the input values (so that in the initial short example the results would be in ascending order), we can pack each input value and the resulting value for its square in a tuple, and save the tuples in the Queue<\/code>\u00a0object.<\/p>\n
\n
\n
def<\/span> square(<\/span>x,<\/span> q)<\/span>:<\/span>\r\n    sleep(<\/span>0.01<\/span> *<\/span> randint(<\/span>0<\/span>,<\/span> 100<\/span>)<\/span>)<\/span>\r\n    q.<\/span>put(<\/span>(<\/span>x,<\/span> x *<\/span> x)<\/span>)<\/span>\r\n<\/pre>\n<\/div>\n<\/div>\n
After the processes have been executed, the ascending order is recovered by sorting the results saved in qout<\/code>.<\/span><\/p>\n
\n
\n
unsorted_result =<\/span> [<\/span>qout.<\/span>get(<\/span>)<\/span> for<\/span> p in<\/span> processes]<\/span>\r\nresult =<\/span> [<\/span>t[<\/span>1<\/span>]<\/span> for<\/span> t in<\/span> sorted<\/span>(<\/span>unsorted_result)<\/span>]<\/span>\r\nprint<\/span>(<\/span>result)<\/span>\r\n<\/pre>\n<\/div>\n<\/div>\n
You may have noticed that the above example is special in that the order of the processes coincides with that of the input values. However, more generally, if the input values are not in ascending order, we need to index the output values using the “process index” or “input index”, rather than the actual input value, in order to obtain the output values in the desired order. This is illustrated in the following code.<\/p>\n
\n
\n
import<\/span> multiprocessing as<\/span> mp\r\nfrom<\/span> random import<\/span> randint\r\nfrom<\/span> time import<\/span> sleep\r\n\r\ndef<\/span> square(<\/span>i,<\/span> x,<\/span> q)<\/span>:<\/span>\r\n    sleep(<\/span>0.01<\/span> *<\/span> randint(<\/span>0<\/span>,<\/span> 100<\/span>)<\/span>)<\/span> \r\n\u00a0 \u00a0 q.<\/span>put(<\/span>(<\/span>i,<\/span> x *<\/span> x)<\/span>)<\/span>\r\n\r\ninput_values =<\/span> [<\/span>2<\/span>,<\/span> 4<\/span>,<\/span> 6<\/span>,<\/span> 8<\/span>,<\/span> 3<\/span>,<\/span> 5<\/span>,<\/span> 7<\/span>,<\/span> 9<\/span>]<\/span>\r\nqout =<\/span> mp.<\/span>Queue(<\/span>)<\/span>\r\nprocesses =<\/span> [<\/span>mp.<\/span>Process(<\/span>target=<\/span>square,<\/span> args=<\/span>(<\/span>ind,<\/span> val,<\/span> qout)<\/span>)<\/span>\r\n\u00a0\u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 for<\/span> ind,<\/span> val in<\/span> enumerate<\/span>(<\/span>input_values)<\/span>]<\/span>\r\n\r\nfor<\/span> p in<\/span> processes:<\/span>\r\n\u00a0 \u00a0 p.<\/span>start(<\/span>)<\/span>\r\n\r\nfor<\/span> p in<\/span> processes:<\/span>\r\n\u00a0 \u00a0 p.<\/span>join(<\/span>)<\/span>\r\n\r\nunsorted_result =<\/span> [<\/span>qout.<\/span>get(<\/span>)<\/span> for<\/span> p in<\/span> processes]<\/span>\r\nresult =<\/span> [<\/span>t[<\/span>1<\/span>]<\/span> for<\/span> t in<\/span> sorted<\/span>(<\/span>unsorted_result)<\/span>]<\/span> \r\nprint<\/span>(<\/span>result)<\/span>\r\n<\/pre>\n<\/div>\n<\/div>\n
The output would then be as follows, where the values are in one-to-one correspondence with the input values [2, 4, 6, 8, 3, 5, 7, 9]<\/code>.<\/p>\n
\n
\n
[4, 16, 36, 64, 9, 25, 49, 81]<\/pre>\n<\/div>\n<\/div>\n
Example: computing \u03c0<\/h2>\n
Now let’s revisit the example of computing \u03c0 in the previous post<\/a>. As mentioned before, the computed result from each process should be saved in a Queue<\/code>\u00a0object. We therefore need to modify the calc_partial_pi<\/code>\u00a0routine.<\/p>\n
\n
\n
def<\/span> calc_partial_pi(<\/span>rank,<\/span> nprocs,<\/span> nsteps,<\/span> dx,<\/span> qout)<\/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    qout.<\/span>put(<\/span>partial_pi)<\/span>\r\n<\/pre>\n<\/div>\n<\/div>\n
Then we can set the number of steps and step size for numerical integration, as well as the desired number of processes. Ideally we would like to have one process per CPU core.<\/p>\n
\n
\n
nsteps =<\/span> 10000000<\/span>\r\ndx =<\/span> 1.0<\/span> \/<\/span> nsteps\r\nnprocs =<\/span> mp.<\/span>cpu_count(<\/span>)<\/span>\r\n<\/pre>\n<\/div>\n<\/div>\n
Based on the desired number of processes, we prepare a list of input arguments and create the processes.<\/p>\n
\n
\n
qout =<\/span> mp.<\/span>Queue(<\/span>)<\/span>\r\ninputs =<\/span> [<\/span>(<\/span>rank,<\/span> nprocs,<\/span> nsteps,<\/span> dx,<\/span> qout)<\/span> for<\/span> rank in<\/span> range<\/span>(<\/span>nprocs)<\/span>]<\/span>\r\nprocesses =<\/span> [<\/span>mp.<\/span>Process(<\/span>target=<\/span>calc_partial_pi,<\/span> args=<\/span>inp)<\/span>\r\n             for<\/span> inp in<\/span> inputs]<\/span>\r\n<\/pre>\n<\/div>\n<\/div>\n
After asynchronous execution of the processes, the computed value of \u03c0 is obtained.<\/p>\n
\n
\n
for<\/span> p in<\/span> processes:<\/span>\r\n    p.<\/span>start(<\/span>)<\/span>\r\n\r\nfor<\/span> p in<\/span> processes:<\/span>\r\n    p.<\/span>join(<\/span>)<\/span>\r\n\r\nresult =<\/span> [<\/span>qout.<\/span>get(<\/span>)<\/span> for<\/span> p in<\/span> processes]<\/span>\r\npi =<\/span> sum<\/span>(<\/span>result)<\/span>\r\n<\/pre>\n<\/div>\n<\/div>\n
For a simple example like computing \u03c0, the Process<\/code>\u00a0class provides very similar scaling to the Pool<\/code>\u00a0class introduced in the previous post<\/a>.<\/p>\n
Summary<\/h2>\n
    \n
  • The Process<\/code>\u00a0class makes it possible to control the processes directly.<\/li>\n
  • The Queue<\/code>\u00a0class can be used to save results from the processes.<\/li>\n
  • The processes are executed asynchronously.<\/li>\n
  • The order of the output is not guaranteed to correspond to that of the input values.<\/li>\n<\/ul>\n<\/div>","protected":false},"excerpt":{"rendered":"
    In the previous post we introduced the Pool\u00a0class of the multiprocessing\u00a0module. In this post we continue on and introduce the\u00a0Process\u00a0class, which makes it possible to have direct control over individual processes. A process can be created by providing a target function and its input arguments to the Process\u00a0constructor. The process can then be started with […]<\/p>\n","protected":false},"author":1140,"featured_media":0,"comment_status":"closed","ping_status":"closed","sticky":false,"template":"","format":"standard","meta":{"inline_featured_image":false,"jetpack_post_was_ever_published":false,"footnotes":""},"categories":[5],"tags":[9,19],"class_list":["post-306","post","type-post","status-publish","format-standard","hentry","category-performance","tag-parallelization","tag-python"],"jetpack_featured_media_url":"","jetpack_shortlink":"https:\/\/wp.me\/p9W9Im-4W","_links":{"self":[{"href":"https:\/\/www.kth.se\/blogs\/pdc\/wp-json\/wp\/v2\/posts\/306","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/www.kth.se\/blogs\/pdc\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/www.kth.se\/blogs\/pdc\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/www.kth.se\/blogs\/pdc\/wp-json\/wp\/v2\/users\/1140"}],"replies":[{"embeddable":true,"href":"https:\/\/www.kth.se\/blogs\/pdc\/wp-json\/wp\/v2\/comments?post=306"}],"version-history":[{"count":34,"href":"https:\/\/www.kth.se\/blogs\/pdc\/wp-json\/wp\/v2\/posts\/306\/revisions"}],"predecessor-version":[{"id":382,"href":"https:\/\/www.kth.se\/blogs\/pdc\/wp-json\/wp\/v2\/posts\/306\/revisions\/382"}],"wp:attachment":[{"href":"https:\/\/www.kth.se\/blogs\/pdc\/wp-json\/wp\/v2\/media?parent=306"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/www.kth.se\/blogs\/pdc\/wp-json\/wp\/v2\/categories?post=306"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/www.kth.se\/blogs\/pdc\/wp-json\/wp\/v2\/tags?post=306"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}