Towards Practical Overlay Networks for Decentralized Federated Learning

An important notion in DFL, or general decentralized optimization algorithms, is the mixing matrix ${\bm{M}}$ of the graph $G$. The $i$-th row of ${\bm{M}}$ denotes the weights used for aggregating local models of the neighboring nodes to update the model of the $i$-th client. Hence the adjacency matrix of an overlay network and its Metropolis-Hastings matrix are both mixing matrices [5]. The symmetric property of ${{\bm{M}}}$ indicates that its eigenvalues are real and can be sorted in non-increasing order. Let $\lambda_{i}({{\bm{M}}})$ denote the $i$-th largest eigenvalue of ${{\bm{M}}}$, then we have $\lambda_{1}({{\bm{M}}})=1>\lambda_{2}({{\bm{M}}})\geq\cdots\geq\lambda_{N}({{\bm{M}}})>-1$ based on the spectral property of the mixing matrix [5]. The constant $\lambda=\lambda({{\bm{M}}}):=\max\{|\lambda_{2}({{\bm{M}}})|,|\lambda_{N}({{\bm{M}}})|\}$ has been used to characterize optimization error (a measure of training loss) and generalization gap (a measure of test accuracy) of DFL. In particular, it is shown that the optimization error and generalization gap – for a typical DFL framework, Decentralized Federated Averaging (DFedAvg) – are bounded, respectively, by $\mathcal{O}\Big{(}\frac{1}{(1-\lambda)^{2}}\Big{)}$ and $\mathcal{O}\Big{(}2\lambda^{2}+4\lambda^{2}\ln\frac{1}{\lambda}+2\lambda+\frac{2}{\ln\frac{1}{\lambda}}\Big{)}$ – in terms of $\lambda$ [33, 13]. Notice that both $\frac{1}{(1-\lambda)^{2}}$ and $2\lambda^{2}+4\lambda^{2}\ln\frac{1}{\lambda}+2\lambda+\frac{2}{\ln\frac{1}{\lambda}}$ are increasing functions of $\lambda\in(0,1)$.

Per the above discussion, to achieve good convergence and generalization, a topology needs to have a $\lambda$ sufficiently smaller than 1 and hence achieve a small value of $\frac{1}{(1-\lambda)^{2}}$ and $2\lambda^{2}+4\lambda^{2}\ln\frac{1}{\lambda}+2\lambda+\frac{2}{\ln\frac{1}{\lambda}}$. Thus we define the first topology metric, called the convergence factor of $G$: $c_{G}=\frac{1}{(1-\lambda)^{2}}$.

Note that when $\frac{1}{(1-\lambda)^{2}}$ is minimized, $2\lambda^{2}+4\lambda^{2}\ln\frac{1}{\lambda}+2\lambda+\frac{2}{\ln\frac{1}{\lambda}}$ is also minimized. Hence for the sake of simplicity, we do not need another factor.

The diameter of a network is the longest length of all shortest paths calculated in the network. It reflects the network distance between the two most distant nodes. The intuition of considering this metric is that the network diameter can represent the maximum latency that the local model trained on the data of a client can propagate to all clients in the network.

The third metric is the average length of all shortest paths in the network. The intuition of considering this metric is that the average length can represent the average latency that a local model can propagate to a random client.

The $\mathtt{join}$ protocol is designed to achieve the following correctness property: given an existing FedLay overlay network, a new joining node runs the $\mathtt{join}$ protocol. When the $\mathtt{join}$ protocol finishes, the joining node is guaranteed to find its correct neighbors, i.e., the adjacent nodes in all virtual spaces. The above property will be proved later and it ensures that a correct FedLay overlay can be achieved after every new node join. If a FedLay network with $n$ node is correct, after a new node joins, the $(n+1)$-node network is also correct. This provides a way to recursively construct a large overlay network correctly from a small-scale network even with 2 or 3 nodes. Note that such recursive property is a key module to ensure the correctness of many P2P overlays such as Chord [30] and distributed DT graph [19, 17].

FedLay builds upon the concept of random virtual coordinates and circular distance introduced by SpaceShuffle [38], extending it to achieve a fully decentralized topology construction. Unlike [38], where an administrator is required to access each node and switch, FedLay allows any client to join the network through any existing node, enhancing scalability and ease of use. Additionally, we optimize the NDMP leave protocol to minimize overhead by ensuring that maintenance operations are only triggered when necessary, thus reducing the resource consumption during network changes.

When a new node $u$ joins the existing correct FedLay network, we assume $u$ knows one existing node $v$ in the overlay, which can be an arbitrary node – this is the minimum assumption for any overlay network. If $u$ knows no existing node, it has no way to join any overlay. $u$ first computes a random coordinate as its position in the first virtual space, say $x_{1}^{u}$. $u$ will let $v$ sends a message Neighbor_discovery to the current Fedlay network using greedy routing to the destination location $x_{1}^{u}$. Neighbor_discovery also includes $u$’s IP address.

We first define the concept of circular distance, which is a metric used in greedy routing of FedLay.

For two coordinates $x$ and $y$ on a ring, the circular distance is the length of the smaller arc between $x$ and $y$, normalized by the perimeter of the ring that is 1. We say $x$ is closer to $y$ than $w$ on a ring space, if $CD(x,y)<CD(w,y)$. If $x$ and $w$ have the same circular distance to $y$, we always break the tie to one of $x$ and $w$ with a smaller value of their IP addresses (considering each IP address is a 32-bit value). Hence, there is only one node that is closest to a given coordinate $x$.

Upon receiving Neighbor_discovery, the greedy routing protocol to the destination location $x_{i}^{u}$ in Space $i$ is executed by a node $v$ as following:

1. 1.

   Node $v$ finds a neighbor $w$, such as $w$’s coordinate in Space $i$, $x_{i}^{w}$, has the smallest circular distance to $x_{i}^{u}$ among all neighbors of $v$.

2. 2.

   If $CD(x_{i}^{u},x_{i}^{v})>CD(x_{i}^{u},x_{i}^{w})$, $v$ forwards the Neighbor_discovery message to $w$.

3. 3.

   If $CD(x_{i}^{u},x_{i}^{v})\leq CD(x_{i}^{u},x_{i}^{w})$, ${Neighbor\_discovery}$ stops at $v$. From $v$’s two adjacent nodes, $v$ finds the adjacent node $p$ such that $x_{i}^{u}$ is on $\wideparen{v,p}$, the smaller arc between $v$ and $p$. Then $v$ sends a message to $u$ to tell $u$ that $v$ and $p$ are $u$’s adjacent nodes on this virtual ring and let $u$ add $v$ and $p$ to $u$’s neighbor set.

The greedy routing presented above will make each node forward Neighbor_discovery to its neighbor that has the shortest circular distance to the destination location $x_{i}^{u}$. When a node $v$ cannot find a neighbor that is closer to $x_{i}^{u}$ than $v$ itself, $v$ must be the node that has the shortest circular distance to $x_{i}^{u}$ among all nodes in FedLay (will be formally proved later). Hence, $v$ and one of its adjacent nodes $p$ will be $u$’s neighbors.

We show an example of FedLay $\mathtt{join}$ in Fig. 5. $u$ joins FedLay and it knows an existing node $H$ in the network. $u$ computes a random coordinate 0.15 and asks $H$ to run greedy routing of Neighbor_discovery to 0.15. $H$ will forward Neighbor_discovery to $B$, which is closest to 0.15 in the space among all $H$’s neighbors. Eventually, the message arrives at $G$, the node that is closest to 0.15 in the space and $G$ tells $u$ to add $G$ and $D$ as neighbors. Note greedy routing has much smaller hop-count than traveling nodes one after another through the ring because there are many shortcuts like the link $HB$.

We have the following property.

Note that every adjacent node of $v$ is its neighbor in a FedLay network. This lemma tells that if $v$ is not the node that is closest to the destination coordinate $x_{i}^{u}$, then the greedy routing algorithm must execute Step 2 and forwards the message to a neighbor. Hence, if the greedy routing algorithm goes to Step 3 and stops at $v$, $v$ must be the node that has the smallest circular distance to $x_{i}^{u}$ in the space. So we have,

The Neighbor_discovery message will stop at the node $v$ closest to $x_{i}^{u}$ and $v$ must be an adjacent node to the joining node $u$, because no other node is closer to $u$’s coordinate $x_{i}^{u}$. $v$ also knows the other adjacent node $w$ of $u$, because $w$ is a current adjacent node of $v$. $v$ will send a message to $u$ by TCP including the information of $v$ and $w$. $u$ will then add $v$ and $w$ to its neighbor set.

The joining node $u$ can find all its neighbors by running the above $\mathtt{join}$ protocol in all spaces. Therefore, if $u$ joins a correct FedLay network, the new FedLay network after this join is also correct.

In some extreme cases, there could be multiple nodes joining the network simultaneously. This situation will be handled by both the $\mathtt{join}$ and $\mathtt{maintenance}$ protocols.

The $\mathtt{leave}$ protocol of NDMP is quite straightforward. When a user wants to leave and closes the client program of FedLay, for every virtual space, the leaving node sends messages to its two adjacent nodes and tells them to add each other to their neighbor sets. As shown in Fig. 6, node $G$ wants to leave the network and tells its two adjacent nodes $A$ and $D$ about its leaving. Then $A$ and $D$ will consider each other as adjacent nodes and neighbors. In another space, $G$’s two adjacent nodes $F$ and $C$ will also add each other to their neighbor sets. Hence, the new FedLay network after a node leave is also correct.

In addition to planned leaves, An overlay network may also experience node failures due to various reasons such as an Internet service outage and end system failures. A failed node disappears without notice. In order to detect and fix these situations, each node in FedLay also runs the $\mathtt{maintenance}$ protocol.

The $\mathtt{maintenance}$ protocol requires every node to send each of its neighbors a heartbeat message periodically. Suppose the time period between two heartbeat messages is $T$. If a node $p$ has not received any heartbeat message from a neighbor $u$ for $3T$ time, it considers that $u$ has failed. $p$ then sends a Neighbor_repair message by greedy routing in the opposite direction of $u$ on the virtual space $i$ where $u$ and $p$ are adjacent nodes. We use an example to explain this protocol, as shown in Fig. 7. In the example of Fig. 7(a), node $A$ detects the failure of $G$. Since $G$ is an adjacent node of $A$ on $A$’s clockwise side, $A$ sends a Neighbor_repair message by greedy routing in the counterclockwise direction (the opposite direction of $G$). By “counterclockwise direction”, it requires that all hops of such greedy routing, $A-E-B-D$ in this example, should follow the counterclockwise order.

We give a formal description of the counterclockwise direction: Upon receiving Neighbor_repair to $x_{i}^{u}$ in Space $i$, a node $v$ runs the following algorithm:

1. 1.

   Node $v$ considers a subset of its neighbors, such that for every neighbor $w$ in the subset, $w$’s coordinate in Space $i$, $x_{i}^{w}$, satisfies $x_{i}^{w}<x_{i}^{v}$ or $x_{i}^{w}>x_{i}^{u}$. For each neighbor $w$, consider the length of the arc from $x_{i}^{w}$ to $x_{i}^{u}$ in the counterclockwise direction $L(\wideparen{x_{i}^{w},x_{i}^{u}})$. From the above subset, node $v$ finds a neighbor $w^{\prime}$, such that $w^{\prime}$’s coordinate in Space $i$, $x_{i}^{w^{\prime}}$, has the smallest arc length $L(\wideparen{x_{i}^{w^{\prime}},x_{i}^{u}})$ among all neighbors in the subset.

2. 2.

   If $L(\wideparen{x_{i}^{v},x_{i}^{u}})>L(\wideparen{x_{i}^{w^{\prime}},x_{i}^{u}})$, $v$ forwards Neighbor_repair to $w$.

3. 3.

   If $L(\wideparen{x_{i}^{v},x_{i}^{u}})<L(\wideparen{x_{i}^{w^{\prime}},x_{i}^{u}})$, Neighbor_repair stops at $v$. $v$ then tells $p$ that $v$ is $p$’s adjacent nodes on this virtual ring and let $p$ add $v$ to $p$’s neighbor set.

When Neighbor_repair stops, it arrives at another adjacent node of $G$ before $G$’s failure (stated in a later theorem). Then $G$’s previous two adjacent nodes can be connected.

In Fig. 7(a), node $D$ also detects $G$’s failure, independent of $A$’s detection. Then $D$ sends a Neighbor_repair message to the clockwise direction and the message will travel on a path $D-F-A$. The algorithm to forward Neighbor_repair in the clockwise direction can be specified in a similar way.

By changing the subset selection condition in Step 1 to “$x_{i}^{w}>x_{i}^{v}$ or $x_{i}^{w}<x_{i}^{u}$”. We can prove the following theorem:

Based on the theorem, in every virtual space, the two adjacent nodes of the failed $u$ can find and connect each other. Hence, if a node fails in a correct FedLay network, after the node failure FedLay is still correct.

Neighbor repair for concurrent joins and failures. Note the above property cannot be proved for multiple failures that happen at the same time, called concurrent failures. For concurrent joins and failures, we allow each node $u$ to periodically send two Neighbor_repair messages with destination $x_{i}^{u}$ to both counterclockwise and clockwise directions in every virtual space $i$, even without detecting any neighbor failure. For each node $v$ in $u$’s neighbor set, if they are indeed adjacent in a virtual space, $v$ will receive a Neighbor_repair message that stops at $v$. The Neighbor_repair stops at a node $w$ that is not in $u$’s neighbor set, $w$ and $u$ will add each other to their neighbor sets. In fact there is no way to prove any property for concurrent joins and failures in any structured P2P network [30, 17]. Hence we conduct experiments of extreme concurrent joins and failures and the above method always allows the network to recover to a correct FedLay.

Previous work assumes synchronous, round-based communication, in which all clients use the same time period to exchange models with neighbors [11, 10, 3, 34, 8, 13]. However, due to client heterogeneity, some low-resource clients may become “stragglers” that will fail to perform model exchanges in the given time period, while powerful or newly joined clients prefer shorter time period (or higher frequency) of model exchanges.

MEP uses asynchronous communication and allows each client $u$ to use a different communication time period $T_{u}$. $T_{u}$ can be set in two ways: 1) Coarse-grained settings. Each client may configure a period based on their device and communication types, for example, $\mathtt{Server}$-$\mathtt{LAN}$, $\mathtt{PC}$-$\mathtt{LAN}$, $\mathtt{Laptop}$-$\mathtt{WLAN}$, $\mathtt{Phone}$-$\mathtt{LTE}$, and $\mathtt{IoT}$-$\mathtt{WLAN}$. These values are pre-specified in the client program. 2) Fine-grained settings. Based on the monitoring of available bandwidth and computing resources, client $u$ estimates the minimum time duration $T_{u,{\min}}$ to produce an updated ML model and transmit it to all neighbors. Then its communication period $T_{u}=\eta T_{u,{\min}}$ for constant $\eta>1$.

For two neighbors with periods $T_{u}$ and $T_{v}$, their model exchange period is set to $\max(T_{u},T_{v})$. Hence, a client may have different exchange periods to different neighbors.

One key innovation in MEP is to introduce confidence parameters. Each node has a set of confidence parameters that present its self-evaluation of the local model accuracy.

We define the data divergence confidence $c_{d}$ on a client $u$:

where $DKL(\cdot)$ is the Kullback-Leibler divergence [16] to evaluate the statistical distance between two probability distributions $P$ and $Q$, $D_{loc}$ denotes the local data distribution, and $D_{std}$ denotes the estimated iid distribution of the dataset. The uniform distribution is widely used [42, 28] to estimate the iid data because the majority of publicly-available datasets for classification follow uniform distributions, such as (MNIST [18] and CIFAR-10 [15]). The Kullback-Leibler divergence can effectively represent the richness of a local dataset. $c_{d}\in(0,1]$ and a higher value represents a higher quality of local data and local models.

In addition, we define the communication confidence $c_{c}$ on a client $u$: $c_{c}^{u}=\frac{1}{T_{u}}$. The intuition of using $c_{c}$ is that when a client has more frequent model exchanges with its neighbors, its models are more likely to have higher qualities.

Hence, the overall confidence of client $u$ is

where $\max(c_{d})$ and $\max(c_{c})$ are the maximum values of $c_{d}$ and $c_{c}$ respectively, from all $u$’s neighbors. $\alpha_{d}$ and $\alpha_{c}$ are two constants to balance the weights of the two confidence parameters. The specific values of $\alpha_{d}$ and $\alpha_{c}$ can just be 0.5 and 0.5. We try a variety of combinations of $\alpha_{d}$ and $\alpha_{c}$, and in all cases, FedLay achieves fast model convergence on different nodes.

The models from $u$’s neighbors are aggregated as follows:

The above aggregation will be computed once every local time period $T_{u}$ and the models from each neighbor are always the most updated ones from the neighbor. In this way, clients with low confidence in their model accuracy will have less impact on other clients. In this work, we do not consider the situation where a client might intentionally set a large confidence value to mislead other clients.

In the neighbor set of each client, other than the IP addresses, coordinates, and confidence values of the neighbors, the client also stores the fingerprint $f$ of the most recent models of each neighbor, computed by hashing the models by a public hash function. Before starting a model exchange, the client first sends the fingerprint to its neighbor. If the neighbor finds the fingerprint matches the models that are going to be sent, the neighbor will consider the models to be a duplicate of a copy sent earlier and stop sending the models. This method reduces unnecessary traffic of exchanges of duplicated models.

We conduct three types of evaluation of FedLay for different scales of DFL networks.

1) Real experiments. We conduct experiments with real packet exchanges and data training. We deploy 16 instances to public clouds (we used both Oracle OCI and Amazon EC2), each with a 2GHz CPU and 2GB RAM. Each instance is connected to the Internet and runs a FedLay client. Each client sends and receives NDMP and MEP messages using TCP. Clients train ML models on their local datasets with Pytorch [25] and exchange the models with active neighbors. There is no central server for any purpose, and the system is completely decentralized. The purpose of this type of experiment is to present a prototype and demonstrate that FedLay can run with real ML data training in practice.

2) Medium-scale emulation with real data training. In this type of experiment, we use real data training and simulated packet exchanges as discrete events to evaluate networks with up to 100 clients. The simulation and real data training and testing run on a machine with an NVIDIA GeForce RTX3080 graphic card for training acceleration. The purpose of this type of experiment is to evaluate the overlay construction/maintenance, model accuracy, convergence speed, and message cost of FedLay and other DFL methods.

3) Large-scale simulation with trained models. For more than 100 clients, conducting all data training on a few machines takes very long time. For networks with 200 to 1000 nodes, we re-use the models trained from the above two types of experiments and assign them to the simulated clients. Packet exchanges are simulated as discrete events.

We evaluate the performance of FedLay for three ML tasks, including 1) Multilayer Perceptron (MLP) for digit classification on the MNIST dataset [9]. 2) Convolutional Neural Networks (CNN) for image classification on the CIFAR-10 dataset [15]. 3) Long Short-Term Memory (LSTM) for role forecasting on the Shakespeare dataset [6] built from The Complete Works of William Shakespeare. Details are described in Table II. All three are standard datasets for FL benchmarks [6].

Learning with non-iid data. We generate non-iid MNIST and CIFAR-10 datasets by selecting limited labels for the local training sets using the sharding method. Each shard contains only one label, and each local dataset includes a limited number of shards, resulting in a non-iid distribution and heterogeneity among clients’ local datasets. For large-scale simulations, data among clients may overlap, but in real experiments and medium-scale simulations, clients have unique data. In the original Shakespeare dataset, each speaking role in a play is considered a unique shard.

Client heterogeneity. We also assumed that clients have different computation and communication resources. We set 3 tiers of clients. For the 16-client real-world experiments, we set 10 medium-capacity, 3 high-capacity, and 3 low-capacity clients. For simulations, each experiment includes 60% medium-, 20% high-, and 20% low-capacity clients. The training time and communication time period of a high-capacity client are 2/3 of those of a medium-capacity user, and those of a low-capacity client are 2x of those of a medium-capacity user. The communication period for medium-capacity clients of each dataset is shown in Table II.

Besides the topology metrics discussed in Sec. II-B, we use the following metrics to evaluate FedLay and other methods.

Model accuracy: We evaluate the individual accuracy and average accuracy of local ML models based on separate test datasets that are different from the training datasets.

Topology correctness: It is defined as the number of correct neighbors of all nodes over the total number of neighbors. Hence correctness equal to 1 means a correct FedLay.

Communication Cost: We evaluate the average communication cost per client, by counting the number of NDMP messages sent by each client and the total size of the models sent by each client in bytes.

There is no existing DFL topology that allows decentralized construction and maintenance. Hence we compare FedLay with the following methods: 1) Gaia [12] is an ML method for geo-distributed clouds and still uses central servers. Hence it is not DFL. It runs server-based ML in each region and lets servers from different regions connect as a complete graph. It includes no aggregation method to handle non-iid data. 2) DFL-DDS [31] is a DFL method without a fixed topology. Instead, it simulates mobile nodes in a road network and considers two geographically close nodes as neighbors. 3) Chord [30]. 4) FedAvg [23] is a standard centralized FL method. We use its accuracy as the upper bound of DFL model accuracy because the central server knows all models from the clients.