{"id":484,"date":"2019-11-06T09:15:49","date_gmt":"2019-11-06T08:15:49","guid":{"rendered":"https:\/\/www.kth.se\/blogs\/pdc\/?p=484"},"modified":"2019-11-06T10:34:52","modified_gmt":"2019-11-06T09:34:52","slug":"parallel-programming-in-python-mpi4py-part-2","status":"publish","type":"post","link":"https:\/\/www.kth.se\/blogs\/pdc\/2019\/11\/parallel-programming-in-python-mpi4py-part-2\/","title":{"rendered":"Parallel programming in Python: mpi4py (part 2)"},"content":{"rendered":"

In part 1 of this post<\/a>, we introduced the mpi4py<\/code><\/a> module (MPI for Python) which provides an object-oriented interface for Python resembling the message passing interface (MPI)<\/a> and enables Python programs to exploit multiple processors on multiple compute nodes.<\/p>\n

The mpi4py<\/code> module provides methods for communicating various types of Python objects in different ways. In part 1 of this post<\/a> we showed you how to communicate generic Python objects between MPI processes \u2013 the methods for doing this have names that are all lowercase letters. Some of these methods were introduced in part 1 of this post<\/a>. It is also possible to directly send buffer-like objects, where the data is exposed in a raw format and can be\u00a0accessed without copying, between MPI processes. The methods for doing this start with an uppercase letter.<\/p>\n

In this post we continue introducing the mpi4py<\/code><\/a> module, with a focus on the direct communication of buffer-like objects using the latter type of methods (that is, those starting with a capital letter), including Send<\/code>, Recv<\/code>, Isend<\/code>, Irecv<\/code>, Bcast<\/code>, and Reduce<\/code>, as well as Scatterv<\/code>\u00a0and Gatherv<\/code>, which are vector variants of Scatter<\/code>\u00a0and Gather<\/code>, respectively.<\/p>\n

<\/p>\n

Buffer-like objects<\/h2>\n

The mpi4py<\/code> module provides methods for directly sending and receiving buffer-like objects. The advantage of working with buffer-like objects is that the communication is fast (close to the speed of MPI communication in C). However, there is a disadvantage in that the user needs to be more explicit when it comes to handling the allocation of memory space. For example,<\/span> the memory of the receiving buffer needs to be allocated prior to the communication, and the size of the sending buffer should not exceed that of the receiving buffer. One should also be aware that mpi4py<\/code> <\/span>expects the buffer-like objects to have contiguous memory. Fortunately, this is usually the case with<\/span>\u00a0Numpy<\/a> arrays, which are probably the most commonly used buffer-like objects in scientific computing in Python.<\/p>\n

In mpi4py<\/code>, a buffer-like object can be specified using a list or tuple with 2 or 3 elements (or 4 elements for the vector variants, which will be elaborated in later sections). For example, for a Numpy array that is named \u201cdata<\/code>\u201d and that consists of double-precision numbers, we can use [data, n, MPI.DOUBLE]<\/code>, where n<\/code> is a positive integer, to refer to the buffer of the first n<\/code> elements. It is also possible to use [data, MPI.DOUBLE]<\/code>, or simply data<\/code>, to refer to the buffer of the whole array. In the following sections, we’ll demonstrate the communication of Numpy<\/a> arrays using mpi4py<\/code>.<\/p>\n

<\/p>\n

Point-to-point communication<\/h2>\n

In the previous post<\/a> we have introduced point-to-point communication using all-lowercase methods (send<\/code> and recv<\/code>) in mpi4py<\/code>. The use of methods with a leading uppercase letter (Send<\/code> and Recv<\/code>) is quite similar, except that the receiving buffer needs to be initialized before the Recv<\/code> method is called. An example code for passing a Numpy array from the master node (which has rank = 0) to the worker processes (that all have rank > 0) is shown below.<\/p>\n

\n
\n
from<\/span> mpi4py import<\/span> MPI\r\nimport<\/span> numpy as<\/span> np\r\n\r\ncomm =<\/span> MPI.<\/span>COMM_WORLD\r\nrank =<\/span> comm.<\/span>Get_rank(<\/span>)<\/span>\r\nsize =<\/span> comm.<\/span>Get_size(<\/span>)<\/span>\r\n\r\n# master process<\/span>\r\nif<\/span> rank ==<\/span> 0<\/span>:<\/span>\r\n    data =<\/span> np.<\/span>arange(<\/span>4.<\/span>)<\/span>\r\n    # master process sends data to worker processes by<\/span>\r\n    # going through the ranks of all worker processes<\/span>\r\n    for<\/span> i in<\/span> range<\/span>(<\/span>1<\/span>,<\/span> size)<\/span>:<\/span>\r\n        comm.<\/span>Send(<\/span>data,<\/span> dest=<\/span>i,<\/span> tag=<\/span>i)<\/span>\r\n        print<\/span>(<\/span>'Process {} sent data:'<\/span>.<\/span>format(<\/span>rank)<\/span>,<\/span> data)<\/span>\r\n\r\n# worker processes<\/span>\r\nelse<\/span>:<\/span>\r\n    # initialize the receiving buffer<\/span>\r\n    data =<\/span> np.<\/span>zeros(<\/span>4<\/span>)<\/span>\r\n    # receive data from master process<\/span>\r\n    comm.<\/span>Recv(<\/span>data,<\/span> source=<\/span>0<\/span>,<\/span> tag=<\/span>rank)<\/span>\r\n    print<\/span>(<\/span>'Process {} received data:'<\/span>.<\/span>format(<\/span>rank)<\/span>,<\/span> data)<\/span>\r\n<\/pre>\n<\/div>\n<\/div>\n
Note that the data<\/code> array was initialized on the worker processes before the Recv<\/code> method was called, and that the Recv<\/code> method takes data<\/code> as the first argument (in contrast to the recv<\/code> method which returns the data<\/code> object). The output from running the above example (send-recv.py) on 4 processes is as follows:<\/p>\n
\n
\n
$ module load mpi4py\/3.0.2\/py37\r\n\r\n$ salloc --nodes 1 -t 00:10:00 -A <your-project-account>\r\nsalloc: Granted job allocation <your-job-id>\r\n\r\n$ srun -n 4 python3 send-recv.py\r\nProcess 0 sent data: [0. 1. 2. 3.]\r\nProcess 0 sent data: [0. 1. 2. 3.]\r\nProcess 0 sent data: [0. 1. 2. 3.]\r\nProcess 1 received data: [0. 1. 2. 3.]\r\nProcess 2 received data: [0. 1. 2. 3.]\r\nProcess 3 received data: [0. 1. 2. 3.]<\/pre>\n<\/div>\n<\/div>\n
When using the Send<\/code> and Recv<\/code> methods, one thing to keep in mind is that the sending buffer and the receiving buffer should match in size. An error will occur if the size of the sending buffer is greater than that of the receiving buffer. It is, however, possible to have a receiving buffer that is larger than the sending buffer. This is illustrated in the following example, where the sending buffer has 4 elements and the size of the receiving buffer is 6. The communication will complete without error, but only the first 4 elements of the receiving buffer will be overwritten (the last 2 elements will remain zero).<\/p>\n
\n
\n
if<\/span> rank ==<\/span> 0<\/span>:<\/span>\r\n    data =<\/span> np.<\/span>arange(<\/span>4.<\/span>)<\/span>\r\n    for<\/span> i in<\/span> range<\/span>(<\/span>1<\/span>,<\/span> size)<\/span>:<\/span>\r\n        comm.<\/span>Send(<\/span>data,<\/span> dest=<\/span>i,<\/span> tag=<\/span>i)<\/span>\r\n        print<\/span>(<\/span>'Process {} sent data:'<\/span>.<\/span>format(<\/span>rank)<\/span>,<\/span> data)<\/span>\r\n\r\nelse<\/span>:<\/span>\r\n    # note: the size of the receiving buffer is larger than<\/span>\r\n    # that of the sending buffer<\/span>\r\n    data =<\/span> np.<\/span>zeros(<\/span>6<\/span>)<\/span>\r\n    comm.<\/span>Recv(<\/span>data,<\/span> source=<\/span>0<\/span>,<\/span> tag=<\/span>rank)<\/span>\r\n    print<\/span>(<\/span>'Process {} has data:'<\/span>.<\/span>format(<\/span>rank)<\/span>,<\/span> data)<\/span>\r\n<\/pre>\n<\/div>\n<\/div>\n
The output is as follows. Note that the last two elements of the receiving buffers are zero.<\/p>\n
\n
\n
$ srun -n 4 python3 send-recv.py\r\nProcess 0 sent data: [0. 1. 2. 3.]\r\nProcess 0 sent data: [0. 1. 2. 3.]\r\nProcess 0 sent data: [0. 1. 2. 3.]\r\nProcess 1 has data: [0. 1. 2. 3. 0. 0.]\r\nProcess 2 has data: [0. 1. 2. 3. 0. 0.]\r\nProcess 3 has data: [0. 1. 2. 3. 0. 0.]<\/pre>\n<\/div>\n<\/div>\n
In part 1 of this post<\/a> we also discussed blocking and non-blocking methods for point-to-point communication. In mpi4py<\/code>, the non-blocking communication methods for buffer-like objects are Isend<\/code> and Irecv<\/code>. The use of non-blocking methods are shown in the example below.<\/p>\n
\n
\n
if<\/span> rank <\/span>==<\/span> 0<\/span>:<\/span>\r\n    data <\/span>=<\/span> np<\/span>.<\/span>arange<\/span>(<\/span>4.<\/span>)<\/span>\r\n    for<\/span> i <\/span>in<\/span> range<\/span>(<\/span>1<\/span>,<\/span> size<\/span>)<\/span>:<\/span>\r\n        req <\/span>=<\/span> comm<\/span>.<\/span>Isend<\/span>(<\/span>data<\/span>,<\/span> dest<\/span>=<\/span>i<\/span>,<\/span> tag<\/span>=<\/span>i<\/span>)<\/span>\r\n        req<\/span>.<\/span>Wait<\/span>(<\/span>)<\/span>\r\n        print<\/span>(<\/span>'Process {} sent data:'<\/span>.<\/span>format<\/span>(<\/span>rank<\/span>)<\/span>,<\/span> data<\/span>)<\/span>\r\n\r\nelse<\/span>:<\/span>\r\n    data <\/span>=<\/span> np<\/span>.<\/span>zeros<\/span>(<\/span>4<\/span>)\r\n<\/span>    req <\/span>=<\/span> comm<\/span>.<\/span>Irecv<\/span>(<\/span>data<\/span>,<\/span> source<\/span>=<\/span>0<\/span>,<\/span> tag<\/span>=<\/span>rank<\/span>)<\/span>\r\n    req<\/span>.<\/span>wait<\/span>(<\/span>)<\/span>\r\n    print<\/span>(<\/span>'Process {} received data:'<\/span>.<\/span>format<\/span>(<\/span>rank<\/span>)<\/span>,<\/span> data<\/span>)<\/span>\r\n<\/pre>\n<\/div>\n
Collective communication<\/h2>\n
As we mentioned in part 1 of this post<\/a>, collective communication methods are very useful in parallel programming. In the example below we use the Bcast<\/code> method to broadcast a buffer-like object data<\/code> from the master process to all the worker processes. Note that data<\/code> needs to be initialized on the worker processes before Bcast<\/code>\u00a0is called.<\/p>\n
\n
\n
from<\/span> mpi4py import<\/span> MPI\r\nimport<\/span> numpy as<\/span> np\r\n\r\ncomm =<\/span> MPI.<\/span>COMM_WORLD\r\nrank =<\/span> comm.<\/span>Get_rank(<\/span>)<\/span>\r\n\r\nif<\/span> rank ==<\/span> 0<\/span>:<\/span>\r\n    data =<\/span> np.<\/span>arange(<\/span>4.0<\/span>)<\/span>\r\nelse<\/span>:<\/span>\r\n    data =<\/span> np.<\/span>zeros(<\/span>4<\/span>)<\/span>\r\n\r\ncomm.<\/span>Bcast(<\/span>data,<\/span> root=<\/span>0<\/span>)<\/span>\r\n\r\nprint<\/span>(<\/span>'Process {} has data:'<\/span>.<\/span>format(<\/span>rank)<\/span>,<\/span> data)<\/span>\r\n<\/pre>\n<\/div>\n<\/div>\n
The output from running the above code (broadcast.py) on 4 processes is as follows:<\/p>\n
\n
\n
$ srun -n 4 python3 broadcast.py\r\nProcess 0 has data: [0. 1. 2. 3.]\r\nProcess 1 has data: [0. 1. 2. 3.]\r\nProcess 2 has data: [0. 1. 2. 3.]\r\nProcess 3 has data: [0. 1. 2. 3.]<\/pre>\n<\/div>\n<\/div>\n
Another collective communication method is Scatter<\/code>, which sends slices of a large array to the worker processes. In part 1 of this post<\/a>, we showed that the all-lowercase scatter<\/code> method is convenient when sending slices of Python objects. However,\u00a0 Scatter<\/code> is not as convenient in practice since it requires the size of the large array to be divisible by the number of processes. For instance, if we know beforehand that the array we are going to distribute has 16 elements, it will be straightforward to use Scatter<\/code> to distribute the array on 4 processes in such a way that each process gets 4 elements. The problem is that, in practice, the size of the array is not known beforehand and is therefore not guaranteed to be divisible by the number of available processes. It is more practical to use Scatterv<\/code>, the vector version of Scatter<\/code>, which offers a much more flexible way to distribute the array. The code below distributes 15 numbers over 4 processes. Note that we use \u201c[sendbuf, count, displ, MPI.DOUBLE]<\/code>\u201d to specify the buffer-like object, where count<\/code> contains the number of elements to be sent to each process and displ<\/code> contains the starting indices of the sub-tasks. (These indices are often known as displacements.)<\/p>\n
\n
\n
from<\/span> mpi4py import<\/span> MPI\r\nimport<\/span> numpy as<\/span> np\r\n\r\ncomm =<\/span> MPI.<\/span>COMM_WORLD\r\nrank =<\/span> comm.<\/span>Get_rank(<\/span>)<\/span>\r\nnprocs =<\/span> comm.<\/span>Get_size(<\/span>)<\/span>\r\n\r\nif<\/span> rank ==<\/span> 0<\/span>:<\/span>\r\n    sendbuf =<\/span> np.<\/span>arange(<\/span>15.0<\/span>)<\/span>\r\n\r\n    # count: the size of each sub-task<\/span>\r\n    ave,<\/span> res =<\/span> divmod<\/span>(<\/span>sendbuf.<\/span>size,<\/span> nprocs)<\/span>\r\n    count =<\/span> [<\/span>ave +<\/span> 1<\/span> if<\/span> p <<\/span> res else<\/span> ave for<\/span> p in<\/span> range<\/span>(<\/span>nprocs)<\/span>]<\/span>\r\n    count =<\/span> np.<\/span>array(<\/span>count)<\/span>\r\n\r\n    # displacement: the starting index of each sub-task<\/span>\r\n    displ =<\/span> [<\/span>sum<\/span>(<\/span>count[<\/span>:<\/span>p]<\/span>)<\/span> for<\/span> p in<\/span> range<\/span>(<\/span>nprocs)<\/span>]<\/span>\r\n    displ =<\/span> np.<\/span>array(<\/span>displ)<\/span>\r\nelse<\/span>:<\/span>\r\n    sendbuf =<\/span> None<\/span>\r\n    # initialize count on worker processes<\/span>\r\n    count =<\/span> np.<\/span>zeros(<\/span>nprocs,<\/span> dtype=<\/span>np.<\/span>int<\/span>)<\/span>\r\n    displ =<\/span> None<\/span>\r\n\r\n# broadcast count<\/span>\r\ncomm.<\/span>Bcast(<\/span>count,<\/span> root=<\/span>0<\/span>)<\/span>\r\n\r\n# initialize recvbuf on all processes<\/span>\r\nrecvbuf =<\/span> np.<\/span>zeros(<\/span>count[<\/span>rank]<\/span>)<\/span>\r\n\r\ncomm.<\/span>Scatterv(<\/span>[<\/span>sendbuf,<\/span> count,<\/span> displ,<\/span> MPI.<\/span>DOUBLE]<\/span>,<\/span> recvbuf,<\/span> root=<\/span>0<\/span>)<\/span>\r\n\r\nprint<\/span>(<\/span>'After Scatterv, process {} has data:'<\/span>.<\/span>format(<\/span>rank)<\/span>,<\/span> recvbuf)<\/span>\r\n<\/pre>\n<\/div>\n<\/div>\n
The output from running the above code on 4 processes is as follows. Note that, as there are 15 elements in total, each of the first three processes received 4 elements, and the last process received 3 elements.<\/p>\n
\n
\n
$ srun -n 4 python3 scatter-array.py\r\nAfter Scatterv, process 0 has data: [0. 1. 2. 3.]\r\nAfter Scatterv, process 1 has data: [4. 5. 6. 7.]\r\nAfter Scatterv, process 2 has data: [8. 9. 10. 11.]\r\nAfter Scatterv, process 3 has data: [12. 13. 14.]<\/pre>\n<\/div>\n<\/div>\n
Gatherv<\/code> is the reverse operation of Scatterv<\/code>. When using Gatherv<\/code>, one needs to specify the receiving buffer as \u201c[recvbuf2, count, displ, MPI.DOUBLE]<\/code>\u201d, as shown in the following code. The sendbuf2<\/code> arrays will be gathered into a large array recvbuf2<\/code> on the master process.<\/p>\n
\n
\n
sendbuf2 =<\/span> recvbuf\r\nrecvbuf2 =<\/span> np.<\/span>zeros(<\/span>sum<\/span>(<\/span>count)<\/span>)<\/span>\r\ncomm.<\/span>Gatherv(<\/span>sendbuf2,<\/span> [<\/span>recvbuf2,<\/span> count,<\/span> displ,<\/span> MPI.<\/span>DOUBLE]<\/span>,<\/span> root=<\/span>0<\/span>)<\/span>\r\n\r\nif<\/span> comm.<\/span>Get_rank(<\/span>)<\/span> ==<\/span> 0<\/span>:<\/span>\r\n    print<\/span>(<\/span>'After Gatherv, process 0 has data:'<\/span>,<\/span> recvbuf2)<\/span>\r\n<\/pre>\n<\/div>\n<\/div>\n
The output from running the Gatherv<\/code> code on 4 processes is as follows.<\/p>\n
\n
\n
$ srun -n 4 python3 scatter-gather.py\r\nAfter Gatherv, process 0 has data: [ 0.  1.  2.  3.  4.  5.  6.  7.  8.  9. 10. 11. 12. 13. 14.]<\/pre>\n<\/div>\n<\/div>\n
Reduce<\/code>, on the other hand, can be used to compute the sum of (or to perform another operation on) the data collected from all processes. For example, after calling Scatterv<\/code>, we can compute the sum of the numbers in recvbuf<\/code> on each process, and then call Reduce<\/code> to add all of those partial contributions and store the result on the master process.<\/p>\n
\n
\n
partial_sum =<\/span> np.<\/span>zeros(<\/span>1<\/span>)<\/span>\r\npartial_sum[<\/span>0<\/span>]<\/span> =<\/span> sum<\/span>(<\/span>recvbuf)<\/span>\r\nprint<\/span>(<\/span>'Partial sum on process {} is:'<\/span>.<\/span>format(<\/span>rank)<\/span>,<\/span> partial_sum[<\/span>0<\/span>]<\/span>)<\/span>\r\n\r\ntotal_sum =<\/span> np.<\/span>zeros(<\/span>1<\/span>)<\/span>\r\ncomm.<\/span>Reduce<\/span>(<\/span>partial_sum,<\/span> total_sum,<\/span> op=<\/span>MPI.<\/span>SUM<\/span>,<\/span> root=<\/span>0<\/span>)<\/span>\r\nif<\/span> comm.<\/span>Get_rank(<\/span>)<\/span> ==<\/span> 0<\/span>:<\/span>\r\n    print<\/span>(<\/span>'After Reduce, total sum on process 0 is:'<\/span>,<\/span> total_sum[<\/span>0<\/span>]<\/span>)<\/span>\r\n<\/pre>\n<\/div>\n<\/div>\n
The output from running the Reduce<\/code> code on 4 processes is as follows.<\/p>\n
\n
\n
$ srun -n 4 python3 scatter-reduce.py\r\nPartial sum on process 0 is: 6.0\r\nPartial sum on process 1 is: 22.0\r\nPartial sum on process 2 is: 38.0\r\nPartial sum on process 3 is: 39.0\r\nAfter Reduce, total sum on process 0 is: 105.0<\/pre>\n<\/div>\n<\/div>\n
Summary<\/h2>\n
We have shown how to directly communicate buffer-like objects using the mpi4py<\/code> module and its methods that start with an uppercase letter. The communication of buffer-like objects is faster, but less flexible, than the communication of Python objects<\/span>.\u00a0<\/span><\/p>\n
You may read the MPI for Python tutorial page<\/a> for more information about the mpi4py<\/code> module.<\/p>\n<\/div>\n<\/div>","protected":false},"excerpt":{"rendered":"
In part 1 of this post, we introduced the mpi4py module (MPI for Python) which provides an object-oriented interface for Python resembling the message passing interface (MPI) and enables Python programs to exploit multiple processors on multiple compute nodes. The mpi4py module provides methods for communicating various types of Python objects in different ways. In […]<\/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-484","post","type-post","status-publish","format-standard","hentry","category-performance","tag-parallelization","tag-python"],"jetpack_featured_media_url":"","jetpack_shortlink":"https:\/\/wp.me\/p9W9Im-7O","_links":{"self":[{"href":"https:\/\/www.kth.se\/blogs\/pdc\/wp-json\/wp\/v2\/posts\/484","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=484"}],"version-history":[{"count":33,"href":"https:\/\/www.kth.se\/blogs\/pdc\/wp-json\/wp\/v2\/posts\/484\/revisions"}],"predecessor-version":[{"id":543,"href":"https:\/\/www.kth.se\/blogs\/pdc\/wp-json\/wp\/v2\/posts\/484\/revisions\/543"}],"wp:attachment":[{"href":"https:\/\/www.kth.se\/blogs\/pdc\/wp-json\/wp\/v2\/media?parent=484"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/www.kth.se\/blogs\/pdc\/wp-json\/wp\/v2\/categories?post=484"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/www.kth.se\/blogs\/pdc\/wp-json\/wp\/v2\/tags?post=484"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}