Skip to content

Latest commit

 

History

History
276 lines (207 loc) · 11.2 KB

algorithms.md

File metadata and controls

276 lines (207 loc) · 11.2 KB
Raw

Algorithms

Index of algorithms provided by Gloo and their semantics.

Variables used:

  • P: Number of processes/machines
  • N: Number of buffers per process
  • S: Size of buffer

Terms used:

  • Communication steps: number of communication steps. Every communication step has some latency, depending on the transport. Therefore, the fewer steps an algorithm uses, the better it is suited towards higher latency transports. Lower latency transports better tolerate more communication steps.
  • Bytes on the wire: total number of bytes transmitted per participating process. The higher this number, the sooner an algorithm will be bound by the network bandwidth.

Allreduce

Compute user-specified reduction operation (for e.g. sum) of N arrays per process across P processes. This computation happens in place; all input arrays contain the resulting reduction after the algorithm completes.

There are 3 phases to each implementation of this algorithm:

  1. Local reduction of N buffers
  2. Allreduce between processes
  3. Broadcast result back to N buffers

allreduce_ring

  • Communication steps: P-1
  • Bytes on the wire: P*S

Phase 2 is implemented as follows:

  1. Transmit local result to right side neighbor
  2. Receive buffer from left side neighbor and reduce into local result
  3. Transmit incoming buffer to right side neighbor
  4. Repeat 2-3 until process has seen all data

allreduce_ring_chunked

  • Communication steps: 4*P
  • Bytes on the wire: 2*S

Phase 2 is implemented in 2 sub-phases:

  1. First, the algorithm iterates over the local reduction, transmitting chunks of the buffer and reducing at every step. The number of chunks is equal to 2*P, allowing double buffering to be used. This means there is always one chunk in flight while reduction is done on another chunk concurrently. At the end of this phase, every process P holds 1/P of the reduced result.
  2. Second, the algorithm iterates over the local reduction again, now broadcasting the local results.

With 2*P chunks and two sub-phases, we arrive at 4*P communication steps.

These sub-phases are implemented as followed (roughly):

First:

  1. Compute offset into local reduction buffer based on process rank
  2. Transmit chunk at offset to right side neighbor
  3. Receive chunk at offset-1 from left side neighbor and reduce into local result
  4. Subtract 1 from offset, wrapping when needed
  5. Repeat 2-4 until process has walked entire buffer

Second:

  1. Transmit chunk at offset+1 (containing the global reduction) to right side neighbor
  2. Receive chunk at offset from left side neighbor and copy into local result
  3. Subtract 1 from offset, wrapping when needed
  4. Repeat 1-3 until process has walked entire buffer

allreduce_halving_doubling

  • Communication steps: 2*lg(P)
  • Bytes on the wire: 2*S

Phase 2 is implemented in two sub-phases:

  1. First, a reduce-scatter is performed in lg(P) steps using a recursive vector-halving, distance-doubling approach. In the first step of this algorithm processes communicate in pairs (rank 0 with 1, 2 with 3, etc.), sending and receiving for different halves of their input buffer. For example, process 0 sends the second half of its buffer to process 1 and receives and reduces data for the first half of the buffer from process 1. A reduction over the received data is performed before proceeding to the next communication step, where the distance to the destination rank is doubled while the data sent and received is halved. After the reduce-scatter phase is finished, each process has a portion of the final reduced array.

  2. The second sub-phase of Phase 2 performs an allgather. This is again done using a recursive algorithm, retracing the communication steps from the reduce-scatter in reverse, but this time simply concatenating the received data at each step. At each process and step, the portion of the buffer that was being sent in the reduce-scatter is received in the allgather, and the portion that was being received in the reduce-scatter is now sent.

Across the steps of the reduce-scatter, data is received into different buffers and there is no potential for race conditions. However, mirrored steps of the reduce-scatter and allgather (e.g. last step of the reduce-scatter and first step of the allgather) write into the same buffers. To prevent race conditions, a notification is sent after data is processed in the reduce-scatter subphase. This notification is processed in the allgather subphase prior to performing the send. In the majority of cases these notification messages will arrive long before the step of the allgather where they are processed, so their effect on performance should be minimal.

When running on non-power-of-two number of processes, the algorithm works by breaking up execution into blocks that are powers of two and communicating interblock after the intrablock reduce-scatter. Non-power-of-two cases will have some degree of load imbalance compared to power-of-two, but cases with few large blocks (e.g. 8 + 4 or 16 + 8) should still perform relatively well.

The halving-doubling / binary-blocks algorithm is described and analyzed in (Thakur et al., Optimization of Collective Communication Operations in MPICH, IJHPCA, 2005).

allreducube_bcube

Additional variables used:

  • B: Base (maximum number of peers per step)

  • Communication steps: 2*log_B(P)

  • Bytes on the wire: 2*Sum(S/B^s) {s: 0 to log_B(P) - 1}

This is another allreduce implementation. Bcube is a scheme where nodes are divided in groups. In reduce-scatter stage, in each group, a node peers with base - 1 other nodes. In the first step data is reduced between nodes within the group. In the next step each node of a group peers with base - 1 nodes from other exclusively different groups. Since each node would start with reduced data communicating with it would be like communicating with base number of nodes/groups from the previous step. This process continues until all the groups are covered and to be able to do that the algorithm would have log_base(n) number of steps. Each step the node reduces totalNumElems / (base^step) amount of elements. At the end of reduce-scatter stage each node would have reduced a chunk of elements. Now, in all-gather we follow a reverse process of reduce-scatter to communicate the reduced data with other nodes.

cuda_allreduce_ring

CUDA-aware implementation of allreduce_ring. GPU side buffers are copied to system memory in parallel, prior to running local reduction on CPU. After phase 2 completes, CPU side result is copied back to GPU side buffers in parallel.

cuda_allreduce_ring_chunked

CUDA-aware implementation of allreduce_ring_chunked. GPU side buffers are reduced into GPU buffer 0 (using NCCL). The result is copied to system memory asynchronously. After phase 2 completes, the CPU side result is copied back to GPU buffer 0, and then broadcast to other GPU buffers in parallel (using NCCL).

Both local reduction in phase 1 and broadcast in phase 3 is pipelined with the communication steps where this data is needed or becomes available.

cuda_allreduce_halving_doubling

CUDA-aware implementation of allreduce_halving_doubling with no pipelining between reduction/broadcast steps and the communication.

cuda_allreduce_halving_doubling_pipelined

CUDA-aware implementation of allreduce_halving_doubling with pipelining between local reduction/broadcast steps and communication. Local reduction step is split into two steps (since the first communication step sends half the buffer size). Final broadcast is pipelined across lgP steps, with each step corresponding to a receive during the allgather phase.

Reduce-Scatter

Compute user-specified reduction operation (for e.g. sum) of N arrays per process across P processes. This computation happens in place. The result is scattered to all processes as specified by the user; all input arrays contain the scattered result after the algorithm completes.

There are 3 phases to each implementation of this algorithm:

  1. Local reduction of N buffers

  2. Reduce-Scatter between processes

  3. Broadcast result back to N buffers

reduce_scatter_halving_doubling

  • Communication steps: lg(P)
  • Bytes on the wire: S (for scattering result evenly among P processes)

Phase 2 is implemented in two sub-phases:

  1. First, a reduce-scatter is performed in lg(P) steps using a recursive vector-halving, distance-doubling approach. In the first step of this algorithm processes communicate in pairs (rank 0 with 1, 2 with 3, etc.), sending and receiving for different halves of their input buffer. For example, process 0 sends the second half of its buffer to process 1 and receives and reduces data for the first half of the buffer from process 1. A reduction over the received data is performed before proceeding to the next communication step, where the distance to the destination rank is doubled while the data sent and received is halved.

  2. After the reduce-scatter phase is finished, each process in the largest binary block has a portion of the final reduced array. Next step is to scatter/distribute based on user-specified distribution. Note that, due to nature of recursive halving algorithm in the largest binary block, the blocks are not ordered in correct order. Enforced correct reorder by exchanging data between processes p and p', where p' is the bit-reverse of p.

When running on non-power-of-two number of processes, the algorithm works by breaking up execution into blocks that are powers of two and communicating interblock after the intrablock reduce-scatter. Non-power-of-two cases will have some degree of load imbalance compared to power-of-two, but cases with close to power-of-two (for e.g. 16 + 2) should still perform relatively well.

The halving-doubling / binary-blocks algorithm is described and analyzed in (Thakur et al., Optimization of Collective Communication Operations in MPICH, IJHPCA, 2005).

Data re-ordering is described in (Sack et al., Faster topology-aware collective algorithms through non-minimal communication, PPoPP, 2012).

Barrier

Synchronization point between processes.

barrier_all_to_all

  • Communication steps: 1
  • Bytes on the wire: P

Every process sends a notification to every other process. Then, it waits for a notification from every other process.

barrier_all_to_one

  • Communication steps: 2
  • Bytes on the wire: 1 for non-root, P for root

Non-root processes: send notification to root, wait for notification from root.

Root process: wait for notification from P-1 processes, send notification to P-1 processes.

Broadcast

Broadcast contents of buffer on one process to other P-1 processes.

broadcast_one_to_all

  • Communication steps: 1
  • Bytes on the wire: P*S

Non-root processes: receive buffer from root.

Root process: send buffer to P-1 processes.

pairwise_exchange

  • Communication steps: variable
  • Bytes on the wire: S

A communication pattern similar to the halving-doubling reduce-scatter, simplified for benchmarking purposes. The number of communication steps, C (which must be between 1 and lg(P)) is specified as input to the algorithm. In each step, the set of nodes is partitioned into pairs as in the corresponding step of halving-doubling reduce-scatter. Unlike the reduce-scatter, however, the buffer size sent in each step is the same (equal to S/C).