Information Theoretic Limits of Data Shuffling for Distributed Learning

I Introduction

Distributed computing systems for large data-sets have gained a lot of interest recently as they enable the processing of data-intensive tasks for machine learning, model tracing, and data analysis over a large number of commodity machines, and servers (e.g., Apache Spark , and MapReduce ). Generally speaking, a master node, which has the entire data-set, sends data blocks to be processed at distributed worker nodes. The workers subsequently respond with locally computed functions to the master node for the desired data analysis. This enables the processing of many terabytes of data over thousands of distributed servers in a time efficient manner.

One of the core elements in distributed learning algorithms is data shuffling . Before each iteration of the learning process, the entire data is randomly shuffled before being assigned to the worker nodes. This shuffling operation enables the worker nodes to process different data batches at each iteration, which presents large statistical gains . The statistical advantages provided by data shuffling come at the unavoidable cost of the communication overhead between the master and the worker nodes which must be incurred for every shuffling iteration. Thus, there exists a fundamental tradeoff between the communication overhead and the storage capacity of each worker node. To exemplify this tradeoff, consider two extreme scenarios: an ideal scenario in which the storage at the distributed workers is large enough to store the entire data-set, thus no communication has to be done from the master node for any shuffle. On the other extreme, when the storage is just enough to store the batch under processing, the communication load is expected to be maximum.

The focus of this paper is on formalizing and understanding this fundamental information-theoretic tradeoff between storage and the worst-case communication overhead for the data shuffling problem. Each iteration of data shuffling can be divided into two phases: data delivery, and storage update. In the data delivery phase, depending on the shuffled data points, the master node communicates a function of the data to the workers, so that each worker obtains its assigned data points. The second phase is termed as the storage update phase, which as shown in this paper is extremely critical in reducing the communication overhead of subsequently shuffling iterations. We next summarize the main contributions of this paper:

$\bullet$ We first present an information-theoretic formulation of the data shuffling problem involving both data delivery and update phases, accounting for the respective constraints and formalizing the tradeoff between worst-case communication overhead and the storage capacity of distributed workers.

$\bullet$ We also completely characterize this tradeoff for $K=2$ and $K=3$ workers, for any value of storage capacity. One of the most interesting aspects of the result for $K=3$ worker problem is the design of data delivery and storage update algorithms. In particular, for data delivery phase, we show that transmitting coded data from the master node to the workers can significantly reduce the communication overhead. More interestingly, the proposed storage update algorithm maintains the structural properties of the storage at the workers over time. This structural invariance placement is extremely critical in leveraging the gains of coding for different shuffles.

Related Work: In the past few years, there has been a flurry of research acitivity in understanding the benefits of coding for caching starting from the work of Maddah-Ali and Niesen who showed that exploiting multi-casting opportunities by coding can reduce the communication for caching. In , coding for MapReduce was proposed in order to reduce the communication cost between mappers and reducers, however the underlying focus of is significantly different than the problem considered in this work, where we care about the communication between the master node and the workers. The paper most closely related to this work is , where the idea of coding for data shuffling problem is presented to reduce the communication overhead between the master node and worker nodes. provides a probabilistic scheme of leveraging coding based on a random storage placement. In contrast to , in this paper we provide a deterministic and systematic storage update scheme, which increases the coding opportunities in the delivery phase. The underlying metric used here is the worst-case communication cost over all the possible shuffles, unlike the average cost considered in . Finally, we also present the first information theoretic lower bounds on the communication overhead for the data shuffling problem.

II System Model

We assume a master node which has access to the entire data-set $A=[x_{1}^{T},x_{2}^{T},\ldots,x_{N}^{T}]^{T}$ of size $Nd$ bits, i.e., $A$ is a matrix containing $N$ data points, denoted by $x_{1},x_{2},\ldots,x_{N}$, where $d$ is the dimensionality of each data point. Treating $A$, and its data points $x_{n}$ as random variables, we therefore have the entropies of these random variables as

At each iteration, indexed by $t$, the master node divides the data-set $A$ into $K$ data batches given as $A^{t}_{1},A^{t}_{2},\ldots,A^{t}_{K}$, where the batch $A^{t}_{k}$ is designated to be processed by worker $w_{k}$, and these batches correspond to the random permutation of the data-set, $\pi^{t}:A\rightarrow\{A^{t}_{1},\ldots,A^{t}_{K}\}$. Note that these data chunks are disjoint, and span the whole data-set, i.e.,

Hence, the entropy of any batch $A^{t}_{k}$ is given as

After getting the data batch, each worker locally computes a function (as an example, this function could correspond to the gradient or sub-gradients of the data points assigned to the worker). The local functions from the $K$ workers are processed subsequently at the master node. We assume that each worker $w_{k}$ has a storage $Z^{t}_{k}$ of size $Sd$ bits, for a real number $S$. For processing purposes, the assigned data blocks are needed to be stored by the workers, therefore, each worker $w_{k}$ must at least store the data block $A^{t}_{k}$ at time $t$. If we consider $Z^{t}_{k}$ as a random variable then the storage constraint is given by

According to (3) and (4), we get the minimum storage per worker $S\geq\frac{N}{K}$. We also have the processing constraint as

In the next epoch $t+1$, the data-set is randomly reshuffled at the master node according to a random permutation $\pi^{t+1}:A\rightarrow\{A^{t+1}_{1},A^{t+1}_{2},\ldots,A^{t+1}_{K}\}$. The main communication bottleneck occurs during Data Delivery since the master node needs to communicate the new data batches to the workers. Trivially, if the storage (per worker) exceeds $Nd$ bits, i.e., $S\geq N$, then each worker can store the whole data-set, and no communication has to be done between the master node and the workers for any shuffle. Therefore from the constraint on minimum storage per worker, we can write the possible range for storage as $\frac{N}{K}\leq S\leq N$.

We next proceed to describe the data delivery mechanism, and the associated encoding and decoding functions. The main process can be divided into 2 phases, namely the data delivery phase and the storage update phase as described next:

At time $t+1$, the master node sends a function of the data batches for the subsequent shuffles $(\pi_{t},\pi_{t+1})$, $X_{(\pi_{t},\pi_{t+1})}=\phi(A^{t}_{1},\ldots,A^{t}_{K},A^{t+1}_{1},\ldots,A^{t+1}_{K})=\phi_{(\pi_{t},\pi_{t+1})}(A)$ over the shared link, where $\phi$ is the data delivery encoding function

where $R_{(\pi_{t},\pi_{t+1})}$ is the rate of the shared link based on the shuffles $(\pi_{t},\pi_{t+1})$. Therefore, we have

Each worker $w_{k}$ should decode the desired batch $A^{t+1}_{k}$ out of the transmitted function $X_{(\pi_{t},\pi_{t+1})}$, and the data stored in the previous time slot denoted as $Z^{t}_{k}$. Therefore, the desired data is given by $A^{t+1}_{k}=\psi(X_{(\pi_{t},\pi_{t+1})},Z^{t}_{k})$, where $\psi$ is the decoding function at the workers

which can be written in terms of a decodability constraint, at each worker as follows

II-B Storage Update Phase

At each iteration, every worker updates its stored content as follows: the new storage content $Z^{t+1}_{k}$ is a function of the old storage content $Z^{t}_{k}$ as well as transmitted function $X_{(\pi_{t},\pi_{t+1})}$, i.e., $Z^{t+1}_{k}=\mu(X_{(\pi_{t},\pi_{t+1})},Z^{t}_{k})$, where $\mu$ is the update function

This implies the following storage-update constraint

The excess storage, if any, can be used to store opportunistically a function of the remaining data batches. Since the shuffling process at each time is done randomly, all the remaining batches are of equal importance. Consequently, the amount of excess storage, given by $(S-\frac{N}{K})d$ bits, is divided equally among the remaining $K-1$ batches. For the scope of this work, we assume that the placement of the excess storage is uncoded, which means that $\frac{(S-\frac{N}{K})}{K-1}d$ bits of the excess storage are dedicated to store a function of only one of the remaining $K-1$ batches. We give the notation $A^{t+1}_{i,k}$, where $i\neq k$, as the part of data that worker $w_{k}$ stores about $A^{t+1}_{i}$ in the excess storage at time $t+1$. Considering $A^{t+1}_{i,k}$ as a random variable, then

for $k,i\in\{1,\ldots,K\}$, and $i\neq k$.

We next define the worst-case communication as follows:

For any achievable scheme characterized by the functions $(\phi,\psi,\mu)$, the worst-case communication overhead over all possible consecutive data shuffles $(\pi_{t},\pi_{t+1})$ is defined as

Our goal in this work is to characterize the optimal worst-case communication $R_{\textsf{worst-case}}^{*}(K,S)$ defined as

We next present a claim which shows that the optimal communication $R^{*}$ (for any shuffle including the worst-case) is a convex function of the storage $S$:

$R^{*}$ is a convex function of $S$, where $S$ is the available storage at each worker.

Claim 1 follows from a simple memory sharing argument which shows that for any two available storage values $S_{1}$ and $S_{2}$, if $(S_{1},R^{*}(K,S_{1}))$, and $(S_{2},R^{*}(K,S_{2}))$ are achievable optimal schemes, then for any storage $\bar{S}=\alpha S_{1}+(1-\alpha)S_{2}$, $0\leq\alpha\leq 1$, there is a scheme which achieves a communication overhead of $\bar{R}=\alpha R^{*}(K,S_{1})+(1-\alpha)R^{*}(K,S_{2})$.

This is done as follows: First, we divide the data-set $A$ across $d$ dimensions into 2 batches namely; $A^{(\alpha)}$, and $A^{(1-\alpha)}$ of dimensions $\alpha d$, and $(1-\alpha)d$, for each point respectively. Then, we divide the storage for every worker $w_{k}$ into 2 parts namely; $Z_{k}^{(\alpha)}$, and $Z_{k}^{(1-\alpha)}$ of size $S_{1}\alpha d$, and $S_{2}(1-\alpha)d$, respectively. The former batch $A^{(\alpha)}$ will be shuffled among the former part of the storage $Z_{k}^{(\alpha)}$ to achieve the point $(S_{1},R^{*}(K,S_{1}))$, while the latter batch $A^{(1-\alpha)}$ will be shuffled among the latter part of the storage $Z_{k}^{(1-\alpha)}$ to achieve the point $(S_{2},R^{*}(K,S_{2}))$. Therefore, the total achievable load is given by

We next note that the optimal communication rate $R^{*}(K,\bar{S})$ is upper bounded by $\bar{R}(K,\bar{S})$, the rate of the memory sharing scheme, which completes the proof. ∎

III Main Results

For the distributed shuffling problem with $K=2$ workers of storage $Sd$ bits each, and a data-set of size $Nd$ bits, the optimal communication versus storage tradeoff is given by

For a distributed shuffling problem system with $K=3$ workers, the optimal communication versus storage tradeoff is given by

One interesting implication of Theorem 2 for $K=3$ workers is that the corner point $(\frac{2N}{3},\frac{N}{6})$ as in Fig. 1 is better than memory sharing between the two points $(\frac{N}{3},\frac{2N}{3})$, and $(N,0)$, which falls on the line connecting the two points, i.e., memory sharing here is not optimal, and coding can be leveraged to reduce the communication overhead.

IV Proof of Theorem 1 (K=2𝐾2K=2 workers)

We start with the achievablity of the corner points $S=N$ and $S=\frac{N}{2}$. The point $S=N$ is trivial and represents the case when the workers can store the whole data-set. In this case, no communication is necessary, i.e., $(S,R^{*})=(N,0)$ is achievable.

The point corresponding to $S=\frac{N}{2}$ is not trivial, where each worker can only store half of the data-set. Let us assume the data batches at time $t$ for the workers $w_{1}$, and $w_{2}$ are $A^{t}_{1}$, and $A^{t}_{2}$, respectively. These batches should be stored at the corresponding workers, which are just enough to store them, i.e., $Z^{t}_{1}=A^{t}_{1}$, and $Z^{t}_{2}=A^{t}_{2}$. Recall that these batches are equally sized of $\frac{N}{2}d$ bits. For the next iteration $t+1$, the data is randomly shuffled at the master node such that the new batches are $A^{t+1}_{1}$, and $A^{t+1}_{2}$, also of the same equal size $\frac{N}{2}d$. In the delivery phase, the master node sends

worker $w_{1}$ uses $X_{(\pi_{t},\pi_{t+1})}$ and $Z^{t}_{1}=A^{t}_{1}$ to decode $A^{t}_{2}$, and get access to the whole data-set $A=\{A^{t}_{1},A^{t}_{2}\}=\{A^{t+1}_{1},A^{t+1}_{2}\}$. The storage update is only storing the desired batch $A^{t+1}_{1}$. The same procedure applies for $w_{2}$. Note that this choice of the transmitted function in (18) works for any possible shuffling, which gives a constant communication load $H(X_{(\pi_{t},\pi_{t+1})})=\frac{N}{2}d$, and hence the corner point $(\frac{N}{2},\frac{N}{2})$ is achievable.

With the achievability of the corner points, any point in between for any real number $S$ can be simply achieved by memory sharing (see Claim 1). Therefore, the upper bound for the optimal worst-case communication rate is given by

If there is an overlap between $A^{t}_{k}$ and $A^{t+1}_{k}$ for $k\in\{1,2\}$, then for $S=\frac{N}{2}$, the communication cost of the above scheme can be further improved by sending

where $A^{t}_{k}\setminus A^{t+1}_{k}$ represents the part of the old batch $A^{t}_{k}$ at worker $w_{k}$, which is not needed any more in the new batch $A^{t+1}_{k}$. For an overlap $|A^{t}_{k}\cap A^{t+1}_{k}|=b$ data points, where $b$ is an integer number $b\in\{0,\ldots,\frac{N}{2}\}$, then the we achieve the corner point $(\frac{N}{2},\frac{N}{2}-b)$, with the worst-case rate when $b=0$.

IV-B Converse for K=2𝐾2K=2 workers

In this section, we present an information theoretic lower bound for the worst-case communication rate which matches the above scheme for $K=2$ workers.

Since we do not know a priori the shuffle that gives the worst-case communication, we assume first a shuffle $(\pi_{t},\pi_{t+1})$ with the optimal rate $R_{(\pi_{t},\pi_{t+1})}^{*}$, and then we lower bound $R_{(\pi_{t},\pi_{t+1})}^{*}$. Since the optimal worst-case rate is larger than the rate for any shuffle, i.e., $R_{\textsf{worst-case}}^{*}\geq R_{(\pi_{t},\pi_{t+1})}^{*}$, the lower bound found over $R_{(\pi_{t},\pi_{t+1})}^{*}$ serves also as a lower bound for the worst-case communication $R_{\textsf{worst-case}}^{*}$. The novel part in our proof is choosing the right shuffle which leads to the optimal lower bound (also see Section V-B for the application of this idea to the converse proof for Theorem $2$).

Now let us assume the data is shuffled such that $A^{t+1}_{1}=A^{t}_{2}$, and $A^{t+1}_{2}=A^{t}_{1}$, then from decodability constraint in (9):

Hence, we can find that $H(A|Z^{t}_{1},X_{(\pi_{t},\pi_{t+1})})=0$ as follows

where $(a)$ follows from (2), $(b)$ follows from the fact that $H(A,B|C)\leq H(A|C)+H(B|C)$, $(c)$ is because conditioning reduces entropy, and $(d)$ is from (5) and (21). We next prove the lower bound as follows

where $(a)$ follows from (1), $(b)$, and $(c)$ follow from the fact that $I(A;B)=H(A)-H(A|B)=H(B)-H(B|A)$ as well as (IV-B), $(d)$ is due to the fact that $H(A,B)\leq H(A)+H(B)$ and the fact that the $Z^{t}_{1}$ and $X_{(\pi_{t},\pi_{t+1})}$ are functions of the whole data-set $A$, i.e., $H(Z^{t}_{1},X_{(\pi_{t},\pi_{t+1})}|A)=0$, and $(e)$ follows from (4) and (7). Hence, from Remark 2, the lower bound for the worst-case rate is characterized as

Hence, the proof of Theorem 1 is complete from (19) and (24).

V Proof of Theorem 2 (K=3𝐾3K=3 workers)

From Fig. 1, the achievability involves three corner points: $S=\frac{N}{3}$, $S=\frac{2N}{3}$, and $S=N$. Achieving the point $(N,0)$ is trivial. The scheme for the point $S=\frac{N}{3}$ is similar to the point $S=\frac{N}{2}$ in the $K=2$ worker case, where each worker can only store the desired data batches. Similar to (18) the transmission at time $t+1$ is $X_{(\pi_{t},\pi_{t+1})}=\{A^{t}_{1}\oplus A^{t}_{2},A^{t}_{2}\oplus A^{t}_{3}\}$, which is sufficient for each worker to access all the data-set and store what it needs, achieving the corner point $(\frac{N}{3},\frac{2N}{3})$.

In this section, we focus on the achievability of the corner point $\left(\frac{2N}{3},\frac{N}{6}\right)$, which is perhaps the most interesting aspect of this result. For each worker $w_{k}$ at time $t$, $k\in\{1,2,3\}$, half of the storage ($N/3$ points) is used to store the desired batch $A^{t}_{k}$, while the remaining half (excess storage of $N/3$ points) is used opportunistically in order to minimize the communication overhead by storing some parts of the remaining batches given by the sub-batches $A^{t}_{i,k},\;i\in\{1,2,3\}\setminus k$.

These parts will be formed as follows: if we consider the data batch $A^{t}_{k}$ assigned for $w_{k}$ of $\frac{N}{3}$ points, each point $x\in A^{t}_{k}$ is divided across $d$ dimensions into two equal subdivisions labeled as $x^{(i)},\;i\in\{1,2,3\}\setminus k$. Then, each one of these subdivisions $x^{(i)}$ is placed in the corresponding sub-batch $A^{t}_{k,i}$ (the part of $x\in A^{t}_{k}$ stored in the excess storage of $w_{i}$). For instance, $A^{t}_{1}$ (of size $\frac{N}{3}$) stored in the processing half of the storage of worker $w_{1}$ is divided into two equal non-overlapping parts $A^{t}_{1}=\{A^{t}_{1,2},A^{t}_{1,3}\}$ (of size $\frac{N}{6}$ each). Worker $w_{2}$ will use half of its excess storage to store $A^{t}_{1,2}$, while worker $w_{3}$ will use half of its excess storage to store $A^{t}_{1,3}$.

The storage update procedure we present here maintains the above structural property of the stored data over time. The consequence of such structural invariance is that for any data point is required to be at a worker, at least half of this point is guaranteed to be already present at the worker, which decreases the communication overhead of the shuffling process.

Data Delivery: With this placement strategy, for the subsequent shuffles $(\pi_{t},\pi_{t+1})$, the transmitted function is given as

We claim that the above transmission is sufficient for all the three workers to obtain the required new points for any shuffle. Without loss of generality, let us consider worker $w_{1}$. According to (25), for $w_{1}$ to obtain the needed points not available in its storage, $A^{t+1}_{1}\setminus Z^{t}_{1}$, it must have $A^{t+1}_{2}\setminus Z^{t}_{2}$ and $A^{t+1}_{3}\setminus Z^{t}_{3}$ in its storage $Z_{1}^{t}$. This is indeed the case and can be proved according to the following argument:

In the following, we prove that $A^{t+1}_{2}\setminus Z^{t}_{2}\in Z_{1}^{t}$, and using the same argument we can show that $A^{t+1}_{3}\setminus Z^{t}_{3}\in Z_{1}^{t}$. Consider a data point $x\in A_{2}^{t+1}$, which is newly assigned to worker $w_{2}$ at time $t+1$ and is not fully present in its storage $Z_{2}^{(t)}$, i.e., $x\not\in A_{2}^{t}$. Therefore, there are two possibilities: a) $x\in A_{1}^{t}$ was being processed by worker $w_{1}$ at time $t$, which directly implies that $x$ is already available at $w_{1}$; or b) $x\in A_{3}^{t}$ was being processed by worker $w_{3}$ at time $t$, which implies that the sub-divisions of $x$ are $\{x^{(1)},x^{(2)}\}$ and the needed part by $w_{2}$, $x^{(1)}=x\cap(A^{t+1}_{2}\setminus Z^{t}_{2})$, is available at $w_{1}$ by definition (since $x^{(2)}$ is the subdivision that is already present at $w_{2}$).

According to (25), the worst-case scenario for this scheme happens if there is no overlap between $A^{t}_{k}$ and $A^{t+1}_{k}$ (completely new assignments). However, half of each data point $x\in A^{t+1}_{k}$ is already stored in the excess storage at $w_{k}$, labeled $x^{(k)}$, then the worst-case rate of $A^{t+1}_{k}\setminus Z^{t}_{k}$ and eventually $X_{(\pi_{t},\pi_{t+1})}$ is $\frac{N}{6}$, achieving the corner point $(\frac{2N}{3},\frac{N}{6})$.

Storage Update: Now, we present a deterministic storage update strategy, which maintains the structural properties of the storage at time $t$. Without loss of generality, let us analyze a data point $x=\{x^{(2)},x^{(3)}\}\in A^{t+1}_{1}$ . The update procedure is done according to the following three cases at time $t$

$\bullet$ Case 1: ${x=\{x^{(2)},x^{(3)}\}\in A^{t}_{1}}$. In this case, since the point $x$ was already being processed at worker $w_{1}$ at time $t$, hence no storage update is necessary at time $t+1$ for this point across workers.

$\bullet$ Case 2: $x=\{x^{(1)},x^{(3)}\}\in A^{t}_{2}$, $x^{(1)}\in A^{t}_{2,1}$, $x^{(3)}\in A^{t}_{2,3}$. After receiving $x^{(3)}$ from the delivery phase, worker $w_{1}$ stores the full-point $x$ in $A^{t+1}_{1}$. For worker $w_{3}$, $x^{(3)}$ leaves $A^{t}_{2,3}$ and enters $A^{t+1}_{1,3}$, and we can notice that $x^{(3)}$ remains within the excess storage of $Z_{3}^{t+1}$. Worker $w_{2}$ removes the data point $x$ from the processing batch $A^{t+1}_{2}$, and stores $x^{(1)}$ in $A_{1,2}^{t+1}$ after relabelling it as $x^{(2)}$ (now stored at the excess storage of $w_{2}$), i.e., it simply moves one half of $x$ in its excess storage.

$\bullet$ Case 3: $x=\{x^{(1)},x^{(2)}\}\in A^{t}_{3}$, $x^{(1)}\in A^{t}_{3,1}$, $x^{(2)}\in A^{t}_{3,2}$. This case is similar to case 2, where $w_{1}$ stores $x$ in $A^{t+1}_{1}$ and removes $x^{(1)}$ from $A^{t+1}_{3,1}$, worker $w_{2}$ moves $x^{(2)}$ into $A^{t+1}_{1,2}$ instead of $A_{3,2}^{t}$, and worker $w_{3}$ removes $x$ from $A^{t+1}_{3}$ and stores $x^{(1)}$ (now labeled as $x^{(3)}$) in $A^{t+1}_{1,3}$.

Let us now take a representative example depicted in Fig. 2 to illustrate our proposed data-delivery and storage-update phases for the corner point $(\frac{2N}{3},\frac{N}{6})$. Consider a system with $K=3$ workers, and $N=3$ data points, $\{x_{1},x_{2},x_{3}\}$. We assume that storage per worker is $S=\frac{2N}{K}=2$ points, i.e., each worker can store one extra data point in addition to the one under processing. We first clarify the color code used in this example to indicate the data point assigned to a certain worker labeled with these colors: blue for $x_{1}$, red for $x_{2}$, and yellow for $x_{3}$.

At time $t$, consider the dataset is shuffled such that $A_{1}^{t}=x_{1}$, $A_{2}^{t}=x_{2}$, and $A_{3}^{t}=x_{3}$. The corresponding storage placement at time $t$ in Fig. 2(a) is as follows: After using half the storage to store the desired data point (which can be depicted in this example as the desired color), each data point is divided equally among the unintended workers (depicted in this example as the unintended colors). For example, if we take the batch $A_{1}^{t}=x_{1}=\{x^{(2)}_{1},x^{(3)}_{1}\}$, worker $w_{2}$ stores $A_{1,2}^{t}=x_{1}^{(2)}$, and worker $w_{3}$ stores $A_{1,3}^{t}=x_{1}^{(3)}$.

At time $t+1$ in Fig. 2(b), the data is randomly shuffled again such that the new batches are: $A_{1}^{t+1}=x_{3}$, $A_{2}^{t+1}=x_{1}$, and $A_{3}^{t+1}=x_{2}$. We take this particular shuffle since it represents one of the possible worst cases, where every worker is assigned a completely different batch. According to the previous storage content shown in Fig. 2(a), worker $w_{1}$ already has $x_{3}^{(1)}$ but still needs $x_{3}^{(2)}$, which is stored at $w_{2}$ and $w_{3}$. Similarly, worker $w_{2}$ needs $x_{1}^{(3)}$ which is stored at $w_{1}$ and $w_{3}$, and worker $w_{3}$ needs $x_{2}^{(1)}$ which is stored at $w_{1}$ and $w_{2}$. Following the data delivery as in (25), the master node transmits:

Each worker has two out of these three subdivisions, therefore it can decode the remaining needed one. The rate of this transmission is $R=\frac{1}{2}$, which is $\frac{N}{6}$ where $N=3$.

For the storage update in Fig. 2(b), we only discuss the changes in $A^{t+1}_{1}$ and the corresponding $A^{t+1}_{1,2}$, and $A^{t+1}_{1,3}$. The storage update of the remaining parts can be done in a similar manner. From the delivery phase and the previous storage, worker $w_{1}$ gets $x_{3}$ (labeled yellow) and stores it in the processing half $A^{t+1}_{1}$ (above the dotted line). Worker $w_{2}$ already has a part of $A^{t+1}_{1}$, $x^{(2)}_{3}$ (labeled yellow), which was previously stored as $A^{t}_{3,2}$. Therefore, it remains in the excess storage (below the dotted line) as $A^{t+1}_{1,2}$. Worker $w_{3}$ already has $x_{3}$ previously labeled as $A^{t}_{3}$, so it keeps in its excess storage the part that is not stored in $A^{t+1}_{1,2}$, i.e., $x_{3}^{(1)}$, to be stored in $A^{t+1}_{1,3}$, after relabelling it to $x_{3}^{(3)}$ (now stored in $w_{3}$).

Using the achievability of the three corner points for $K=3$, and Claim 1, we get an upper bound on $R^{*}_{\textsf{worst-case}}(3,S)$ as

V-B Converse for K=3𝐾3K=3 workers

We now present the information theoretic lower bounds for the three-worker case, which matches the above scheme. Following Remark 2, we first assume subsequent data shuffles at times $\{t,t+1,t+2\}$ such that $A^{t+1}_{1}=A^{t}_{2}$, and $A^{t+2}_{1}=A^{t}_{3}$, then from the decodability constraint in (9), we have

Hence, in a similar proof to (IV-B) we get

where $(a)$ follows from (2) and the storage update constraint in (11), $(b)$ follows from the fact that $H(A,B,C|D)\leq H(A|D)+H(B|D)+H(C|D)$, and also the fact that conditioning reduces entropy, $(c)$ from (5) and (28), and $(d)$ from (28). Now, using (V-B) we can find the upper bound as follows

where $(a)$ follows from (1), $(b)$, and $(c)$ follow from (V-B), and due to the fact that $I(A;B)=H(A)-H(A|B)=H(B)-H(B|A)$, $(d)$ is due to the fact that $H(A,B,C)\leq H(A)+H(B)+H(C)$ and the fact that $Z^{t}_{1}$, $X_{(\pi_{t},\pi_{t+1})}$, and $X_{(\pi_{t+1},\pi_{t+2})}$ are all functions of the whole data-set $A$, and $(e)$ follows from (4) and (7). Hence, following Remark 2, we get a lower bound for the worst-case rate characterized as

The lower bound in (31) matches the upper bound obtained in (27) for the range $\frac{2N}{3}\leq S\leq N$.

We next present another lower bound on $R^{*}(3,S)$ which proves the optimality of our scheme for the range $\frac{N}{3}\leq S\leq\frac{2N}{3}$. To this end, we now assume a data shuffle such that $A_{1}^{t+1}=A_{3}^{t}$. Similar to (IV-B) and (V-B), we have

where $(a)$ follows from (2), $(b)$ from the facts that $H(A,B,C|D)\leq H(A|D)+H(B|D)+H(C|D)$, and $(c)$ using (5), and (9). We now proceed to obtain the second lower bound on $R^{*}_{\textsf{worst-case}}$ as follows

where $(a)$ follows from (1), $(b)$, and $(c)$ from (V-B), and due to the fact that $I(A;B)=H(A)-H(A|B)=H(B)-H(B|A)$, $(d)$ from the chain rule of entropy and the fact that $Z^{t}_{1}$, $Z^{t}_{2}$, and $X_{(\pi_{t},\pi_{t+1})}$ are all functions of the whole data-set $A$, $(e)$ from (9) where $A_{1}^{t+1}=A_{3}^{t}$ must be decoded from $Z_{1}^{t}$, and $X_{(\pi_{t},\pi_{t+1})}$, and because $A_{1}^{t}$ and $A^{t}_{2,1}$ are stored within $Z^{t}_{1}$, $(f)$ because conditioning reduces entropy, $(g)$ because after obtaining $A_{1}^{t}$, $A_{3}^{t}$, and $A_{2,1}^{t}$, the only remaining part in $Z_{2}^{t}$ is $A_{2,3}^{t}$, and finally $(h)$ follows from (4), (7), and (12). From Remark 2, and by rearranging (V-B), we get the following bound

Therefore, from (34), and (31), we get the following lower bound on $R^{*}_{\textsf{worst-case}}(3,S)$:

Finally, the proof of Theorem 2 follows from (27) and (35).

VI Conclusions

In this paper, we presented information theoretic formulation of the data shuffling problem, where we studied the tradeoff between the worst-case communication overhead and the storage available at the worker nodes. We completely characterized the optimal worst-case communication for $K=2$, and $K=3$ workers with any storage capacity, where we leveraged excess storage and coding to minimize the communication overhead in subsequent data shuffling iteration. A systematic storage update and delivery scheme was presented, which preserves the structural properties of the storage across workers. Generalizing these results for any number of workers $(K>3)$ is part of our ongoing work.

References