# Optimal Broadcast Schedules in Logarithmic Time with Applications to
Broadcast, All-Broadcast, Reduction and All-Reduction

Jesper Larsson Träff
TU Wien
Faculty of Informatics
Institute of Computer Engineering, Research Group Parallel Computing 191-4
Treitlstrasse 3, 5th Floor, 1040 Vienna, Austria

(July 22nd, 2024)

###### Abstract

We give optimally fast $O(\log p)$ time (per processor) algorithms for
computing round-optimal broadcast schedules for message-passing parallel
computing systems. This affirmatively answers difficult questions posed in a
SPAA 2022 BA and a CLUSTER 2022 paper. We observe that the computed schedules
and circulant communication graph can likewise be used for reduction, all-
broadcast and all-reduction as well, leading to new, round-optimal algorithms
for these problems. These observations affirmatively answer open questions
posed in a CLUSTER 2023 paper.

The problem is to broadcast $n$ indivisible blocks of data from a given root
processor to all other processors in a (subgraph of a) fully connected network
of $p$ processors with fully bidirectional, one-ported communication
capabilities. In this model, $n-1+\lceil\log_{2}p\rceil$ communication rounds
are required. Our new algorithms compute for each processor in the network
receive and send schedules each of size $\lceil\log_{2}p\rceil$ that determine
uniquely in $O(1)$ time for each communication round the new block that the
processor will receive, and the already received block it has to send.
Schedule computations are done independently per processor without
communication. The broadcast communication subgraph is an easily computable,
directed, $\lceil\log_{2}p\rceil$-regular circulant graph also used elsewhere.
We show how the schedule computations can be done in optimal time and space of
$O(\log p)$, improving significantly over previous results of $O(p\log^{2}p)$
and $O(\log^{3}p)$, respectively. The schedule computation and broadcast
algorithms are simple to implement, but correctness and complexity are not
obvious. The schedules are used for new implementations of the MPI (Message-
Passing Interface) collectives MPI_Bcast, MPI_Allgatherv,MPI_Reduce and
MPI_Reduce_scatter. Preliminary experimental results are given. Carefully
engineered and extensively evaluated implementations will be presented
elsewhere.

##  1 Introduction

The broadcast problem is a fundamental problem in message-passing computing
and broadcasting a fundamental, collective operation in programming interfaces
for distributed-memory parallel computing. In this paper, we give a particular
solution in terms of pipelining that can be used to implement this operation
(MPI_Bcast) as defined for the Message-Passing Interface (MPI) [11] for any
number of processes, and show that our resulting, optimal schedules (in terms
of preprocessing time and space and in terms of number of communication
rounds) and circulant communication graphs can also be used to implement the
related collective communication operations all-broadcast (MPI_Allgatherv),
reduction (MPI_Reduce) and all-reduction (MPI_Reduce_scatter_block and
MPI_Reduce_scatter). This addresses and solves a number of problems and
conjectures of recent papers [13, 14, 16].

The broadcasting problem considered here is the following. In a distributed
memory system with $p$ processors, a designated root processor has $n$
indivisible blocks of data that have to be communicated to all other
processors in the system. Each processor can in a communication operation send
an already known block to some other processor and at the same time receive
a(n unknown, new) block from some other, possibly different processor. Blocks
can be sent and received in unit time, where the time unit depends on the size
of the blocks which are assumed to all have (roughly) the same size. Since
communication of blocks takes the same time, the complexity of an algorithm
for solving the broadcast problem can be stated in terms of the number of
communication rounds that are required for the last processor to have received
all $n$ blocks from the root. In this fully-connected, one-ported, fully
(send-receive) bidirectional $p$ processor system [1, 3], any broadcast
algorithm requires $n-1+\lceil\log_{2}p\rceil$ communication rounds as is well
known. A number of algorithms reach this optimal number of communication
rounds with different communication patterns in a fully connected network [3,
8, 10, 14, 17].

The optimal communication round algorithm given in [17] was used to implement
the MPI_Bcast operation for MPI [11]. Thus a concrete, implementable solution
was given, unfortunately with a much too high schedule precomputation cost of
$O(p\log^{2}p)$ sequential steps which was partially amortized through caching
of schedules with communicators [12]. An advantage of this algorithm compared
to other solutions to the broadcast problem [3, 8, 10] is its simple,
$\lceil\log_{2}p\rceil$-regular circulant graph communication pattern, where
all processors throughout the broadcasting operation operate symmetrically.
Circulant graphs with different skips are used in many attractive algorithms
for different collective communication problems [2, 4, 5, 6, 16]. The
circulant graph pattern combined with optimal broadcast schedules makes it
possible to implement also the the difficult, irregular all-broadcast
operation MPI_Allgatherv more efficiently as was shown recently in [14]. As we
observe in this paper, we can likewise implement the reduction (MPI_Reduce)
and the difficult, irregular as well as regular all-reduction
(MPI_Reduce_scatter, MPI_Reduce_scatter_block) operations. These
implementations answer questions posed in [16].

In [13, 14], a substantial improvement of the schedule computation cost
compared to [17] was given, from super-linear $O(p\log^{2}p)$ to
$O(\log^{3}p)$ time steps, thus providing a practically much more relevant
algorithm. However, no adequate correctness proof was presented. A major
challenge was to get the schedule computation down to $O(\log p)$ time steps.
This is optimal, since at least $\lceil\log_{2}p\rceil$ communication rounds
are required independently of $n$ in which each processor sends and receives
different blocks.

In this paper, we constructively prove the conjecture that round optimal send
and receive schedules can be computed in $O(\log p)$ operations per processor
(without any communication) by stating and analyzing the corresponding
algorithms. The new algorithms use the same circulant graph communication
pattern and give rise to the same schedules as those constructed by previous
algorithms [13, 14, 17] but the algorithmic ideas are different. They are
readily implementable and of great practical relevance, as indicated by
preliminary experiments, for the full spectrum of MPI_Bcast, MPI_Allgather,
MPI_Allgatherv, MPI_Reduce, MPI_Reduce_scatter_block and MPI_Reduce_scatter
collectives. The paper focuses on the schedule properties and algorithms for
schedule computation. More careful, tuned implementation and experimental
evaluation of the collective operations will be given elsewhere. Nevertheless,
we think the algorithms presented here can become algorithms of choice for MPI
libraries and implementers.

##  2 Broadcast(ing with) Schedules and Circulant Graphs

Our problem is to distribute (broadcast) $n$ indivisible, disjoint blocks of
data from a single, designated _root_ processor to all other processors in a
$p$-processor system with processors $r,0\leq r<p$ that each communicate with
certain other processors by simultaneously sending and receiving data blocks.
The $n$ blocks for instance constitute a buffer of $m$ data units, $m\geq n$,
which can be broadcast by broadcasting the $n$ blocks, each of size at most
$\lceil m/n\rceil$ units. Choosing a best $n$ for a given $m$ is a system
dependent tuning problem that is not addressed at all in this paper.

We first show how broadcast from a designated root, $r=0$ (without loss of
generality) can be done with a regular, symmetric communication pattern and
explicit send and receive schedules that in constant time determine for each
communication operation by each processor which block is received and which
block is sent. We use this algorithm to formulate the correctness conditions
on possible send and receive schedules. Based on this, we then in the next
section show how correct schedules can be computed fast in $O(\log p)$ time
per processor, independently of all other processors and with no
communication.

In all of the following, we let $p$ denote the number of processors and take
$q=\lceil\log_{2}p\rceil$.

Algorithm 1 The $n$-block broadcast algorithm for processor $r,0\leq r<p$ of
data blocks in array buf. Round $x$ numbers the first round where actual
communication takes place. Blocks smaller than $0$ are neither sent nor
received, and for blocks larger than $n-1$, block $n-1$ is sent and received
instead. Also blocks to the root processor are not sent. This is assumed to be
taken care of by the bidirectional $\textsf{Send}()\parallel\textsf{Recv}()$
communication operations.

procedure Broadcast($\mathtt{buf}[n]$)

$\textsc{recvschedule}(r,\mathtt{recvblock[}\mathtt{]})$

$\textsc{sendschedule}(r,\mathtt{sendblock[}\mathtt{]})$

$x\leftarrow(q-(n-1)\bmod q)\bmod q$

for $i=0,1,\ldots,q-1$ do

$\mathtt{recvblock}[i]\leftarrow\mathtt{recvblock[}i\mathtt{]}-x$

$\mathtt{sendblock}[i]\leftarrow\mathtt{sendblock[}i\mathtt{]}-x$

if $i<x$ then $\triangleright$ Virtual rounds before $x$ already done

$\mathtt{recvblock}[i]\leftarrow\mathtt{recvblock[}i\mathtt{]}+q$

$\mathtt{sendblock}[i]\leftarrow\mathtt{sendblock[}i\mathtt{]}+q$

end if

end for

for $i=x,x+1,\ldots,(n+q-1+x)-1$ do

$k\leftarrow i\bmod q$

$t\leftarrow(r+\mathtt{skip[}k\mathtt{]})\bmod p$ $\triangleright$ to- and
from-processors

$f\leftarrow(r-\mathtt{skip[}k\mathtt{]}+p)\bmod p$

$\textsf{Send}(\mathtt{buf}[\mathtt{sendblock[}k\mathtt{]}],t)\parallel\textsf{Recv}(\mathtt{buf}[\mathtt{recvblock[}k\mathtt{]}],f)$

$\mathtt{sendblock[}k\mathtt{]}\leftarrow\mathtt{sendblock[}k\mathtt{]}+q$

$\mathtt{recvblock[}k\mathtt{]}\leftarrow\mathtt{recvblock[}k\mathtt{]}+q$

end for

end procedure

The generic $n$-block broadcast algorithm is shown as Algorithm 1. It proceeds
in rounds and is symmetric in the sense that all processors follow the same
regular, circulant graph communication pattern and do the same communication
operations in each round. For the rooted, asymmetric broadcast operation, this
is perhaps surprising. Rounds start from some round $x$ (to be explained
shortly) and end at round $n-1+q+x-1$ for a total of the required $n-1+q$
communication rounds. For round $i,x\leq i<n-1+q+x$, we take $k=i\bmod q$,
such that always $0\leq k<q$. In round $i$, each processor $r,0\leq r<p$
simultaneously sends a block to a _to-processor_
$t^{k}_{r}=(r+\mathtt{skip[}k\mathtt{]})\bmod p$ and receives a different
block from a _from-processor_ $f^{k}_{r}=(r-\mathtt{skip[}k\mathtt{]}+p)\bmod
p$, determined by a skip per round $\mathtt{skip[}k\mathtt{]},0\leq k<q$.
These skips are the same for all processors and define the circulant graph.
The blocks that are sent and received are numbered consecutively from $0$ to
$n-1$ and stored in a buf array indexed by the block number. The block that
processor $r,0\leq r<p$ sends in round $i$ is determined by a send schedule
array $\mathtt{sendblock[}k\mathtt{]}_{r}$ and likewise the block that
processor $r$ will receive in round $i$ by a receive schedule array
$\mathtt{recvblock[}k\mathtt{]}_{r}$. Since the blocks are thus fully
determinate, no block indices or other meta-data information needs to be
communicated. The $\mathtt{sendblock[}\mathtt{]}_{r}$ and
$\mathtt{recvblock[}\mathtt{]}_{r}$ arrays are computed such that blocks are
effectively broadcast from root processor $r=0$ that initially has all $n$
blocks, and such that all $n$ blocks are received and sent further on by all
the other processors. The starting round $x$ is chosen such that
$(n-1+q+x)\bmod q=(n-1+x)\bmod q=0$ and after this last round, which is a
multiple of $q$, all processors will have received all $n$ blocks. The
assumption that processor $r=0$ is the root can be made without loss of
generality: Should some other processor $r^{\prime}\neq 0$ be root, the
processors are simply renumbered by subtracting $r^{\prime}$ (modulo $p$) from
the processor indices.

To make Algorithm 1 efficient and correct, no block is ever sent back to the
root (which has all blocks in the first place), so there is no send operation
if $t^{k}_{r}=0$ for processor $r$ in round $k$. Non-existent, negatively
indexed blocks are neither sent nor received, so if
$\mathtt{sendblock[}k\mathtt{]}_{r}<0$ or
$\mathtt{recvblock[}k\mathtt{]}_{r}<0$ for some $k$, the corresponding send or
receive communication is simply ignored. For block indices larger than the
last block $n-1$, block $n-1$ is instead sent or received. These cases are
assumed to be handled by the concurrent send- and receive operations denoted
by $\textsf{Send}()\parallel\textsf{Recv}()$. The receive and send block
schedules $\mathtt{recvblock[}\mathtt{]}_{r}$ and
$\mathtt{sendblock[}\mathtt{]}_{r}$ are computed by the calls to
recvschedule() and sendschedule() to be derived in Section 2.2 and Section
2.3. Receive and send schedules for $p=17$ (a non-power of $2$) with $q=5$ is
shown in Table 1, which can be used to trace the communication when
broadcasting with Algorithm 1.

Table 1: Receive and send schedule for a non-power-of-two number of processors, $p=17$, $q=\lceil\log_{2}p\rceil=5$. The table shows for each processor $r,0\leq r<p$ the baseblock $b$ and the $\mathtt{recvblock[}k\mathtt{]}_{r}$ and $\mathtt{sendblock[}k\mathtt{]}_{r}$ schedules for $k=0,1,2,3,4$. The vertical bars correspond to $\mathtt{skip[}k\mathtt{]}$ $r$: | 0 | 1 | 2 | 3 | 4 | 5 | 6 | 7 | 8 | 9 | 10 | 11 | 12 | 13 | 14 | 15 | 16
---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---
$b$: | 5 | 0 | 1 | 2 | 0 | 3 | 0 | 1 | 2 | 4 | 0 | 1 | 2 | 0 | 3 | 0 | 1
$\mathtt{recvblock[}0\mathtt{]}$: | -4 | 0 | -5 | -4 | -3 | -5 | -2 | -5 | -4 | -3 | -1 | -5 | -4 | -3 | -5 | -2 | -5
$\mathtt{recvblock[}1\mathtt{]}$: | -5 | -4 | 1 | -5 | -4 | -3 | -3 | -2 | -5 | -4 | -3 | -1 | -5 | -4 | -3 | -3 | -2
$\mathtt{recvblock[}2\mathtt{]}$: | -2 | -2 | -2 | 2 | 0 | -4 | -4 | -3 | -2 | -2 | -4 | -3 | -1 | -1 | -4 | -4 | -3
$\mathtt{recvblock[}3\mathtt{]}$: | -1 | -3 | -3 | -2 | -2 | 3 | 0 | 1 | 2 | -5 | -2 | -2 | -2 | -2 | -1 | -1 | -1
$\mathtt{recvblock[}4\mathtt{]}$: | -3 | -1 | -1 | -1 | -1 | -1 | -1 | -1 | -1 | 4 | 0 | 1 | 2 | 0 | 3 | 0 | 1
$\mathtt{sendblock[}0\mathtt{]}$: | 0 | -5 | -4 | -3 | -5 | -2 | -5 | -4 | -3 | -1 | -5 | -4 | -3 | -5 | -2 | -5 | -4
$\mathtt{sendblock[}1\mathtt{]}$: | 1 | -5 | -4 | -3 | -3 | -2 | -5 | -4 | -3 | -1 | -5 | -4 | -3 | -3 | -2 | -5 | -4
$\mathtt{sendblock[}2\mathtt{]}$: | 2 | 0 | -4 | -4 | -3 | -2 | -2 | -4 | -3 | -1 | -1 | -4 | -4 | -3 | -2 | -2 | -2
$\mathtt{sendblock[}3\mathtt{]}$: | 3 | 0 | 1 | 2 | -5 | -2 | -2 | -2 | -2 | -1 | -1 | -1 | -1 | -3 | -3 | -2 | -2
$\mathtt{sendblock[}4\mathtt{]}$: | 4 | 0 | 1 | 2 | 0 | 3 | 0 | 1 | -3 | -1 | -1 | -1 | -1 | -1 | -1 | -1 | -1

For the algorithm to be correct in the sense of broadcasting all blocks from
root processor $r=0$ to all other processors, the following conditions must
hold:

  1. 1.

The block that is received in round $i$ with $k=i\bmod q$ by some processor
$r$ must be the block that is sent by the from-processor $f_{r}^{k}$,
$\mathtt{recvblock[}k\mathtt{]}_{r}=\mathtt{sendblock[}k\mathtt{]}_{f_{r}^{k}}$.

  2. 2.

The block that processor $r$ sends in round $i$ with $k=i\bmod q$ must be the
block that the to-processor $t_{r}^{k}$ will receive,
$\mathtt{sendblock[}k\mathtt{]}_{r}=\mathtt{recvblock[}k\mathtt{]}_{t_{r}^{k}}$.

  3. 3.

Over any $q$ successive rounds, each processor must receive $q$ different
blocks. More concretely,
$\bigcup_{k=0}^{q-1}\\{\mathtt{recvblock[}k\mathtt{]}_{r}\\}=(\\{-1,-2,\ldots,-q\\}\setminus\\{b_{r}-q\\})\cup\\{b_{r}\\}$
where $b_{r},0\leq b_{r}<q$ is a non-negative, real block received by
processor $r$ in one of the first $q$ rounds. This block $b_{r}$ is called the
_baseblock_ for processor $r$.

  4. 4.

The block that a processor sends in round $i$ with $k=i\bmod q$ must be a
block that has been received in some previous round, so either
$\mathtt{sendblock[}k\mathtt{]}_{r}=\mathtt{recvblock[}j\mathtt{]}_{r}$ for
some $j,0\leq j<k$, or $\mathtt{sendblock[}k\mathtt{]}_{r}=b_{r}-q$ where
$b_{r}\geq 0$ is the _baseblock_ for processor $r$.

The last correctness condition implies that
$\mathtt{sendblock[}0\mathtt{]}_{r}=b_{r}-q$ for each processor. With receive
and send schedules fulfilling these four conditions, it is easy to see that
Algorithm 1 correctly broadcasts the $n$ blocks over the $p$ processors.

######  Theorem 1.

Let $K,K>0$ be a number of communication phases each consisting of $q$
communication rounds for a total of $Kq$ rounds. Assume that in each round
$i,0\leq i<Kq$, each processor $r,0\leq r<p$ receives a block
$\mathtt{recvblock[}i\bmod q\mathtt{]}_{r}+\lfloor i/q\rfloor q$ and sends a
block $\mathtt{sendblock[}i\bmod q\mathtt{]}_{r}+\lfloor i/q\rfloor q$. By the
end of the $Kq$ rounds, processor $r$ will have received all blocks
$\\{0,1,\ldots,(K-1)q-1\\}\cup\\{b_{r}+(K-1)q\\}$ where $b_{r}$ is the first
(non-negative) block received by processor $r$.

###### Proof.

The proof is by induction on the number of phases. For $K=1$, there are $q$
rounds $i=0,1,\ldots,q-1$ over which each processor will receive its non-
negative baseblock $b$; all other receive blocks are negative (Correctness
Condition 3). For $K>1$, in the last phase $K-1$, each processor will receive
the blocks
$(\\{(K-2)q,(K-2)q+1,\ldots,(K-2)q+q-1\\}\setminus\\{b+(K-2)q\\})\cup\\{b+(K-1)q\\}$
since the set $\bigcup_{k=0}^{q-1}\\{\mathtt{recvblock[}k\mathtt{]}\\}$
contains $q$ different block indices, one of which is positive. The block
$b+(K-2)q$ has been received in phase $K-2$ by the induction hypothesis, in
its place block $b+(K-1)q$ is received. Therefore, at the end of phase $K-1$
using the induction hypothesis, blocks
$\\{0,1,\ldots,(K-2)q-1\\}\cup\\{(K-2)q,(K-2)q+1,\ldots,(K-2)q+q-1\\}=\\{0,1,\ldots,(K-1)q-1\\}$
plus the block $b+(K-1)q$ have been received, as claimed. By Correctness
Condition 4, no block is sent that has not been received in a previous round
or phase. ∎

In order to broadcast a given number of blocks $n$ in the optimal number of
rounds $n-1+q$, we use the smallest number of phases $K$ such that $Kq\geq
n-1+q$, and introduce a number of dummy blocks $x=Kq-(n-1+q)$ that do not have
to be broadcast. In the $K$ phases, all processors will have received the
$n+x-1$ blocks $0,1,\ldots,n+x-2$ plus one larger block. We perform $x$
initial, virtual rounds with no communication to handle the $x$ dummy blocks,
and broadcast the actual, positive blocks in the following $n-1+q$ rounds.
This is handled by subtracting $x$ from all computed block indices. Recall
that negative blocks are neither received nor sent by the algorithm. Blocks
with index larger than $n-1$ in the last phase are capped to $n-1$.

The symmetric communication pattern where each processor (node) $r$ has
incoming (receive) edges $(f_{r}^{k},r)$ and outgoing (send) edges
$(r,t^{k}_{r})$ is a _circulant graph_ with skips (jumps)
$\mathtt{skip[}k\mathtt{]},k=0,1,\ldots,q-1$. The skips are computed in
Section 2.1.

By Theorem 1, we make the following observations.

######  Observation 1.

  1. 1.

Algorithm 1 can be used to implement the MPI_Bcast operation for any number of
processes and any problem size $m$. The data of size $m$ units is divided into
$n$ blocks to give the shortest completion time (empirical tuning problem).
With only $n=1$ block, the algorithm takes $q=\lceil\log_{2}p\rceil$
communication rounds and is similar to a binomial tree.

  2. 2.

Since the communication pattern is symmetric and all processes do the
structurally same communication operations in each round, simultaneous
broadcast from all processes as roots can be done at the same time. For this,
each processes computes send and receive schedules with each other process as
root. In each send and receive operations, $p-1$ blocks corresponding to the
$p-1$ other roots are concatenated and treated as one unit. Thus, the all-
broadcast (all-to-all broadcast, allgather) problem where each process has $n$
blocks (possibly of different sizes) to broadcast can be solved with $n-1+q$
communication rounds. The algorithm is shown in the appendix. By this, the
difficult, irregular MPI_Allgatherv operation can be implemented efficiently.

  3. 3.

By reversing the direction of all communication operations, working downwards
from round $(n-1+q+x)-1$ to round $x$ and applying some given binary,
associative and commutative reduction operator on each received block, the
schedule gives a round-optimal solution to the reduction problem for
commutative operators. Each non-root process will send each partial result
block exactly once, and the root will receive and reduce partial results for
all blocks. This observation can be used to implement MPI_Reduce in the
optimal number of rounds for any problem size $m$ and any number of processes.

  4. 4.

Simultaneous reduction to all processes as roots can likewise be done, giving
round-optimal solutions for the all-reduction operations
MPI_Reduce_scatter_block and the difficult, irregular MPI_Reduce_scatter. Each
partial result block for all processes is sent and received once for a total
volume of $p-1$ blocks sent and $p-1$ blocks received. For $n=1$ block, we
believe that this is the first such algorithm with logarithmic number of
rounds for any number of processors. Previous algorithms use either a linear
ring [7, 18] or have almost twice the communication volume [16] for certain
number of processes $p$.

Table 2: Receive and send schedules for $p=9$ processors with $q=4$. $r$ | 0 | 1 | 2 | 3 | 4 | 5 | 6 | 7 | 8
---|---|---|---|---|---|---|---|---|---
$b$ | 4 | 0 | 1 | 2 | 0 | 3 | 0 | 1 | 2
$\mathtt{recvblock[}0\mathtt{]}$: | -2 | 0 | -4 | -3 | -2 | -4 | -1 | -4 | -3
$\mathtt{recvblock[}1\mathtt{]}$: | -3 | -2 | 1 | -4 | -3 | -2 | -2 | -1 | -4
$\mathtt{recvblock[}2\mathtt{]}$: | -1 | -3 | -2 | 2 | 0 | -3 | -3 | -2 | -1
$\mathtt{recvblock[}3\mathtt{]}$: | -4 | -1 | -1 | -1 | -1 | 3 | 0 | 1 | 2
$\mathtt{sendblock[}0\mathtt{]}$: | 0 | -4 | -3 | -2 | -4 | -1 | -4 | -3 | -2
$\mathtt{sendblock[}1\mathtt{]}$: | 1 | -4 | -3 | -2 | -2 | -1 | -4 | -3 | -2
$\mathtt{sendblock[}2\mathtt{]}$: | 2 | 0 | -3 | -3 | -2 | -1 | -1 | -3 | -2
$\mathtt{sendblock[}3\mathtt{]}$: | 3 | 0 | 1 | 2 | -4 | -1 | -1 | -1 | -1
Table 3: Receive and send schedules for $p=18$ processors with $q=5$. $r$ | 0 | 1 | 2 | 3 | 4 | 5 | 6 | 7 | 8 | 9 | 10 | 11 | 12 | 13 | 14 | 15 | 16 | 17
---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---
$b$ | 5 | 0 | 1 | 2 | 0 | 3 | 0 | 1 | 2 | 4 | 0 | 1 | 2 | 0 | 3 | 0 | 1 | 2
$\mathtt{recvblock[}0\mathtt{]}$: | -3 | 0 | -5 | -4 | -3 | -5 | -2 | -5 | -4 | -3 | -1 | -5 | -4 | -3 | -5 | -2 | -5 | -4
$\mathtt{recvblock[}1\mathtt{]}$: | -4 | -3 | 1 | -5 | -4 | -3 | -3 | -2 | -5 | -4 | -3 | -1 | -5 | -4 | -3 | -3 | -2 | -5
$\mathtt{recvblock[}2\mathtt{]}$: | -2 | -4 | -3 | 2 | 0 | -4 | -4 | -3 | -2 | -2 | -4 | -3 | -1 | -1 | -4 | -4 | -3 | -2
$\mathtt{recvblock[}3\mathtt{]}$: | -5 | -2 | -2 | -2 | -2 | 3 | 0 | 1 | 2 | -5 | -2 | -2 | -2 | -2 | -1 | -1 | -1 | -1
$\mathtt{recvblock[}4\mathtt{]}$: | -1 | -1 | -1 | -1 | -1 | -1 | -1 | -1 | -1 | 4 | 0 | 1 | 2 | 0 | 3 | 0 | 1 | 2
$\mathtt{sendblock[}0\mathtt{]}$: | 0 | -5 | -4 | -3 | -5 | -2 | -5 | -4 | -3 | -1 | -5 | -4 | -3 | -5 | -2 | -5 | -4 | -3
$\mathtt{sendblock[}1\mathtt{]}$: | 1 | -5 | -4 | -3 | -3 | -2 | -5 | -4 | -3 | -1 | -5 | -4 | -3 | -3 | -2 | -5 | -4 | -3
$\mathtt{sendblock[}2\mathtt{]}$: | 2 | 0 | -4 | -4 | -3 | -2 | -2 | -4 | -3 | -1 | -1 | -4 | -4 | -3 | -2 | -2 | -4 | -3
$\mathtt{sendblock[}3\mathtt{]}$: | 3 | 0 | 1 | 2 | -5 | -2 | -2 | -2 | -2 | -1 | -1 | -1 | -1 | -5 | -2 | -2 | -2 | -2
$\mathtt{sendblock[}4\mathtt{]}$: | 4 | 0 | 1 | 2 | 0 | 3 | 0 | 1 | 2 | -1 | -1 | -1 | -1 | -1 | -1 | -1 | -1 | -1

We can make an important, structural observation on receive schedules that
fulfill the four correctness conditions which helps to understand the
intuition of the broadcast algorithm described as Algorithm 1 and the
construction of receive and send schedules to be developed in the next
sections.

######  Observation 2.

Let a correct receive schedule for $p$ processors be given. A correct receive
schedule for $2p$ processors can be constructed by the following
modifications. First, for each processor $r,p\leq r<2p$, copy the receive
schedule for processor $r-p$, such that
$\mathtt{recvblock[}k\mathtt{]}_{r}=\mathtt{recvblock[}k\mathtt{]}_{r-p}$ for
each round $k,0\leq k<q$. Thus, the block that a processor would have received
in one of the first $q$ rounds in the $p$ processor schedule is the same as it
will now receive in the $2p$ processor schedule. Since $q$ is one larger in
the $2p$ than in the $p$ processor schedule, subtract $1$ from all the
negative blocks in the receive schedules. Since $p$ is doubled, set
$\mathtt{skip[}q+1\mathtt{]}=2p$, all other $\mathtt{skip[}k\mathtt{]},0\leq
k\leq$ will stay. The receive blocks $\mathtt{recvblock[}q\mathtt{]}_{r}$ for
all processors $r,0\leq r<2p$ are determined as follows. For processor $r=p$,
this shall be the new baseblock $q$ that has not been used so far, so
$\mathtt{recvblock[}q\mathtt{]}_{p}=q$. For all other large processors
$r,p<r<2p$, find the positive baseblock $b_{r}$ in
$\mathtt{recvblock[}k\mathtt{]}_{r}$, replace this with $-1$ (which will
correspond to receiving baseblock $q$ received previously by processor $r$)
and set $\mathtt{recvblock[}q\mathtt{]}_{r}=b_{r}$. For processors $r,0\leq
r<p$, we only need to set $\mathtt{recvblock[}q\mathtt{]}_{r}=-1$.

Table 2 and Table 3 show receive and send schedules for $p=9$ and $p=18$
processors. The doubling of $p$ from $9$ to $18$ gives rise to the shown
schedule by following Observation 2 as can readily be verified. If the receive
schedule for $p=9$ is correct, so is obviously and by construction the
schedule for $p=18$. The observation shows that receive schedules where
$p=2^{q}$ is a power of two indeed exist and can readily be computed.
Extending a $p$ processor receive schedule to a $2p-1$ processor receive
schedule is not as obvious. Had it been, it would be easy to construct $p$
processor receive schedules for any $p$.

###  2.1 The communication pattern

Algorithm 2 Computing the skips (jumps) for a $p$-processor circulant graph
($q=\lceil\log_{2}p\rceil$).

$k\leftarrow q$

$\mathtt{skip[}k\mathtt{]}\leftarrow p$

while $k>0$ do

$\mathtt{skip[}k-1\mathtt{]}\leftarrow\mathtt{skip[}k\mathtt{]}-\mathtt{skip[}k\mathtt{]}/2$

$k\leftarrow k-1$ $\triangleright$ Now
$\mathtt{skip[}k\mathtt{]}=\lceil\mathtt{skip[}k+1\mathtt{]}/2\rceil$

end while

The skips for the circulant graph pattern are computed by repeated halving
with rounding up of $p$ and shown as Algorithm 2. This obviously takes
$q=\lceil\log_{2}p\rceil$ iterations. We observe that for any $p,p>1$ it is
always the case that $\mathtt{skip[}0\mathtt{]}=1$ and
$\mathtt{skip[}1\mathtt{]}=2$. For convenience, we take
$\mathtt{skip[}q\mathtt{]}=p$. Thus, in the regular, circulant graph, each
processor has $q$ incoming and $q$ outgoing neighbor processors. We make some
simple observations needed for the following computations of the send and
receive schedules.

######  Observation 3.

For each $k,0\leq k<q$ it holds that
$\mathtt{skip[}k+1\mathtt{]}\leq\mathtt{skip[}k\mathtt{]}+\mathtt{skip[}k\mathtt{]}\leq\mathtt{skip[}k+1\mathtt{]}+1$

This follows directly from the halving scheme of Algorithm 2.

######  Lemma 1.

For any $k,k<q$ it holds that

$\mathtt{skip[}k+1\mathtt{]}-1\leq\sum_{i=0}^{k}\mathtt{skip[}i\mathtt{]}<\mathtt{skip[}k+1\mathtt{]}+k$

###### Proof.

The prof is by induction on $k$. For $k=0$, the claim holds since
$\mathtt{skip[}0\mathtt{]}=1$ and $\mathtt{skip[}1\mathtt{]}=2$. Assume the
claim holds for $k$. By the induction hypothesis and Observation 3, we get

$\sum_{i=0}^{k+1}\mathtt{skip[}i\mathtt{]}=\mathtt{skip[}k+1\mathtt{]}+\sum_{i=0}^{k}\mathtt{skip[}i\mathtt{]}<\mathtt{skip[}k+1\mathtt{]}+\mathtt{skip[}k+1\mathtt{]}+k\leq\mathtt{skip[}k+2\mathtt{]}+k+1$

and

$\sum_{i=0}^{k+1}\mathtt{skip[}i\mathtt{]}=\mathtt{skip[}k+1\mathtt{]}+\sum_{i=0}^{k}\mathtt{skip[}i\mathtt{]}\geq\mathtt{skip[}k+1\mathtt{]}+\mathtt{skip[}k+1\mathtt{]}-1\geq\mathtt{skip[}k+2\mathtt{]}-1\quad.$

∎

######  Observation 4.

For any $p$, there are at most two $k,k>1$ such that
$\mathtt{skip[}k-2\mathtt{]}+\mathtt{skip[}k-1\mathtt{]}=\mathtt{skip[}k\mathtt{]}$.

For $\mathtt{skip[}2\mathtt{]}=3$, Algorithm 2 gives
$\mathtt{skip[}1\mathtt{]}=2$ and $\mathtt{skip[}0\mathtt{]}=1$ for which the
observation holds. For $\mathtt{skip[}3\mathtt{]}=5$, Algorithm 2 gives
$\mathtt{skip[}2\mathtt{]}=3$ and $\mathtt{skip[}1\mathtt{]}=2$ for which the
observation holds. Any $p$ for which $\mathtt{skip[}2\mathtt{]}=3$ or
$\mathtt{skip[}3\mathtt{]}=5$ will have this property, and none other. We see
for $p\geq 3$ that $\mathtt{skip[}2\mathtt{]}\geq 3$, for $p>4$ that
$\mathtt{skip[}3\mathtt{]}\geq 5$ and for $p>7$ that
$\mathtt{skip[}4\mathtt{]}\geq 9$, and therefore
$\mathtt{skip[}k-2\mathtt{]}+\mathtt{skip[}k-1\mathtt{]}<\mathtt{skip[}k\mathtt{]}$
for $k>3$ for all $p$.

######  Observation 5.

For some $p$ and $k>0$, there is an $r,r<\mathtt{skip[}k\mathtt{]}$ with
$r+\mathtt{skip[}k\mathtt{]}=\mathtt{skip[}k+1\mathtt{]}$.

If $\mathtt{skip[}k+1\mathtt{]}$ is odd,
$r=\mathtt{skip[}k+1\mathtt{]}-\mathtt{skip[}k\mathtt{]}$ fulfills the
observation.

######  Lemma 2.

For any $r,0\leq r<p$ there is a (possibly empty) sequence
$[e_{0},e_{1},\ldots,e_{j-1}]$ of $j,j<q$, different skip indices such that
$r=\sum_{i=0}^{j-1}\mathtt{skip[}e_{i}\mathtt{]}$.

We call a (possibly empty) sequence $[e_{0},e_{1},\ldots,e_{j-1}]$ for which
$r=\sum_{i=0}^{j-1}\mathtt{skip[}e_{i}\mathtt{]}$ and where
$e_{0}<e_{1}<\ldots<e_{j-1}$ a _skip sequence_ for $r$. A non-empty skip
sequence $[e_{0},e_{1},\ldots,e_{j-1}]$ for $r$ defines a path from processor
$0$ to processor $r>0$. We will use the terms skip sequence and path
interchangeably.

###### Proof.

The proof is by induction on $k$. When $r=0$ the claim holds for the empty
sequence. Assuming the claim holds for any $r,0\leq
r<\mathtt{skip[}k\mathtt{]}$, we show that it holds for $0\leq
r<\mathtt{skip[}k+1\mathtt{]}$. If already $0\leq r<\mathtt{skip[}k\mathtt{]}$
the claim holds by assumption. If $\mathtt{skip[}k\mathtt{]}\leq
r<\mathtt{skip[}k+1\mathtt{]}$, then $0\leq
r-\mathtt{skip[}k\mathtt{]}<\mathtt{skip[}k+1\mathtt{]}-\mathtt{skip[}k\mathtt{]}\leq\mathtt{skip[}k\mathtt{]}$
by Observation 3. By the induction hypothesis, there is a sequence of
different skips not including $\mathtt{skip[}k\mathtt{]}$ and summing to
$r-\mathtt{skip[}k\mathtt{]}$, and $\mathtt{skip[}k\mathtt{]}$ can be appended
to this sequence to sum to $r$. ∎

The lemma indicates how to recursively compute in $O(q)$ steps a specific,
_canonical skip sequence_ for any $r,0\leq r<p$. By Observation 4 and
Observation 5, for some $p$ there may be more than one skip sequence for some
$r$; the decomposition of $r$ into a sum of different skips is not unique for
all $p$ (actually, the decomposition is unique only when $p$ is a power of
$2$). A canonical skip sequence will contain $\mathtt{skip[}k\mathtt{]}$ and
not $\mathtt{skip[}k-2\mathtt{]}$ and $\mathtt{skip[}k-1\mathtt{]}$ if
$\mathtt{skip[}k-2\mathtt{]}+\mathtt{skip[}k-1\mathtt{]}=\mathtt{skip[}k\mathtt{]}$
(Observation 4), and $\mathtt{skip[}k+1\mathtt{]}$ instead of
$\mathtt{skip[}k\mathtt{]}$ if
$r+\mathtt{skip[}k\mathtt{]}=\mathtt{skip[}k+1\mathtt{]}$ (Observation 5).

Algorithm 3 Finding the baseblock for processor $r,0\leq r<p$.

function baseblock($r$)

$k,r^{\prime}\leftarrow q,0$

repeat

$k\leftarrow k-1$

if $r^{\prime}+\mathtt{skip[}k\mathtt{]}=r$ then return $k$

else if $r^{\prime}+\mathtt{skip[}k\mathtt{]}<r$ then $r^{\prime}\leftarrow
r^{\prime}+\mathtt{skip[}k\mathtt{]}$

end if

until $k=0$

return $q$ $\triangleright$ Only processor $r=0$ will return $q$ as baseblock

end function

The canonical skip sequence for an $r,0\leq r<p$ is implicitly computed
iteratively by the baseblock() function of Algorithm 3 which explicitly
returns the first (smallest) skip index in the canonical skip sequence. This
index will be the _baseblock_ $b_{r}$ for $r$. Baseblocks are essential for
the broadcast schedules. For convenience, we define $q$ to be the baseblock of
$r=0$ for which the skip sequence is otherwise empty; for other $r>0$ the
baseblock $b_{r}$ satisfies $0\leq b_{r}<q$ and is a legal skip index.

If we let the root processor $r=0$ send out blocks one after the other (that
is, $\mathtt{sendblock[}k\mathtt{]}_{0}=k$ for $k=0,1,\ldots,q-1$) to
processors $\mathtt{skip[}k\mathtt{]}$, the canonical skip sequence to any
other processor $r,r>0$ gives a path through which the baseblock $b_{r}$ can
be sent from the root to processor $r$. Processor $r$ will receive its
baseblock in communication round $e$ where $e$ is the last, largest skip index
in the canonical skip sequence for $r$, therefore we can set
$\mathtt{recvblock[}e\mathtt{]}_{r}=b_{r}$.

The distribution of baseblocks for processors $r=0,1,2,\ldots,p-1$ follows a
somewhat, and for $p=2^{q}$ particularly regular pattern. The next lemma shows
that for any consecutive sequence of processors, there are enough
corresponding baseblocks that any processor can in $k$ communication rounds
receive $k$ different baseblocks when using the circulant graph communication
pattern.

######  Lemma 3.

Any consecutive sequence of baseblocks $b_{r},b_{r+1},\ldots
b_{r+\mathtt{skip[}k\mathtt{]}-1}$ of length $\mathtt{skip[}k\mathtt{]}$ for
processors $r,r+1,\ldots,r+\mathtt{skip[}k\mathtt{]}-1$ has at least $k+1$
different baseblocks.

###### Proof.

The proof is by induction on $k$. For this we list the baseblocks in sequence
for all processors $r,0\leq r<p$ which can be done in $O(p)$ steps as follows.
At first, only the root processor $r=0$ has baseblock $0$. Assume that the
baseblocks for processors $0,1,\ldots,\mathtt{skip[}k\mathtt{]}-1$ have been
listed. Append this list to itself but possibly remove the last element,
giving now a list of $\mathtt{skip[}k+1\mathtt{]}$ baseblocks. Increment the
baseblock for processor $r=0$ to $k+1$. Example for
$\mathtt{skip[}0\mathtt{]}=1$, $\mathtt{skip[}1\mathtt{]}=2$,
$\mathtt{skip[}2\mathtt{]}=3$, $\mathtt{skip[}3\mathtt{]}=6$,
$\mathtt{skip[}4\mathtt{]}=11$: $0\rightarrow 10\rightarrow 201\rightarrow
301201\rightarrow 40120130120$. As can be seen, each sequence of
$\mathtt{skip[}k\mathtt{]}$ baseblocks, starting from $r=0$ has exactly $k+1$
different baseblocks. Furthermore, any sequence starting from
$r=\mathtt{skip[}k\mathtt{]}$ has likewise $k+1$ different baseblocks, even
when one block is missing from the end when
$\mathtt{skip[}k\mathtt{]}+\mathtt{skip[}k\mathtt{]}=\mathtt{skip[}k+1\mathtt{]}+1$.
From the construction it can also be seen that in any sequence of baseblocks
for a consecutive sequence of processors, there is a unique, largest baseblock
appearing once. Now consider a sequence of $\mathtt{skip[}k\mathtt{]}$
baseblocks. Find the unique, largest baseblock corresponding to some processor
$r$. Within the $\mathtt{skip[}k\mathtt{]}$ length sequence, there is a
consecutive, shorter sequence of length $\mathtt{skip[}k-1\mathtt{]}+1$ either
starting or ending at $r$. By the induction hypothesis, the sequence of length
$\mathtt{skip[}k-1\mathtt{]}$ has at least $k$ different baseblocks and does
not include the baseblock of processor $r$. In total, there are therefore at
least $k+1$ different baseblocks. ∎

The way the $p$ baseblocks are listed correspond to what Algorithm 3 is doing
for each individual processor $r$. The proof shows that all baseblocks can be
listed more efficiently, in linear time, than by application of baseblock for
each $r$. This is relevant when the schedule computations are used for all-
broadcast and all-reduction.

###  2.2 The receive schedule

Algorithm 4 Computing receive blocks for processor $r,p\leq r<2p$ by depth-
first search with removal of accepted skip indices.

function allblocks($r,r^{\prime},s,e,k,\mathtt{recvblock[}q\mathtt{]}$)

while $e\neq-1$ do

if $r^{\prime}+\mathtt{skip[}e\mathtt{]}\leq r-\mathtt{skip[}k\mathtt{]}\wedge
r^{\prime}+\mathtt{skip[}e\mathtt{]}<s$ then

if $r^{\prime}+\mathtt{skip[}e\mathtt{]}\leq r-\mathtt{skip[}k+1\mathtt{]}$
then

$k\leftarrow\textsc{allblocks}(r,r^{\prime}+\mathtt{skip[}e\mathtt{]},s,e,k,\mathtt{recvblock[}\mathtt{]})$

end if

if $r^{\prime}>r-\mathtt{skip[}k+1\mathtt{]}$ then return $k$

end if

$s\leftarrow r^{\prime}+\mathtt{skip[}e\mathtt{]}$ $\triangleright$ Canonical
skip sequence found, keep it in $s$

$\mathtt{recvblock[}k\mathtt{]},k\leftarrow e,k+1$ $\triangleright$ Accept
$e$, next $k$

$\mathtt{next[}\mathtt{prev[}e\mathtt{]}\mathtt{]},\mathtt{prev[}\mathtt{next[}e\mathtt{]}\mathtt{]}\leftarrow\mathtt{next[}e\mathtt{]},\mathtt{prev[}e\mathtt{]}$
$\triangleright$ Unlink $e$

end if

$e\leftarrow\mathtt{next[}e\mathtt{]}$

end while

return $k$

end function

We now show how to compute the receive schedule
$\mathtt{recvblock[}k\mathtt{]},\allowbreak k=0,\ldots q-1$ for any processor
$r,0\leq r<p$ in $O(q)$ operations. More precisely, we compute for any given
$r$, the $q$ blocks that $r$ will receive in the $q$ successive communication
rounds $k=0,1,\ldots,q-1$ when processor $0$ is the root processor. The basis
for the receive schedule computation is to find $q$ canonical skip sequences
to different intermediate processors $r^{\prime}_{k}$ for $k=0,1,\ldots,q-1$
with $r^{\prime}_{k}\leq r-\mathtt{skip[}k\mathtt{]}$ and
$r_{0}>r^{\prime}_{1}>\ldots>r^{\prime}_{q-1}>0$. That all $r^{\prime}_{k}$
are positive will be ensured by considering processor $p+r$ instead of $r$ as
justified by Observation 2 since processors $r$ and $r+p$ have essentially the
same receive schedules. Hereby, modulo computations in the circulant graphs
are avoided. The $k$th skip sequence to $r^{\prime}_{k}$ is not allowed to
contain the baseblock of any of the previously found skip sequences.
Therefore, the skip sequences may not fulfill exactly
$r^{\prime}_{k}=r-\mathtt{skip[}k\mathtt{]}$, but it must be ensured that
$r^{\prime}_{k}\leq r-\mathtt{skip[}k\mathtt{]}$ since
$r-\mathtt{skip[}k\mathtt{]}$ is the processor from which processor $r$ will
receive a new block in round $k$ in the broadcast algorithm listed in
Algorithm 1. As will be seen, the skip sequences furthermore fulfil that
$r-\mathtt{skip[}k+1\mathtt{]}\leq r^{\prime}_{k}\leq
r-\mathtt{skip[}k\mathtt{]}$. These properties will ensure Correctness
Condition 3 on the receive schedule that over $q$ rounds, a processor will
receive $q$ different blocks, and since $r^{\prime}_{k}$ is not too far away
from $r-\mathtt{skip[}k\mathtt{]}$, this block can indeed have been delivered
to processor $r-\mathtt{skip[}k\mathtt{]}$ prior to communication round $k$
where $r$ is receiving from $r-\mathtt{skip[}k\mathtt{]}$ (Lemma 1).

For $q$ skips, there are $2^{q}\geq p$ canonical skip sequences, so exploring
them all will give a linear time (or worse) and not an $O(\log p)$ time
algorithm. Also, using Algorithm 3 $q$ times in succession will give an
$O(q^{2})$ time algorithm. Observation 2 is only for even $p$ and will,
naively implemented, also give an $O(\log^{2}p)$ algorithm. We instead do a
greedy search through the canonical skip sequences by the special backtracking
algorithm shown as Algorithm 4. The complexity of the computation is reduced
to $O(q)$ steps by removing the smallest (last found) skip index,
corresponding to the baseblock for $r^{\prime}_{k}$, each time a good
canonical skip sequence has been found, and then backtracking only enough to
find the next, $k+1$st skip sequence to $r^{\prime}_{k+1}$. The smallest (last
found) skip index of the $k$th canonical skip sequence will be taken as
$\mathtt{recvblock[}k\mathtt{]}$. The allblocks() function of Algorithm 4
stores the (remaining) skip indices in a doubly linked list in decreasing
order. Thus, for a skip index $e$, $\mathtt{next[}e\mathtt{]}$ is the next,
smaller, remaining skip index. This list is used to try the skips in
decreasing order as in Algorithm 3. Once an $r^{\prime}_{k}$ is found, the
corresponding skip index $e$ (baseblock) is removed by linking it out of the
doubly linked list in $O(1)$ time as also shown in Algorithm 4.

In the allblocks() function, $r^{\prime}$ is the largest intermediate
processor found so far. Each recursive call looks for the largest, remaining
skip index $e$ for which $r^{\prime}+\mathtt{skip[}e\mathtt{]}\leq
r-\mathtt{skip[}k\mathtt{]}$ where furthermore
$r^{\prime}+\mathtt{skip[}e\mathtt{]}$ is different from $r^{\prime}_{k-1}$,
causing a traversal of some of the remaining skip indices in the list. If
furthermore also $r^{\prime}+\mathtt{skip[}e\mathtt{]}\leq
r-\mathtt{skip[}k+1\mathtt{]}$, a recursive call is done on
$r^{\prime}+\mathtt{skip[}e\mathtt{]}$ to find an intermediate processor even
closer to $r-\mathtt{skip[}k\mathtt{]}$. Upon return, processor
$r^{\prime}+\mathtt{skip[}e\mathtt{]}$ is taken as $r^{\prime}_{k}$ if still
$r^{\prime}\leq r-\mathtt{skip[}k+1\mathtt{]}$ and $e$ will be the baseblock
stored as $\mathtt{recvblock[}k\mathtt{]}$.

Algorithm 5 Computing the receive schedule for processor $r,0\leq r<p$.

procedure recvschedule($r,\mathtt{recvblock[}q\mathtt{]}$)

for $e=0,\ldots,q$ do
$\mathtt{next[}e\mathtt{]},\mathtt{prev[}e\mathtt{]}\leftarrow e-1,e+1$

end for$\triangleright$ Doubly linked, circular list of skips in decreasing
order

$\mathtt{prev[}q\mathtt{]},\mathtt{next[}-1\mathtt{]},\mathtt{prev[}-1\mathtt{]}\leftarrow-1,q,0$

$b\leftarrow\textsc{baseblock}(r)$

$\mathtt{next[}\mathtt{prev[}b\mathtt{]}\mathtt{]},\mathtt{prev[}\mathtt{next[}b\mathtt{]}\mathtt{]}\leftarrow\mathtt{next[}b\mathtt{]},\mathtt{prev[}b\mathtt{]}$
$\triangleright$ Unlink baseblock index $b$

allblocks($p+r,0,p+p,q,0,\mathtt{recvblock[}\mathtt{]}$)

for $k=0,\ldots,q-1$ do $\triangleright$ Make baseblock $b$ only non-negative
block (Condition 3)

if $\mathtt{recvblock[}k\mathtt{]}=q$ then
$\mathtt{recvblock[}k\mathtt{]}\leftarrow b$

else
$\mathtt{recvblock[}k\mathtt{]}\leftarrow\mathtt{recvblock[}k\mathtt{]}-q$

end if

end for

end procedure

The allblocks() function is now used to compute the receive schedule for a
processor $r,0\leq r<p$ as shown in Algorithm 5. The algorithm uses the $q+1$
skips computed by Algorithm 2 (including $\mathtt{skip[}q\mathtt{]}=p$) and
searches for canonical paths to processor $p+r$. In order to exclude the
canonical path leading to $r$ itself with baseblock $b$ (as computed by
Algorithm 3), $b$ is removed from the list of skip indices before calling
allblocks(). The search starts from the largest skip index $e$ with $k=0$ and
the previous processor set to $s=p+p$. Upon return from allblocks(),
$\mathtt{recvblock[}k\mathtt{]}$ contains $q$ different, non-negative skip
indices. In order to arrive at a schedule where only the index for the
baseblock is positive as required by Theorem 1, $q$ is subtracted from all
$\mathtt{recvblock[}k\mathtt{]}$, except for the $k$ where
$\mathtt{recvblock[}k\mathtt{]}=p$: This is the round where processor $r$
receives its baseblock from root $r=p$.

######  Lemma 4.

A call $\textsc{allblocks}(p+r,0,p+p,q,0,\mathtt{recvblock[}\mathtt{]})$
implicitly determines $q$ processors $r^{\prime}_{k}$ with
$r-\mathtt{skip[}k+1\mathtt{]}\leq r^{\prime}_{k}\leq
r-\mathtt{skip[}k\mathtt{]}$ and stores the corresponding $q$ different
baseblocks in $\mathtt{recvblock[}k\mathtt{]}$. Each
$\mathtt{recvblock[}k\mathtt{]}$ is the largest baseblock in its interval that
can be found without using the skips of the baseblocks found in the intervals
before $k$.

The existence of such a sequence of processors and baseblocks (skip indices)
is guaranteed by Lemma 3.

###### Proof.

The recursive calls maintain the invariant that $r^{\prime}\leq
r-\mathtt{skip[}k\mathtt{]}$ before the while-loop, which holds prior to the
first call where $r^{\prime}=0$ and $k=0$. A recursive call is done only if
$r^{\prime}+\mathtt{skip[}e\mathtt{]}\leq
r-\mathtt{skip[}k+1\mathtt{]}<r-\mathtt{skip[}k\mathtt{]}$ which will maintain
the invariant for the call. Otherwise, $r-\mathtt{skip[}k\mathtt{]}\geq
r^{\prime}+\mathtt{skip[}e\mathtt{]}>r-\mathtt{skip[}k+1\mathtt{]}$. Upon
return from a recursive call the invariant is rechecked since $k$ can have
changed by the recursive call. Thus, whenever
$r^{\prime}+\mathtt{skip[}e\mathtt{]}$ is accepted as $r^{\prime}_{k}$, then
$r-\mathtt{skip[}k+1\mathtt{]}\leq r^{\prime}_{k}\leq
r-\mathtt{skip[}k\mathtt{]}$. Each time a new baseblock for
$\mathtt{recvblock[}k\mathtt{]}$ is found, it is removed from the list of skip
indices. Therefore, the returned baseblocks will be different. The conditions
on the receive blocks that $r^{\prime}\leq r-\mathtt{skip[}k+1\mathtt{]}$ and
$r^{\prime}+\mathtt{skip[}e\mathtt{]}\geq r-\mathtt{skip[}k+1\mathtt{]}$ means
that $e$ is the largest baseblock in the interval found not using prior
baseblocks; no further edges inside the interval are explored. ∎

######  Lemma 5.

Function allblocks() as called from Algorithm 5 performs at most $q-1$
recursive calls.

###### Proof.

The allblocks() function maintains the invariant that prior to each call,
$r^{\prime}<r-\mathtt{skip[}k\mathtt{]}$. After a recursive call, either
$r^{\prime}+\mathtt{skip[}e\mathtt{]}$ is accepted as $r^{\prime}_{k}$ and $e$
removed from the doubly linked list never to be considered again, or, if not,
then $k$ has changed because some other $e$ has been accepted by the recursive
call. Therefore, each recursive call causes the elimination of at least one
skip index $e$, and since there are only $q$ such indices, the claim follows
since at least one index $e$ in the list will not lead to a recursive call
(namely, when
$r^{\prime}+\mathtt{skip[}e\mathtt{]}=r-\mathtt{skip[}k\mathtt{]}$ which
happens at least once when $k=0$). ∎

######  Lemma 6.

Let $R$ be the number of recursive calls caused by the first invocation of
allblocks(). Then there are at most $2q+R$ iterations in total of the while-
loop over all calls.

###### Proof.

All skip indices must be scanned at least once in the while-loops of the
recursive calls. When for some skip index $e$,
$r^{\prime}+\mathtt{skip[}e\mathtt{]}>r-\mathtt{skip[}k\mathtt{]}$ and no
recursive call is performed, this index will be scanned again in some earlier,
pending recursive call (backtrack). When
$r^{\prime}+\mathtt{skip[}e\mathtt{]}$ is not accepted after the return from
the recursive call, because the invariant that $r^{\prime}\leq
r-\mathtt{skip[}k+1\mathtt{]}$ is violated, the algorithm backtracks to a
point where $e$ can be accepted. Thus, the total number of scans is at most
$2q+R$. ∎

We have now established that Algorithm 5 with the help of Algorithm 4 in
$O(q)$ steps computes $\mathtt{recvblock[}k\mathtt{]},k=0,1,\ldots,q-1$
fulfilling Correctness Condition 3 that
$\mathtt{recvblock[}\mathtt{]}=\\{-1,-2,\ldots,-q\\}\setminus\\{b-q\\}\cup\\{b\\}$.
It remains to be shown that Correctness Condition 4 is also fulfilled. To this
end, we define for processor $r$ that
$\mathtt{sendblock[}k\mathtt{]}_{r}=\mathtt{recvblock[}k\mathtt{]}_{t^{k}_{r}}$
where as in Algorithm 1, $t^{k}_{r}=(r+\mathtt{skip[}k\mathtt{]})\bmod p$.
With this definition, the first two correctness conditions from Section 2 are
obviously satisfied. It needs to be shown that
$\mathtt{recvblock[}k\mathtt{]}_{r}$ for processor $r$ is either
$\mathtt{recvblock[}j\mathtt{]}_{r-\mathtt{skip[}k\mathtt{]}}$ for a $j,j<k$
or the baseblock $b_{r-\mathtt{skip[}k\mathtt{]}}$ for processor
$r-\mathtt{skip[}k\mathtt{]}$.

######  Lemma 7.

For each processor $r,p\leq r<2p$, either
$\mathtt{recvblock[}k\mathtt{]}_{r}=\mathtt{recvblock[}j\mathtt{]}_{r-\mathtt{skip[}k\mathtt{]}}$
for some $j,j<k$ or
$\mathtt{recvblock[}k\mathtt{]}_{r}=b_{r-\mathtt{skip[}k\mathtt{]}}$ where
$b_{r-\mathtt{skip[}k\mathtt{]}}$ is the baseblock for processor
$r-\mathtt{skip[}k\mathtt{]}$.

###### Proof.

We first note that since
$(r-\mathtt{skip[}k\mathtt{]})-\mathtt{skip[}(k-1)+1\mathtt{]}=r-\mathtt{skip[}k\mathtt{]}-\mathtt{skip[}k\mathtt{]}\geq
r-\mathtt{skip[}k+1\mathtt{]}$, the intermediate processor $r^{\prime}_{k}$
chosen by processor $r$ in round $k$ is in the range of possible intermediate
processors chosen by processor $r-\mathtt{skip[}k\mathtt{]}$. The baseblock
$\mathtt{recvblock[}k\mathtt{]}_{r}$ is the largest in the range
$r-\mathtt{skip[}k+1\mathtt{]}$ to $r-\mathtt{skip[}k\mathtt{]}$ not using any
prior baseblocks for smaller $k$. If
$\mathtt{recvblock[}k\mathtt{]}_{r}=b_{r-\mathtt{skip[}k\mathtt{]}}$ there is
nothing to show. Otherwise, $\mathtt{recvblock[}k\mathtt{]}_{r}$ is a largest
block in the range excluding processor $r-\mathtt{skip[}k\mathtt{]}$, and
indeed one of the baseblocks found by processor $r-\mathtt{skip[}k\mathtt{]}$
in the rounds up to $k-1$ since initially only skip index
$b_{r-\mathtt{skip[}k\mathtt{]}}$ is excluded from the search. ∎

The lemmas together establish our main theorem which states that correct
receive schedules can be computed in optimal time.

######  Theorem 2.

For any number of processors $p$, a correct receive schedule for a
$p$-processor circulant graph with $\lceil\log_{2}p\rceil$ skips computed by
Algorithm 2, fulfilling the four correctness conditions, can be computed by
Algorithm 5 in $O(\log p)$ time steps for any processor $r,0\leq r<p$.

###  2.3 The send schedule

A straightforward computation of send schedules from the receive schedules by
for processor $r$ setting
$\mathtt{sendblock[}k\mathtt{]}_{r}=\mathtt{recvblock[}k\mathtt{]}_{t^{k}_{r}}$
with each $\mathtt{recvblock[}k\mathtt{]}$ computed by Algorithm 5 will take
$O(\log^{2}p)$ operations. To reach $O(\log p)$ operations, a different
approach is required. This structural approach can be seen as a generalization
of the well-known, easy case where $p=2^{q}$ is a power of two [9]. The
starting point is captured by Observation 6 below which complements
Observation 2.

######  Observation 6.

Let a correct send schedule for $p$ processors be given. A correct send
schedule for $2p$ processors can be constructed by the following
modifications. First, for the large processors $r,p\leq r<2p$, copy the send
schedule for processor $r-p$, such that
$\mathtt{sendblock[}k\mathtt{]}_{r}=\mathtt{sendblock[}k\mathtt{]}_{r-p}$ for
each round $0\leq k<q$. Since $q$ is one larger in the $2p$ processor schedule
than in the $p$ processor schedule, subtract $1$ from all negative blocks in
the send schedules. For the last round, set
$\mathtt{sendblock[}q\mathtt{]}_{r}=b_{r}$ for the small processors $r,0\leq
r<p$ where $b_{r}$ is the baseblock for $r$. For the large processors $r,p\leq
r<2p$, replace all positive send blocks with $-1$ and set
$\mathtt{sendblock[}q\mathtt{]}=-1$ for the last round.

The correctness of Observation 6 can be verified with the aid of Table 2 and
Table 3 which doubles $p$ from $9$ to $18$. There is no similarly simple
observation on how to construct a send schedule for $2p-1$ processors from a
$p$ processor send schedule. Nevertheless, the send schedule computation that
we adopt here builds on Observation 6 as much as possible.

Algorithm 6 Computing the send schedule for processors $r=0$ (root) and
$r,0\leq r<p$.

procedure Sendschedule($r,\mathtt{sendblock[}q\mathtt{]}$)

if $r=0$ then

for $k=0,\ldots,q-1$ do $\mathtt{sendblock[}k\mathtt{]}\leftarrow k$

end for

else

$b\leftarrow\textsc{baseblock}(r)$

$r^{\prime},c,e\leftarrow r,b,p$

for $k=q-1,\ldots,1$ do $\triangleright$ Obvious invariant: $r^{\prime}<e$

if $r^{\prime}<\mathtt{skip[}k\mathtt{]}$ then $\triangleright$ Lower part

if $r^{\prime}+\mathtt{skip[}k\mathtt{]}<e\vee
e<\mathtt{skip[}k-1\mathtt{]}\vee(k=1\wedge b>0)$ then
$\mathtt{sendblock[}k\mathtt{]}\leftarrow c$

else$\triangleright$ Violation

recvschedule($(r+\mathtt{skip[}k\mathtt{]})\bmod p,\mathtt{block}[]$)

$\mathtt{sendblock[}k\mathtt{]}\leftarrow\mathtt{block}[k]$

end if

if $e>\mathtt{skip[}k\mathtt{]}$ then $e\leftarrow\mathtt{skip[}k\mathtt{]}$

end if

else$\triangleright$ Upper part, $r^{\prime}\geq\mathtt{skip[}k\mathtt{]}$

$c\leftarrow k-q$

if $k=1\vee r^{\prime}>\mathtt{skip[}k\mathtt{]}\vee
e-\mathtt{skip[}k\mathtt{]}<\mathtt{skip[}k-1\mathtt{]}$ then
$\mathtt{sendblock[}k\mathtt{]}\leftarrow c$

else if $r^{\prime}+\mathtt{skip[}k\mathtt{]}>e$ then $\triangleright$
Violation

recvschedule($(r+\mathtt{skip[}k\mathtt{]})\bmod p,\mathtt{block}[]$)

$\mathtt{sendblock[}k\mathtt{]}\leftarrow\mathtt{block}[k]$

else $\mathtt{sendblock[}k\mathtt{]}\leftarrow c$

end if

$r^{\prime},e\leftarrow
r^{\prime}-\mathtt{skip[}k\mathtt{]},e-\mathtt{skip[}k\mathtt{]}$

end if

end for

$\mathtt{sendblock[}0\mathtt{]}\leftarrow b-q$

end if

end procedure

The root processor $r=0$ greedily sends the blocks $0,1,\ldots q-1$ in the
first $q$ communication rounds. For the send schedule computation for the
other processors, we go through the $q$ communication rounds starting from
$k=q-1$ down to $k=0$. In each round, we divide the processors into a lower
and and upper part, those with $r^{\prime}<\mathtt{skip[}k\mathtt{]}$ and
those with $\mathtt{skip[}k\mathtt{]}\leq r^{\prime}$, where $r^{\prime}$ is a
virtual processor index to be explained. Initially, $r^{\prime}=r$, and we
maintain an invariant upper bound $e$ on the range of virtual processors,
$0\leq r^{\prime}<e$. Initially, $e=\mathtt{skip[}q\mathtt{]}=p$. To maintain
the invariant from round $k$ to round $k-1$, if $r^{\prime}$ is in the upper
part, both $r^{\prime}$ and $e$ are decreased by $\mathtt{skip[}k\mathtt{]}$
at the end of round $k$, and if $r^{\prime}$ is in the lower part, then
$e=\mathtt{skip[}k\mathtt{]}$ for round $k-1$ unless already
$e<\mathtt{skip[}k\mathtt{]}$. For processors in the upper part, $e$ keeps
track of the number of times that
$\mathtt{skip[}k\mathtt{]}+\mathtt{skip[}k\mathtt{]}\neq\mathtt{skip[}k+1\mathtt{]}$,
that is, the number of times that $\mathtt{skip[}k+1\mathtt{]}$ is odd.
Virtual processors in the lower part aim to send the same block $c$ in round
$k$ as they already sent in round $k+1$. Initially, for these processors,
$c=b_{r}$ where $b_{r}$ is the baseblock for processor $r$ as found by
Algorithm 3. Virtual processors in the upper part aim to send block $c=k-q$ in
round $k$. These processors start with block $c=(q-1)-q=-1$ and decrease $c$
by $1$ in each round as indicated by Observation 6. The complete algorithm
maintaining the invariant on the virtual processor indices is shown as
Algorithm 6.

If in round $k$, following Observation 6, the $r^{\prime}$ for processor $r$
is in the lower part, $r^{\prime}<\mathtt{skip[}k\mathtt{]}$, the processors
$r^{\prime}+\mathtt{skip[}k\mathtt{]}<e$ have not yet received block $c$, so
$c$ is to be sent if $r^{\prime}+\mathtt{skip[}k\mathtt{]}<e$. Otherwise, it
we do not know which block processor $(r+\mathtt{skip[}k\mathtt{]})\bmod p$ is
missing, so in that case the receive block (as computed by Algorithm 5) for
round $k$ for processor $(r+\mathtt{skip[}k\mathtt{]})\bmod p$ is taken as the
block to send. This takes $O(q)$ steps, and is called a _violation_. If there
are more than a constant number of such violations for some processor $r$, a
logarithmic number of operations in total for that processor cannot be
guaranteed. If $e$ is very small, $e<\mathtt{skip[}k-1\mathtt{]}$, processor
$(r+\mathtt{skip[}k\mathtt{]})\bmod p$ will not have received $c$ and this
block can therefore be sent.

If instead $r^{\prime}$ is in the upper part for round $k$, then only the
processor with $r^{\prime}=\mathtt{skip[}k\mathtt{]}$ may have to use the
receive block for processor $(r+\mathtt{skip[}k\mathtt{]})\bmod p$ as the
block to send.

######  Theorem 3.

Algorithm 6 computes for any $r,0\leq r<p$ a send schedule in $O(\log p)$
operations.

###### Proof.

The loop performs $q-1$ iterations. Iterations that are not violations take
constant time. We will show that there can be only a constant number of
violations for $r$, namely at most four. Each violation takes $O(\log p)$
steps by the receive schedule Theorem 2. Therefore, the send schedule
computation takes $\Theta(\log p)$ steps.

Assume that a violation happens at iteration $k$ for $r^{\prime}$, under what
conditions can a violation happen again for a smaller $k$? If $r^{\prime}$ is
in the upper part, $r^{\prime}\geq\mathtt{skip[}k\mathtt{]}$, the violation
can happen only for $r^{\prime}=\mathtt{skip[}k\mathtt{]}$ by the first
condition (indeed, $\mathtt{sendblock[}k\mathtt{]}=c$ when
$r^{\prime}>\mathtt{skip[}k\mathtt{]}$). For the next, lower iteration,
$r^{\prime}$ is decreased by $\mathtt{skip[}k\mathtt{]}$ to $r^{\prime}=0$ and
will remain so for all remaining iterations, thus it will never again hold
that $r^{\prime}\geq\mathtt{skip[}k\mathtt{]}$. We therefore only have to
consider violations for the lower part case where
$r^{\prime}\leq\mathtt{skip[}k\mathtt{]}$. A violation happens if
$r^{\prime}+\mathtt{skip[}k\mathtt{]}\geq e$. If
$e\geq\mathtt{skip[}k\mathtt{]}$, $e=\mathtt{skip[}k\mathtt{]}$ for the next,
lower iteration. Thus, a violation cannot happen in iteration $k-1$ if
$r^{\prime}<\mathtt{skip[}k-1\mathtt{]}$. If
$r^{\prime}\geq\mathtt{skip[}k-1\mathtt{]}$, $r^{\prime}$ will be in the upper
part for iteration $k-1$ with $e=\mathtt{skip[}k\mathtt{]}$ and so
$r^{\prime}+\mathtt{skip[}k-1\mathtt{]}>\mathtt{skip[}k\mathtt{]}=e$ can
happen only for $r^{\prime}=\mathtt{skip[}k-1\mathtt{]}$ (and by Observation 3
only if $\mathtt{skip[}k\mathtt{]}$ is odd, which is then taken care of in the
upper part for iteration $k-1$). Thus, as long as
$e\geq\mathtt{skip[}k\mathtt{]}$ there can be at most two violations for $r$.
The difficult cases are when $e<\mathtt{skip[}k\mathtt{]}$ which will cause a
violation for $\mathtt{skip[}k-1\mathtt{]}\leq r^{\prime}$. These $r^{\prime}$
will be in the upper part for the next iteration $k-1$, but there could
possibly be a violation for some later, smaller $k$. At which iterations can
it happen that $\mathtt{skip[}k-1\mathtt{]}\leq
e\leq\mathtt{skip[}k\mathtt{]}$?

The smallest possible $e$ in iteration $k$ is

$\displaystyle p-\sum_{i=k+1}^{q-1}\mathtt{skip[}i\mathtt{]}$ $\displaystyle=$
$\displaystyle
p-(\sum_{i=0}^{q-1}\mathtt{skip[}i\mathtt{]}-\sum_{i=0}^{k}\mathtt{skip[}i\mathtt{]})$
$\displaystyle=$ $\displaystyle
p-(\mathtt{skip[}q\mathtt{]}+q-1-(\mathtt{skip[}k+1\mathtt{]}+k))$
$\displaystyle=$ $\displaystyle p-(p+q-1-\mathtt{skip[}k+1\mathtt{]}-k)$
$\displaystyle=$ $\displaystyle\mathtt{skip[}k+1\mathtt{]}+1-q+k$

by the upper bound of Lemma 1.

A violation always happens if $\mathtt{skip[}k-1\mathtt{]}\leq
e\leq\mathtt{skip[}k\mathtt{]}$. For the smallest possible $e$, we have
$\mathtt{skip[}k+1\mathtt{]}+1-q+k\leq\mathtt{skip[}k\mathtt{]}$ implying
$\mathtt{skip[}k+1\mathtt{]}-\mathtt{skip[}k\mathtt{]}\leq q-1-k$ and with
Observation 3 that $\mathtt{skip[}k\mathtt{]}\leq q-k$ and with Observation 3
again that $\mathtt{skip[}k-1\mathtt{]}\leq\frac{q+1-k}{2}$. Likewise, for a
violation to happen, $\mathtt{skip[}k-1\mathtt{]}\leq e$ gives
$\mathtt{skip[}k-1\mathtt{]}\leq\mathtt{skip[}k+1\mathtt{]}+1-q+k$. Using
Observation 3 twice, this implies $\mathtt{skip[}k-1\mathtt{]}\leq
4\mathtt{skip[}k-1\mathtt{]}+1-q+k$ and therefore
$\frac{q-1-k}{3}\leq\mathtt{skip[}k-1\mathtt{]}$. Since
$\mathtt{skip[}k\mathtt{]}$ are roughly halving with decreasing $k$, both
inequalities are fulfilled for a very small number of $k$ on the order of
$O(\log\log p)$. This is small enough, that such a violation cannot happen
again.

Counting, there are at most four times that a violation can happen for $r$. ∎

Theorem 2 and Theorem 3 are crucial, also for practical use of the schedule
computations. To verify the implementations, finite, exhaustive proof for $p$
up to some millions shows that the computed receive and send schedules are
correct and consistent according to the four conditions in Section 2, that the
number of violations claimed in Theorem 3 is indeed at most $4$ and that the
stated invariants all hold (see the appendix). Table 1 shows receive and send
schedules as computed by the algorithms for $p=17$ processors (not a power of
two). There are, for instance, send schedule violations in round $k=2$ for
processor $r=3$ and in round $k=3$ for processor $r=8$.

##  3 Empirical Results

As experimental evidence, the algorithms for computing receive and send
schedules in $O(\log p)$ time steps are being used in new implementations of
MPI_Bcast, MPI_Allgatherv, MPI_Reduce, MPI_Reduce_scatter_block and
MPI_Reduce_scatter. All implementations are available from the author. We show
here preliminary results with MPI_Bcast (Algorithm 1) and MPI_Reduce on a
large HPC system and MPI_Allgatherv with irregular problems on a small system.
We compare our new implementations against a native MPI library
implementation. The best number of blocks $n$ leading to the smallest
broadcast time is chosen based on a linear cost model. For MPI_Bcast, the size
of the blocks is chosen as $F\sqrt{m/\lceil\log p\rceil}$ for a constant $F$
chosen experimentally. For MPI_Allgatherv, the number of blocks to be used is
chosen as $\sqrt{m\lceil\log p\rceil}/G$ for another, experimentally
determined constant $G$. The constants $F$ and $G$ depend on context, system,
and MPI library. Finding a best $n$ in practice is a highly interesting
problem outside the scope of this work.

Figure 1: Broadcast and reduce results, native versus new, with the OpenMPI
4.1.5 library with $p=200\times 1,p=200\times 4$ and $p=200\times 128$ MPI
processes. The constant factor $F$ for the size of the blocks has been chosen
as $F=70$. The MPI datatype is MPI_INT.

The large system is the VEGA system in Slowenia, which is comprised of 768 CPU
based compute nodes each two AMD EPYC 7H12 Rome 64-core CPUs running at 2.6
GHz for a total of 128 cores per node. We have run with 200 nodes and in
Figure 1 show results with $p=200\times 1$, $p=200\times 4$ and $p=200\times
128$ MPI processes. Our new MPI_Bcast and MPI_Reduce implementations are run
against the native OpenMPI 4.1.5 library implementation. Even though our
implementation is not specifically tuned, and in particular not adapted to the
hierarchical structure of the system, our implementation is faster than the
native MPI library by a factor of more than $4$ and $3$, respectively for the
one and four processes per node configurations for the large problem sizes.
Even with full nodes, our (non-hierarchical, fully-connected) implementation
of MPI_Bcast is a factor of 3 faster than the native implementation. Part of
the reason for this must be poor implementations in the OpenMPI 4.1.5 library.
We will focus on fine tuning and hierarchical implementation of our new
algorithms elsewhere.

The small system is a $36\times 32$ processor cluster with 36 dual socket
compute nodes, each with two Intel(R) Xeon(R) Gold 6130F 16-core processors.
The nodes are interconnected via dual Intel Omnipath interconnects each with a
bandwidth of 100 GigaBytes/s. The implementations and benchmarks were compiled
with gcc 8.3.0 with the -O3 option. The MPI library used is OpenMPI 4.0.5.

Figure 2: Irregular all-broadcast (MPI_Allgatherv) results, native versus new,
with the OpenMPI 4.0.5 library with $p=36\times 32$ MPI processes and
different types of input problems (regular, irregular, degenerate). The
constant factor $G$ for the number of blocks has been chosen as $G=40$. The
MPI datatype is MPI_INT.

The results for MPI_Allgatherv are shown in Figure 2 for $p=36\times 32$ MPI
processes and different types of input problems (see the appendix for a
detailed algorithm description). The _regular problem_ divides the given input
size $m$ roughly evenly over the processes in chunks of $m/p$ elements. The
_irregular problem_ divides the input in chunks of size roughly $(i\bmod
3)m/p$ for process $i=0,1,\ldots,p-1$. The _degenerate problem_ has one
process contribute the full input of size $m$ and all other process no input
elements. For the degenerate problem, the performance of the OpenMPI 4.0.5
baseline library indeed degenerates and is a factor of close to 100 slower
than the new implementation, where the running time is largely independent of
the problem type. The running time of the new MPI_Allgatherv implementation is
in the ballpark of MPI_Bcast for the same total problem size. This is highly
attractive for MPI libraries and not guaranteed by other implementations. The
overheads for the schedule computations are one to two orders of magnitude
smaller than those shown in [13].

##  4 Summary

We showed that round-optimal broadcast schedules on fully connected, one-
ported, fully bidirectional $p$-processor systems can indeed be computed in
$O(\log p)$ time steps per processor. This is a surprising and highly non-
trivial result that affirmatively answers the long standing open questions
posed in [13, 14, 17]. One reason for the difficulty and the awkward proofs is
that there is no obvious counterpart to Observation 2 and Observation 6 for
odd, $2p-1$, number of processors. We furthermore observed that such schedules
that rely on regular, circulant graph communication patterns likewise give
rise to round-optimal, pipelined algorithms for the related all-broadcast,
reduction and all-reduction operations. We outlined how to use the schedules
for implementation of the MPI collectives MPI_Bcast, MPI_Allgatherv,
MPI_Reduce, MPI_Reduce_scatter_block and MPI_Reduce_scatter, and gave
promising, initial experiments. A more careful evaluation of these
implementations, also in versions that are more suitable to systems with
hierarchical, non-homogeneous communication systems [15] is ongoing and should
be found elsewhere.

## References

  * [1] Amotz Bar-Noy and Shlomo Kipnis.  Broadcasting multiple messages in simultaneous send/receive systems.  Discrete Applied Mathematics, 55(2):95–105, 1994.
  * [2] Amotz Bar-Noy, Shlomo Kipnis, and Baruch Schieber.  An optimal algorithm for computing census functions in message-passing systems.  Parallel Processing Letters, 3(1):19–23, 1993.
  * [3] Amotz Bar-Noy, Shlomo Kipnis, and Baruch Schieber.  Optimal multiple message broadcasting in telephone-like communication systems.  Discrete Applied Mathematics, 100(1–2):1–15, 2000.
  * [4] Massimo Bernaschi, Giulio Iannello, and Mario Lauria.  Efficient implementation of reduce-scatter in MPI.  Journal of Systems Architecture, 49(3):89–108, 2003.
  * [5] Jehoshua Bruck and Ching-Tien Ho.  Efficient global combine operations in multi-port message-passing systems.  Parallel Processing Letters, 3(4):335–346, 1993.
  * [6] Jehoshua Bruck, Ching-Tien Ho, Schlomo Kipnis, Eli Upfal, and Derrick Weathersby.  Efficient algorithms for all-to-all communications in multiport message-passing systems.  IEEE Transactions on Parallel and Distributed Systems, 8(11):1143–1156, 1997.
  * [7] Ernie Chan, Marcel Heimlich, Avi Purkayastha, and Robert A. van de Geijn.  Collective communication: theory, practice, and experience.  Concurrency and Computation: Practice and Experience, 19(13):1749–1783, 2007.
  * [8] Bin Jia.  Process cooperation in multiple message broadcast.  Parallel Computing, 35(12):572–580, 2009.
  * [9] S. Lennart Johnsson and Ching-Tien Ho.  Optimum broadcasting and personalized communication in hypercubes.  IEEE Transactions on Computers, 38(9):1249–1268, 1989.
  * [10] Oh-Heum Kwon and Kyung-Yong Chwa.  Multiple message broadcasting in communication networks.  Networks, 26:253–261, 1995.
  * [11] MPI Forum.  MPI: A Message-Passing Interface Standard. Version 4.1, November 2nd 2023.  www.mpi-forum.org.
  * [12] Hubert Ritzdorf and Jesper Larsson Träff.  Collective operations in NEC’s high-performance MPI libraries.  In 20th International Parallel and Distributed Processing Symposium (IPDPS), page 100, 2006.
  * [13] Jesper Larsson Träff.  Brief announcement: Fast(er) construction of round-optimal $n$-block broadcast schedules.  In 34th ACM Symposium on Parallelism in Algorithms and Architectures (SPAA), pages 143–146. ACM, 2022.
  * [14] Jesper Larsson Träff.  Fast(er) construction of round-optimal $n$-block broadcast schedules.  In IEEE International Conference on Cluster Computing (CLUSTER), pages 142–151. IEEE Computer Society, 2022.
  * [15] Jesper Larsson Träff and Sascha Hunold.  Decomposing MPI collectives for exploiting multi-lane communication.  In IEEE International Conference on Cluster Computing (CLUSTER), pages 270–280. IEEE Computer Society, 2020.
  * [16] Jesper Larsson Träff, Sascha Hunold, Nikolaus Manes Funk, and Ioannis Vardas.  Uniform algorithms for reduce-scatter and (most) other collectives for MPI.  In IEEE International Conference on Cluster Computing (CLUSTER). IEEE Computer Society, 2023.
  * [17] Jesper Larsson Träff and Andreas Ripke.  Optimal broadcast for fully connected processor-node networks.  Journal of Parallel and Distributed Computing, 68(7):887–901, 2008\.
  * [18] Jesper Larsson Träff, Andreas Ripke, Christian Siebert, Pavan Balaji, Rajeev Thakur, and William Gropp.  A pipelined algorithm for large, irregular all-gather problems.  International Journal of High Performance Computing Applications, 24(1):58–68, 2010.

##  Appendix A All-Broadcast

Algorithm 7 The $n$-block all-to-all broadcast algorithm for processor
$r,0\leq r<p$ for data in the arrays $\mathtt{bufs}[j],0\leq j<p$. The count
$x$ is the number of empty first rounds. Blocks smaller than $0$ are neither
sent nor received, and for blocks larger than $n-1$, block $n-1$ is sent and
received instead.

procedure AllBroadcast($\mathtt{bufs}[p][n]$)

for $j=0,1,\ldots,p-1$ do

$r^{\prime}\leftarrow(r-j+p)\bmod p$

$\textsc{recvschedule}(r^{\prime},\mathtt{recvblocks}[j][])$

end for

for $j=0,1,\ldots,p-1$ do

for $k=0,1,\ldots,q-1$ do

$f\leftarrow(j-\mathtt{skip[}k\mathtt{]}+p)\bmod p$

$\mathtt{sendblocks}[j][k]\leftarrow\mathtt{recvblocks}[f][k]$

end for

end for

$x\leftarrow(q-(n-1)\bmod q)\bmod q$ $\triangleright$ Virtual rounds, only
when $x>0$

for $j=0,1,\ldots,p-1$ do

for $i=0,1,\ldots,q-1$ do

$\mathtt{recvblocks}[j][i]\leftarrow\mathtt{recvblocks}[j][i]-x$

$\mathtt{sendblocks}[j][i]\leftarrow\mathtt{sendblocks}[j][i]-x$

if $i<x$ then $\triangleright$ Virtual rounds before $x$ already done

$\mathtt{recvblocks}[j][i]\leftarrow\mathtt{recvblocks}[j][i]+q$

$\mathtt{sendblocks}[j][i]\leftarrow\mathtt{sendblocks}[j][i]+q$

end if

end for

end for

for $i=x,x+1,\ldots,(n+q-1+x)-1$ do

$k\leftarrow i\bmod q$

$t,f\leftarrow(r+\mathtt{skip[}k\mathtt{]})\bmod
p,(r-\mathtt{skip[}k\mathtt{]}+p)\bmod p$

$j^{\prime}\leftarrow 0$

for $j=0,1,\ldots,p-1$ do $\triangleright$ Pack

if $j\neq t$ then

$\mathtt{tempin}[j^{\prime}]\leftarrow\mathtt{bufs}[j][\mathtt{sendblocks}[j][k]]$

$j^{\prime}\leftarrow j^{\prime}+1$

end if

$\mathtt{sendblocks}[j][k]\leftarrow\mathtt{sendblocks}[j][k]+q$

end for

$\textsf{Send}(\mathtt{tempin},t)\parallel\textsf{Recv}(\mathtt{tempout},f)$

$j^{\prime}\leftarrow 0$

for $j=0,1,\ldots,p-1$ do $\triangleright$ Unpack

if $j\neq r$ then

$\mathtt{bufs}[j][\mathtt{recvblocks}[j][k]],\leftarrow\mathtt{tempout}[j^{\prime}]$

$j^{\prime}\leftarrow j^{\prime}+1$

end if

$\mathtt{recvblocks}[j][k]\leftarrow\mathtt{recvblocks}[j][k]+q$

end for

end for

end procedure

The explicit send and receive schedules can be used for all-broadcast as shown
as Algorithm 7. In this problem, each processor has $n$ blocks of data to be
broadcast to all other processors. The $n$ blocks for each processor $r$ are
as for the broadcast algorithm assumed to be roughly of the same size, but
blocks from different processors may be of different size as long as the same
number of $n$ blocks are to be broadcast from each processor. The algorithm
can therefore handle the irregular case where different processors have
different amounts of data to be broadcast as long as each divides its data
into $n$ roughly equal-sized blocks. Due to the fully symmetric, circulant
graph communication pattern, this can be done by doing the $p$ broadcasts for
all $p$ processors $r,0\leq r<p$ simultaneously, in each communication step
combining blocks for all processors into a single message. The blocks for
processor $j$ are assumed to be stored in the buffer array
$\mathtt{bufs}[j][]$ indexed by block numbers from $0$ to $n-1$. Initially,
processor $r$ contributes its $n$ blocks from $\mathtt{bufs}[r][]$. The task
is to fill all other blocks $\mathtt{bufs}[j][]$ for $j\neq r$. Each processor
$r$ computes a receive schedule $\mathtt{recvblocks}[j]$ for each other
processor as root processor $j,0\leq j<p$ which is the receive schedule for
$r^{\prime}=(r-j+p)\bmod p$. Before each communication operation, blocks for
all processors $j,0\leq j<p$ are packed consecutively into a temporary buffer
tempin, except the block for the to-processor $t^{k}$ for the communication
round. This processor is the root for that block, and already has the
corresponding block. After communication, blocks from all processors are
unpacked from the temporary buffer tempout into the $\mathtt{bufs}[j][]$
arrays for all $j,0\leq j<p$ except for $j=r$: A processor does not receive
blocks that it already has. As in the broadcast algorithm in Algorithm 1, it
is assumed that the packing and unpacking will not pack for negative block
indices, and that indices larger than $n-1$ are taken as $n-1$. Also packing
and unpacking blocks for processors not contributing any data (as can be the
case for highly irregular applications of all-to-all broadcast) shall be
entirely skipped (not shown in Algorithm 7), so that the total time spent in
packing and unpacking per processor over all communication rounds is bounded
by the total size of all $\mathtt{bufs}[j][],j\neq t^{k}$ and
$\mathtt{bufs}[j][],j\neq r$.

##  Appendix B Overheads

Table 4: Timings of old $O(\log^{3}p)$ and new $O(\log p)$ time step receive and send schedule algorithms for different ranges of processors $p$. Receive and send schedules are computed for all processors $0\leq r<p$ for all $p$ in the given ranges. The expected running times are thus bounded by $O(p\log^{3}p)$ (old) and $O(p\log p)$ (new) time, respectively. Times are in seconds and measured with the clock() function. We also estimate the average time spent per processor for computing its send and receive schedules of $\lceil\log_{2}p\rceil$ entries. This is done by measuring for each $p$ the total time for the schedule computation, dividing by $p$ and averaging over all $p$ in the range. These times are in micro seconds. Proc. range $p$ | Total Time (seconds) |  Per proc. ($\mu$seconds)
---|---|---
| $O(p\log^{3}p)$ | $O(p\log p)$ | $O(\log^{3}p)$ | $O(\log p)$
$[1,17\,000]$ | 443.8 | 50.0 | 2.769 | 0.334
$[16\,000,33\,000]$ | 1567.2 | 152.8 | 3.763 | 0.370
$[64,000\,73\,000]$ | 3206.0 | 282.6 | 5.187 | 0.454
$[131\,000,140\,000]$ | 7595.0 | 653.2 | 6.226 | 0.534
$[262\,000,267\,000]$ | 9579.4 | 726.6 | 7.242 | 0.548
$[524\,000,529\,000]$ | 21580.2 | 1492.9 | 8.196 | 0.566
$[1\,048\,000,1\,050\,000]$ | 18934.3 | 1083.8 | 9.024 | 0.516
$[2\,097\,000,2\,099\,000]$ | 44714.9.0 | 2554.6 | 10.656 | 0.608

To demonstrate the practical impact of the improvement from $O(\log^{3}p)$
time steps (per processor) which was the bound given in [14] to $O(\log p)$
steps per processor as shown here, we run the two algorithms for different
ranges of processors $p$. For each $p$ in range, we compute both receive and
send schedules for all processors $r,0\leq r<p$, and thus expect total running
times bounded by $O(p\log^{3}p)$ and $O(p\log p)$, respectively. These
runtimes in seconds, gathered on a standard workstation with an Intel Xeon
E3-1225 CPU at 3.3GHz and measured with the clock() function from the time.h C
library, are shown in Table 4. The timings include both the receive and the
send schedule computations, but exclude the time for verifying the correctness
of the schedules, which has also been performed up to some $p\geq 2M,M=2^{20}$
(and for a range of $100\,000$ processors around $16M$), including verifying
the bounds on the number of recursive calls from Theorem 2 and the number of
violations from Theorem 3. When receive and send schedules have been computed
for all processors $r,0\leq r<p$, verifying the four correctness conditions
from Section 2 can obviously be done in $O(p\log p)$ time steps. The
difference between the old and the new implementations is significant, from
close to a factor of 10 to significantly more than a factor of 10. However,
the difference is not by a factor of $\log^{2}p$ as would be expected from the
derived upper bounds. This is explained by the fact that the old send schedule
implementations employ some improvements beyond the trivial computation from
the receive schedules which makes the complexity closer to $O(\log^{2}p)$. We
also give a coarse estimate of the time spent per processor by measuring for
each $p$ in the given range the time for the schedule computations for all $p$
processors, dividing this by $p$ and averaging over all $p$ in the given
range. This is indicative of the overhead for the recvblock() and sendblock()
computations in the implementations of Algorithm 1 and Algorithm 7. These
times (in microseconds) are listed as columns $O(\log^{3}p)$ and $O(\log p)$,
respectively. The difference by a factor of $10$ and more is slowly increasing
with $\log_{2}p$.