#  _AFAFed_ —Protocol analysis

Enzo Baccarelli<EMAIL_ADDRESS>Michele Scarpiniti
<EMAIL_ADDRESS>Alireza Momenzadeh
<EMAIL_ADDRESS>Sima Sarv Ahrabi<EMAIL_ADDRESS>
###### Abstract

In this paper, we design, analyze the convergence properties and address the
implementation aspects of AFAFed. This is a novel Asynchronous Fair Adaptive
Federated learning framework for stream-oriented IoT application environments,
which are featured by time-varying operating conditions, heterogeneous
resource-limited devices (i.e., coworkers), non-i.i.d. local training data and
unreliable communication links. The key new of AFAFed is the synergic co-
design of: (i) two sets of adaptively tuned tolerance thresholds and fairness
coefficients at the coworkers and central server, respectively; and, (ii) a
distributed adaptive mechanism, which allows each coworker to adaptively tune
own communication rate. The convergence properties of AFAFed under (possibly)
non-convex loss functions is guaranteed by a set of new analytical bounds,
which formally unveil the impact on the resulting AFAFed convergence rate of a
number of Federated Learning (FL) parameters, like, first and second moments
of the per-coworker number of consecutive model updates, data skewness,
communication packet-loss probability, and maximum/minimum values of the
(adaptively tuned) mixing coefficient used for model aggregation.

###### keywords:

Adaptive asynchronous FL, heterogeneous Fog computing ecosystems, stream-
oriented distributed ML, non-i.i.d. data, personalization-vs.-generalization
trade-off.

.sp

##  1 Motivations and goals

The incoming booming phase of the Internet of Things (IoT) era is
characterized by an ever more increasing massive utilization of ever-smaller
resource-constrained devices for the personalized and distributed support of
ever more complex Machine Learning (ML)-aided stream wireless applications [1,
2]. Since IoT devices operate mainly at the network edge and connect to the
backbone Internet through bandwidth-limited and failure-prone wireless
(possibly, mobile) access links, the questions of where and how to process the
gathered IoT big data are still open challenges [2].

About where to perform the processing, in the state-of-the-art cloud-centric
approach, streams of data gathered by IoT devices at the network edge are
uploaded to a remote centralized cloud-data center by exploiting the long-
range Internet backbone [2]. However, we note that: (i) due to ever more
stringent user privacy concerns, the recent worldwide legislation limits data
storage only to what is user-consented and strictly necessary for processing
[3]; (ii) the centralized cloud-centric approach suffers from large and
unpredictable network latencies [1]; and, (iii) massive data uploaded to the
remote cloud cause severe traffic congestion in the Internet backbone [4].
Hence, being the sources of IoT data remotely located outside the cloud, Fog
Computing (FC) [5] is emerging as a practical paradigm for the distributed
support of ML applications, in which IoT devices and local servers are
networked [6], to bring data processing closer to where data is produced [7].

Passing to consider how data processing should be performed, we remark that,
in order to guarantee data privacy while simultaneously enforce distributed
training of complex ML models by resource-limited IoT devices, a decentralized
paradigm for ML training called Federated Learning (FL) has been quite
recently introduced [8]. In FC-supported FL systems, IoT devices (hereafter
referred to as coworkers) employ own local data to cooperatively train an ML
model shared with a central server hosted by a proximal Fog node. For this
purpose, from time to time, coworkers upload their updated local ML models to
the Fog-hosted central server for aggregation. The updating-aggregation steps
are repeated in multiple rounds until a target accuracy of the trained ML
model is achieved [8].

Overall, the FL paradigm enables the privacy-preserving distributed training
of complex ML models on resource-limited coworkers by operating only at the
edge network [9, 10, 11]. However, several open issues must be still
adequately addressed before implementing FL at scale in IoT stream realms,
namely [12]: (i) unreliable communication; (ii) heterogeneous IoT devices;
(iii) non-independent and identically distributed (non-i.i.d.) local data
sets; (iv) fairness-related issues; (v) synchronization issues; and, (vi)
time-varying operating environments. The above issues are inter-related,
mainly because: (a) under device heterogeneity, an asynchronous approach to
the FL could be appealing, in order to speed-up global convergence and avoid
the slowdown induced by the slower coworkers (generally referred to as
stragglers) [13]. However, under non-i.i.d. local data sets, the resulting
global FL solution attained at the convergence could be unfair, i.e., it could
be too biased toward the local solutions of the faster coworkers. Then,
suitable fairness mechanisms should be designed for coping with the
asynchronous access of the faster coworkers to the central server; (b) in
order to capitalize the faster convergence promoted by the asynchronous FL
paradigm, each coworker should locally process fresh data without experiencing
stalling phenomena. However, under stream applications, the arrival times of
new data at the coworkers are random and (typically) out-of-control [12]. This
requires, in turn, the design of suitable pro-active local data management
policies, which account for the asynchronous behaviour of the coworkers and
allow them to not stall their data processing; and, (c) under wireless realms,
coworker-to-central server links are typically unreliable and heterogeneous,
i.e., they exhibit non-vanishing packet loss rates that may vary from coworker
to coworker [2]. This may be a potential additional source of unfairness,
because the global FL solution could be biased towards the local solutions of
the coworkers that experience more reliable links. Hence, it demands, in turn,
for the joint design of transmission scheduling policies at the coworkers and
fairness-aware aggregation mechanisms at the central server that account also
for link unreliability.

Overall, since the afforded challenges are inter-related, a unified formal
framework is required for capturing their inter-dependency and, then, a joint
approach should be developed for their integrated solution.

#### Reference scenario

Figure 1: The considered AFAFed reference scenario. Continuous (dashed) arrows
indicate reliable (unreliable) communication links. CW:= CoWorker; K:= Number
of involved coworkers; AS:= Arrival Stream; LM:= Local ML Model; CS:= Central
Server; GM:= Global ML Model; UL:= Up-Link; DL:= Down-Link.

Motivated by these considerations, we consider the reference scenario of Fig.
1, in which:

  1. 1.

the trained ML model is general, i.e. it may range from a simple Linear
Regressor (LR), a binary Support Vector Machine (SVM) to a (complex) multi-
class Deep Neural Network (DNN) [14]. Furthermore, depending on the underlying
ML model, the performed training may be supervised/unsupervised or a mix of
them [14]. The target loss function to minimize through the FL may be non-
convex;

  2. 2.

the involved coworkers may be equipped with heterogeneous
computing/communication/storage local resources, and their local training data
may be non-i.i.d. Furthermore, the arrivals of new data at the coworkers of
Fig. 1 may be non-synchronized, and they may happen at random times;

  3. 3.

coworkers perform multiple consecutive local model updates before sending
their own current local models to the central server in an asynchronous way.
The number of the performed consecutive local updates may be time-varying and
different from coworker to coworker;

  4. 4.

coworker-to-server uplinks in Fig. 1 may be unreliable (i.e., affected by
communication delays, packet-losses, and/or intermittent failure phenomena);
and,

  5. 5.

after receiving an updated model update from a coworker, the central server
aggregates it into the global model without waiting for additional updates
from other coworkers and, then, returns the updated global model to the
sending coworker. The age (i.e., staleness) of each local updating received by
the central server is modeled as a random variable (r.v.) with unbounded
support.

#### Paper contributions

By referring to the depicted scenario, the focus of this paper is on the: (a)
optimized algorithmic design; (b) convergence analysis; and, (c) protocol
design of AFAFed—an adaptive asynchronous scheme for performing FL in stream
realms that are supported by networked FC technological platforms. We expect
that the main key news that characterize the AFAFed proposal are as follows:

  1. 1.

in order to cope with the (aforementioned) non-i.i.d. nature of the local
training data, AFAFed implements a distributed mechanism, which is inspired by
the formal tool of the so called Imperfect Consensus [15]. Its main new is
that it is based on the adaptive setting of some suitable tolerance thresholds
and allows each coworker to develop a personalized local model that accounts
for the statistical heterogeneity of the available local training data. For
this purpose, AFAFed implements also a novel distributed mechanism for the
adaptive tuning of the number of the per-coworker inter-aggregation local
model updates, which aims to reduce the per-coworker transmission rates (i.e.,
save bandwidth) while accounting for the (randomly distributed) arrivals of
new data. At this regard, in order to cope with the (typically, unpredictable)
random fluctuations of the inter-arrival times of the data streams of Fig. 1,
the AFAFed coworkers implement an (ad-hoc designed) distributed and adaptive
policy for the management of their finite-size buffers, which allows the
coworkers to process data at their maximum processing speeds, so to speed up
the resulting AFAFed convergence time;

  2. 2.

in order to cope with the resource-heterogeneity of the coworkers and their
related fairness issues, AFAFed implements a novel mechanism for the adaptive
tuning of some suitably defined fairness coefficients. The goal is to build up
a fair global model at the central server that treats all the participating
coworkers in an egalitarian way. For this purpose, AFAFed implements a novel
adaptive age and fairness-aware asynchronous aggregation mechanism at the
central server, in which each received local model is aggregated by weighting
it by an adaptively tuned coefficient that jointly accounts for its age and
fairness level; and, finally,

  3. 3.

we derive a set of analytical bounds, which guarantee the asymptotic
convergence (in expectation) of the global model built up by the central
server to a neighborhood of a stationary point of the underlying (possibly,
nonconvex) loss function. Since the numerical evaluations of the proposed
bounds rely on some system parameters that could be _a priori_ unknown, we
also design a companion protocol for their online profiling.

#### Adopted notation

We point out that $\vv{w}$ indicates a column vector, $|\mathcal{S}|$ is the
size (i.e., the cardinality) of the set $\mathcal{S}$,
$\left\lceil\cdot\right\rceil$ (resp., $\left\lfloor\cdot\right\rfloor$) is
the ceil (resp., floor) function, while $\triangleq$ means “equal by
definition”. Furthermore, $\xi\sim\mathcal{D}$ indicates that the r.v. $\xi$
follows the probability distribution $\mathcal{D}$, and
$E_{\mathcal{D}}\left\\{\cdot\right\\}$ (resp.,
$E_{X|Y}\left\\{\cdot\right\\}$) denotes the expectation with respect to
(w.r.t.) the probability distribution $\mathcal{D}$ (resp., w.r.t. the
distribution of the r.v. $X$ conditioned to $Y$), while
$E\left\\{\cdot\right\\}$ means joint expectation w.r.t. all the r.v.’s
involved.

##  2 Asynchronous FL: Preliminaries and Reference Models

The main goal of this section is to formally introduce the AFAFed learning
problem and the adopted solving approach. The main taxonomy of the paper and
the definition of the main adopted notation is recapped in Appendix A.

###  2.1 Local and global loss functions

ML models for statistical inference require that a set of model parameter
$\vv{w}\in\mathbb{R}^{l}$ is learnt (possibly online) on the basis of
(possibly time varying streams of distributed) training data. Hence, let:

$S_{k}\triangleq\left\\{\left(\vv{x}_{i}\in\mathbb{R}^{d},\,\vv{y}_{i}\in\mathbb{R}^{\mathcal{C}},\;1\leq
i\leq\left|S_{k}\right|\right)\right\\},\quad k=1,\ldots,K,$ (1)

be the set of (possibly time-varying) training data available at the $k$-th
coworker (i.e. $coworker\\#k$) of Fig. 1. Under supervised training [14], each
data of the set in Eq. (1) is composed of: (i) a feature vector
$\vv{x}_{i}\in\mathbb{R}^{d}$, which acts as input of the ML model to be
trained; and (ii) a vector $\vv{y}_{i}\in\mathbb{R}^{\mathcal{C}}$, which
represents the $\mathcal{C}$-ary ground-truth label associated to the feature
$\vv{x}_{i}$. After assuming $S_{k}\cap S_{j}=\emptyset$ for $k\neq j$, in FL
stream application scenarios, the elements of $S_{k}$ may vary over the time
but they may be accessed only by $coworker\\#k$. Furthermore, in stationary
non-i.i.d. environments, the probability distribution $\mathcal{D}_{k}$,
$k=1,\ldots,K$, of the training data set $S_{k}$ is time invariant, but it may
depend on the coworker index $k$.

After indicating by $\vv{w}_{k}\in\mathbb{R}^{l}$, $k=1,\ldots,K$, the weight
vector of the ML model estimated by $coworker\\#k$, let:
$\mathcal{T}:\mathbb{R}^{d}\times\mathbb{R}^{l}\rightarrow\mathbb{R}^{\mathcal{C}}$
be the $\vv{w}_{k}$-dependent input-output transformation implemented by the
considered ML model. Hence, by design, the label
$\widehat{\vv{y}}_{i}\in\mathbb{R}^{\mathcal{C}}$ generated by the ML model
when its input is $\vv{x}_{i}\in S_{k}$ is formally defined as follows [14]:
$\widehat{\vv{y}}_{i}\triangleq\mathcal{T}\left(\vv{x}_{i},\vv{w}_{k}\right)$.
The deviation of the estimated label from the ground-truth one is measured by
an assigned (possibly non-convex but differentiable) scalar real-valued loss
function, formally defined as follows [14]:
$\ell\left(\vv{y}_{i},\widehat{\vv{y}}_{i},\vv{w}_{k}\right):\mathbb{R}^{\mathcal{C}}\times\mathbb{R}^{\mathcal{C}}\times\mathbb{R}^{l}\rightarrow\mathbb{R}$.
In the case of ML models that require unsupervised learning (for example the
K-means model), the corresponding loss function does not longer depend on the
ground-truth label $\vv{y}_{i}$. Some examples of convex/non-convex loss
functions for supervised/unsupervised training are reported, for example, in
Table I of [16].

Since the probability distribution $\mathcal{D}_{k}$ is typically a priori
unknown, in practical application scenarios, the expectation:
$E_{\left(\vv{x}_{i},\vv{y}_{i}\right)\sim\mathcal{D}_{k}}\left(\ell\left(\vv{y}_{i},\widehat{\vv{y}}_{i},\vv{w}_{k}\right)\right)$
of the loss function w.r.t. $\mathcal{D}_{k}$ is replaced by the corresponding
$k$-th empirical risk [14]:

$F_{k}\\!\left(\vv{w}_{k}\right)\triangleq\frac{1}{\left|S_{k}\right|}\sum_{\left(\vv{x}_{i},\vv{y}_{i}\right)\in
S_{k}}\ell\left(\vv{y}_{i},\widehat{\vv{y}}_{i},\vv{w}_{k}\right),$ (2)

whose gradient w.r.t the weight vector reads as in:

$\vv{\nabla}F_{k}\\!\left(\vv{w}_{k}\right)\triangleq\frac{1}{\left|S_{k}\right|}\sum_{\left(\vv{x}_{i},\vv{y}_{i}\right)\in
S_{k}}\vv{\nabla}_{\vv{w}}\ell\left(\vv{y}_{i},\widehat{\vv{y}}_{i},\vv{w}_{k}\right).$
(3)

Hence, according to the (aforementioned) paradigm of the Agnostic FL [17], the
global empirical risk $F\left(\vv{w}\right)$ over all data sets of Eq. (1) is:

$F\left(\vv{w}\right)\triangleq\sum_{k=1}^{K}\lambda_{k}F_{k}\left(\vv{w}\right),$
(4)

where: $\left\\{\lambda_{k}\in\left[0,\,1\right],\,k=1,\ldots,K\right\\}$ is a
set of fairness coefficients which sum up to the unit. These coefficients play
the role of (possibly, a priori unknown and/or time-varying) hyper-parameters,
which must be suitably optimized, in order to guarantee the attainment of some
target fairness level among the involved coworkers [17]. According to the
definition in Eq. (4), the resulting Jain’s Fairness Index ($FI$) for
measuring the inter-coworker fairness reads as in:

$FI=\frac{\left(\sum_{k=1}^{K}\lambda_{k}\right)^{2}}{K\left(\sum_{k=1}^{K}\lambda_{k}^{2}\right)}.$

In our setting, it ranges from $1/K$ (most unfair case) to the unit (perfect
fairness), and it is maximum when all $\lambda$’s coefficients share a same
value.

###  2.2 Learning problem formulation and model personalization

The goal of the learning problem is to minimize the global risk function in
Eq. (4), i.e., to compute:

$\vv{w}^{\ast}\triangleq\operatorname*{arg\,min}_{\vv{w}}\left\\{F\left(\vv{w}\right)\right\\},$
(5)

so that,

$F^{\ast}\triangleq
F\left(\vv{w}^{\ast}\right)\equiv\min_{\vv{w}}\left\\{F\left(\vv{w}\right)\right\\}>-\infty,$
(6)

is the global minimum of $F(\cdot)$. The key challenge of carrying out the
above minimization is that, from Eqs. (2) and (3), it follows that the
computation of $F(\cdot)$ in Eq. (4) and its gradient:

$\vv{\nabla}F\\!\left(\vv{w}\right)\triangleq\sum_{k=1}^{K}\lambda_{k}\vv{\nabla}F_{k}\\!\left(\vv{w}\right),$
(7)

would require that the full set $S\triangleq\cup_{k=1}^{K}S_{k}$ of training
data is available at the central server of Fig. 1. Since this requirement
contrasts, indeed, with the privacy preserving nature of the FL paradigm, we
will pursue a solving approach that is inspired by the so-called Imperfect
Consensus paradigm for the distributed optimization of global objective
function under proximity local constraints [18]. Shortly, we point that
Imperfect Consensus is a generalization of the original Consensus paradigm
[19]. This last one is a formal technique for turning an additive global
objective function which does not split due to the presence of a global
variable shared by the involved additive terms into a summation of separable
local objectives, which are amenable to distributed local optimization.
However, when the local training sets in Eq. (1) are non-i.i.d., enforcing
“perfect” consensus among the local models evaluated by the coworkers may
degrade the resulting local (i.e., per-coworker) prediction accuracy [18].
This is the reason for introducing “soft” local proximity constraints, which
enforce “imperfect” consensus by allowing local models to be close but not
necessary equal.

Motivated by these considerations, we formally introduce a set of per-coworker
local variables:
$\Big{\\{}\vv{w}_{k}\in\mathbb{R}^{l},\,k=1,\ldots,K\Big{\\}}$ and a single
global variable $\vv{\overline{w}}\in\mathbb{R}^{l}$ tied to the central
server. Afterwards, we replace the centralized minimization problem in Eq. (5)
by the following distributed one:

$\displaystyle\underset{\mbox{\normalsize{$\vv{w}_{1},\ldots,\vv{w}_{K},\vv{\overline{w}}$}}}{\mbox{\large{$\min$}}}\>\;\sum_{k=1}^{K}\lambda_{k}F_{k}\\!\left(\vv{w}_{k}\right)\,,$
(8a)
$\displaystyle\text{s.t.:}\quad\left\|\vv{w}_{k}-\vv{\overline{w}}\right\|^{2}\leq
B_{k}\,,\hphantom{======}$ (8b)

where $\left\\{B_{k},\,k=1,\ldots,K\right\\}$ are non-negative scalar-valued
tolerance thresholds. This last formulation is referred to as the Imperfect
Consensus form of the learning problem in Eq. (5), since the set of
constraints in Eq. (8b) forces all the local variables
$\left\\{\vv{w}_{k}\right\\}$ to be close enough to a same global variable
$\vv{\overline{w}}$. Perfect consensus is attained for vanishing values of the
tolerance thresholds at the r.h.s.’s of Eq. (8b).

Remark 1 – Model Personalization and adaptive per-coworker ball constraints. A
key reason for adopting the Imperfect Consensus-based formulation in Eqs. (8a)
and (8b) is that, under non-i.i.d. FL scenarios, the sets of local proximity
constraints in Eq. (8b) with non-vanishing tolerance thresholds
$\left\\{B_{k}\right\\}$ may be used for personalizing the models
$\left\\{\vv{w}_{k}\right\\}$ locally evaluated by coworkers. Several
algorithms have been recently proposed to mitigate data heterogeneity by
adding a local regularization term to the co-worker’s objective functions, and
popular examples of such algorithms include FedProx in [20] and FedAsync in
[21], to name a few. However, we argue that it would be worth to explore
algorithms that add adaptive ball constraints when minimizing local
objectives, in order to adaptively control the variability of local models and
eventually attempt to align the steady-state values of global and local models
when the radius $\left\\{B_{k}\right\\}$ of the ball constraints vanish in the
limit. In this regard, we anticipate that a peculiar and new feature of the
AFAFed proposal is that the radius $\left\\{B_{k}\right\\}$ of the ball
constraints in Eq. (8b) are not a priori fixed but they are adaptively tuned,
in order to allow the implemented FL scheme to attain a right trade-off among
the contrasting requirements of model personalization and model
generalization. ∎

Being the Imperfect Consensus learning problem in Eqs. (8a) and (8b) a
constrained optimization problem, its solution may be characterized by
resorting to the evaluation of the saddle-points of the corresponding
Lagrangian function. This last one is defined as in [22]:

$\mathcal{L}\equiv\mathcal{L}\\!\left(\left\\{\vv{w}_{k}\right\\},\vv{w}_{k}\left\\{\mu_{k}\right\\}\right)\triangleq\sum_{k=1}^{K}\lambda_{k}F_{k}\\!\left(\vv{w}_{k}\right)+\frac{\mu_{k}}{2}\left(\left\|\vv{w}_{k}-\vv{\overline{w}}\right\|^{2}-B_{k}\right),$
(9)

where: (i) $\left\\{\vv{w}_{k}\right\\}$ and $\vv{\overline{w}}$ play the role
of $(K+1)$ primal optimization variables; while, (ii)
$\left\\{\mu_{k}\right\\}$ are non-negative Lagrange multipliers associated to
the proximity constraints in Eq. (8b) (i.e., dual variables). Let:

$\left\\{\left\\{\vv{w}_{k}^{\ast}\right\\},\left\\{\mu_{k}^{\ast}\right\\},\vv{\overline{w}}^{\ast}\right\\},$
(10)

be the (possibly, not unique) optimal values of the primal and dual variables
which solve the constrained problem in Eq. (8). Hence, these optimal values
must be a (possibly, not unique) saddle-point of the Lagrangian function in
Eq. (9), that is, they must be a (possibly, not unique) solution of the
following max-min problem:

$\max_{\vv{\mu}\geq\vv{0}}\left\\{\min_{\vv{w}_{1},\ldots,\vv{w}_{K},\vv{\overline{w}}}\>\mathcal{L}\\!\left(\left\\{\vv{w}_{k}\right\\},\vv{w}_{k}\left\\{\mu_{k}\right\\}\right)\right\\}.$
(11)

This means, in turn, that the $\left((K+1)l+K\right)$-dimensional vector:
$\vv{\nabla}\mathcal{L}$ of the gradients of the Lagrangian function in Eq.
(9) performed w.r.t. the primal and dual variables must vanish at the optimum,
that is111In the case of non-convex global loss function $F(\cdot)$ in Eq.
(4), the set of solutions of Eq. (12) embraces local/global maxima,
local/global minima and/or saddle-points of $F(\cdot)$ [22].:

$\vv{\nabla}\mathcal{L}=\vv{0}_{(K+1)l+K}.$ (12)

According to Eq. (9), the partial gradients of the Lagrangian function
$\mathcal{L}(\cdot)$ w.r.t. the primal and dual variables read as follows:

$\displaystyle\vv{\nabla}_{\vv{w}_{k}}\mathcal{L}$
$\displaystyle=\lambda_{k}\vv{\nabla}F_{k}\\!\left(\vv{w}_{k}\right)+\mu_{k}\left(\vv{w}_{k}-\vv{\overline{w}}\right),\quad
k=1,\ldots,K,$ (13a) $\displaystyle\vv{\nabla}_{\vv{\overline{w}}}\mathcal{L}$
$\displaystyle=\sum_{k=1}^{K}\mu_{k}\left(\vv{\overline{w}}-\vv{w}_{k}\right),\quad
k=1,\ldots,K,$ (13b) and, $\displaystyle\vv{\nabla}_{\mu_{k}}\mathcal{L}$
$\displaystyle=\left(\left\|\vv{w}_{k}-\vv{\overline{w}}\right\|^{2}-B_{k}\right),\quad
k=1,\ldots,K.$ (13c)

###  2.3 Asynchronous distributed updating of local models

In the AFAFed framework, the solution of the distributed leaning problem in
Eqs. (8a) and (8b) is numerically evaluated by implementing a set of
distributed SGD iterations, where the coworkers of Fig. 1 proceed in a fully
asynchronous way. This means that each coworker and the central server are
equipped with own local clocks which run without any synchronization. Formally
speaking, the following assumption holds:

######  Assumption 1 (The AFAFed asynchronous setting).

We assume that:

  1. 1.

$coworker\\#k$ is equipped with a non-negative integer-valued local time index
$t^{(k)}$ that counts the number of performed local iterations, so that
$\left\\{\vv{w}_{k}\left(t^{(k)}\right),\,t^{(k)}=0,1,\ldots\right\\}$ is the
corresponding $k$-th sequence of local model weights;

  2. 2.

the central server is equipped with a global non-negative integer-valued time
index $t$ that counts the number of performed global updates, so that
$\left\\{\vv{w}(t),\,t=0,1,\ldots\right\\}$ is the corresponding sequence of
global model weights;

  3. 3.

all the introduced time indexes evolve in an un-coordinated way, i.e., there
is not present any master clock.

$\blacksquare$

When the volumes $\left\\{\left|S_{k}\right|\right\\}$ of local training data
are large, it may be computationally challenging to compute the full gradients
in Eq. (3). In such cases, SGD is typically used [14]. It employs the
gradients [14]:

$\vv{\nabla}\widetilde{F}_{k}\\!\left(\vv{w}_{k};t^{(k)}\right)\triangleq\frac{1}{\left|MB_{k}\right|}\sum_{\left(\vv{x}_{i},\vv{y}_{i}\right)\in
MB_{k}\left(t^{(k)}\right)}\vv{\nabla}_{\vv{w}}\ell\\!\left(\vv{y}_{i},\widehat{\vv{y}}_{i},\vv{w}_{k}\right),$
(14)

of the partial loss functions:

$\widetilde{F}_{k}\\!\left(\vv{w}_{k};t^{(k)}\right)\triangleq\frac{1}{\left|MB_{k}\right|}\sum_{\left(\vv{x}_{i},\vv{y}_{i}\right)\in
MB_{k}\left(t^{(k)}\right)}\ell\\!\left(\vv{y}_{i},\widehat{\vv{y}}_{i},\vv{w}_{k}\right),$
(15)

which are defined on the basis of time-varying randomly sampled mini-batches:
$MB_{k}\left(t^{(k)}\right)$, $k=1,\ldots,K$, of the local data. Hence, after
replacing the full gradient $\vv{\nabla}F_{k}\\!\left(\vv{w}_{k}\right)$ in
Eq. (13a) by its stochastic approximation
$\vv{\nabla}\widetilde{F}_{k}\\!\left(\vv{w}_{k};t^{(k)}\right)$ in Eq. (14),
the SGD iterations on the $k$-th primal and dual variables $\vv{w}_{k}$ and
$\mu_{k}$ carried out by $coworker\\#k$ at the local time $t^{(k)}$ read as
follows (see Eqs. (13a) and (13c)):

$\displaystyle\vv{w}_{k}\left(t^{(k)}+1\right)$
$\displaystyle=\vv{w}_{k}\left(t^{(k)}\right)-\eta_{k}^{(0)}\left(t^{(k)}\right)\left[\lambda_{k}^{(LAST)}\vv{\nabla}\widetilde{F}_{k}\\!\left(\vv{w}_{k}\left(t^{(k)}\right)\right)+\mu_{k}\left(t^{(k)}\right)\left(\vv{w}_{k}\left(t^{(k)}\right)-\vv{\overline{w}}_{k}^{(LAST)}\right)\right],$
(16a) and, $\displaystyle\mu_{k}\left(t^{(k)}+1\right)$
$\displaystyle=\left[\mu_{k}\left(t^{(k)}\right)+\eta_{k}^{(1)}\left(t^{(k)}\right)\left[\left\|\vv{w}_{k}\left(t^{(k)}\right)-\vv{\overline{w}}_{k}^{(LAST)}\right\|^{2}-B_{k}\right]\right]_{+},\quad
t^{(k)}=0,1,\ldots$ (16b)

In Eqs. (16a) and (16b), we have that: (i) $[z]_{+}$ means:
$\max\left\\{0,z\right\\}$; (ii) $\eta_{k}^{(0)}\left(t^{(k)}\right)$ and
$\eta_{k}^{(1)}\left(t^{(k)}\right)$ are (possibly, time-varying) scalar step-
sizes to be suitably designed for improving the convergence speed of the
iterations; (iii) $\lambda_{k}^{(LAST)}$ is the most recent value of the
$k$-th fairness coefficient $\lambda_{k}$ in Eq. (8b) communicated by the
central server to $coworker\\#k$; and, (iv) $\vv{\overline{w}}_{k}^{(LAST)}$
is the most recent instance of the global weight vector $\vv{\overline{w}}$
that $coworker\\#k$ received from the central server. In this regard, we point
out that, due to the asynchronous setting considered here (see Assumption 1),
$\vv{\overline{w}}_{k}^{(LAST)}$ may depend on the coworker index $k$, i.e.,
at a same wall clock time, different instances of $\vv{\overline{w}}^{(LAST)}$
may be locally available at different coworkers[23].

###  2.4 Unreliable uplink communication

In the scenario of Fig. 1, coworkers are equipped with resource-constrained
(typically, battery-powered) devices, while the central sever is hosted by a
fixed resource-rich Fog node. Hence, the coworker-to-central server uplinks
are the bottleneck, since they are typically more power and bandwidth
constrained than the corresponding central server-to-coworker downlinks [24].
According to this consideration, we introduce the following formal assumption:

######  Assumption 2 (AFAFed communication modeling).

By referring to Fig. 1, we assume that:

  1. 1.

the downlinks are ideal, i.e., delay and packet-loss free;

  2. 2.

the $k$-th uplink may be affected by both communication delays and packet-loss
phenomena, where:

     * •

communication delays are featured by an i.i.d random sequence:
$\left\\{T_{k}^{(COM)}\\!\left(t^{(k)}\right),\,t^{(k)}=0,1,\ldots\right\\}$,
where  $T_{k}^{(COM)}\\!\left(t^{(k)}\right)\triangleq
b_{k}/R_{k}\\!\left(t^{(k)}\right)$, with: (i) $b_{k}$ (bit) being the size of
the model updated by $coworker\\#k$; and, (ii) $R_{k}\\!\left(t^{(k)}\right)$
(bit/link-use) being the corresponding randomly time-varying uplink
transmission rate at $t^{(k)}$;

     * •

packet loss phenomena are modeled by an i.i.d. binary random sequence:
$\left\\{CH_{k}\\!\left(t^{(k)}\right)\in\left\\{0,\,1\right\\},t^{(k)}=0,1,\ldots\right\\}$,
where $CH_{k}\\!\left(t^{(k)}\right)$ is zero when a packet loss happens at
$t^{(k)}$.

$\blacksquare$

According to this assumption, we denote by:

$P_{k}^{(LOSS)}\triangleq Prob\left\\{CH_{k}=0\right\\}\equiv
1-E\left\\{CH_{k}\right\\},\quad k=1,\ldots,K,$ (17)

the resulting steady-state packet-loss probability of the $k$-th uplink of
Fig. 1.

###  2.5 Asynchronous model aggregation

In principle, $coworker\\#k$ could provide updates to the central server in
form of Lagrangian gradients or model vectors. Although these two alternatives
are equivalent under ideal communication, nevertheless, it has been
experienced that model-based updates are more robust than the corresponding
gradient-based ones against the impairing effects induced by communication
packet-loss phenomena [25, 26]. Hence, in order to formally characterize the
asynchronous access to the central server performed by the involved AFAFed
coworkers, let:
$\left\\{\xi_{k}(t)\in\left\\{0,\,1\right\\},\,k=1,\ldots,K\right\\}$ be the
set of binary r.v.’s so defined:

$\displaystyle\xi_{k}\left(t\right)\triangleq\begin{cases}1,&\text{if the
central server is updated by the $k$-th coworker at global time $t$},\\\
0,&\text{otherwise}\end{cases}$ (18a) where the constraint:
$\displaystyle\sum_{k=1}^{K}\xi_{k}(t)=1,$ (18b)

assures that, at each global index time $t$, the central server is updated by
a single coworker. Then, by leveraging the gradient expression in Eq. (13b)
and the definition in Eq. (18a), the gradient descent update of the global
model $\vv{\overline{w}}(\cdot)$ at the central server reads as in:

$\vv{\overline{w}}(t+1)=\vv{\overline{w}}(t)+\eta^{(2)}(t)\left[\sum_{k=1}^{K}\xi_{k}(t+1)\left(\vv{w}_{k}(t+1)-\vv{\overline{w}}(t)\right)\right],\quad
t=0,1,\ldots$ (19)

After exploiting the constraint in Eq. (18b), Eq. (19) can be rewritten, in
turn, in the following two equivalent forms:

$\displaystyle\vv{\overline{w}}(t+1)$
$\displaystyle=\left(1-\beta(t)\right)\vv{\overline{w}}(t)+\beta(t)\left[\sum_{k=1}^{K}\xi_{k}(t+1)\vv{w}_{k}(t+1)\right]$
(20a)
$\displaystyle\equiv\vv{\overline{w}}(t)-\beta(t)\,\widehat{\vv{G}}\\!\left(\vv{\overline{w}}(t);\left\\{\xi_{k}(t+1)\right\\}\right),$
(20b)

where:

$\widehat{\vv{G}}\\!\left(\vv{\overline{w}}(t);\left\\{\xi_{k}(t+1)\right\\}\right)\triangleq\left[\sum_{k=1}^{K}\xi_{k}(t+1)\left(\vv{\overline{w}}(t)-\vv{w}_{k}(t+1)\right)\right],$
(21)

plays the role of a global stochastic gradient, and:

$\beta(t)\equiv\beta\left(\left\\{\xi_{k}(t+1)\right\\}\right)\in[0,\,1],$
(22)

is a (possibly time varying) non-negative mixing parameter, which is limited
to the unit and may also depend on the current values of the binary access
r.v.’s in Eq. (18a). Interestingly enough, the companion relationships in Eqs.
(20a) and (20b) point out that the asynchronous model aggregation performed by
the central server may be viewed under the equivalent forms of moving average
or SGD iterates.

##  3 The AFAFed proposal–Algorithms and Protocols

The modeling carried out in Section 2 points out that, in the AFAFed
framework:

  1. 1.

the variables $\left\\{\vv{w}_{k}\right\\}$ and $\vv{\overline{w}}$ (resp.,
$\left\\{\mu_{k}\right\\}$) are the primal (resp., dual) optimization
variables of the learning problem in Eqs. (8a) and (8b);

  2. 2.

the fairness coefficients $\left\\{\lambda_{k}\right\\}$ play the role of
(possibly, time varying) hyper-parameters, which must be suitably designed, in
order to throttle the global model $\vv{\overline{w}}$ towards a suitable fair
solution;

  3. 3.

the (possibly time varying) tolerance thresholds $\left\\{B_{k}\right\\}$ in
Eq. (8b) are non-negative hyper-parameters, to be designed for attaining a
right personalization-vs.-generalization trade-off;

  4. 4.

the (possibly time varying) sets of step sizes
$\left\\{\eta_{k}^{(0)}\right\\}$ and $\left\\{\eta_{k}^{(1)}\right\\}$ in
Eqs. (16a) and (16b) must be suitably designed, to speed up the convergence of
the primal and dual iterates carried out by the coworkers; and,

  5. 5.

the (possibly time varying) mixing parameter $\beta\left(\cdot\right)$ in Eqs.
(20a) and (20b) is a hyper-parameter to be designed for performing fair and
age-aware model aggregation.

The above remarks fix the main directions pursued for the optimized design of
the AFAFed protocol.

###  3.1 Overall view of the AFAFed protocol

Algorithm 1 provides an overall view of the protocol that implements AFAFed.
It is composed of two main segments, i.e. a local segment to be run in
parallel by coworkers (see the rows of Algorithm 1 numbered in azure), and a
global segment to be run by the central server in a sequential way (see the
rows of Algorithm 1 numbered in green). These two protocol segments will be
described in the sequel. In Algorithm 1, gray-shadowed equation numbering
highlights the main algorithmic news of the AFAFed proposal.

Algorithm 1 — Overall view of the proposed AFAFed protocol

Input:

  * •

Number $T$ of global iterations to be carried out by the central server;

  * •

Number $K$ of the involved coworkers.

Output:

  * •

Set of the fairness coefficients $\left\\{\lambda_{k}(T)\right\\}$;

  * •

Set of the weights vectors $\left\\{\vv{w}_{k}(T)\right\\}$ of the local
models;

  * •

Weight vector $\vv{\overline{w}}(T)$ of the global model.

1:$\triangleright$ Initialization phase

2:Set $\vv{\overline{w}}$ and $\left\\{\vv{w}_{k}\right\\}$ to a same initial
value $\vv{w}_{0}$;

3:Set all $\lambda_{k}$’s to $1/K$.

4:$\triangleright$ Sequential iterative phase at the central server

5:for $t=0:\left(T-1\right)$ do

6:$\triangleright$ Parallel iterative phase at the coworkers

7: parfor $k=1:K$ do

8: $coworker\\#k$ runs $Iter_{k}$ primal/dual local iterations in Eqs. (16a)
and (16b) by updating the step-sizes as in Eqs. (27a) and (27b);

9: $coworker\\#k$ updates its local clock index as:
$t^{(k)}:=t^{(k)}+Iter_{k}$;

10: $coworker\\#k$ updates its tolerance threshold $B_{k}$ as in Eq. (29);

11: $coworker\\#k$ updates its averaged Lagrange multiplier
$\overline{\mu}_{k}$ as in Eq. (23);

12: $coworker\\#k$ sends the 3–ple:
$\left\\{\vv{w}_{k},\,\overline{\mu}_{k},\,timestamp_{k}\right\\}$ to the
central server and starts its $Timer_{k}$;

13: $coworker\\#k$ updates the number of local iterations $Iter_{k}$ to be
carried out in the next batch as in Eq. (25);

14: if ($Timer_{k}$ expired) & (no feedback is received from the central
server) then

15: go to Step#5;

16: end if$\triangleright$ End-if at Step#11

17: if a new arrival happens at the $coworker\\#k$ then

18: $coworker\\#k$ applies the buffer management policy of Section 5.2;

19: end if$\triangleright$ End-if at Step#14

20:

21: if central server receives an update from $coworker\\#k$ then

22: central server updates the global upper and lower fairness thresholds
$TH_{U}$ and $TH_{L}$ as in Eqs. (34) and (35);

23: central server updates the global up/down scaling factors $\Psi$ and $V$
as in Eq. (39);

24: if $\left(\overline{\mu}_{k}>TH_{U}\right)$ then

25: central server updates $\lambda_{k}$ as in Eq. (36) and normalizes the
overall set $\left\\{\lambda_{j}\right\\}$ as in Eq. (40);

26: else if $\left(\overline{\mu}_{k}<TH_{L}\right)$ then

27: central server updates $\lambda_{k}$ as in Eq. (37) and normalizes the
overall set $\left\\{\lambda_{j}\right\\}$ as in Eq. (40);

28: end if$\triangleright$ End-if at Step#20

29: central server updates the model $age\left(\cdot\right)$ and mixing
parameter $\beta\left(\cdot\right)$ as in Eqs. (41) and (44), respectively;

30: central server updates the global model weight vector $\vv{\overline{w}}$
as in Eq. (20a);

31: central server broadcasts the updated set
$\left\\{\lambda_{j}^{(NEW)},\,j=1,\ldots,K\right\\}$ of the fairness
coefficients to all coworkers;

32: central server sends a timestamped copy of the updated model
$\vv{\overline{w}}$ to $coworker\\#k$ and updates its global time index as:
$t:=t+1$;

33: end if$\triangleright$ End-if at Step#17

34:

35: if $coworker\\#k$ receives a new timestamped global model
$\vv{\overline{w}}$ from the central server then

36: $coworker\\#k$ updates its $timestamp_{k}$ and set:
$\vv{w}_{k}^{(LAST)}:=\vv{\overline{w}}$ in Eqs. (16a) and (16b);

37: $coworker\\#k$ restarts its primal/dual iterations in Eqs. (16a) and (16b)
by setting: $\vv{w}_{k}\left(t^{(k)}\right):=\vv{\overline{w}}$;

38: go to Step#5;

39: end if$\triangleright$ End-if at Step#30

40:

41: if $coworker\\#k$ receives an updated fairness coefficient
$\lambda_{k}^{(NEW)}$ from the central server then

42: $coworker\\#k$ sets $\lambda_{k}^{(LAST)}:=\lambda_{k}^{(NEW)}$ in Eq.
(16a);

43: end if$\triangleright$ End-if at Step#35

44:

45: end parfor$\triangleright$ End parallel-for at Step#4

46:end for$\triangleright$ End for at Step#3

47:

48:return $\left\\{\lambda_{k}(T)\right\\}$, $\left\\{\vv{w}_{k}(T)\right\\}$,
and $\vv{\overline{w}}(T)$.

###  3.2 AFAFed procedures at the co-workers

In the AFAFed framework, each $coworker\\#k$ maintains and updates a local
model vector $\vv{w}_{k}$ (see $line\\#5$ of Algorithm 1), while a main goal
of the central server is to update the corresponding global model vector
$\vv{\overline{w}}$ (see $line\\#26$ of Algorithm 1). All these local and
global models are initialized to a same value in the bootstrap phase (see
$line\\#1$ of Algorithm 1). The distributed nature of the AFAFed framework is
featured by the parallel-for cycle in $line\\#4$, which forces all the
embraced steps to be executed in parallel by the coworkers. At the same time,
its asynchronous nature stems from the fact that there is not a centralized
master clock for aligning the local and global time indexes in $line\\#6$ and
$line\\#28$ of Algorithm 1.

#### Distributed and adaptive tuning of the Lagrange multipliers

Unlike previous contributions [27, 21, 28, 29], a main (and, indeed, new)
feature of the AFAFed proposal is that each coworker is equipped with its own
Lagrange multiplier $\mu_{k}$, which is time-varying and is locally updated in
$line\\#5$ of Algorithm 1 according to the iterations of Eq. (16b). In the
AFAFed ecosystem, its role is, indeed, twofold. First, it provides an adaptive
and personalized (i.e., per-coworker) measure of the penalization to be
imposed to the deviation of the current local model $\vv{w}_{k}$ from the last
received global one: $\vv{\overline{w}}_{k}^{(LAST)}$ (see Eq. (16b)). Second,
it enforces imperfect consensus by adaptively throttling $\vv{w}_{k}$ to be
close to (but not necessarily coincident with) the last received version of
the global model $\vv{\overline{w}}_{k}^{(LAST)}$ (see Eq. (16a)). In order to
smooth the oscillatory phenomena typically induced by too abrupt environmental
changes, a second pursued design choice is to utilize the corresponding per-
coworker sample average:

$\overline{\mu}_{k}\left(t^{(k)}+1\right)\triangleq\frac{1}{t^{(k)}+2}\sum_{i=0}^{t^{(k)}+1}\mu_{k}(i),\quad
t^{(k)}=0,1,\ldots,$ (23)

for the adaptive tuning of the fairness coefficients. For this purpose,
$coworker\\#k$ locally updates the sample average in Eq. (23) at each local
time index $t^{(k)}$, and from time to time, communicates it to the central
server for further processing (see $line\\#8$ of Algorithm 1).

#### Distributed and adaptive tuning of the number of local iterations

Each $coworker\\#k$ may perform a cluster of $Iter_{k}\geq 1$ local iterations
between two consecutive communications to the central server. A new feature of
the AFAFed proposal is that the current value of the sample average in Eq.
(23) is used as scaling coefficient, in order to make both distributed and
adaptive the updating of $Iter_{k}$ in $line\\#10$ of Algorithm 1.
Intuitively, the pursued design principle is to increase (resp., decrease)
$Iter_{k}$ when the current value of $\overline{\mu}_{k}$ in Eq. (23) is low
(resp., high). The rationale is that, since $\overline{\mu}_{k}$ is an
(averaged) measure of the penalty to impose to the current deviation of the
local model $\vv{w}_{k}$ from the global one $\vv{\overline{w}}_{k}^{(LAST)}$,
it may be, indeed, reasonable to lower the communication coworker-to-central
server communication rate (that is, to increase the number of performed
consecutive local iterations) when the local model is sufficiently aligned
with the global one. In order to formally implement this design principle,
let:

$\Omega_{k}\left(\overline{\mu}_{k}\right):\mathbb{R}_{0}^{+}\longrightarrow[1,\,+\infty),$
(24)

be a (possibly, $k$-dependent) non-negative function of the averaged Lagrange
multiplier $\overline{\mu}_{k}$ in Eq. (23), which: (i) is not decreasing for
increasing $\overline{\mu}_{k}$; and, (ii) is unit valued in the origin.
Hence, we design to adaptively tune $Iter_{k}$ as follows:

$Iter_{k}\left(t^{(k)}+1\right)=\max\left\\{1;\,\left\lceil\frac{Iter_{k}^{(MAX)}}{\Omega_{k}\left(\overline{\mu}_{k}\left(t^{\left(k\right)}+1\right)\right)}\right\rceil\right\\},\quad
t^{(k)}=0,1,\ldots,$ (25)

where the positive integer-valued constant $Iter_{k}^{(MAX)}$ fixes the
maximum number of consecutive local iterations allowed $coworker\\#k$. Before
proceeding, two explicative remarks are in order. First, an eligible example
of $\Omega_{k}$-function is the following max-log one:

$\Omega_{k}\left(\overline{\mu}_{k}\right)=\max\left\\{1;\,\min\left\\{Iter_{k}^{(MAX)};\,a^{c\times\overline{\mu}_{k}}\right\\}\right\\},\quad\text{with}\;a>1,\;\text{and}\;c>0.$
(26)

Second, from a statistical point of view, $Iter_{k}$ in Eq. (25) plays the
role of an r.v., which assumes values over the integer-valued set:
$\left\\{1,2,\ldots,Iter_{k}^{(MAX)}\right\\}$. According to the relationship
in (25), the probability density function (pdf) of $Iter_{k}$ depends on the
pdf of the r.v. $\overline{\mu}_{k}$ in Eq. (23), which, in turn, may be a
priori unknown. In this regard, we anticipate that the issue of the online
profiling of the statistics of $Iter_{k}$ will be addressed in Section 5.

#### Distributed and adaptive tuning of step-sizes

The adaptive design of the $\eta$’s step-sizes in Eqs. (16a) and (16b) is
driven by the following two considerations. First, in order to allow both fast
responsiveness in the transient-state and stable behavior in the steady-state,
it is likelihood that $\eta_{k}^{(0)}$ and $\eta_{k}^{(1)}$ scale up with the
modules: $\left\|\vv{\nabla}\widetilde{F}_{k}\right\|$ and
$\left\|\left\|\vv{w}_{k}\\!\left(t^{(k)}\right)-\vv{\overline{w}}_{k}^{(LAST)}\right\|^{2}-B_{k}\\!\left(t^{(k)}\right)\right\|$
of the gradients of the involved Lagrangian function. This setting is
supported, indeed, by the formal results reported in Theorem 3.3 of [30] about
the convergence of primal/dual iterations under nonstationary environments.
Second, in order to attain a good personalization-vs.-generalization trade-
off, it is reasonable to allow the step-sizes in Eq. (16a) and (16b) to scale
up for increasing values of $\Omega_{k}$ in Eq. (24), so to increase the
personalization (resp., generalization) capability of $coworker\\#k$ when the
current value of the corresponding averaged Lagrange multiplier
$\overline{\mu}_{k}$ in Eq. (23) is large (resp., small). Overall, on the
basis of these considerations, we design the following two relationships for
the per-iteration adaptive tuning of the step-sizes in Eqs. (16a) and (16b):

$\displaystyle\eta_{k}^{(0)}\\!\left(t^{(k)}\right)$
$\displaystyle=\max\left\\{\eta_{MIN};\,\min\left\\{\eta_{MAX};\,\Omega_{k}\\!\left(t^{(k)}\right)\left\|\vv{\nabla}\widetilde{F}_{k}\\!\left(t^{(k)}\right)\right\|\right\\}\right\\},\quad
t^{(k)}=0,1,\ldots,$ (27a) and,
$\displaystyle\eta_{k}^{(1)}\\!\left(t^{(k)}\right)$
$\displaystyle=\max\left\\{\eta_{MIN};\,\min\left\\{\eta_{MAX};\,\Omega_{k}\\!\left(t^{(k)}\right)\left\|\left\|\vv{w}_{k}\\!\left(t^{(k)}\right)-\vv{\overline{w}}_{k}^{(LAST)}\right\|^{2}-B_{k}\\!\left(t^{(k)}\right)\right\|\right\\}\right\\},\quad
t^{(k)}=0,1,\ldots,$ (27b)

where $\eta_{MIN}$ (resp., $\eta_{MAX}$) is a non-negative lower (resp.,
upper) bound on the allowed step-size values, which acts as a regularization
term [14].

#### Distributed and adaptive tuning of the tolerance thresholds

It is expected that the personalization (resp., generalization) capability of
the local models $\left\\{\vv{w}_{k}\right\\}$ (resp., of the global model
$\vv{\overline{w}}$) increases for increasing (resp., decreasing) values of
$\left\\{B_{k}\right\\}$ in Eq. (8b). This requires, in turn, that the overall
set of these tolerance thresholds is jointly optimized, in order to attain a
good personalization-vs.-generalization trade-off. In this regard, a
remarkable (and, at the best of the authors’ knowledge, new) design feature of
AFAFed is that even the values of the tolerance thresholds are tuned in a
distributed and adaptive way. Specifically, the pursued design criterion is
allow $coworker\\#k$ to autonomously increase (resp., decrease) the current
value of $B_{k}\\!\left(t^{(k}\right)$ when the corresponding average Lagrange
multiplier $\overline{\mu}_{k}$ in Eq. (23) increases (resp., decreases), so
to enhance model personalization (resp., model generalization). For this
purpose, let:

$\Theta\\!\left(\overline{\mu}_{k}\right):\mathbb{R}_{0}^{+}\longrightarrow\mathbb{R}_{0}^{+},$
(28)

be a non-negative non-decreasing function of $\overline{\mu}_{k}$, which
vanishes in the origin. Hence, in $line\\#7$ of Algorithm 1, we design to
adapt $B_{k}$ as follows:

$B_{k}\\!\left(t^{(k)}\right)=\Theta\\!\left(\overline{\mu}_{k}\\!\left(t^{(k)}\right)\right),\quad
t^{(k)}=0,1,\ldots.$ (29)

Just as an example, we point out that an eligible choice for the
$\Theta(\cdot)$ function in Eq. (29) is the following one:

$\Theta\\!\left(\overline{\mu}_{k}\\!\left(t^{(k)}\right)\right)=B_{0}\times\left(\overline{\mu}_{k}\\!\left(t^{(k)}\right)\right)^{\gamma},$
(30)

where $B_{0}$ is a positive threshold reference value and $\gamma$ is a non-
negative power-shaping coefficient.

#### Distributed detection of communication failures and time-stamp management

Assumption 2 points out that, in the reference scenario of Fig. 1, packet-loss
events are random and (typically) unpredictable. Hence, in order to detect
them in a distributed way, $coworker\\#k$ starts its local $Timer_{k}$
whenever it sends a new model update to the central server (see $line\\#9$ of
Algorithm 1). By doing so, $coworker\\#k$ autonomously detects a packet-loss
event whenever $Timer_{k}$ expires and feedback has been not still received
from the central server (see $line\\#11$ of Algorithm 1). Interestingly
enough, the re-transmission policy implemented by AFAFed is of non-persistent
type [31]. This means that, after detecting a packet-loss, $coworker\\#k$
refrains to immediately attempt a retransmission of the loss packet, but it
starts a new cluster of local iterations (see the go to statement in
$line\\#12$ of Algorithm 1). This design choice aims to randomize the
transmission times of different coworkers, in order to avoid multiple
consecutive packet-loss events in communication-congested FL scenarios.

Finally, we anticipate that, in order to allow the central server to perform
model aggregation in an age-aware way (see Eq. (44)), $coworker\\#k$ manages a
local $timestamp_{k}$ variable for storing the value of the global time index
$t$ at which $coworker\\#k$ received the last global model
$\vv{\overline{w}}^{(LAST)}$. By doing so, after receiving a new global model
from the central server, $coworker\\#k$: (i) updates
$\vv{\overline{w}}^{(LAST)}$ and $timestamp_{k}$ (see $line\\#31$ of Algorithm
1); (ii) sets its local model $\vv{w}_{k}$ to $\vv{\overline{w}}^{(LAST)}$
(see $line\\#32$ of Algorithm 1); and, then: (iii) starts a new cluster of
local primal/dual iterations (see $line\\#33$ of Algorithm 1). After finishing
the current iteration cluster, $coworker\\#k$ communicates the last updated
3-ple: $\left\\{\vv{w}_{k},\,\overline{\mu}_{k},\,timestamp_{k}\right\\}$ to
the central server (see $line\\#9$ of Algorithm 1).

Overall, the resulting per-iteration computational effort to be sustained by
$coworker\\#k$ scales up as $\mathcal{O}(l+6)$, where $l$ is the size of the
vector update in Eq. (16a), while the additive term accounts for the scalar
updates in Eqs. (16b), (23), (26), (27a), (27b), and (30).

###  3.3 AFAFed procedures at the central server

The main task of the central server is to adaptively tune the set of fairness
coefficients $\left\\{\lambda_{k}\right\\}$ (see $lines\\#18-24$ of Algorithm
1), as well as the mixing parameter $\beta$ and the global model
$\vv{\overline{w}}$ (see $lines\\#25-26$ of Algorithm 1). In order to perform
this task in an age-aware way, let:

$ac(t+1):\left\\{1,2,\ldots,T\right\\}\longrightarrow\left\\{1,\ldots,K\right\\},$
(31)

be an auxiliary function, which is implemented at the central server and
returns the index of the coworker from which the central server receives the
model update at the global time index $(t+1)$, $t=0,1,\ldots,T$222From a
formal point of view, we use $(t+1)$ in place of $t$, because, in the AFAFed
framework, the central server is not updated at the bootstrapping instant
$t=0$..

#### Adaptive distributed tuning of fairness coefficients

Let us assume that, at the global time index $t+1$, the central server
receives a new updated 3-ple:
$\left\\{\vv{w}_{ac(t+1)},\,\overline{\mu}_{ac(t+1)},\,timestamp_{ac(t+1)}\right\\}$
from the $ac(t+1)$-th coworker. The pursued principle behind the proposed
adaptive design of the AFAFed fairness coefficients is to scale up (resp.,
scale down) the current value of $\lambda_{ac(t+1)}$ when the corresponding
value $\overline{\mu}_{ac(t+1)}$ of the averaged Lagrange multiplier in Eq.
(23) is larger (resp., smaller) than a suitable upper (resp., lower) fairness
threshold. The rationale is that too high (resp., too small) values of
$\overline{\mu}_{ac(t+1)}$ indicate that the $ac(t+1)$-th coworker is
receiving a too unfair (resp., too biased) service from the central server,
i.e. the current global model $\vv{\overline{w}}(t)$ is too far from (resp.,
too close to) the current local model $\vv{w}_{ac(t+1)}$ of the $ac(t+1)$-th
coworker. However, due to both dynamically time-varying stream nature of new
arrivals and coworker heterogeneity, the aforementioned upper and lower
fairness thresholds should retain the following two main properties. First, at
each $t$, they should measure the average level of fairness globally
experienced by the full set of coworkers. Second, these thresholds should be
dynamically tuned, in order to track the current fairness trend of the
coworkers.

In order to attain the first property, let us assume that the central server
implements and updates the following global ensemble average:

$\widetilde{\mu}(t+1)\triangleq\frac{1}{t+1}\sum_{\tau=1}^{t+1}\overline{\mu}_{ac(\tau)},\quad
t=0,1,\ldots,$ (32)

and the corresponding global ensemble standard deviation:

$\sigma_{\tilde{\mu}}(t+1)\triangleq\frac{1}{t+1}\sum_{\tau=1}^{t+1}\left|\,\widetilde{\mu}(\tau)-\overline{\mu}_{ac(\tau)}\right|,\quad
t=0,1,\ldots,$ (33)

of the received sequence of per-coworker average Lagrange multipliers in (23).
Hence, we propose that, at each $t$, the central server adaptively tunes the
upper: $TH_{U}(t)$ and lower: $TH_{L}(t)$ fairness thresholds as follows (see
$line\\#18$ of Algorithm 1):

$TH_{U}(t)\equiv\left|\,\widetilde{\mu}(t)+4.0\,\sigma_{\tilde{\mu}}(t)\right|,$
(34)

and

$TH_{L}(t)\equiv\left|\,\widetilde{\mu}(t)-4.0\,\sigma_{\tilde{\mu}}(t)\right|,$
(35)

where the terms: $\pm 4.0\,\sigma_{\tilde{\mu}}(t)$ act as safety margins.
Afterwards, as reported in $lines\\#20-24$ of Algorithm 1, the central server
scales up the current value of $\lambda_{ac(t+1)}$ as in:

$\lambda_{ac(t+1)}\times\Psi\\!\left(\overline{\mu}_{ac(t+1)},\widetilde{\mu}(t)\right),\quad\text{if
}\leavevmode\nobreak\ \overline{\mu}_{ac(t+1)}>TH_{U}(t),$ (36)

or scale down it as in:

$\lambda_{ac(t+1)}\times
V\\!\left(\overline{\mu}_{ac(t+1)},\widetilde{\mu}(t)\right),\quad\text{if
}\leavevmode\nobreak\ \overline{\mu}_{ac(t+1)}<TH_{L}(t),$ (37)

where:

$\Psi\\!\left(\overline{\mu},\widetilde{\mu}\right):\mathbb{R}_{0}^{+}\times\mathbb{R}_{0}^{+}\rightarrow[0,\,+\infty),\quad\text{and}\quad
V\\!\left(\overline{\mu},\widetilde{\mu}\right):\mathbb{R}_{0}^{+}\times\mathbb{R}_{0}^{+}\rightarrow(0,\,1],$
(38)

are two scaling functions, which meet the following defining properties:

  1. 1.

$\Psi\\!\left(\overline{\mu},\widetilde{\mu}\right)$ assumes values in
$[1,\,+\infty)$ (i.e., it is a scaling-up function). It does not decreases for
increasing values of $\left(\overline{\mu}-\widetilde{\mu}\right)\geq 0$;
while,

  2. 2.

$V\\!\left(\overline{\mu},\widetilde{\mu}\right)$ assumes values in $(0,\,1]$
(i.e., it is a scaling-down function). It does not increase for increasing
values of $\left(\overline{\mu}-\widetilde{\mu}\right)\geq 0$.

Examples of eligible up/down scaling functions are the following ones:

$\Psi\\!\left(\overline{\mu},\widetilde{\mu}\right)\equiv\frac{1}{V\\!\left(\overline{\mu},\widetilde{\mu}\right)}=1+\log\\!\left(1+\frac{\left|\overline{\mu}-\widetilde{\mu}\right|}{1+\widetilde{\mu}}\right).$
(39)

Therefore, after performing the scaling in Eqs. (36) and (37), the central
server re-normalizes the full set
$\Big{\\{}\lambda_{j},\,j=1,\ldots,K\Big{\\}}$ of the fairness coefficients to
be unit-sum as in (see $line\\#21$ and $line\\#23$ of Algorithm 1):

$\lambda_{j}^{(NEW)}=\frac{\lambda_{j}}{\sum_{m=1}^{K}\lambda_{m}},\quad
j=1,\ldots,K.$ (40)

Finally, the server broadcasts the new set of fairness coefficients
$\left\\{\lambda_{j}^{(NEW)}\right\\}$ to all coworkers (see $line\\#27$ of
Algorithm 1). So doing, it is guaranteed that central server and coworkers
share the same updated set of $\lambda$’s coefficients (see $line\\#36$ of
Algorithm 1).

#### Age and fairness-aware tuning of the model-mixing parameter

A new feature of the pursued criterion for the design of the updating
relationship of the mixing parameter $\beta(t)$ in Eqs. (20a) and (20b) is
that it jointly accounts for the model age and fairness coefficients,
according to the following three considerations. First, $\beta(t)$ should
scale up/down proportionally the current value $\lambda_{ac(t+1)}$ of the
fairness coefficient of the communicating coworker, in order to enforce
fairness in the corresponding updating of the global model in Eq. (20a).
Second, under stationary (resp., non-stationary) application environments,
$\beta(t)$ should asymptotically vanish (resp., approach a non-zero positive
value), in order to guarantee a stable behavior in the steady-state (resp.,
quick responsiveness to abrupt environmental changes). Third, since too aged
(i.e., stale) model updates may be less reliable, $\beta(t)$ should scale down
for increasing values of the age of the current model update, which is
formally defined as follows [28, 32]:

$age_{ac(t+1)}\triangleq(t+1)-timestamp_{ac(t+1)}-1.$ (41)

In order to formalize the above design criteria, let:

$\Phi(age):\mathbb{N}_{0}\longrightarrow\left[0,\,1\right],$ (42)

be a non-negative function that does not decrease for increasing values of
$age$, like, for example, the following ones [21]:

$\Phi(age)\equiv\left(1+age\right)^{-\alpha},\leavevmode\nobreak\
\alpha>0;\quad\Phi(age)\equiv e^{-\alpha\times age},\leavevmode\nobreak\
\alpha>0,\quad\text{and:}\quad\Phi(age)\equiv\begin{cases}1,&0\leq age\leq
b,\\\\[4.30554pt] (1+age)^{-\alpha},&age\geq b.\end{cases}$ (43)

Hence, we propose the above relationship for the adaptive age and fairness-
aware updating of $\beta(\cdot)$ in $line\\#25$ of Algorithm 1:

$\beta(t)=\max\left\\{\beta_{MIN};\,\min\left\\{\beta_{MAX};\,\frac{\lambda_{ac(t+1)}\times\Phi\\!\left(age_{ac(t+1)}\right)}{\left(1+t\right)^{de}}\right\\}\right\\},\quad
t=0,1,\ldots,$ (44)

where: (i) $de$ is a non-negative decay-exponent333Vanishing (resp., non-
vanishing) values of the $de$ exponent may be suitable under stationary
(resp., nonstationary) operating environments.; while, (ii) $\beta_{MIN}$ and
$\beta_{MAX}$ with $0<\beta_{MIN}\leq\beta_{MAX}\leq 1$ are lower/upper
clipping parameters.

After updating $\beta(\cdot)$, the central server: (i) performs model
aggregation as in Eqs. (20a) and (20b) (see $line\\#26$ of Algorithm 1); (ii)
sends a timestamped copy (with time stamp $(t+1)$) of the updated global model
$\vv{\overline{w}}(t+1)$ to the currently communicating $ac(t+1)$-th coworker;
(iii) increases its time index $t$; and, finally, (iv) stalls for waiting for
a new model update (see $line\\#28$ of Algorithm 1).

Overall, the resulting per-aggregation computational effort to be sustained by
the central server scales up as $\mathcal{O}(l+4)$, where $l$ is the size of
the vector updating in Eq. (20a), while the additive term accounts for the
scalar updates in Eqs. (27a), (36), (37), (40), and (44).

##  4 AFAFed Convergence Analysis

The goal of this section is to prove the convergence (in expectation) of the
sequence $\left\\{\vv{\overline{w}}(t),\,t\geq 0\right\\}$ of the global
models evaluated by the central server. In this regard, in order to account
for the packet-loss events affecting the uplinks of Fig. 1, we indicate by
$P_{k}$, $k=1,\ldots,K$, the steady-state probability of successful
transmissions from $coworker\\#k$, formally defined as follows (see Eq.
(18a)):

$P_{k}\triangleq
E\\!\left\\{\xi_{k}(t)\right\\}\equiv\lim_{t\rightarrow\infty}\frac{1}{t}\sum_{\tau=1}^{t}\xi_{k}(\tau),\quad
k=1,\ldots,K.$ (45)

Since $P_{k}$ vanishes when the corresponding packet loss probability in Eq.
(17) approaches the unit (see Eq. (18a)), we refer to the set
$\left\\{P_{k}\right\\}$ as the coworker access probabilities.

Regarding the methodology pursued for proving the convergence of the sequence
$\left\\{\vv{\overline{w}}(t)\right\\}$, let us consider, as a benchmark, the
Synchronous Centralized Gradient-Descent scheme (SyCGD), in which: (i) all the
training data $S\triangleq\cup_{k=1}^{K}S_{k}$ stays at the central server;
and, (ii) the full gradient in Eq. (7) is used to carry out the following
synchronous gradient-descent iterations:

$\vv{\overline{w}}(t+1)=\vv{\overline{w}}(t)-\beta(t)\vv{\nabla}F\\!\left(\vv{\overline{w}}(t)\right),\quad
t=0,1,\ldots$ (46)

Hence, a comparison of the asynchronous SGD iterations of Eq.(20b) and the
SyCGD ones of Eq. (46) supports the conclusion that asynchronous and
synchronous iterations may follow similar time-paths, provided that the
stochastic gradient $\widehat{\vv{G}}(\cdot)$ in Eq. (21) is a reliable
estimate of the corresponding full gradient $\vv{\nabla}F(\cdot)$ in Eq. (46).
Hence, we expect that, if we would capable to suitably limit the gap between
these two gradients, then, the convergence rate of the asynchronous SGD
iterations in Eqs. (20a) and (20b) would approach the one of the SyCGD
benchmark in Eq. (46).

###  4.1 Preliminaries and baseline properties

Motivated by the above expectation, we introduce the following baseline
assumptions on the formal properties of the considered local and global loss
functions:

######  Assumption 3 (Baseline assumptions on the global/local loss
functions).

By referring to the defining relationships in Eqs. (2)–(4), we assume that:

  1. 1.

there exists at least a global minimizer in Eq. (5) with finite global minimum
Eq. (6) in correspondence of the considered (possibly, non-convex) local
functions $\left\\{F_{k}(\cdot),\,k=1,\ldots,K\right\\}$;

  2. 2.

the local loss functions are $\zeta$-smooth, i.e., there exists a positive
scalar constant $\zeta$ such that:

$\left\|\vv{\nabla}F_{k}\\!\left(\vv{w}_{1}\right)-\vv{\nabla}F_{k}\\!\left(\vv{w}_{2}\right)\right\|\leq\zeta\left\|\vv{w}_{1}-\vv{w}_{2}\right\|,\quad\forall\vv{w}_{1},\vv{w}_{2}\in\mathbb{R}^{l},\leavevmode\nobreak\
k=1,\ldots,K;$ (47)

  3. 3.

the stochastic gradients in Eq. (14) provide unbiased estimates of the
corresponding full gradients in Eq. (3), that is:

$E_{\mathcal{D}_{k}}\left\\{\vv{\nabla}\widetilde{F}_{k}\\!\left(\vv{w}\right)\right\\}\equiv\vv{\nabla}F_{k}\\!\left(\vv{w}\right),\quad\forall\vv{w}\in\mathbb{R}^{l},\leavevmode\nobreak\
k=1,\ldots,K;$ (48)

  4. 4.

the variances of the stochastic gradients are bounded and scale down with the
sizes of the corresponding mini-batches as in:

$E_{\mathcal{D}_{k}}\left\\{\left\|\vv{\nabla}\widetilde{F}_{k}\\!\left(\vv{w}\right)-\vv{\nabla}F_{k}\\!\left(\vv{w}\right)\right\|^{2}\right\\}\leq\frac{\sigma_{k,1}^{2}}{\left|MB_{k}\right|},\quad\forall\vv{w}\in\mathbb{R}^{l},\leavevmode\nobreak\
k=1,\ldots,K,$ (49)

where $\sigma_{k,1}^{2}\geq 0$ is the variance of the noise affecting the
$k$-th stochastic gradient when the mini-batch is unit-sized;

  5. 5.

under any setting of the fairness coefficients $\left\\{\lambda_{j}\right\\}$,
the gradient variances induced by non-i.i.d. data are limited as follows:

$E_{\mathcal{D}_{k}}\left\\{\left\|\vv{\nabla}F_{k}\\!\left(\vv{w}\right)-\vv{\nabla}F\\!\left(\vv{w}\right)\right\|^{2}\right\\}\leq\sigma_{k,2}^{2},\quad\forall\vv{w}\in\mathbb{R}^{l},\leavevmode\nobreak\
k=1,\ldots,K.$ (50)

$\blacksquare$

Although quite standard (see, for example, [19, 21, 16]), the above
assumptions merit some auxiliary remarks.

First, as formally proved, for example, in [16], the $\zeta$-smooth property
of the local loss functions in Eq. (47) implies that also the resulting global
function in Eq. (4) is $\zeta$-smooth, so that we have:

$\left\|\vv{\nabla}F\\!\left(\vv{w}_{1}\right)-\vv{\nabla}F\\!\left(\vv{w}_{2}\right)\right\|\leq\zeta\left\|\vv{w}_{1}-\vv{w}_{2}\right\|,\quad\forall\vv{w}_{1},\vv{w}_{2}\in\mathbb{R}^{l},\leavevmode\nobreak\
\forall\left\\{\lambda_{j}\right\\}.$ (51)

Second, from the bounds in Eqs. (49) and (50), it follows that the variances
of the local stochastic gradients in Eq. (14) around the full global gradient
in Eq. (7) are limited as in (see the Appendix B for the proof):

$E_{\mathcal{D}_{k}}\left\\{\left\|\vv{\nabla}\widetilde{F}_{k}\\!\left(\vv{w}\right)-\vv{\nabla}F\\!\left(\vv{w}\right)\right\|^{2}\right\\}\leq\Sigma_{k}^{2},\quad\forall\vv{w}\in\mathbb{R}^{l},\leavevmode\nobreak\
k=1,\ldots,K,$ (52)

where:

$\Sigma_{k}^{2}\triangleq\frac{\sigma_{k,1}^{2}}{\left|MB_{k}\right|}+\sigma_{k,2}^{2},\quad
k=1,\ldots,K,$ (53)

accounts for both mini batch-induced and data skewness-induced gradient
variances.

Finally, we remark that the introduced Assumption 3 allows us to define three
main ensemble averages over the spectrum of the access probabilities in Eq.
(45). The first index is the ensemble average $\overline{\Sigma^{2}}$ of the
per-coworker variances of Eq. (53), formally defined as follows:

$\overline{\Sigma^{2}}\triangleq E_{\xi\sim
P}\left\\{\Sigma_{k}^{2}\right\\}\equiv\sum_{k=1}^{K}P_{k}\Sigma_{k}^{2}.$
(54)

The second and third indexes are the first $\overline{I}$ and second
$\overline{I^{2}}$ moments of the numbers of per-cluster local iterations
performed by the coworkers, i.e.,

$\overline{I}\triangleq E_{\xi\sim
P}\left\\{Iter_{k}\right\\}\equiv\sum_{k=1}^{K}P_{k}\overline{Iter_{k}},\quad\text{and:}\quad\overline{I^{2}}\triangleq
E_{\xi\sim
P}\left\\{Iter_{k}^{2}\right\\}\equiv\sum_{k=1}^{K}P_{k}\overline{Iter_{k}^{2}},$
(55)

where: $\overline{Iter_{k}}\triangleq
E_{\mathcal{D}_{k}}\\!\left\\{Iter_{k}\right\\}$, and:
$\overline{Iter_{k}^{2}}\triangleq
E_{\mathcal{D}_{k}}\\!\left\\{Iter_{k}^{2}\right\\}$ are the corresponding
first and second moments of the r.v. $Iter_{k}$ in Eq. (25).

###  4.2 Convergence assumptions and their meaning/roles

State-of-the-art literature [33] has already pointed out that, under non-
convex settings, commonly metrics used for evaluating the convergence rate in
convex optimization (like: $F\\!\left(\vv{\overline{w}}(t)\right)-F^{\ast}$,
or $\left\|\vv{\overline{w}}-\vv{\overline{w}}^{\ast}\right\|$) are no longer
eligible. Hence, according to, for example, [33, 34, 26], we resort to the
convergence (in expectation) of the sequence:
$\left\\{\left\|\vv{\nabla}F\\!\left(\vv{\overline{w}}\right)\right\|^{2},\,t\geq
0\right\\}$ along the path: $\left\\{\vv{\overline{w}}(t),\,t\geq 0\right\\}$
followed by the iterations in Eqs. (20a) and (20b) as the metric for
evaluating the AFAFed convergence. This means, in turn, that our target is to
formally characterize the convergence rate of the following global average:

$E\\!\left\\{\frac{1}{T}\sum_{t=0}^{T-1}\left\|\vv{\nabla}F\\!\left(\vv{\overline{w}}(t)\right)\right\|^{2}\right\\},$
(56)

for increasing values of $T$. The pursued formal methodology is inspired by
the so called Gradient Coherence principle in [35]. It measures the similitude
between the stochastic: $\widehat{\vv{G}}(\cdot)$ and full:
$\vv{\nabla}F(\cdot)$ gradients in Eqs. (20b) and (46) through the
corresponding average gap between the cross-product:
$\widehat{\vv{G}}(\cdot)^{T}\,\vv{\nabla}F(\cdot)$ and the corresponding
benchmark value:
$\vv{\nabla}F(\cdot)^{T}\,\vv{\nabla}F(\cdot)\equiv\left\|\vv{\nabla}F(\cdot)\right\|^{2}$.
Hence, motivated by this consideration, we introduce the following formal
assumption:

######  Assumption 4 (Assumptions on the gradient coherence).

By referring to the stochastic and full gradient-descent iterations in Eqs.
(20b) and (46), let us assume that:

  1. 1.

there exist two scalar positive constants $\Gamma\geq C>0$ such that:

$\left(E_{\xi\sim
P}\\!\left\\{\widehat{\vv{G}}\\!\left(\vv{\overline{w}};\left\\{\xi_{k}\right\\}\right)\right\\}\right)^{T}\vv{\nabla}F\left(\vv{\overline{w}}\right)\geq
C\left\|\vv{\nabla}F\\!\left(\vv{\overline{w}}\right)\right\|^{2},\quad\forall\vv{\overline{w}},\leavevmode\nobreak\
\forall\left\\{\lambda_{k}\right\\},$ (57)

and:

$\left\|E_{\xi\sim
P}\\!\left\\{\widehat{\vv{G}}\\!\left(\vv{\overline{w}};\left\\{\xi_{k}\right\\}\right)\right\\}\right\|\leq\Gamma\left\|\vv{\nabla}F\\!\left(\vv{\overline{w}}\right)\right\|,\quad\forall\vv{\overline{w}},\leavevmode\nobreak\
\forall\left\\{\lambda_{k}\right\\};$ (58)

  2. 2.

there exists a non-negative scalar constant $A\geq 0$ such that the
stochastic-gradient variance:

$\sigma_{\widehat{\vv{G}}}^{2}\triangleq
var\\!\left\\{\widehat{\vv{G}}\right\\}\equiv E_{\xi\sim
P}\\!\left\\{\left\|\widehat{\vv{G}}\\!\left(\cdot\right)\right\|^{2}\right\\}-\left\|E_{\xi\sim
P}\\!\left\\{\widehat{\vv{G}}\\!\left(\cdot\right)\right\\}\right\|^{2},$ (59)

is upper bounded as follows:

$\sigma_{\widehat{\vv{G}}}^{2}\leq
A+\left\|\vv{\nabla}F\\!\left(\vv{\overline{w}}\right)\right\|^{2},\quad\leavevmode\nobreak\
\forall\left\\{\lambda_{k}\right\\}.$ (60)

$\blacksquare$

The meaning/role and feasible values of the constants $\Gamma$, $C$, and $A$
involved in the above Assumption 4 are featured by the following Lemma 1 (see
Appendix B for the proof):

######  Lemma 1 (On the meaning/role and feasible values of the constants
$\Gamma$, $C$, and $A$).

Let us assume that the global stochastic gradient
$\widehat{\vv{G}}\\!\left(\cdot\right)$ in Eq. (21) provides a (possibly,
biased) estimate of the full gradient $\vv{\nabla}F(\cdot)$ in Eq. (4), i.e.,
let us assume that:
$\widehat{\vv{G}}\\!\left(\cdot\right)\equiv\vv{\nabla}F(\cdot)+\vv{n}$, where
$\vv{n}\in\mathbb{R}^{l}$ is a noise r.v., which meets the following two
assumptions:

$E_{\xi\sim P}\\!\left\\{\vv{n}\right\\}=\gamma_{0}\vv{\nabla}F(\cdot),$ (61)

with $\gamma_{0}>-1$, and,

$E_{\xi\sim
P}\\!\left\\{\left\|\vv{n}\right\|^{2}\right\\}=\gamma_{1}\left\|\vv{\nabla}F(\cdot)\right\|^{2}+b_{0},$
(62)

with $b_{0}\geq 0$ and $0\leq\left(\gamma_{1}-\gamma_{0}^{2}\right)\leq 1$.
Then, the spectra of feasible values of $C$, $\Gamma$, and $A$ in Eqs.
(57)–(60) are related to the $\gamma_{0}$, $\gamma_{1}$, and $b_{0}$
parameters as follows:

$0<C\leq\left(1+\gamma_{0}\right),\quad\Gamma\geq\left(1+\gamma_{0}\right),\quad\text{and}\quad
A\geq b_{0}.$ (63)

Furthermore, when $\widehat{\vv{G}}\\!\left(\cdot\right)$ provides an unbiased
estimate of $\vv{\nabla}F(\cdot)$, the relationships in Eq. (63) still hold
with $\gamma_{0}=\gamma_{1}=0$. $\blacksquare$

The bounds reported in Eq. (63) highlight how the constants $C$, $\Gamma$ and
$A$ in Assumption 4 are, indeed, related to the statistical properties of the
noise affecting the stochastic global gradient
$\widehat{\vv{G}}\\!\left(\cdot\right)$.

###  4.3 The AFAFed convergence bounds

In Appendix C, the following main result is proved:

######  Proposition 1 (AFAFed convergence rate).

Let us assume that $\beta_{MAX}$ in Eq. (44) is limited as in:

$\beta_{MAX}\leq\frac{2C\varepsilon}{\zeta\left(1+\Gamma^{2}\right)},$ (64)

where $\varepsilon$ is a regularizing parameter, which may be set by the
central server over the open interval $(0,\,1)$. Then, under the (previously
reported) Assumptions 1–4, the following upper bound holds on the convergence
rate of the AFAFed Algorithm 1:

$E\\!\left\\{\frac{1}{T}\sum_{t=0}^{T-1}\beta(t)\left\|\vv{\nabla}F\\!\left(\vv{\overline{w}}(t)\right)\right\|^{2}\right\\}\leq\frac{E\\!\left\\{F\\!\left(\vv{\overline{w}}(0)\right)\right\\}-F^{\ast}}{C\left(1-\varepsilon\right)T}+\frac{A\zeta}{2C\left(1-\varepsilon\right)}\,E\\!\left\\{\frac{1}{T}\sum_{t=0}^{T-1}\beta(t)^{2}\right\\}.$
(65)

Furthermore, in the case in which $\beta_{MIN}$ in Eq. (44) is positive, we
have that:

$E\\!\left\\{\frac{1}{T}\sum_{t=0}^{T-1}\left\|\vv{\nabla}F\\!\left(\vv{\overline{w}}(t)\right)\right\|^{2}\right\\}\leq\frac{E\\!\left\\{F\\!\left(\vv{\overline{w}}(0)\right)\right\\}-F^{\ast}}{C\left(1-\varepsilon\right)\beta_{MIN}T}+\frac{A\zeta\beta_{MAX}^{2}}{2C\left(1-\varepsilon\right)\beta_{MIN}},$
(66)

which, in turn, in the case of constant $\beta$ (i.e., in the case of
$\beta_{MAX}\equiv\beta_{MIN}$) reduces to:

$E\\!\left\\{\frac{1}{T}\sum_{t=0}^{T-1}\left\|\vv{\nabla}F\\!\left(\vv{\overline{w}}(t)\right)\right\|^{2}\right\\}\leq\frac{E\\!\left\\{F\\!\left(\vv{\overline{w}}(0)\right)\right\\}-F^{\ast}}{C\left(1-\varepsilon\right)\beta
T}+\frac{A\zeta\beta}{2C\left(1-\varepsilon\right)}.$ (67)

$\blacksquare$

The reported bounds merit two main explicative remarks. First, all the bounds
in Eqs. (65)–(67) are the summation of two non-negative terms. The first terms
guarantee that the convergence rate of Algorithm 1 is (at least):
$\mathcal{O}\left(1/T\right)$ for increasing $T$, while the second terms at
the r.h.s.’s of Eqs. (65)–(67) fix the AFAFed convergence performance for
$T\rightarrow\infty$. They point out that the asymptotic convergence is
guaranteed up to a spherical neighborhood of a stationary point of the global
loss function $F(\cdot)$, whose squared radius scales up (resp., down) with
the product $A\zeta$ (resp., $C\left(1-\varepsilon\right)$). Second, the
version of AFAFed reported in Algorithm 1 refers to local primal/dual iterates
that are implemented through first-order SGD. However, the analytical
derivation in Appendices B and C point out that the reported convergence
bounds still apply verbatim under any arbitrary local solver, provided that
the corresponding $C$, $\Gamma$, and $A$ parameters of Assumption 4 refers to
the actually implemented solver.

###  4.4 Scaled forms of the convergence bounds

Our next step is to evaluate the impact of the ensemble-averaged global
indexes in Eqs. (54) and (55) and the average packet-loss probability
$\overline{P}^{\,(LOSS)}$ of the uplinks of Fig. 1 on the convergence bounds
of Eqs. (65)–(67). To this end, let us indicate by $C_{0}$, $\Gamma_{0}$, and
$A_{0}$ the values assumed by the parameters $C$, $\Gamma$ and $A$ in Eqs.
(57)–(60) under the following normalized operating conditions:
$\overline{\Sigma^{2}}=\overline{I}=\overline{I^{2}}=1$. Therefore, the
developments reported in B lead to the conclusion that $C$, $\Gamma$, and $A$
scale with $\overline{\Sigma^{2}}$, $\overline{I}$, and $\overline{I^{2}}$ as
in:

$\displaystyle C$ $\displaystyle=C_{0}\overline{I},$ (68a)
$\displaystyle\Gamma$ $\displaystyle=\Gamma_{0}\overline{I},$ (68b) and,
$\displaystyle A$
$\displaystyle=A_{0}\,\overline{\Sigma^{2}}\>\overline{I^{2}}.$ (68c)

Hence, by inserting the above relationships in Eqs. (65)–(67), we formally
characterize the dependence of the AFAFed convergence bounds on the ensemble-
averaged global indexes $\overline{\Sigma^{2}}$, $\overline{I}$, and
$\overline{I^{2}}$. Just as an example, we may rewrite the bound of Eq. (66)
in the following equivalent form:

$E\\!\left\\{\frac{1}{T}\sum_{t=0}^{T-1}\left\|\vv{\nabla}F\\!\left(\vv{\overline{w}}(t)\right)\right\|^{2}\right\\}\leq\frac{E\\!\left\\{F\\!\left(\vv{\overline{w}}(0)\right)\right\\}-F^{\ast}}{C_{0}\overline{I}\left(1-\varepsilon\right)\beta_{MIN}T}+\frac{A\overline{\Sigma^{2}}\>\overline{I^{2}}\,\zeta\,\beta_{MAX}^{2}}{2C_{0}\overline{I}\left(1-\varepsilon\right)\beta_{MIN}},\quad\text{with:}\quad\beta_{MAX}\leq\frac{2C_{0}\overline{I}\,\varepsilon}{\zeta\left(1+\left(\Gamma_{0}\overline{I}\right)^{2}\right)}.$
(69)

In practical application scenarios, the online profiling of the (possibly, a
priori unknown) parameters $C$, $\Gamma$ and $A$ may be carried out as
detailed in the next Section 5.

Remark 2 – On the contrasting effects of $\overline{I}$ and
$\overline{I^{2}}$. An application of the Jensen’s inequality [31] guarantees
that: $\overline{I^{2}}/\overline{I}\geq\overline{I}$, so that the second
terms at the r.h.s. of the bound in Eq. (69) scale up (at least) as
$\mathcal{O}\left(\overline{I}\right)$ for increasing values of
$\overline{I}$. Since the first term at the r.h.s. of this bound scale down as
$\mathcal{O}\left(1/\overline{I}\right)$ for growing $\overline{I}$, we
conclude that the effects of $\overline{I}$ on the convergence properties of
AFAFed are, indeed, contrasting. Specifically, large (resp., small) values of
$\overline{I}$ speed up (resp., slow down) the resulting AFAFed convergence
rate at small $T$’s, but, at the same time, they increase (resp., decrease)
the residual average squared fluctuations around the reached stationary-point
of the global loss function $F(\cdot)$ for large $T$’s. This supports the
conclusion that, since the optimized setting of $\overline{I}$ is typically a
priori unknown, an adaptive and tuning of the per-cluster number of local
iterations carried out by each coworker is, indeed, mandatory under the FL
scenario of Fig. 1. This provides formal motivation of the choice pursued in
Section 3.2 for the adaptive design of the AFAFed in Algorithm 1 (see, in
particular, Eq. (25) and related text). ∎

Remark 3 – On the contrasting effects of the ensemble-averaged variance
$\overline{\Sigma^{2}}$. By definition, $\overline{\Sigma^{2}}$ in Eq. (54)
measures the ensemble-averaged squared jitter of the local gradients in Eq.
(14) around the global full gradient of Eq. (7), which is caused by both
stochastic gradients and data skewness (see Eqs. (53) and (54)). Accordingly,
the impact of large values of $\overline{\Sigma^{2}}$ on the bound in Eq. (69)
is negative, because large $\overline{\Sigma^{2}}$’s increase the average
squared residual errors experienced by AFAFed for large $T$’s. However, this
(negative) conclusion is somewhat counterbalanced by the analysis carried out
in [11]. Specifically, the main conclusion of this analysis is that, when the
Hessian matrix of the global loss function $F(\cdot)$ retains at least one
negative eigenvalue at each saddle point $\overline{\vv{w}}_{sdl}$, then, a
random perturbation of $\overline{\vv{w}}_{sdl}$ allows $F(\cdot)$ to escape
it with high probability through a finite number of perturbed iterations.
Hence, since, in the AFAFed framework, the global sequence
$\left\\{\vv{\overline{w}}(t)\right\\}$ in Eqs. (20a) and (20b) is driven by
the global stochastic gradient $\widehat{\vv{G}}(\cdot)$ and
$var\left\\{\widehat{\vv{G}}\right\\}$ in (59) increases as
$\mathcal{O}\left(\overline{\Sigma^{2}}\right)$ (see Eqs. (60) and (68c)), we
expect that large enough values of $\overline{\Sigma^{2}}$ may help the
gradient sequence
$\left\\{\vv{\nabla}F\\!\left(\vv{\overline{w}}(t)\right)\right\\}$ to escape
saddle points, so to converge to neighborhoods of local/global minima of
$F(\cdot)$. ∎

A last remark concerns the effect of the packet-loss probabilities in Eq. (17)
on the convergence rate of the bound in Eq. (69). After noting that the number
$T$ of updates performed by the central server does not depend, by definition,
on the probability spectrum $\left\\{P_{k}^{\,(LOSS)}\right\\}$, let
$\overline{P}^{(LOSS)}$ be the average packet-loss probability, and let
$N^{(TR)(TOT)}$ indicate the total number of successful-plus-failed
transmissions performed by all coworkers of Fig. 1. Hence, since $T$ in Eq.
(69) equates, by definition, the (average) fraction of the number of the total
transmissions which do not undergo packet-loss events, we obtain the following
scaling law [31]:

$T\propto\left(1-\overline{P}^{\,(LOSS)}\right)N^{(TR)(TOT)}.$ (70)

Thus, after inserting Eq. (70) at the denominator of the first term of Eq.
(69), we conclude that the effect of the packet-loss probabilities is to scale
down the average convergence rate of the bound by a factor:
$\left(1-\overline{P}^{\,(LOSS)}\right)$, while they do not modify the
attained asymptotic value.

##  5 AFAFed Implementation Aspects

The goal of this section is to consider two main implementation aspects of the
AFAFed Algorithm 1, which concern: (i) the online profiling of the (possibly,
_a priori_ unknown) parameters involved by the bounds of Proposition 1; and,
(ii) the optimized design and implementation of a throughput-maximizing
distributed policy for the dynamic management of the streams of arrivals in
Fig. 1.

###  5.1 Profiling the AFAFed parameters

The developed procedure for the online profiling of the parameters involved in
the bounds of Proposition 1 relies on the following formal results on the
feasibility of their estimates (see the last part of Appendix B for the
proof):

######  Lemma 2 (Feasible estimates of the AFAFed parameters).

Let $\overline{\vv{\widehat{G}}}$,
$\overline{\left\|\vv{\widehat{G}}\right\|}$,
$\overline{\left\|\vv{\widehat{G}}\right\|^{2}}$, and $\widehat{\vv{\nabla}F}$
indicate estimates of the corresponding parameters: $E_{\xi\sim
P}\left\\{\vv{\overline{G}}\right\\}$, $\left\|E_{\xi\sim
P}\left\\{\vv{\overline{G}}\right\\}\right\|$, $E_{\xi\sim
P}\left\\{\left\|\vv{\overline{G}}\right\|^{2}\right\\}$, and $\vv{\nabla}F$.
Hence, if and only if the following three conditions are simultaneously met:

$\displaystyle\overline{\vv{\widehat{G}}}^{T}\,\widehat{\vv{\nabla}F}>0,$
(71a) $\displaystyle
k_{0}\triangleq\frac{\left\|\overline{\vv{\widehat{G}}}\right\|\,\left\|\widehat{\vv{\nabla}F}\right\|}{\left(\overline{\vv{\widehat{G}}}\right)^{T}\,\left(\widehat{\vv{\nabla}F}\right)}-1\geq
0,$ (71b)
$\displaystyle\overline{\left\|\vv{\widehat{G}}\right\|^{2}}-\left(\overline{\left\|\vv{\widehat{G}}\right\|}\right)^{2}-\left\|\widehat{\vv{\nabla}F}\right\|^{2}\geq
0,$ (71c)

then, the following three relationships:

$\displaystyle\widehat{C}$
$\displaystyle\triangleq\frac{\overline{\vv{\widehat{G}}}^{T}\,\widehat{\vv{\nabla}F}}{\left\|\widehat{\vv{\nabla}F}\right\|^{2}},$
(72a) $\displaystyle\widehat{\Gamma}$
$\displaystyle\triangleq\left(1+k_{0}\right)\widehat{C},$ (72b)
$\displaystyle\widehat{A}$
$\displaystyle\triangleq\overline{\left\|\vv{\widehat{G}}\right\|^{2}}-\left(\overline{\left\|\vv{\widehat{G}}\right\|}\right)^{2}-\left\|\widehat{\vv{\nabla}F}\right\|^{2},$
(72c)

provide feasible estimates of the parameters $C$, $\Gamma$ and $A$ in Eqs.
(57)–(60). $\blacksquare$

On the basis of Lemma 2, we conclude that when a priori information about the
actual parameters $C$, $\Gamma$, $A$ is not available, the only viable means
of computing their estimates in Eqs. (72a)–(72c) is to run a (suitably
modified version) of the AFAFed protocol, in order to perform the online
profiling of the estimates: $\overline{\vv{\widehat{G}}}$,
$\overline{\left\|\vv{\widehat{G}}\right\|}$,
$\overline{\left\|\vv{\widehat{G}}\right\|^{2}}$, and $\widehat{\vv{\nabla}F}$
on the r.h.s of Eqs. (72a)–(72c).

Algorithm 2 — Online profiling of the AFAFed parameters

1:$\triangleright$ Skeleton of the modified AFAFed protocol for online
parameter profiling

2:Central server broadcasts the initialization model $\vv{\overline{w}}(0)$ to
all coworkers;

3:for $t=0:\left(T-1\right)$ do

4: central server and coworkers perform the $t$-th iteration of Algorithm 1;

5: if $coworker\\#k$ communicates to the central server then

6: $coworker\\#k$ sends its last computed local stochastic gradient
$\vv{\nabla}\widetilde{F}_{k}\\!\left(\vv{w}_{k}(t+1)\right)$ to the central
server;

7: central server stores the received gradient;

8: end if$\triangleright$ End-if at Step#4

9: central server evaluates and stores:
$\vv{\widehat{G}}(t)=\sum_{k=1}^{K}\xi_{k}(t+1)\left[\vv{\overline{w}}(t)-\vv{w}_{k}(t+1)\right]$;

10:end for$\triangleright$ End for at Step#2

11:

12:$\triangleright$ AFAFed profiling phase

13:$\triangleright$ After ending to run Algorithm 1, the central server and
coworkers synchronously perform the following steps:

14:Central server broadcasts the final global model $\vv{\overline{w}}(T)$;

15:parfor $k=1:K$ do

16: $coworker\\#k$ evaluates: $F_{k}(0)\triangleq
F_{k}\\!\left(\vv{\overline{w}}(0)\right)$, $F_{k}(T)\triangleq
F_{k}\\!\left(\vv{\overline{w}}(T)\right)$, and
$\widehat{\zeta}_{k}\triangleq\frac{\left\|\vv{\nabla}\widetilde{F}_{k}\\!\left(\vv{w}_{k}(T)\right)-\vv{\nabla}\widetilde{F}_{k}\\!\left(\vv{w}_{k}(0)\right)\right\|}{\left\|\vv{\overline{w}}(T)-\vv{\overline{w}}(0)\right\|}$;

17: $coworker\\#k$ sends the 3-ple:
$\left\\{F_{k}(0),\,F_{k}(T),\,\widetilde{\zeta}_{k}\right\\}$ of the locally
estimated parameters to the central server;

18:end parfor$\triangleright$ End parfor at Step#11

19:Central server evaluates the following four sample averages:
$\overline{\vv{\widehat{G}}}=\frac{1}{T}\sum_{t=0}^{T-1}\vv{\widehat{G}}(t),$
$\overline{\left\|\vv{\widehat{G}}\right\|}=\frac{1}{T}\sum_{t=0}^{T-1}\left\|\vv{\widehat{G}}(t)\right\|,$
$\overline{\left\|\vv{\widehat{G}}\right\|^{2}}=\frac{1}{T}\sum_{t=0}^{T-1}\left\|\vv{\widehat{G}}(t)\right\|^{2},$
and
$\widehat{\vv{\nabla}F}=\frac{1}{T}\sum_{k=1}^{K}\xi_{k}(t+1)\vv{\nabla}\widetilde{F}_{k}\\!\left(\vv{w}_{k}(t+1)\right);$

20:Central server computes the following estimates:
$\widehat{F}^{\ast}\triangleq\sum_{k=1}^{K}\lambda_{k}(T)F_{k}(T);$
$\widehat{F}(0)\triangleq\sum_{k=1}^{K}\lambda_{k}(T)F_{k}(0);$ and
$\widehat{\zeta}\triangleq\max_{1\leq k\leq
K}\left\\{\widehat{\zeta}_{k}\right\\},$ of the actual parameters: $F^{\ast}$,
$E\\!\left\\{F\\!\left(\vv{\overline{w}}(0)\right)\right\\}$, and $\zeta$,
respectively;

21:if conditions in Eqs.(71a)–(71c) are met then

22: central server computes the estimates: $\widehat{C}$, $\widehat{\Gamma}$,
and $\widehat{A}$ as in Eqs. (72a), (72b), and (72c), respectively;

23:end if$\triangleright$ End-if at Step#17

24:

25:return the parameter estimates: $\widehat{F}^{\ast}$, $\widehat{F}(0)$,
$\widehat{\zeta}$, $\widehat{C}$, $\widehat{\Gamma}$, and $\widehat{A}$.

To this end, we design the procedure in Algorithm 2, whose implementation
merits, indeed, three main explicative remarks. First, at each $t$-indexed
global iteration, the modified AFAFed protocol in the first part of Algorithm
2 performs the same steps already reported in the baseline Algorithm 1 and,
then, it carries out the additional steps detailed in $line\\#5$–$line\\#8$.
These additional steps allow the central server to: (i) store the more recent
stochastic gradient communicated by the coworkers (see $line\\#6$); and, (ii)
compute and store the $t$-th stochastic global gradient $\vv{\widehat{G}}(t)$
(see $line\\#8$). Second, after finishing to run the first part of Algorithm
2, central server and coworkers enter the online profiling phase, which is
reported in the second part of Algorithm 2 and is carried out in a
synchronized way (see the parfor statement at $line\\#11$). Third, the vector:
$\widehat{\vv{\nabla}F}$ in $line\\#15$ is the sample average of the sequence:
$\left\\{\vv{\nabla}\widetilde{F}_{k}(\cdot)\right\\}$ of the stochastic local
gradients sent to the central server by coworkers during the execution of the
first part of Algorithm 2. Hence, $\widehat{\vv{\nabla}F}$ must be properly
understood as a noisy estimate of the average value of the full gradient
$\vv{\nabla}F$ present in the bounds of Proposition 1. This is, indeed, the
main reason why the Eqs. (72a)–(72c) provide only estimates of the (typically,
unknown) actual values of the corresponding AFAFed parameters.

Before proceeding, we remark that the reason why the profiling phase of
Algorithm 2 requires coworker synchronization is that $\zeta$ in Eq. (47) and
$F\\!\left(\vv{\overline{w}}(0)\right)$ and $F^{\ast}$ in Eqs. (65)–(67) are
global parameters, which characterize the analytical behaviour of the (_a
priori_ unknown) global loss function $F(\cdot)$ in Eq. (4). For this reason,
designing an asynchronous procedure for their online estimates seems to be (if
feasible) a challenging task. However, an estimation procedure that does not
require inter-coworker synchronization could be designed in the case in which:
(i) the global constraint in Eq. (47) is relaxed and replaced by a set of $K$
per-coworker local constraints, each one depending on a local parameter:
$\zeta_{k}$, $k=1,\ldots,K$; and, (ii) the global system parameter
$F\\!\left(\vv{\overline{w}}(0)\right)$ (resp., $F^{\ast}$) is replaced by its
upper (resp., lower) bound:
$\max_{k}\left\\{F_{k}\\!\left(\vv{\overline{w}}(0)\right)\right\\}$ (resp.,
$\min_{k}\left\\{F_{k}\\!\left(\vv{\overline{w}}(T)\right)\right\\}$). The
drawback is that the resulting convergence bounds would be somewhat looser
than the ones reported in Eqs. (65)–(67). Finally, we stress that, since
Algorithm 2 is required only for the numerical evaluation of the bounds, it
does not impact on the AFAFed protocol of Algorithm 1, which is asynchronous,
regardless from the synchronous/asynchronous nature of Algorithm 2.

###  5.2 Throughput-maximizing distributed management of stream arrivals

In the IoT-oriented AFAFed ecosystem of Fig. 1, coworkers are typically
equipped with local sensors, which perform sensing actions on the surrounding
environment according to some specified (possibly, energy-conserving) event-
driven policy for acquiring new training data. Therefore, since the sensing
instants are not known in advance, they should be properly modeled as streams
of random arrival times. As a matter of this fact, in order to avoid that the
execution of the primal/dual iterations in Eqs. (16a) and (16b) slows down (or
even stalls at all) due to the stretching of the data inter-arrival times, the
main guideline pursued in the design of the AFAFed stream management policy is
to make the execution rate of the local iterations at the coworkers as much as
possible uncoupled from the corresponding (uncontrollable) arrival rate of new
training data. To this end, after designing the coworker functional
architecture of Fig. 2, we propose the distributed stream management policy
composed by the Access Control Policy (ACP) and Buffer Control Policy (BCP)
detailed by the flow charts in Figs. 3 and 4, respectively.

Figure 2: Designed functional architecture of an AFAFed co-worker. S:= Storage
buffer; BC := Buffer Controller; AC := Access Controller; LP := Local
Processor; TC := Trans-Ceiver. Figure 3: Flowchart of the proposed Access
Control Policy (ACP). Figure 4: Flowchart of the proposed Buffer Control
Policy (BCP). S:= Storage buffer; BC := Buffer Controller; LP := Local
Processor.

According to Fig. 2, each AFAFed coworker is assumed to be equipped with: (i)
a buffer $\mathcal{S}$, which is capable of storing a finite number
$|\mathcal{S}|\geq 1$ of training data; (ii) an Access Controller (AC), which
performs selective admission of new training data; (iii) a Buffer Controller
(BC), which manages the current content of buffer $\mathcal{S}$ by removing
the oldest stored data; (iv) a Local Processor (LP), which runs the
primal/dual local iterations of Eqs. (16a) and (16b), and, at each iteration,
builds up a mini-batch of size $|MB|\geq 1$ by randomly sampling data from the
current content of buffer $\mathcal{S}$; and, (v) a Trans-Ceiver (TC), which
manages the up/down communications between coworker and central server.

The task of the AC is to implement the ACP reported in Fig. 3. This last
guarantees that the number of stored training data does not exceed the
(finite) capacity $|\mathcal{S}|$ of the available buffer, while the current
content of buffer $\mathcal{S}$ is always composed of the last arrived (i.e.
newest) $|\mathcal{S}|$ training data. Task of the BC and LP of Fig. 2 is to
jointly implement the BCP of Fig. 4. In this regard, an examination of the
corresponding flow chart points out that, by design, the proposed BCP
dynamically removes the oldest entry of buffer $\mathcal{S}$, but only when
the number of currently stored data is larger than the mini-batch size $|MB|$.
By doing so, the designed BCP: (i) dynamically frees space for storing new
arrivals, in order to minimize the drop rate of the AC; while: (ii)
guaranteeing that at least $|MB|$ training data are always stored by the
buffer. This last feature is, indeed, crucial to guarantee that the designed
BCP of Fig. 4 is computing throughput-maximizing. By definition, in our
setting, computing throughput-maximizing means that the steady-state computing
throughput (i.e., the computing rate) at which the local iterations of Eqs.
(16a) and (16b) are run by each coworker may equate the corresponding per-
coworker maximum processor’s speed, regardless from the (random and out-of-
control) rate of arrivals of new stream data (see Fig. 2). Hence, the expected
key benefit arising from the designed throughput-maximizing policy is that it
would shorten the resulting AFAFed average training times, because it allows
the coworkers to run their local iterations without suffering from stalling
phenomena. To check this property, we observe that, in order to run each local
iteration in Eq. (16a), $coworker\\#k$ must evaluate the per-minibatch
gradient $\vv{\nabla}\widetilde{F}_{k}(\cdot)$ in Eq. (14). This last
evaluation requires, in turn, to perform the average of the underlying local
loss function of Eq. (14) over a set of $\left|MB_{k}\right|$ data randomly
sampled from the local buffer of $coworker\\#k$, and this last operation may
be carried out without stalling only if the designed BCP of Fig. 4 guarantees
that the $k$-th local buffer always stores at least $\left|MB_{k}\right|$
elements.

##  6 Conclusion

In this paper, we designed and analyzed the convergence properties of AFAFed,
a novel asynchronous distributed and adaptive scheme for performing FL in
heterogeneous stream-oriented IoT scenarios affected by packet-loss
communication phenomena. The key feature of AFAFed is the implementation of
distributed sets of local adaptive tolerance thresholds and global centralized
adaptive fairness coefficients, whose symbiotic action allows AFAFed to attain
right personalization-vs.-fairness trade-offs in learning scenarios featured
by coworker heterogeneity, skewed training data, and unreliable
communications. AFAFed numerical performance evaluation and performance
comparisons against FL benchmark schemes are work in progress, and will be
published in a fiture work.

## Acknowledgment

This work has been supported by the projects: “DeepFog – Optimized distributed
implementation of Deep Learning models over networked multitier Fog platforms
for IoT stream applications” funded by Sapienza University of Rome Bando 2020
and 2021; and, “End-to-End Learning for 3D Acoustic Scene Analysis (ELeSA)”
funded by Sapienza University of Rome Bando Acquisizione di medie e grandi
attrezzature scientifiche 2018. The work is also partially supported by the
research collaboration with the Institute of Informatics and Telematics (IIT-
CNR) on the theme: “Design and software development of Fog Computing platforms
for the support of Deep Learning algorithms for the real-time image analysis”

##  Appendix A Taxonomy and default parameter setting

Table LABEL:tab:Tab_A1 details the main symbols used in this paper, their
meaning/role, and the corresponding default setup.

Table 1: Main taxonomy and default setup. AF:= AFAFed; FAG:= FedAvg; FPR:= FedProx; CS-SGD:= Cent.Sync.-SGD. |  |
---|---|---
Parameter | Definition | Default Settings
$K$ |  Number of involved coworkers |  100
$T$ |  Number of aggregations performed by the central server |  Application dependent
$t^{(k)}$ (resp., $t$) |  Integer-valued non-negative dimensionless $k$-th local time index (resp., global time index) |  Application dependent
$\ell(\cdot)$ |  Loss function |  Application dependent
$F(\cdot)$ and $\vv{\nabla}F(\cdot)$ |  Global loss function and related vector gradient |  Application dependent
$\widetilde{F}_{k}(\cdot)$ and $\vv{\nabla}\widetilde{F}_{k}(\cdot)$ |  Local $k$-th empirical risk function and related vector gradient |  Application dependent
$\left\\{\lambda_{k}\right\\}$ |  Set of the (dimensionless non-negative) fairness coefficients |  Adaptively tuned
$\left\\{\mu_{k}\right\\}$ and $\left\\{\overline{\mu}_{k}\right\\}$ |  Set of the (dimensionless non-negative) local Lagrange multipliers and their time averages |  Adaptively tuned
$\left\\{B_{k}\right\\}$ |  Set of the (dimensionless non-negative) local tolerance thresholds |  Adaptively tuned
$\left\\{Iter_{k}\right\\}$ |  Set of the (integer-valued positive) numbers of consecutive local model updatings |  Adaptively tuned
$\left\\{S_{k}\right\\}$ |  Spectrum of the local training data sets |  Application dependent
$\left\\{MB_{k}\right\\}$ and $\left\\{\left|MB_{k}\right|\right\\}$ |  Spectra of local minibatches and their sizes |  $\left|MB_{k}\right|=$ 16, 32
$\left\\{\Sigma_{k}\right\\}$ and $\overline{\Sigma^{2}}$ |  Spectrum of the local gradient variances and its corresponding ensemble average |  Application dependent
$\left\\{P_{k}\right\\}$ and $\left\\{\xi_{k}\right\\}$ |  Spectra of the access probabilities and related binary-valued indicator functions |  See Eqs. (45) and (18a)
$B_{0}$, $de$, and $\gamma$ |  Non-negative dimensionless reference tolerance threshold, decay exponent and power exponent |  $B_{0}=1.0$, $de\in\left[0,\,0.6\right]$, and $\gamma=0.1$
$\widetilde{\mu}$ and $\sigma_{\tilde{\mu}}^{2}$ |  Non-negative dimensionless ensemble-averaged Lagrange multiplier and corresponding variance at the central server |  Adaptively tuned
$\overline{I}$ (resp., $\overline{I^{2}}$) |  Non-negative dimensionless ensemble-average number (resp., squared number) of per-coworker consecutive local model updatings |  Adaptively tuned
$age(t)$ |  Non-negative integer-valued age of the $t$-th aggregation at the central server |  Application dependent
$C$, $\Gamma$, and $A$ |  Dimensionless non-negative system parameters |  Profiled online as in Eqs. (72a), (72b), and (72c)
$\beta(t)$ |  Dimensionless non-negative mixing parameter for model aggregation |  Adaptively tuned as in Eq. (44)
$\left\\{\eta_{k}^{(0)}\right\\}$ and $\left\\{\eta_{k}^{(1)}\right\\}$ |  Sets of the local primal and dual step-sizes |  Adaptively tuned as in Eqs. (27a) and (27b)
$Cat\\#i$, $i=1,2,3$ |  Set of coworkers of $i$-th category |  $\left|Cat\\#1\right|=\left|Cat\\#3\right|=30$ and $\left|Cat\\#2\right|=40$
$\left\\{s_{k}\right\\}$ |  Set of coworker processing speeds (in (CPU cycles/round)) |  $50\times 10^{7}$, $25\times 10^{7}$, and $1.0\times 10^{7}$ (CPU cycles/round) for coworkers of $Cat\\#1$, #2 and #3, respectively
$\left\\{R_{k}\right\\}$ |  Set of coworker communication rates in uplink (in (bit/round)) |  $100\times 10^{4}$, $50\times 10^{4}$, and $2\times 10^{4}$ (bit/round) for coworkers of $Cat\\#1$, #2 and #3, respectively
$\left\\{P_{k}^{(LOSS)}\right\\}$ |  Set of the per-uplink packet-loss probabilities |  0.00, 0.25, and 0.50
$\left\\{\mathcal{D}_{k}\right\\}$ |  Spectrum of the probability distributions of the training data sets |  IID, 1-nonIID, and 2-nonIID
$\Omega^{(AF)}$, $\Theta^{(AF)}$, $\Psi^{(AF)}$, $V^{(AF)}$, and $\Phi^{(AF)}$ |  Designed functions for the adaptive tuning of the AFAFed numbers of consecutive local updates, tolerance thresholds, fairness coefficients, and mixing parameter |  Set as in Eqs. (26), (30), (39), and (43)
$Iter_{MAX}^{(AF)}$ |  AFAFed maximum number of consecutive local updates |  1 and 30
$I^{(FAG)}$ |  FedAvg number of consecutive local updates |  Statically set to $\left\lceil\overline{I}^{(AF)}\right\rceil$
$I^{(FPR)}$ |  FedProx number of consecutive local updates |  Statically set to $\left\lceil\overline{I}^{(AF)}\right\rceil$
$\beta^{(FAS)}$ |  FedAsync mixing parameter |  Set to: $1/\sqrt{1+age(t)}$
$I^{(FAS)}$ |  FedAsync number of consecutive local updates |  Statically set to $\left\lceil\overline{I}^{(AF)}\right\rceil$
$s^{(CS-SGD)}$ |  CS-SGD processing speed (in (CPU cycles/round)) |  $3.28\times 10^{10}$ (CPU cycles/round)
$\left|MB^{(CS-SGD)}\right|$ |  CS-SGD mini-batch size |  $K\times\left|MB\right|$

##  Appendix B Derivations of some auxiliary results

In this Appendix B, we provide the proofs of some auxiliary results.

Proof of the bound in Eq. (52) on the gradient variance. The derivation of the
bound in Eq. (52) relies on the following chain of developments:

$\begin{split}E_{\mathcal{D}_{k}}\left\\{\left\|\vv{\nabla}\widetilde{F}_{k}\\!\left(\vv{w}\right)-\vv{\nabla}F\\!\left(\vv{w}\right)\right\|^{2}\right\\}&=E_{\mathcal{D}_{k}}\left\\{\left\|\left(\vv{\nabla}\widetilde{F}_{k}\\!\left(\vv{w}\right)-\vv{\nabla}F_{k}\\!\left(\vv{w}\right)\right)-\left(\vv{\nabla}F\\!\left(\vv{w}\right)-\vv{\nabla}F_{k}\\!\left(\vv{w}\right)\right)\right\|^{2}\right\\}\\\
&\overset{(a)}{=}E_{\mathcal{D}_{k}}\left\\{\left\|\vv{\nabla}\widetilde{F}_{k}\\!\left(\vv{w}\right)-\vv{\nabla}F_{k}\\!\left(\vv{w}\right)\right\|^{2}+\left\|\vv{\nabla}F_{k}\\!\left(\vv{w}\right)-\vv{\nabla}F\\!\left(\vv{w}\right)\right\|^{2}\right\\}\overset{(b)}{\leq}\frac{\sigma_{k,1}^{2}}{\left|MB_{k}\right|}+\sigma_{k,2}^{2},\end{split}$
(73)

where: (a) follows from the assumption of unbiased gradient estimation in Eq.
(48); and, (b) stems from the assumption in Eqs. (49) and (50) of bounded
gradient variances. ∎

Proof of Lemma 1. In order to prove the bounds on the $C$ parameter in Eq.
(63), we begin to observe that the following chain of inequalities holds:

$\left(E_{\xi\sim
P}\left\\{\widehat{\vv{G}}(\cdot)\right\\}\right)^{T}\vv{\nabla}F(\cdot)\overset{(a)}{=}\left(E_{\xi\sim
P}\left\\{\vv{\nabla}F(\cdot)+\vv{n}\right\\}\right)^{T}\vv{\nabla}F(\cdot)\overset{(b)}{=}\left\|\vv{\nabla}F(\cdot)\right\|^{2}\left(1+\gamma_{0}\right),$
(74)

where: (a) follows from the definition of the estimation noise:
$\widehat{\vv{G}}(\cdot)\equiv\vv{\nabla}F(\cdot)+\vv{n}$; and, (b) stems from
Eq. (61). Hence, by leveraging Eq. (74), we may re-write the condition in Eq.
(57) as in:
$\left\|\vv{\nabla}F(\cdot)\right\|^{2}\left(1+\gamma_{0}\right)\geq
C\left\|\vv{\nabla}F(\cdot)\right\|^{2}$, which, in turn, leads to the upper
bound: $\left(1+\gamma_{0}\right)\geq C$. Furthermore, since $C$ must be, by
definition, positive, we must have: $\gamma_{0}>-1$.

In order to derive the bound in Eq. (63) on the $\Gamma$ parameter, we observe
that the following equality chain holds:

$\left\|E_{\xi\sim
P}\left\\{\widehat{\vv{G}}(\cdot)\right\\}\right\|=\left\|\vv{\nabla}F(\cdot)+E_{\xi\sim
P}\left\\{\vv{n}\right\\}\right\|\overset{(a)}{=}\left\|\vv{\nabla}F(\cdot)+\gamma_{0}\vv{\nabla}F(\cdot)\right\|\overset{(b)}{=}\left(1+\gamma_{0}\right)\vv{\nabla}F(\cdot),$
(75)

where: (a) arises from Eq. (61); and, (b) is assured by the fact that
$\gamma_{0}>-1$. Therefore, after inserting Eq. (75) into the l.h.s. of Eq.
(58), we arrive at the lower bound on $\Gamma$ of Eq. (63).

Finally, in order to derive the lower bound on $A$ in Eq. (63), we develop the
defining relationship of Eq. (59) of $\sigma_{\widehat{\vv{G}}}^{2}$ as
follows:

$\begin{split}\sigma_{\widehat{\vv{G}}}^{2}&=E_{\xi\sim
P}\left\\{\left\|\widehat{\vv{G}}(\cdot)\right\|^{2}\right\\}-\left\|E_{\xi\sim
P}\left\\{\widehat{\vv{G}}(\cdot)\right\\}\right\|^{2}\overset{(a)}{=}E_{\xi\sim
P}\left\\{\left\|\vv{\nabla}F(\cdot)+\vv{n}\right\|^{2}\right\\}-\left\|E_{\xi\sim
P}\left\\{\vv{\nabla}F(\cdot)+\vv{n}\right\\}\right\|^{2}\\\
&\overset{(b)}{=}\left(1+\gamma_{1}+2\gamma_{0}\right)\left\|\vv{\nabla}F(\cdot)\right\|^{2}+b_{0}-\left(1+\gamma_{0}\right)^{2}\left\|\vv{\nabla}F(\cdot)\right\|^{2}\equiv\left(\gamma_{1}-\gamma_{0}^{2}\right)\left\|\vv{\nabla}F(\cdot)\right\|^{2}+b_{0},\end{split}$
(76)

where: (a) follows from the definition:
$\widehat{\vv{G}}(\cdot)=\vv{\nabla}F(\cdot)+\vv{n}$ of the estimation noise;
and, (b) stems from Eqs. (61) and (62). Hence, after noting that the
condition: $0\leq\left(\gamma_{1}-\gamma_{0}^{2}\right)\leq 1$, in Eq. (62)
guarantees that: (i) the last expression in Eq. (76) is non-negative for non-
negative $b_{0}$; and, (ii) the factor
$\left(\gamma_{1}-\gamma_{0}^{2}\right)$ in Eq. (76) is upper bounded by the
unit, the following upper bound follows from Eq. (76):

$\sigma_{\widehat{\vv{G}}}^{2}\leq\left\|\vv{\nabla}F(\cdot)\right\|^{2}+b_{0}.$
(77)

Therefore, the comparison of the bounds in Eqs. (60) and (77) proves the
validity of the lower bound on $A$ in Eq. (63). Finally, we note that, in the
case in which the stochastic gradient $\widehat{\vv{G}}(\cdot)$ would provide
an unbiased estimate of the full gradient $\vv{\nabla}F(\cdot)$, the
expectation of the corresponding estimation noise $\vv{n}$ vanishes and its
second-order moment must be constant. This requires, in turn, that both
$\gamma_{0}$ and $\gamma_{1}$ in Eqs. (61) and (62) must vanish. The proof of
Lemma 1 is now completed. ∎

Derivation of the scaling relationships in Eqs. (68a), (68b), and (68c). We
begin to rewrite the primal iteration of Eq. (16a) in the following compact
form:

$\vv{w}_{k}\\!\left(t^{(k)}+1\right)=\vv{w}_{k}\\!\left(t^{(k)}\right)-\eta_{k}^{(0)}\vv{\nabla}Z_{k}\\!\left(\vv{w}_{k}\left(t^{(k)}\right)\right),\quad
t^{(k)}=0,1,\ldots,$ (78)

with the dummy position:

$\vv{\nabla}Z_{k}\\!\left(\vv{w}_{k}\left(t^{(k)}\right)\right)\triangleq\lambda_{k}^{(LAST)}\vv{\nabla}\widetilde{F}_{k}\\!\left(\vv{w}_{k}\\!\left(t^{(k)}\right)\right)+\mu_{k}\\!\left(t^{(k)}\right)\left(\vv{w}_{k}\left(t^{(k)}\right)-\vv{\overline{w}}_{k}^{(LAST)}\right).$
(79)

Afterwards, without loss of generality, let us assume that $coworker\\#k$: (i)
starts a cluster of $Iter_{k}\geq 1$ iterations at its local index time
$t^{(k)}$ by moving from $\vv{\overline{w}}_{k}^{(LAST)}$; and, then, (ii)
communicates its updated local model:
$\vv{w}_{k}\\!\left(t^{(k)}+Iter_{k}\right)$ to the central server at the
global time index: $t+1$. Hence, by iterating Eq. (78) from: $t^{(k)}$ up to:
$t^{(k)}+Iter_{k}$, we obtain:

$\vv{w}_{k}\\!\left(t^{(k)}+Iter_{k}\right)=\vv{w}_{k}^{(LAST)}-\sum_{j=t^{(k)}}^{t^{(k)}+Iter_{k}-1}\eta_{k}^{(0)}(j)\,\vv{\nabla}Z_{k}\\!\left(\vv{w}_{k}(j)\right).$
(80)

Since, by assumption, the local time index: $t^{(k)}+Iter_{k}$ at which
$coworker\\#k$ ends its current iteration cluster equates to the global time
index $(t+1)$ at which it communicates its updated model to the central server
(i.e., $t^{(k)}+Iter_{k}\equiv t+1$, and $\xi_{k}(t+1)\equiv 1$), the
resulting expression assumed by the global stochastic gradient
$\widehat{\vv{G}}(\cdot)$ may be obtained by inserting Eq. (80) into the
defining relationship of Eq. (21) , and, then, it reads as follows:

$\widehat{\vv{G}}\\!\left(\vv{\overline{w}}(t);\xi_{k}(t+1)\equiv
1\right)=\vv{\overline{w}}(t)-\vv{w}_{k}^{(LAST)}+\sum_{j=t^{(k)}}^{t^{(k)}+Iter_{k}\left(t^{(k)}\right)-1}\eta_{k}^{(0)}(j)\,\vv{\nabla}Z_{k}\\!\left(\vv{w}_{k}(j)\right).$
(81)

Hence, after performing the expectation of both sides of Eq. (81) over the set
of the access probabilities in Eq. (45), we arrive (after some tedious but
quite standard algebras) at the following scaling relationships for the first
and second-order moments of $\widehat{\vv{G}}(\cdot)$:

$E_{\xi\sim
P}\left\\{\widehat{\vv{G}}(\cdot)\right\\}\propto\mathcal{O}\\!\left(\overline{I}\right),$
(82) $\mathcal{O}\\!\left(\overline{I}^{2}\right)\leq E_{\xi\sim
P}\left\\{\left\|\widehat{\vv{G}}(\cdot)\right\|^{2}\right\\}\leq\mathcal{O}\\!\left(\overline{I^{2}}\right),$
(83)

and,

$\mathcal{O}\\!\left(E_{\xi\sim
P}\left\\{\left\|\widehat{\vv{G}}(\cdot)\right\|^{2}\right\\}\right)\leq\mathcal{O}\\!\left(\overline{\Sigma^{2}}\right).$
(84)

Finally, after observing that, from Eqs. (83) and (84), the variance
$\sigma_{\widehat{\vv{G}}}^{2}$ in Eq. (59) scales (at most) as the product
$\mathcal{O}\left(\overline{I^{2}}\>\overline{\Sigma^{2}}\right)$, the
relationships in Eqs. (68a), (68b), and (68c) follow by accounting for the
scaling expressions in Eqs. (82)–(84) in the bounds of Eqs. (57), (58), and
(60). ∎

Proof of Lemma 2. From the bound in Eq. (57) on $C$, we obtain:
$C\leq\left(E_{\xi\sim
P}\left\\{\widehat{\vv{G}}\right\\}\right)^{T}\,\vv{\nabla}F/\left\|\vv{\nabla}F\right\|^{2}$,
so that, since we must have $C>0$, we conclude that the condition in Eq. (71a)
is both necessary and sufficient for the feasibility of the estimate in Eq.
(72a). Regarding the parameter $\Gamma$ in Eq. (57), from Assumption 4.1, we
must have: $\Gamma>C$, so that without loss of generality, we may pose:

$\Gamma=\left(1+k_{0}\right)C,$ (85)

with $k_{0}\geq 0$. Furthermore, from the bound in Eq. (57), we obtain the
following limit value on the feasible values of $\Gamma$:

$\Gamma\geq\frac{\left\|E_{\xi\sim
P}\left\\{\widehat{\vv{G}}\right\\}\right\|}{\left\|\vv{\nabla}F\right\|}.$
(86)

Hence, a comparison of the relationships in Eqs. (85) and (86) directly leads
to the conclusion that $\widehat{\Gamma}$ in Eq. (72b) provides a feasible
estimate of $\Gamma$ if and only if the condition in Eq. (71b) is met.
Finally, the bound in Eq. (60) may be rewritten as:
$A\geq\sigma_{\widehat{\vv{G}}}^{2}-\left\|\vv{\nabla}F\\!\left(\vv{\overline{w}}\right)\right\|^{2}$,
with $\sigma_{\widehat{\vv{G}}}^{2}$ defined as in Eq. (59). Hence, since $A$
must be non-negative by Assumption 4.2, we conclude that $\widehat{A}$ in Eq.
(72c) is a feasible estimate of $A$ if and only if the condition in Eq. (71c)
is met. ∎

##  Appendix C Proof of Proposition 1

Assumption 3.2 on the $\zeta$-smoothness of the local loss functions
guarantees that also the resulting global loss function $F(\cdot)$ in Eq. (4)
is $\zeta$-smooth (see, for example, Lemma 1 of [16]), so that, from Lemma 3.4
of [36], it meets the following inequality:

$F\\!\left(\vv{w}_{1}\right)\leq
F\\!\left(\vv{w}_{2}\right)+\vv{\nabla}F\\!\left(\vv{w}_{2}\right)^{T}\left(\vv{w}_{1}-\vv{w}_{2}\right)+\frac{\zeta}{2}\left\|\vv{w}_{1}-\vv{w}_{2}\right\|^{2},\quad\forall\vv{w}_{1},\vv{w}_{2}\in\mathbb{R}^{l}.$
(87)

Hence, after posing: $\vv{w}_{1}=\vv{\overline{w}}(t+1)$, and:
$\vv{w}_{2}=\vv{\overline{w}}(t)$ in Eq. (87), and, then, replacing
$\vv{\overline{w}}(t+1)$ by the r.h.s. of Eq. (20b), we obtain:

$F\\!\left(\vv{\overline{w}}(t+1)\right)-F\\!\left(\vv{\overline{w}}(t)\right)\leq-\beta(t)\vv{\nabla}F\\!\left(\vv{\overline{w}}(t)\right)^{T}\,\widehat{\vv{G}}\\!\left(\vv{\overline{w}}(t);\left\\{\xi_{k}(t+1)\right\\}\right)+\frac{\zeta}{2}\beta(t)^{2}\left\|\widehat{\vv{G}}\\!\left(\vv{\overline{w}}(t);\left\\{\xi_{k}(t+1)\right\\}\right)\right\|^{2},$
(88)

which, by performing the expectation of both sides of Eq. (88) the access
probabilities of Eq. (45) and, then, exploiting the gradient coherence
assumption of Eq. (57), leads, in turn, to:

$E_{\xi\sim
P}\left\\{F\\!\left(\vv{\overline{w}}(t+1)\right)-F\\!\left(\vv{\overline{w}}(t)\right)\right\\}\leq-\beta(t)C\left\|\vv{\nabla}F\\!\left(\vv{\overline{w}}(t)\right)\right\|^{2}+\frac{\zeta}{2}\beta(t)^{2}E_{\xi\sim
P}\left\\{\left\|\widehat{\vv{G}}\\!\left(\vv{\overline{w}}(t);\left\\{\xi_{k}(t+1)\right\\}\right)\right\|^{2}\right\\}.$
(89)

Furthermore, since the introduction of the inequalities in Eqs. (58) and (60)
into the defining relationship of Eq. (59) allows us to upper limit the last
expectation in Eq. (89) as in:

$E_{\xi\sim
P}\left\\{\left\|\widehat{\vv{G}}\\!\left(\vv{\overline{w}}(t);\left\\{\xi_{k}(t+1)\right\\}\right)\right\|^{2}\right\\}\leq
A+\left(1+\Gamma^{2}\right)\left\|\vv{\nabla}F\\!\left(\vv{\overline{w}}(t)\right)\right\|^{2},$
(90)

we may further rework the bound in (89) as follows:

$E_{\xi\sim
P}\left\\{F\\!\left(\vv{\overline{w}}(t+1)\right)-F\\!\left(\vv{\overline{w}}(t)\right)\right\\}\leq-\beta(t)\left\|\vv{\nabla}F\\!\left(\vv{\overline{w}}(t)\right)\right\|^{2}\left[C-\frac{\zeta}{2}\beta(t)\left(1+\Gamma^{2}\right)\right]+\frac{\zeta}{2}\beta(t)^{2}A.$
(91)

Our next task is to upper bound $\beta(t)$ by a suitable constant, in order to
guarantee that the term into the square brackets of Eq. (91) is positive. To
this end, we enforce the following upper bound:

$\frac{\zeta}{2}\beta(t)\left(1+\Gamma^{2}\right)\leq\varepsilon C,$ (92)

where $\varepsilon$ is an user-defined constant, which must fall into the open
interval $(0,\,1)$. So doing, we have that: (i) the term into bracket in Eq.
(91) is guaranteed to be positive, while the mixing parameter $\beta(t)$ in
Eq. (92) is not forced to vanish; and, (ii) $\beta(t)$ is upper limited as in
Eq. (64), so that the bound in Eq. (91) may be reworked as in:

$E_{\xi\sim
P}\left\\{F\\!\left(\vv{\overline{w}}(t+1)\right)-F\\!\left(\vv{\overline{w}}(t)\right)\right\\}\leq-\beta(t)C\left(1-\varepsilon\right)\left\|\vv{\nabla}F\\!\left(\vv{\overline{w}}(t)\right)\right\|^{2}+\frac{\zeta}{2}\beta(t)^{2}A.$
(93)

Hence, after performing the telescopic sum of the terms at both sides of Eq.
(93) over the interval $t\in\left\\{0,1,\ldots,T\right\\}$, and, then, noting
that (see Eq. (6)): $E_{\xi\sim
P}\left\\{F\\!\left(\vv{\overline{w}}(T)\right)\right\\}\geq F^{\ast}$, we
derive the following bound:

$F^{\ast}-E_{\xi\sim
P}\left\\{F\\!\left(\vv{\overline{w}}(0)\right)\right\\}\leq
C\left(1-\varepsilon\right)\sum_{t=0}^{T-1}\beta(t)\left\|\vv{\nabla}F\\!\left(\vv{\overline{w}}(t)\right)\right\|^{2}+\frac{\zeta}{2}A\sum_{t=0}^{T-1}\beta(t)^{2}.$
(94)

So, after dividing both sides of Eq. (94) by $T$ and noting that the term:
$(1-\varepsilon)$ is positive by design, we directly arrive at the bound in
Eq. (65).

Finally, the bound in Eq. (66) stems from the one in Eq. (65) by accounting
for the following two inequalities:

$E\\!\left\\{\frac{1}{T}\sum_{t=0}^{T-1}\beta(t)\left\|\vv{\nabla}F\\!\left(\vv{\overline{w}}(t)\right)\right\|^{2}\right\\}\geq\beta_{MIN}\,E\\!\left\\{\frac{1}{T}\sum_{t=0}^{T-1}\left\|\vv{\nabla}F\\!\left(\vv{\overline{w}}(t)\right)\right\|^{2}\right\\},$
(95)

and,

$E\\!\left\\{\frac{1}{T}\sum_{t=0}^{T-1}\beta(t)\right\\}\leq
E\\!\left\\{\frac{1}{T}\sum_{t=0}^{T-1}\beta_{MAX}\right\\}\equiv\beta_{MAX},$
(96)

while the last bound in Eq. (67) is obtained by setting
$\beta_{MIN}=\beta_{MAX}\equiv\beta$ at the r.h.s. of Eq. (66). The proof of
Proposition 1 is now completed.

## References

  * [1] Cisco Systems, Fog computing and the internet of things: Extend the cloud to where the things are, White paper (2015).
URL https://www.cisco.com/c/dam/en_us/solutions/trends/iot/docs/computing-
overview.pdf

  * [2] D. Hanes, G. Salgueiro, P. Grossetete, R. Barton, J. Henry, IoT Fundamentals – Networking Technologies, Protocols, and Use Cases for the internet of Things, Cisco Press, 2017.
  * [3] B. Custers, A. Sears, F. Dechesne, I. Georgieva, T. Tani, S. van der Hof, EU personal data protection in policy and practice, Springer, Berlin, 2019.
  * [4] H. Li, K. Ota, M. Dong, Learning IoT in edge: deep learning for the Internet of Things with edge computing, IEEE Network 32 (2018) 96–101.  doi:10.1109/MNET.2018.1700202.
  * [5] E. Baccarelli, P. G. Vinueza Naranjo, M. Scarpiniti, M. Shojafar, J. H. Abawajy, Fog of everything: Energy-efficient networked computing architectures, research challenges, and a case study, IEEE Access 5 (2017) 9882–9910.  doi:10.1109/ACCESS.2017.2702013.
  * [6] E. Baccarelli, M. Scarpiniti, A. Momenzadeh, EcoMobiFog – Design and dynamic optimization of a 5G Mobile-Fog-Cloud multi-tier ecosystem for the real-time distributed execution of stream applications, IEEE Access 7 (2019) 55565–55608.  doi:10.1109/ACCESS.2019.2913564.
  * [7] E. Baccarelli, M. Scarpiniti, A. Momenzadeh, S. Sarv Ahrabi, Learning-in-the-Fog (LiFo): Deep learning meets fog computing for the minimum-energy distributed early-exit of inference in delay-critical IoT realms, IEEE Access 9 (2021) 2571–25757.  doi:10.1109/ACCESS.2021.3058021.
  * [8] H. B. McMahan, E. Moore, D. Ramage, B. Agüera y Arcas, Federated learning of deep networks using model averaging, arXiv:1602.05629v1 (2016) 1–11.
  * [9] T. S. Brisimi, R. Chen, T. Mela, A. Olshevsky, I. C. Paschalidis, W. Shi, Federated learning of predictive models from federated electronic health records, International Journal of Medical Informatics 112 (2018) 59–67.  doi:10.1016/j.ijmedinf.2018.01.007.
  * [10] E. Baccarelli, P. G. Vinueza Naranjo, M. Shojafar, M. Scarpiniti, Q*: Energy and delay-efficient dynamic queue management in tcp/ip virtualized data centers, Computer Communications 102 (2017) 89–106.  doi:10.1016/j.comcom.2016.12.010.
  * [11] C. Jin, R. Ge, P. Netrapalli, S. M. Kakade, M. I. Jordan, How to escape saddle points efficiently, in: Proceedings of the 34th International Conference on Machine Learning (ICML 2017), Vol. 70, Sydney, Australia, 2017, pp. 1724–1732.
  * [12] M. Mohammadi, A. Al-Fuqaha, S. Sorour, M. Guizani, Deep learning for IoT big data and streaming analytics: a survey, IEEE Commun. Surveys Tuts. 20 (4) (2018) 2923–2960.  doi:10.1109/COMST.2018.2844341.
  * [13] M. Chen, B. Mao, T. Ma, Efficient and robust asynchronous federated learning with stragglers, in: Seventh International Conference on Learning Representations (ICLR 2019), 2019, pp. 1–15.
  * [14] I. Goodfellow, Y. Bengio, A. Courville, Deep Learning, MIT Press, 2016.
  * [15] H. B. McMahan, E. Moore, D. Ramage, S. Hampson, B. Agüera y Arcas, Communication-efficient learning of deep networks from decentralized data, in: Proceedings of the 20th International Conference on Artificial Intelligence and Statistics (AISTATS 2017), Vol. PMLR 54, Fort Lauderdale, Florida,USA, 2017, pp. 1273–1282.
  * [16] S. Wang, T. Tuor, T. Salonindis, K. K. Leung, C. Makaya, T. He, K. Chan, Adaptive federated learning in resource constrained edge computing systems, IEEE Journal on Selected Areas in Communications 37 (2019) 1205–1221.  doi:10.1109/JSAC.2019.2904348.
  * [17] M. Mohri, G. Sivek, A. T. Suresh, Agnostic federated learning, in: Proceedings of the 36th International Conference on Machine Learning, 2019, pp. PMLR 97:4615–4625.
  * [18] A. Koppel, B. M. Sadler, A. Ribeiro, Proximity without consensus in online multi-agent optimization, IEEE Transactions on Signal Processing 65 (2017) 3062–3077.  doi:10.1109/TSP.2017.2686368.
  * [19] B. Boyd, N. Parikh, E. Chu, B. Peleato, J. Eckstein, Distributed optimization and statistical learning via the alternating direction method of multipliers, Foundations and Trends® in Machine Learning 3 (2011) 1–122.  doi:10.1561/2200000016.
  * [20] T. Li, A. K. Sahu, M. Zaheer, M. Sanjabi, A. Talwalkar, V. Smith, Federated optimization in heterogeneous networks, in: Proceedings of the 3rd Conference on Machine Learning and Systems (MLSys 2020), Austin, TX, USA, 2020, pp. 429–450.
  * [21] C. Xie, S. Koyejo, I. Gupta, Asynchronous federated optimization, in: 12th Annual Workshop on Optimization for Machine Learning (OPT 2020), 2020, pp. 1–11.
  * [22] M. S. Bazaraa, H. D. Sherali, C. M. Shetty, Nonlinear programming: theory and algorithms, 3rd Edition, John Wiley and Sons, New Jersey, USA, 2017.
  * [23] X. Zhang, X. Zhu, J. Wang, H. Yan, H. Chen, W. Bao, Federated learning with adaptive communication compression under dynamic bandwidth and unreliable networks, Information Sciences 540 (2020) 242–262.  doi:10.1016/j.ins.2020.05.137.
  * [24] S. Niknam, H. S. Dillon, J. H. Reed, Federated learning for wireless communication: motivation, opportunities and challenges, IEEE Communications Magazine 58 (2020) 46–51.  doi:10.1109/MCOM.001.1900461.
  * [25] Q. Yang, Y. Liu, Y. Cheng, Y. Kang, T. Chen, H. Yu, Federated Learning, Morgan & Claypool, 2019.
  * [26] Z. Peng, Y. Xu, M. Yan, W. Yin, On the convergence of asynchronous parallel iteration with unbounded delays, Journal of the Operations Research Society of China 7 (2019) 5–42.  doi:10.1007/s40305-017-0183-1.
  * [27] X. Li, K. Huang, W. Yang, S. Wang, Z. Zhang, On the convergence of FedAvg on non-iid data, in: 2020 International Conference on Learning Representations (ICLR), 2020, pp. 1–26.
  * [28] S. Zhang, A. Choromanska, Y. LeCun, Deep learning with elastic averaging SGD, in: Twenty-ninth Conference on Neural Information Processing Systems (NISP 2015), 2015, pp. 1–9.
  * [29] Y. Chen, Y. Ning, M. Slawski, H. Rangwala, Asynchronous online federated learning for edge devices with non-iid data, in: 2020 IEEE International Conference on Big Data, Atlanta, GA, USA, 2020, pp. 15–24.  doi:10.1109/BigData50022.2020.9378161.
  * [30] Q. Peng, A. Walid, J. Hwang, S. H. Low, Multipath TCP: Analysis, design, and implementation, IEEE/ACM Transactions on Networking 24 (1) (2016) 596–609.  doi:10.1109/TNET.2014.2379698.
  * [31] A. Kumar, D. Manjunath, J. Kuri, Communication networking: An analytical approach, Elsevier, Morgan Kaufman Publishers, 2004.  doi:10.1016/B978-0-12-428751-8.X5000-9.
  * [32] W. Zhang, S. Gupta, X. Lian, J. Liu, Staleness-aware Async-SGD for distributed deep learning, in: Proceedings of the Twenty-Fifth International Joint Conference on Artificial Intelligence (IJCAI’16), 2016, pp. 2350–2356.
  * [33] X. Lian, Y. Huang, Y. Li, J. Liu, Asynchronous parallel stochastic gradient for nonconvex optimization, in: Twenty-ninth Conference on Neural Information Processing Systems (NIPS 2015), 2015, pp. 1–9.
  * [34] H. R. Feyzmahdavian, A. Aytekin, M. Johansson, An asynchronous mini-batch algorithm for regularized stochastic optimization, IEEE Transactions on Automatic Control 61 (2016) 3740–3754.  doi:10.1109/TAC.2016.2525015.
  * [35] W. Dai, Y. Zhou, N. Dong, H. Zhang, E. P. Xing, Toward understanding the impact of staleness in distributed machine learning, in: 7th International Conference on Learning Representations (ICLR 2019), New Orleans, LU, USA, 2019, pp. 1–19.
  * [36] S. Bubeck, Convex optimization: Algorithms and complexity, Foundations and Trends® in Machine Learning 8 (2015) 231–357.  doi:10.1561/2200000050.