Ompi: Which allreduce algorithm is used for large vectors (200MB+)?

[leonardo0lyj]

Background information

Hi, I am Leo, a great fan of the OpenMPI. Currently, I am trying to understand the details of the MPI_Allreduce in the OpenMPI (version 2.0.2). I have learned allreduce algorithms like binomial tree, recursive doubling, recursive halving and doubling, binary block, and the ring algorithm. Most of them are based on the reduce-scatter and then allgather. But I am still wondering which allreduce algorithm is used for large vectors (200MB+) in OpenMPI. Really appreciated : )

What version of Open MPI are you using? (e.g., v1.10.3, v2.1.0, git branch name and hash, etc.)

Version 2.0.2

Describe how Open MPI was installed (e.g., from a source/distribution tarball, from a git clone, from an operating system distribution package, etc.)

tarball

Please describe the system on which you are running

• Operating system/version: Ubuntu 14.04
• Computer hardware: AMD CPU
• Network type: Ethernet

---

Details of the problem

As the background section, I am eager to know: Which allreduce algorithm is used for large vectors (200MB+) in OpenMPI? It would be great to provide details about it. Really appreciated! I have searched regarding this question and didn't find anything meaningful, so I post an issue here. Sorry for the trouble and Thanks a lot!

[bosilca]

Hi Leo, the allreduce algorithm is decided in ompi/mca/coll/tuned/coll_tuned_decision_fixed.c in the function ompi_coll_tuned_allreduce_intra_dec_fixed. For very large data (aka. > #processes * 1Mb) and commutative operations OMPI uses a segmented ring. This is not the best algorithm for such large data sets, as we have highlighted in Efficient Communications in Training Large Scale Neural Networks.

[leonardo0lyj]

Is it "ompi_coll_base_allreduce_intra_ring_segmented"?

[bosilca]

Yes, this is the backend function getting called (and according to the decision function the segment size seems to be 1Mb).

[leonardo0lyj]

Thanks a lot, Bosilca! I just tried the "--mca coll_base_verbose 1" flag and I can see the allreduce algorithm clearly. (i.e. I have 4 processes to perform the allreduce on float vectors. Each process owns its float32 vector with 648010 elementes, which is about 2.47MB. The native sum operation is used to perform the allreduce on these 4 vectors. It turns out to be the "coll:base:allreduce_intra_ring" algorithm. Great ! Just as the source code said.)

However, when I change to my "user-defined reduce operation" (which is also element-wise sum, just I coded myself instead of the native sum.), the mpi printed out two algorithms! "ring" and "recursive doubling". Is there some secrets for user-defined allreduce? I mean the allreduce might change its algorithm to handle user-defined "sum".

Thanks again,
Leo

[bosilca]

For non commutative operations (and user defined operations fall in this category), we are using a 2 steps allreduce, a reduce followed by a bcast (ompi/mca/coll/base/coll_base_allreduce.c function ompi_coll_base_allreduce_intra_nonoverlapping).

[leonardo0lyj]

Thanks a lot, Bosilca. What if I declare my user-defined operation as the "commutative"? I just tried a toy example (my google drive: https://drive.google.com/file/d/0Bxuk2nyTbdcTMjQ0SUNlck9hWWc/view ) and confirmed that my user-defined sum allreduce only prints out the "ring" algorithm, just as the native sum allreduce. So how did the allreduce deal with my user-defined commutative sum?

Really appreciated,
Leo

[bosilca]

The answer is here.

[leonardo0lyj]

Got it. Very clear now : )

May I have one more question about the "Ring Allreduce"? I read the code and found that the ring allreduce performs the ring-based reduce-scatter followed by the ring-based allgather. However, it says the allgather is a variation of ring allgather. I don't get it. Could you provide some details about this allgather? How is it different from the regular ring allgather?

Really appreciated!
Leo

[bosilca]

There are basically 2 differences. First, the sizes are different between iterations (depending if the initiator of the slice was before or after the split_rank, or 1/2 the size). Second, it is a variation because the first data each process receives is the one that was initiated from itself (because all other processes have added their contribution the data that started locally is now on the left neighbor). So, it is basically the same algorithm but adapted for the specificities of the allreduce scenario.

[leonardo0lyj]

Thanks a lot! Bosilca : )

[leonardo0lyj]

Hi Bosilca, may I have last question about the Segmented Ring Allreduce? I have been trying to understand this pipelined ring allreduce and failed to figure out how the phases are pipelined. (i.e. COMPUTATION PHASE 1 (b) seems to perform the two phases concurrently instead of "pipelinely".) Could you provide the motivation behind this Segmented Ring Allreduce and details about the pipeline?

Really appreciated,
Leo

[bosilca]

You do not pipeline the 2 stages, but you pipeline the first stage. There is no need to pipeline the second as every process send and receive data in same time.

[leonardo0lyj]

You mean the pipeline is for the "COMPUTATION PHASE 0 (a)"? But this phase seems to be just regular ring algorithm -- every rank r sends its block "a" to rank (r+1), concurrently. I still don't see the pipeline. Do I miss something? Really appreciated,

[bosilca]

The pipeline is between blocks, the operation is not done on a single piece, but instead divided in blocks to provide a global pipelining.

[leonardo0lyj]

Thanks. So you mean each step in the computation phases is not actually done on a single piece. Rather, each step is executed in block-cyclic fashion for each of the segment groups (a, b in this example).

For the code below, in this Step 0 for segment "a", the pipleline is executed as:
First on block 0: node 0's [00a] is sent to node 1 and perform the reduction.
Then on block 1: node 1's [11a] is sent to node 2 and perform the reduction.
Then on block 2: node 2's [22a] is sent to node 0 and perform the reduction.
It is called pipeline because the reduction time can be overlapped with the data tranfer time.

Am I correct? Thanks again : )

 *        COMPUTATION PHASE 0 (a)
 *         Step 0: rank r sends block ra to rank (r+1) and receives bloc (r-1)a
 *                 from rank (r-1) [with wraparound].
 *    #     0              1             2
 *        [00a]        [00a+10a]       [20a]
 *        [00b]          [10b]         [20b]
 *        [01a]          [11a]       [11a+21a]
 *        [01b]          [11b]         [21b]
 *      [22a+02a]        [12a]         [22a]
 *        [02b]          [12b]         [22b]

[bosilca]

I thought the ASCII art above the code was pretty clear. The pipeline is because we redivide each process's owned block into smaller segments in order to reduce the time where no op can be executed. Yes this also give us the opportunity to overlap communications with computations.

[leonardo0lyj]

Appreciated for your patience. So you mean: the communication and computation of segment "a" and "b" within their block are overlapped due to the pipeline of "a" and "b"?

Thanks again,

[bosilca]

Yes. The goal is to maximize the occupancy of all connections between the processes (obviously this is not topology-aware), by having each process send and receive data at each step. The smaller the steps the more overlap we get between the reception of the next segment and the op with the last one. The communication overheads (o and g on the LogP model) limit the size of the segment size.

[leonardo0lyj]

Thanks Bosilca : ) It seems clear now. I feel like the segmented ring is still the regular ring in high level, because they share the same pattern -- every block right-shift by one rank. However, in the segmented ring, we feel the block is too large so we segment it into multiple segments (a,b,c...). Accordingly, instead of sending the entire block and perform the reduction, each segment is sent and reduced. Here the pipeline method is used such that neighbor segments (ab, bc, cd ...) are overlapped in terms of their communication and computation. In the end, all segments within their block are received and reduced but with shorter time, because the pipeline increases the occupancy of both the communication and computation resources.

Am I correct? Thanks again.

[bosilca]

@leonardo0lyj there are models that show the drawbacks and benefits of the different algorithms, including the pipelining ring. As you pointed out the pipeline increase the resource occupancy, leading to a sharp increase in performance, especially for very large data.

[leonardo0lyj]

Really appreciated!

[Young768]

@bosilca Hi Bosilca. I am new to MPI. I'm using a library Horovod, which applies MPI to synchronize results among processes. They said that they delegate to MPI implementation to do allreduce. Here is the related code. They just call MPI_Allreduce API and say that MPI has its own ways to determine the best order. But there are several algorithms like recursive doubling, recursive halving and doubling, the ring algorithm, etc. Which algorithm is used when we call that API. Can we configure by ourself?

Anyone could help me?

[bosilca]

@Young768 you can configure the underlying collective decision function using coll_tuned_use_dynamic_rules=1 and then provide your own decision via coll_tuned_dynamic_rules_filename (both these can be set in the MCA param file). Here is an example on redefining the all reduce algorithm for very large message sizes

1 # num of collectives
3 # ID = 3 AllReduce collective (ID in ompi/mca/coll/base/coll_base_functions.h)
   # 0:"ignore", 1:"basic_linear", 2:"nonoverlapping", 3:"recursive_doubling",
   # 4:"ring", 5:"segmented_ring", 6:"rabenseifner"
1 # number of com sizes
4 # every comm over size 4
1 # number of msg sizes
0 5 0 131072 # for message size 0, pipelined segmented ring with fragsize 128k)
# end of first collective

[Young768]

@bosilca Thanks for your help. I did see these related tuning codes. It helps me a lot. Should I configure the collective decision when I submit the MPI_run? Or do you have a detailed manual about how to configure this parameter? I want to have a look at what there numbers represent. Many thanks

[Young768]

Hi @bosilca I just found your paper and know how it works. Basically, MPI have two options to configure which collective algorithm to be used. One is forcing choices, another is selective file driven decision functions? Your example is using selective file driven decision functions. Right?

Could you please tell me where are these selective/configurable files? I want to know detail about these passed parameters. Thanks.

[bosilca]

Yes, my example was using a configuration file, provided as an MCA parameter. I might not have been clear enough, you need to write such a file for your specific usage. You might want to start with the example I provided, and work it up from there.

[Young768]

Yep I found there are three ways to set these parameters. I was confused by your example above. If you don't pass the parameter name, like mpi_show_handle_leaks = 1, how can MPI know what these numbers represent?

[bosilca]

The example I provided above should go in a file, let’s say .opemmpi/collective.rules, and then you add the following 2 lines in your .openmpi/mca-params.conf:

coll_tuned_use_dynamic_rules=1
coll_tuned_dynamic_rules_filename=.openmpi/collective.rules