xleftmargin=15pt, frame=l, basicstyle=, framesep=5mm, fillcolor=, rulecolor=,
numberstyle=

11institutetext: UC Berkeley, Berkeley, CA, USA
11email<EMAIL_ADDRESS>11email<EMAIL_ADDRESS>11email:
<EMAIL_ADDRESS>22institutetext: UT Dallas, Richardson, TX, USA
22email<EMAIL_ADDRESS>33institutetext: TU Dresden, Dresden, Germany
33email<EMAIL_ADDRESS>
# Quantifying and Generalizing the CAP Theorem

Edward A. Lee 11 0000-0002-5663-0584 Soroush Bateni 22 0000-0002-5448-3664
Shaokai Lin 11 0000-0001-6885-5572 Marten Lohstroh 11 0000-0001-8833-4117
Christian Menard 33 0000-0002-7134-8384

###### Abstract

In distributed applications, Brewer’s CAP theorem tells us that when networks
become partitioned, there is a tradeoff between consistency and availability.
Consistency is agreement on the values of shared variables across a system,
and availability is the ability to respond to reads and writes accessing those
shared variables. We quantify these concepts, giving numerical values to
inconsistency and unavailability. Recognizing that network partitioning is not
an all-or-nothing proposition, we replace the P in CAP with L, a numerical
measure of apparent latency, and derive the CAL theorem, an algebraic relation
between inconsistency, unavailability, and apparent latency. This relation
shows that if latency becomes unbounded (e.g., the network becomes
partitioned), then one of inconsistency and unavailability must also become
unbounded, and hence the CAP theorem is a special case of the CAL theorem. We
describe two distributed coordination mechanisms, which we have implemented as
an extension of the Lingua Franca coordination language, that support
arbitrary tradeoffs between consistency and availability as apparent latency
varies. With _centralized coordination_ , inconsistency remains bounded by a
chosen numerical value at the cost that unavailability becomes unbounded under
network partitioning. With _decentralized coordination_ , unavailability
remains bounded by a chosen numerical quantity at the cost that inconsistency
becomes unbounded under network partitioning. Our centralized coordination
mechanism is an extension of techniques that have historically been used for
distributed simulation, an application where consistency is paramount. Our
decentralized coordination mechanism is an extension of techniques that have
been used in distributed databases when availability is paramount.

###### Keywords:

Concurrency Distributed Systems Consistency Availability Network Partitioning
Apparent Latency

##  1 Motivation

In the year 2000, Eric Brewer gave a keynote talk [5] at the Symposium on
Principles of Distributed Computing (PODC) in which he introduced the “CAP
Theorem,” which states that you can have only two of the following three
properties in a distributed system:

  * •

Consistency: Distributed components agree on the value of shared state.

  * •

Availability: Ability to respond to user requests.

  * •

tolerance to network Partitions: The ability to keep operating when
communication fails.

This keynote is credited by many in the distributed computing community with
opening up the field, enabling innovative approaches that offer differing
tradeoffs between these properties.

In 2002, Gilbert and Lynch [17] proved a couple of variants of this theorem,
one rather strong result for “asynchronous networks” (which have no clocks)
[34, chapter 8] and one weaker result for “partially synchronous networks”
(which have unsynchronized clocks that measure the passage of time at the same
rate). These results reinforced Brewer’s original interpretation that you
could obtain any two of the three properties, but not all three.

In 2012, Brewer wrote a retrospective [6] in which he observes that tradeoff
is more subtle than this. First, the “P” property is not really one you can
trade off against the others. You cannot choose whether to have network
partitions. Brewer clarified the design problem as one of how to trade off
consistency against availability when network partitions occur.

In the same year, Abadi [1] reacted that CAP is irrelevant when there are no
network partitions, but Brewer [6] points out that network partitions are not
a binary property; all networks have latency, and a complete communication
failure is just the limiting case when the latency goes to infinity. Abadi
[1], in fact, emphasizes the tradeoff in database design between consistency
and latency experienced by the user, arguing that it has had more impact on
actual database system designs than CAP. He then observes that “availability
and latency are arguably the same thing” and calls for unifying the
consistency/latency tradeoff with the CAP tradeoffs.

There are (at least) two latencies of interest: network latency and user
latency. Whereas network latency is the time that it takes messages to
traverse the network, user latency, which we will refer to as unavailability,
is the time it takes to respond to a user. Specifically, unavailability is the
physical time that elapses between when a system first receives a request from
a user and when the system sends a response to the user. Network partitioning
is the limiting case of network latency. When the network latency goes to
infinity, the network is partitioned. Hence, we derive a CAL theorem,
replacing “tolerance for network Partitioning” with “tolerance for network
Latency.” This theorem will establish a numerical relationship between
inconsistency, unavailability, and network latency, thereby unifying Brewer’s
and Abadi’s positions and having the CAP theorem as a special case. Our
resulting formulas make clear the tradeoff between these quantities and enable
a range of designs that vary in their emphasis on one or more of these
quantities.

To establish the relationship, we have to carefully define the three
variables: inconsistency, unavailability, and network latency. Most designers
have strong intuitive understandings of the meanings of these terms, but there
are pitfalls. For example, an intuitive statement about a consistent view of a
shared variable $x$ goes something like this: If component 1 writes to $x$
“before” component 2 reads the value of $x$, then component 2 will read the
updated value. Another such statement is that if component 1 writes to $x$
“before” component 2 also writes to $x$, then all observers of $x$ will
“subsequently” see the value written by component 2. These statements are
difficult to formalize, however, because the words “before” and “subsequently”
are problematic. The order in which two physically separated events occur may
be not only unknown, but also, at a fundamental level, unknowable. From
special relativity, the order in which physically separated events occur
depends on the frame of reference. In modern distributed systems—some of which
span the globe and all of which have communication that is limited by the
speed of light—if many updates and reads of shared variables are occurring,
some of those are sure to have ambiguous ordering. So we will have to do
better than this. We need a more solid foundation for the notion of
consistency.

In Section 2, we give an overview of the concept of consistency, the role of
time, and the use of timestamps. In Section 3, we define the numerical
quantities inconsistency, unavailability, and network latency and derive the
CAL theorem, which relates these three quantities. In Section 4, we give
complete programs illustrating the tradeoffs implied by the CAL theorem, and
show how the Lingua Franca coordination language supports being explicit about
these tradeoffs. In Section 5, we describe two implementations: one that is
capable of bounding inconsistency, and one that is capable of bounding
unavailability in the face of network partitions. Section 6 offers some
concluding remarks.

##  2 Consistency and Time

Consider a shared variable $x$ that is read and written to by various
sequential processes running on nodes distributed across a network. An
intuitive form of consistency requires that all nodes agree on the order of
operations on $x$ (reads and writes). Gilbert and Lynch say “The most natural
way of formalizing the idea of a consistent service is as an atomic data
object” [17], where “atomic” comes from Lamport [25]. “Atomic” here is
interchangeable with the term “linearizable” that comes from Herlihy and Wing
[19]. This interpretation defines “consistent” to mean that all observers see
the same sequence of atomic operations (reads and writes) on the data object.

For most applications, however, this is too strong. It is sufficient to agree
on the order in which writes occur, and to agree on the order in which reads
occur relative to those writes, but otherwise, there is usually no need to
agree on the order in which the reads occur relative to each other. But even
this weaker agreement may be too strong. Consider two independent writes to a
shared variable $x$ that occur nearly simultaneously on two distinct nodes.
Depending on how these writes are to be merged, it may not be necessary to
define the order in which they occur.

There is also the question of what we mean by “agreement.” When should the
nodes agree? And what is agreement?

These questions have as a backdrop some notion of physical time. Intuitively,
physical time has to play a major role. It would be strange to design a system
that considers a write to $x$ that occurs three weeks after another write to
$x$ to have occurred “before.” But physical time is a slippery, poorly
understood thing. Newtonian time, with all its beauty and simplicity, has
limited applicability, and there is no perfect measurement of it.
Fundamentally, the ordering and agreement that we need are logical (or
semantic) properties, not physical properties. But to make them useful in
applications, we will need physical realizations that are reasonably faithful
to our semantic models using only practical, realizable measurements of
physical time. We require, therefore, formal semantic models and physical
realizations that, with high confidence, conform to these models.

The events we will consider in this paper are reading and writing values to or
from some shared variable $x$. The value of $x$ may be a text, a number, or
some data structure, and the semantics of a “write” will matter. In general, a
“write” is a merge of a new value with an old value. For example, $x$ may be a
number and a “write” may replace the old value or simply add to it. The latter
merge operation is associative and commutative, assuming perfect arithmetic,
where there are no overflows or rounding errors. The replacement merge
operation is associative but not commutative. If $x$ is a text, the “write”
may append to the text, again giving a merge operation that is also
associative but not commutative.

If the merge operation is associative and commutative, then we have two of the
three properties of the ACID 2.0 database principle, proposed by Helland and
Campbell [18]. The letters stand for Associative, Commutative, Idempotent, and
Distributed. Idempotence means that operations that are applied more than once
have the same effect as operations that are applied exactly once. We will
assume that idempotence is realized at a layer below, not in the application
code. In Section 4, we describe a system realization where this is the case.
Hence, the application code can assume that writes are applied exactly once
(in the absence of failures).

If we are to have availability in the presence of network partitioning, then
some amount of replication is necessary [1]. That is, the value of $x$ will
need to be stored at multiple nodes. It is this replication that creates the
consistency problem, but without it, we cannot assure availability. We will
assume that reads and writes to $x$ can occur at any of the nodes.

###  2.1 Causal Consistency

An elegant formal model of consistency, called causal consistency, is analyzed
in depth by Schwartz and Mattern [41]. They define a causality relation,
written $e_{1}\rightarrow e_{2}$, between events $e_{1}$ and $e_{2}$ to mean
that $e_{1}$ can causally affect $e_{2}$. The phrase “causally affect” is
rather difficult to pin down (see Lee [27, Chapter 11] for the subtleties
around the notion of causation), but, intuitively, $e_{1}\rightarrow e_{2}$
means $e_{2}$ cannot behave as if $e_{1}$ had not occurred. Put another way,
if the effect of an event is reflected in the state of a local replica of a
variable $x$, then any cause of the event must also be reflected. Put yet
another way, an observer must never observe an effect before its cause.

For example, if $e_{1}$ writes $x=1$ and $e_{2}$ sends the value of $x$ to
another process, then $e_{1}\rightarrow e_{2}$ means that whatever value
$e_{2}$ sends is either the $x=1$ value written by $e_{1}$ or some
subsequently written value.

Following Schwartz and Mattern, we will impose a bit more structure on our
distributed applications. First, we define a process to be a totally ordered
sequence of events occurring on a single machine and maintaining a single
replica of one or more shared variables. A process, therefore, is the trace
(the record of events) of a single thread of execution. Second, for each
shared variable $x$, we will consider four types of events:

  1. 1.

$w_{x}$: Merge a value with the local replica of $x$.

  2. 2.

$d_{x}$: Read the value of the local replica of $x$.

  3. 3.

$s_{x}$: Send the value of the local replica of $x$ to some set of other
processes.

  4. 4.

$r_{x}$: Receive a new value for $x$ and merge it with the local replica.

When the type of event is important, we will use one of the four symbols above
to denote the event. Otherwise, we will use the symbol $e$. If the variable
$x$ is clear from context, we will omit the subscript. Typical merge
operations are replace, append, and add, but the merge operation can be
arbitrarily sophisticated. To maintain consistency, we will find it necessary
for all replicas of $x$ to use the same merge operation.

Formally, the causality relation of Schwartz and Mattern is the smallest
_transitive_ relation such that $e_{1}\rightarrow e_{2}$ if $e_{1}$ precedes
$e_{2}$ in a process, or $e_{1}$ is the sending of a value in one process
(event type $s_{x}$) and $e_{2}$ is the receipt of the value in another
process (event type $r_{x}$). If neither $e_{1}\rightarrow e_{2}$ nor
$e_{2}\rightarrow e_{1}$ holds, then we write $e_{1}||e_{2}$ or $e_{2}||e_{1}$
and say that $e_{1}$ and $e_{2}$ are incomparable. The causality relation is
identical to the “happened before” relation of Lamport [26], but Schwartz and
Mattern prefer the term “causality relation” because even if $e_{1}$ occurs
unambiguously earlier than $e_{2}$ in physical time, they may nevertheless be
incomparable, $e_{1}||e_{2}$.

The causality relation is a strict partial order. Schwartz and Mattern use
their causality relation to define a “consistent global snapshot” of a
distributed computation to be a subset $S$ of all the events $E$ in the
execution that is a downset, meaning that if $e^{\prime}\in S$ and
$e\rightarrow e^{\prime}$, then $e\in S$ (this was previously called a
“consistent cut” by Mattern [36]).

Figure 1: Causal ordering of events in a bulletin board example.

Consider for example a shared online bulletin board, like a Facebook Timeline.
Consider the following events, adapted from Bailis et al. [4]:

  * •

$w_{1}$: Joe posts a picture of Sally at a recent party by writing to a local
copy.

  * •

$w_{2}$: Sally posts that her son Billy is missing, writing to a local copy.

  * •

$s_{1}$: Sally’s message is sent to Joe’s process.

  * •

$r_{1}$: Joe’s machine receives the message and updates his local copy.

  * •

$w_{3}$: Sally posts that her son has been found (on the local copy).

  * •

$s_{2}$: Sally’s message is sent to Joe’s process.

  * •

$r_{2}$: Joe’s machine receives the second message and updates his local copy.

  * •

$d_{1}$: Joe reads Sally’s messages.

  * •

$w_{4}$: Joe posts “That’s good news, a relief.”

The immediate causal relations are depicted by arrows in Fig. 1. The complete
causal relation is the transitive closure of the depicted causal relations.

Figure 2: Observer of the exchange between Joe and Sally.

Suppose that in this application the merge operation is append. Each write to
the local copy of the bulletin board simply appends the message to the end of
the board. This operation is associative but not commutative. After the writes
$w_{1}$ and $w_{2}$, Sally and Joe have inconsistent views of the shared
variable. A third observer, as indicated in Fig. 2, may have yet a third
inconsistent view. In the figure, after the event $r_{4}$, the $p_{3}$
observer sees both messages, albeit in arbitrary order that has been
determined by the happenstances of network latency. However, in this case,
there is no violation of _causal_ consistency because $w_{1}||w_{2}$.
Semantically, Joe’s and Sally’s first writes are unordered, and different
observers may see none, one, or both messages, and in the latter case, they
may see them in either order. Causal consistency tolerates this form of
nondeterminism.

A worse problem, however, is illustrated later in the same figure. After event
$r_{5}$, the $p_{3}$ observer has seen Sally’s first post and Joe’s second
(“That’s good news”) but has not seen Sally’s second post. At the read event
$d_{2}$, this observer will think that Joe has responded “that’s good news” to
Sally’s message that Billy is missing! This observer would have to conclude
that Joe really does not like Sally’s son. After event $r_{5}$, the state of
$p_{3}$’s local replica is _causally inconsistent_ because $p_{3}$ sees the
effects of $w_{4}$, but not the effects of $w_{3}$, and $w_{3}\rightarrow
w_{4}$.

Leveraging the results of Charron-Bost [9], Schwartz and Mattern argue that
there is no simple solution that ensures causal consistency that does not also
impose additional unnecessary constraints. They give a non-simple solution
using vector clocks. It is non-simple in that it requires, in the worst case,
message sizes that are order $N$, where $N$ is the number of communicating
processes. In subsequent sections, we will describe a solution using
timestamps that has constant overhead and exploits static information about
communication patterns between nodes in order to reduce the number of
unnecessary additional constraints.

###  2.2 Causal Consistency is Insufficient

In practice, causal consistency is necessary but not sufficient for most
applications. It may be perfectly reasonable to impose additional constraints.
In the bulletin board application, for example, it may not be acceptable for
different observers to see Sally and Joe’s first posts in arbitrary order. One
common way to prevent that is to timestamp the posts using some clock
measuring physical time on the machine where the post originates. All
observers could then arrange whatever posts they see in timestamp order,
breaking ties with some consistent algorithm (e.g. alphabetical ordering).
These timestamps give a semantics to the problematic “before” and
“subsequently” that we encountered in Section 1. A general form of this
solution is often called “last writer wins,” but that phrase only makes sense
if the merge policy is replace. So we will call the policy timestamp-ordered.

Some care is required here because unsynchronized clocks could result in
rather unexpected orderings. Hence, this solution assumes some measure of
clock synchronization, a topic we discuss further in Section 2.3.
Nevertheless, this solution is fairly common, found in many distributed
database systems, such as Google Spanner [10, 11].

By itself, however, time-stamping events is not enough to ensure causal
consistency. After $r_{5}$ in Fig. 2, $w_{3}$ will still be missing even after
the posts seen are arranged in timestamp order. Ensuring causal consistency
will require more work. We describe two mechanisms in Section 5 that
accomplish this.

An additional requirement in many applications is eventual consistency.
Eventual consistency, informally, means that if writes stop occurring at some
point in time, then, eventually, all replicas of $x$ will have the same value.
Fig. 2 would become eventually consistent if we add send-receive pairs so that
$p_{1}$ sends its writes to $p_{2}$. However, even though it will be
eventually consistent, it will still not be causally consistent unless
something is done about the mis-ordered $r_{5}$ and $r_{6}$.

We will henceforth assume that our distributed system is causally and
eventually consistent with timestamps, which we shorten to CET. The first and
last properties are absolute in that they hold at all times during an
execution. Eventual consistency, however, is rather odd in that an eventually
consistent system may, in fact, never be consistent! It may become consistent
only if all processes stop writing to shared variables, and even then, it may
take an unbounded amount of time to become consistent.

We will formally define a time interval $\bar{C}$ (technically, a timestamp
interval) over which a system is allowed to be inconsistent. $\bar{C}$ will
become a measure of inconsistency in the system, and we will give relations
between $\bar{C}$, a measure $\bar{A}$ of unavailability, and a measure
$\mathcal{L}$ of (apparent) network latency. If $\bar{C}$ is bounded, then the
system is eventually consistent. If $\bar{C}$ is zero, we call the system
strongly consistent. A strongly consistent system is obviously eventually
consistent.

###  2.3 Using Physical Time to Assign Logical Times

To replace the naive “earlier” with something realizable, we borrow an idea
from Lamport [24], who proposed that each update of a shared variable be
assigned a timestamp. The timestamp is based on a local clock, and the local
clocks throughout the system are sufficiently synchronized to provide an
application-dependent reasonable notion of “before” or “earlier.” Once a
timestamp is assigned, it is treated as a logical time and defines the
semantic order of events.

In the bulletin board example, the timestamps could be based on Coordinated
Universal Time (UTC) and synchronized via the Network Time Protocol (NTP) [37,
38], which is designed to synchronize computer clocks to within a few
milliseconds of UTC. NTP is ubiquitous in modern computers, so this is not
likely to present a practical problem. If a post in London gets a timestamp 10
milliseconds less than a post in Singapore, the London post will be deemed to
have occurred before. Whether it physically occurred before the post in
Singapore is irrelevant.111In fact, the physical order of these two events is
unknowable. At the speed of light, Singapore is about 32 milliseconds away
from London, which means that even with perfect clock synchronization, these
two events are outside each other’s light cones. This implies that the
physical order of the two events will be different in different frames of
reference.

Given such timestamped events, the behavior of the distributed system can then
be defined by the numerical order of those timestamps. Once the timestamps are
assigned, their numerical order is a semantic property, and the goal of a CET
system design is to ensure that every component sees events in timestamp
order. This is the semantics adopted by Google Spanner [10, 11], a globally
distributed replicated database operating today.

Because the timestamps are numbers, there is always the possibility of two
events having the same timestamp. These will be deemed to have occurred
“simultaneously.” How to handle simultaneous events is application dependent.
If the merge operation is associative and commutative, then the simultaneous
operations can be applied in different orders without adverse consequences.
If, further, no user can observe the intermediate results of these operations,
then the whole system will be strongly consistent. Without associative and
commutative operations, however, the order of operations will need to be the
same at all nodes in order to achieve consistency. For simultaneous events,
consistent ordering requires a consistent mechanism for assigning distinct
priorities to each event.

Note that even sloppy clock synchronization can be useful because it defines a
semantic ordering. In the bulletin board application, does it really matter if
the clocks are off even by a few seconds (which would be unusually poor
performance for NTP)? We will see that clock synchronization error is
indistinguishable from network latency, and so the sum of the two quantities
is the only meaningful measure. Consequently, a large clock synchronization
error coupled with a requirement for strong consistency will result in high
unavailability. This would be true with any mechanism for assigning a total
order to distributed updates because any such order, in effect, assigns some
kind of logical “time” to each event. The mere act of putting them in order is
isomorphic with assigning them a time.

On local area networks and in data centers, much higher precision clock
synchronization is available with PTP (precision time protocol) [13, 14], a
centerpiece of the TSN (time sensitive network) family of protocols being
developed by the IEEE 802.1 working group [29]. The clock synchronization
standard part of TSN, IEEE 802.1AS-2020, a subset of the older standard IEEE
1588, is widely supported in networking hardware, even if it has not yet
appeared in standard programming APIs. This standard is capable of
synchronizing clocks to within less than a nanosecond on local area networks,
and on the order of microseconds on wide area networks. GPS (the global
positioning system) provides a complementary alternative that is sometimes
combined with PTP. With GPS, clocks can be synchronized to sub-nanosecond
precision globally, in principle [20]. Google Spanner synchronizes master
clocks at data centers using GPS combined with atomic clocks, and then
synchronizes clocks within the data center using PTP [10].

##  3 The CAL Theorem

We can now (finally!) define the terms in the CAL theorem, inconsistency,
unavailability, and (apparent) latency. As we will see, network latency is
only one of several factors that can affect the other two quantities, so we
drop the reference to the “network” and simply consider “latency,” trusting
that the reader will not apply the word “latency” to “unavailability,” as
Abadi does [1]. We will see that network latency is indistinguishable from
clock synchronization error, and that computation times can have just as
detrimental effects as network latency. Hence, we lump all these three
effects, network latency, clock synchronization error, and execution times
under one heading, “latency,” or “apparent latency” when there might be some
ambiguity about which latency we are talking about.

###  3.1 Tags and Physical Time

We are interested in times of events and time intervals between events, but we
have to be careful because these cannot simply be Newtonian time. Newtonian
time is inaccessible to distributed software. We will instead use two distinct
notions of time, logical and physical. A physical time $T\in\mathbb{T}$ will
be an imperfect measurement of time taken from some clock somewhere the
system.

The set $\mathbb{T}$ contains all the possible times that a physical clock can
report. We assume that $\mathbb{T}$ is totally ordered and includes two
special members: $\infty\in\mathbb{T}$ is larger than any time any clock can
report, and $-\infty\in\mathbb{T}$ is smaller than any time any clock can
report. For example, $\mathbb{T}$ could be the set of integers $\mathbb{Z}$
augmented with the two infinite members.

Given any $T_{1},T_{2}\in\mathbb{T}$, the physical time interval (or just time
interval if there is no ambiguity) between the two times is written
$i=T_{1}-T_{2}$. Time intervals are assumed to be members of a group
$\mathbb{I}$ with a largest member $\infty$ and smallest member $-\infty$ and
a commutative and associative addition operation. For example, $\mathbb{I}$
could be the set of integers $\mathbb{Z}$ augmented with the two infinite
members. Addition involving the infinite members behaves in the expected way
in that for any $i\in\mathbb{I}\setminus\\{\infty,-\infty\\}$,

$\displaystyle i+\infty$ $\displaystyle=$ $\displaystyle\infty$ $\displaystyle
i+(-\infty)$ $\displaystyle=$ $\displaystyle-\infty$

We also assume that addition of infinite intervals saturates, as in

$\displaystyle\infty+\infty$ $\displaystyle=$ $\displaystyle\infty$
$\displaystyle(-\infty)+(-\infty)$ $\displaystyle=$ $\displaystyle-\infty$
$\displaystyle\infty+(-\infty)$ is undefined

Note that we use the same symbols $\infty$ and $-\infty$ for the special
members of both the set of physical times $\mathbb{T}$ and the set of
intervals $\mathbb{I}$. We hope this will not create confusion.

Intervals can be added to a physical time value, and we assume that this
addition is associative. I.e., for any $T\in\mathbb{T}$ and any
$i_{1},i_{2}\in\mathbb{I}$,

$T+(i_{1}+i_{2})=(T+i_{1})+i_{2}\in\mathbb{T}.$ (1)

Addition of infinite intervals to a time value saturates in a manner similar
to addition of infinite intervals.

These idealized requirements for physical times and time intervals can be
efficiently approximated in practical implementations. First, it is convenient
to have the set $\mathbb{T}$ represent a common definition of physical time,
such as Coordinated Universal Time (UTC) because, otherwise, comparisons
between times will not correlate with physical reality. In the Lingua Franca
language that we use in Section 4, $\mathbb{T}$ and $\mathbb{I}$ are both the
set of 64-bit integers. A $T\in\mathbb{T}$ is a POSIX-compliant
representations of time, where $T$ represents the number of nanoseconds that
have elapsed since midnight, January 1, 1970, Greenwich mean time. In the
Lingua Franca realization, the largest and smallest 64-bit integers represent
$\infty$ and $-\infty$, respectively, and addition and subtraction respect the
above saturation requirements. Note, however, the set of 64-bit integers is
not the same as the set $\mathbb{Z}$ because it is finite. As a consequence,
addition can overflow. In Lingua Franca, such overflow saturates at $\infty$
of $-\infty$, and as a consequence, addition is no longer associative. For
example, $T+(i_{1}+i_{2})$ may not overflow while $(T+i_{1})+i_{2}$ does
overflow. As a practical matter, however, this will only become a problem with
systems that are running near the year 2270. Only then will the behavior
deviate from the ideal given by our theory.

For _logical_ time, we use an element that we call a tag $g$ of a totally-
ordered set $\mathbb{G}$. Each event in a distributed system is associated
with a tag $g\in\mathbb{G}$. From the perspective of any component of a
distributed system, the order in which events occur is defined by the order of
their tags. If two distinct events have the same tag, we say that they are
logically simultaneous. We assume the tag set $\mathbb{G}$ has an element
$\infty$ that is larger than any other tag and another $-\infty$ that is
smaller than any other tag.

In the Lingua Franca language, $\mathbb{G}=\mathbb{T}\times\mathbb{U}$, where
$\mathbb{U}$ is the set of 32-bit unsigned integers representing the microstep
of a superdense time system [35, 7, 12]. We use the term tag rather than
timestamp to allow for such a richer model of logical time. For the purposes
of this paper, however, the microsteps will not matter, and hence you can
think of a tag as a timestamp and ignore the microstep. We will consistently
denote tags with a lower case $g\in\mathbb{G}$ and measurements of physical
time $T\in\mathbb{T}$ with upper case.

We will need operations that combine tags and physical times. To do this, we
assume a monotonically nondecreasing function
$\mathcal{T}\colon\mathbb{G}\to\mathbb{T}$ that gives a physical time
interpretation to any tag. For any tag $g$, we call $\mathcal{T}(g)$ its
timestamp. In Lingua Franca, for any tag $g=(t,m)\in\mathbb{G}$,
$\mathcal{T}(g)=t$. Hence, to get a timestamp from a tag, you just have to
ignore the microstep.

The set $\mathbb{G}$ also includes infinite elements such that
$\mathcal{T}(\infty_{\mathbb{G}})=\infty_{\mathbb{T}}$ and
$\mathcal{T}(-\infty_{\mathbb{G}})=-\infty_{\mathbb{T}}$, where the subscripts
disambiguate which infinity we are referring to.

An external input from outside the system, such as a user input or query, will
be assigned a tag $g$ such that $\mathcal{T}(g)=T$, where $T$ is a measurement
of physical time taken from the local clock where the input first enters the
system. In Lingua Franca, this tag is normally given microstep 0, $g=(T,0)$.

###  3.2 Inconsistency

We will define inconsistency, $\bar{C}$, to be a non-negative time interval in
$\mathbb{I}$ that has the value zero for strongly consistent systems. When it
has a value greater than zero, it will represent a time interval over which
components in a distributed system can disagree about the values assigned to a
shared data object. If this number is bounded, then we have eventual
consistency.

Assume we are given a trace of an execution of a distributed system. As
before, we assume the distributed system consists of $N$ sequential processes,
and each sequential process is a potentially unbounded sequence of (tagged)
events. The $k$-th event is associated with a tag $g_{k}$ and a physical time
$T_{k}$. The physical time $T_{k}$ is the reading on a local clock at the time
where the event starts being processed.

The events in a process are required to have nondecreasing tags and increasing
physical times. That is, if $g_{k}$ is the tag and $T_{k}$ is the physical
time of the $k$-th event, then $g_{k}\leq g_{k+1}$ and $T_{k}<T_{k+1}$. The
constraint that a process have nondecreasing tags is central to maintaining
causal consistency, but it comes with significant costs. We will show in
Section 5 how a practical runtime infrastructure can enforce this requirement.

The constraint that physical times be increasing is, perhaps surprisingly to
some readers, not trivial to ensure. The physical time of an event is the
reading of a local clock when the processing of the event starts. But local
clocks are not always assured of increasing monotonically. A clock that is
synchronized using NTP, for example, can move backward when updated. The
Lingua Franca realization used in Section 5 provides an interface to the
system clock that is guaranteed to be increasing.

A write event is an update to a local copy of a shared variable. A read event
is a reading of the local copy of a shared variable. A send event is the
launching into the network of a notification of a write event; we assume that,
in the trace, every write event is followed by a corresponding send event. A
receive event is the receipt of such a notification; we assume that every
receive event is also a write event that updates the local copy of the shared
variable with the value received. All updates are mediated by some
application-specific merge operation that combines the update value with the
previous value in some specified way.

Within each process, every read event with tag $g$ yields the value of a
shared variable $x$ that was assigned to the local copy of $x$ by the write or
receive event in the same process with the largest tag $g^{\prime}$ where
$g^{\prime}\leq g$. If $g^{\prime}=g$, we require that $T^{\prime}<T$, where
$T^{\prime}$ is the physical time of the write or receive event and $T$ is the
physical time of the read event. This requirement ensures that a read event
reads a value that was written at an earlier physical time. This requirement
is the principal reason for the requirement that physical times in a process
be monotonically increasing.

We have additional requirements on tags:

###### Requirement 1

A send event has a tag greater than or equal to that of the write event that
it is reporting and a physical time greater than that of the write event. A
receive event has a tag greater than or equal to the tag of the send event
that it is receiving. The physical time of the receive event relative to the
send event is unconstrained.

###### Definition 1

For each write event on process $j$ with tag $g_{j}$, let $g_{i}$ be the tag
of the corresponding receive event on process $i$ or $\infty$ if there is no
corresponding receive event. The inconsistency $\bar{C}_{ij}\in\mathbb{I}$
from $j$ to $i$ is defined to be

$\bar{C}_{ij}=\max(\mathcal{T}(g_{i})-\mathcal{T}(g_{j})),$ (2)

where the maximization is over all write events on process $j$. If there are
no write events on process $j$, then we define $\bar{C}_{ij}=0$.

With requirement 1, it is clear that $\bar{C}_{ij}\geq 0$. If $\bar{C}=0$, we
have strong consistency. We will see that this strong consistency comes at a
price in availability, and that network failures can result in unbounded
unavailability. If $\bar{C}$ is bounded, we have eventual consistency, and
$\bar{C}$ quantifies “eventual.”

###  3.3 Unavailability

Unavailability, $\bar{A}$, is a measure of the time it takes for a system to
respond to user requests [21]. If it takes a long time (or it never responds),
then the system is unavailable, whereas if responses are instantaneous, then
the system is highly available.

A user request is an external event that originates from outside the
distributed system. Assume that a user request triggers a read event in
process $i$ with tag $g_{i}$ such that its timestamp $\mathcal{T}(g_{i})$ is
the reading of a local clock when the external event occurs. Let $T_{i}$ be
the physical time of the read event, i.e., the physical time at which the read
is processed. Hence, $T_{i}\geq\mathcal{T}(g_{i})$.

###### Definition 2

For each read event on process $i$, let $g_{i}$ be its tag and $T_{i}$ be the
physical time at which it is processed. The unavailability
$\bar{A}_{i}\in\mathbb{I}$ at process $i$ is defined to be

$\bar{A}_{i}=\max(T_{i}-\mathcal{T}(g_{i})),$ (3)

where the maximization is over all read events on process $i$ that are
triggered by user requests. If there are no such read events on process $i$,
then $\bar{A}_{i}=0$.

Because we are considering only read events that are triggered by external
user requests, $\mathcal{T}(g_{i})\leq T_{i}$, so $\bar{A}_{i}\geq 0$. If
$\bar{A}_{i}=0$, then we have maximum availability (minimum unavailability).
This situation arises when reads are immediately satisfied. To maintain strong
consistency across a distributed system, however, we will find that
unavailability is typically larger, $\bar{A}_{i}>0$.

###  3.4 Processing Offsets

To maintain causal consistency, we require that a process have nondecreasing
tags. For this reason, in a trace, a read or write event triggered by an
external input may have a physical time $T$ that is significantly larger than
its tag’s timestamp $\mathcal{T}(g)$. While $\mathcal{T}(g)$ is determined by
the physical clock at the time the external input appears, the physical time
at which the event is actually processed may have to be later to ensure that
all events with earlier tags have been processed.

Here, it becomes convenient to assume that write events _with the same tag_
have an associative and commutative merge operation. Consistency of any kind
would not be possible were this not the case because identical tags cannot be
used to determine in which order to apply updates.

A consequence of this assumption is that a process can process a _write_ event
whenever it is sure that no events with _earlier_ tags will later appear. In
contrast, a process can only process a _read_ event when it is sure that no
events with _earlier or equal_ tags will later appear. The result of the read
should reflect _all_ updates to the shared variable with tags equal to or less
than that of the read event.

This motivates the following definition:

###### Definition 3

For process $i$, the processing offset $O_{i}\in\mathbb{I}$ is

$O_{i}=\max(T_{i}-\mathcal{T}(g_{i}))$ (4)

where $T_{i}$ and $g_{i}$ are the physical time and tag, respectively, of a
write event on process $i$ that is triggered by a local external input (and
hence assigned a timestamp drawn from the local clock). The maximization is
over all such write events in process $i$. If there are no such write events,
then $O_{i}=0$.

Notice that the processing offset closely resembles the unavailability of
Definition 2, but the former refers to _write_ events and the latter refers to
_read_ events.

The processing offset, by definition, is greater than or equal to zero. We
will see that it is often zero, but not always.

###  3.5 Apparent Latency

When a write to a shared variable occurs in process $j$, some time will elapse
before a corresponding receive event on process $i$ performs a corresponding
write to its local copy of the shared variable. This motivates the following
definition:

###### Definition 4

Let $g_{j}$ be the tag of a write event in process $j$ that is triggered by an
external input at $j$ (so $\mathcal{T}(g_{j})$ is the physical time of that
external input). Let $T_{i}$ be the physical time of the corresponding receive
event in process $i$ (or $\infty$ if there is no such event). (If $i=j$, we
assume $T_{i}$ is the same as the physical time of the write event.) The
apparent latency or just latency $\mathcal{L}_{ij}\in\mathbb{I}$ for
communication from $j$ to $i$ is

$\mathcal{L}_{ij}=\max(T_{i}-\mathcal{T}(g_{j})),$ (5)

where maximization is over all such write events in process $j$. If there are
no such write events, then $\mathcal{L}_{ij}=0$.

Note that $T_{i}$ and $\mathcal{T}(g_{j})$ are physical times _on two
different clocks_ if $i\neq j$, so this apparent latency is an actual latency
only if those clocks are perfectly synchronized. Unless the two processes are
actually using the same physical clock, they will never be perfectly
synchronized. Hence, the apparent latency may even be negative. Note that
despite these numbers coming from different clocks, if tags are sent along
with messages, this apparent latency is measurable at the receiving end.

The apparent latency can be thought of as the sum of four components,

$\mathcal{L}_{ij}=O_{j}+X_{ij}+L_{ij}+E_{ij},$ (6)

where $X_{ij}$ is execution time overhead at node $j$ for sending a message to
node $i$, $L_{ij}$ is the network latency from $j$ to $i$, and $E_{ij}$ is the
clock synchronization error. The three latter quantities are indistinguishable
and always appear summed together, so there is no point in breaking apparent
latency down in this way. Moreover, these latter three quantities would have
to be measured with some physical clock, and it is not clear what clock to
use. The apparent latency requires no problematic measurement since it
explicitly refers to local clocks and tags.

The clock synchronization error can be positive or negative, whereas $O_{j}$,
$X_{ij}$, and $L_{ij}$ are always nonnegative. If $E_{ij}$ is a sufficiently
large negative number, the apparent latency will itself also be negative.
Because of the use of local clocks, the receive event will appear to have
occurred before the user input that triggered it. This possibility is
unavoidable with imperfect clocks.

###  3.6 The CAL Theorem for Strong Consistency

We can now put everything together and identify the sources of the
unavailability of Definition 2. In that definition, $T_{i}$ must be greater
than $\mathcal{T}(g_{i})$ by enough to ensure that a read event at process $i$
is processed after all writes to the shared variable with tags less than or
equal to that of the read event. From this, we can see that

$\bar{A}_{i}=\max_{j\in N}(\bar{A}_{ij}),$ (7)

where $\bar{A}_{ij}$ is the unavailability at $i$ if only the two processes
$i$ and $j$ are considered.

Suppose that process $j$ has a write event with tag $g_{j}$ triggered by a
local external event. In the worst case, the receive event on $i$
corresponding to this write event will have physical time

$T_{i}=\mathcal{T}(g_{j})+\mathcal{L}_{ij},$ (8)

from (5). Note that $T_{i}$ is measured by the clock at $i$, and, since
$\mathcal{L}_{ij}$ can be negative (provided $i\neq j$), it is possible for
$T_{i}$ to be less than $\mathcal{T}(g_{j})$.

Node $i$ cannot respond to the local read query with tag $g_{i}$ before
physical time reaches $\mathcal{T}(g_{i})+O_{i}$ because $O_{i}$ is the time
it takes to make sure all _earlier_ events have been processed. But the
possibility of incoming network messages means that _additional_ delay may be
required to ensure that earlier _or equal_ tags have been processed. That is,
if there is the possibility of a remote write event on $j$ with tag
$\mathcal{T}(g_{j})=\mathcal{T}(g_{i})$, then (8) implies that $i$ needs to
also wait at least until physical time

$T^{\prime}_{i}=\mathcal{T}(g_{i})+\mathcal{L}_{ij},$ (9)

where $\mathcal{L}_{ij}$ includes $O_{j}$. Therefore, to maintain strong
consistency, in the worst case, node $i$ cannot respond to the local read
query with tag $g_{i}$ before physical time reaches

$R_{ij}=\max(T^{\prime}_{i},\mathcal{T}(g_{i})+O_{i})=\mathcal{T}(g_{i})+\max(\mathcal{L}_{ij},O_{i}),$

using $T^{\prime}_{i}$ given by (9). Hence, the unavailability for node $i$
caused by communication from node $j$ is at most

$\bar{A}_{ij}=R_{ij}-\mathcal{T}(g_{i})=\max(\mathcal{L}_{ij},O_{i}).$ (10)

Combine this with (7) and we get the unavailability at process $i$ for a
strongly consistent system to be

$\displaystyle\bar{A}_{i}$ $\displaystyle=$ $\displaystyle\max_{j\in
N}\max(\mathcal{L}_{ij},O_{i})$ (11) $\displaystyle=$
$\displaystyle\max(O_{i},\max_{j\in N}\mathcal{L}_{ij}).$ (12)

When network latency, clock synchronization error, or execution time increase
sufficiently, the apparent latency will increase, and so will the
unavailability. We can interpret “network partitioning” to mean that
$\mathcal{L}_{ij}$ goes to infinity and the system becomes unavailable at
process $i$, $\bar{A}_{i}=\infty$. When enforcing strong consistency,
therefore, network partitioning implies unavailability, as expected from the
CAP theorem.

###  3.7 The CAL Theorem for Arbitrary Consistency

We can now explore what happens if we relax consistency. An extreme solution
is to eliminate timestamps altogether and have each node just handle messages
in the order received. However, the breach of causal consistency in Fig. 2 at
$d_{2}$ becomes hard to avoid, and assigning an order to distributed user
inputs becomes impossible.

A more controlled way to relax consistency is to explicitly manipulate the
tags. In Section 4, we will describe a language that supports such
manipulation, but for now, assume that each sending of any update message from
node $j$ to $i$ increments the timestamp part of the tag of the message by
some fixed amount $D_{ij}$ called a logical delay from $j$ to $i$. That is, if
the send event at $j$ has tag $g_{j}$ and the corresponding receive event at
node $i$ has tag $g_{i}$, then

$\mathcal{T}(g_{i})=\mathcal{T}(g_{j})+D_{ij}.$ (13)

If there is no communication from $j$ to $i$, then we take $D_{ij}=\infty$.
Solving for $D_{ij}$, we get

$D_{ij}=\mathcal{T}(g_{i})-\mathcal{T}(g_{j}).$ (14)

If each node processes events in tag order, and if the merge operation is
commutative and associative for updates with identical tags, then
$D_{ij}=\bar{C}_{ij}$, the inconsistency from Definition 1. Hence, by
controlling the logical delay $D_{ij}$, we control the amount of inconsistency
between nodes $i$ and $j$. In exchange for increasing the inconsistency, we
potentially reduce the unavailability by $\bar{C}_{ij}$. Equation (10) is
replaced by

$\bar{A}_{ij}=\max(\mathcal{L}_{ij}-\bar{C}_{ij},O_{i}).$ (15)

This represents the unavailability at $i$ due to messages it may receive from
$j$. One way to understand this is to realize that the $\mathcal{L}_{ij}$ in
Definition 4 is the difference between the physical time an update message is
received at $i$ and the tag of the update at $j$. But now, the tag seen at $i$
has been incremented by $D_{ij}$, so if it were to compare the physical time
against the _received_ tag (instead of the sender’s tag), the latency is
reduced by $D_{ij}$. $D_{ij}$ equals $\bar{C}_{ij}$ because the _same update_
has two different logical times on the two nodes and hence $D_{ij}$ measures
the inconsistency these two nodes experience.

The overall unavailability for node $i$ is therefore

$\displaystyle\bar{A}_{i}$ $\displaystyle=$
$\displaystyle\max_{j}\left(\max(\mathcal{L}_{ij}-\bar{C}_{ij},O_{i})\right)$
(16) $\displaystyle=$ $\displaystyle\max\left(O_{i},\max_{j\in
N}(\mathcal{L}_{ij}-\bar{C}_{ij})\right).$

We have derived the CAL theorem (or $\bar{C}\bar{A}\mathcal{L}$ theorem),
which we now state formally:

###### Theorem 3.1

Given a trace as defined in Section 3.2, the unavailability in Definition 2 at
process $i$ is, in the worst case,

$\bar{A}_{i}=\max\left(O_{i},\max_{j\in
N}(\mathcal{L}_{ij}-\bar{C}_{ij})\right),$ (17)

where $O_{i}$ is the processing offset given by Definition 3,
$\mathcal{L}_{ij}$ is apparent latency in Definition 4 (which includes
$O_{j}$), and $\bar{C}_{ij}$ is the inconsistency of Definition 1 .

Note from Definition 1 that if there is a write event on $j$ and no
corresponding receive event on $i$, then the inconsistency becomes
$\bar{C}_{ij}=\infty$. This corresponds to the case of total network
partitioning. When this is the case, however, (17) tells us that node $j$ has
no effect on availability on node $i$, as expected.

###  3.8 More on Processing Offsets

The processing offsets $O_{i}$ and $O_{j}$ deserve more discussion. These are
physical time delays incurred on nodes $i$ and $j$ before they can begin
handling events with a particular tag. Specifically, node $i$ can begin
handling a user input with tag $g_{i}$ at physical time
$T_{i}=\mathcal{T}(g_{i})+O_{i}$. We will see that these offsets are derivable
from program structure, but for this simple bulletin board application, as we
will see in Section 5, they can both be made zero, $O_{i}=O_{j}=0$. This means
that each node can immediately launch local posts into the network, incurring
only an execution time delay $X_{ij}$. This will not compromise causal
consistency in any way.

For strong consistency, we require $\bar{C}_{ij}=0$. In this case, the time it
takes to respond to a user on node $i$ is the largest apparent latency from
another node $j$. If we are willing to tolerate an inconsistency at least as
large as this apparent latency, i.e., if

$\bar{C}_{ij}\geq\mathcal{L}_{ij}\text{ for all nodes }j,$ (18)

then the unavailability becomes zero for the case where the processing offsets
are zero. We can respond immediately to user input. This is a reasonable
design choice for a bulletin board application because users will expect to
see their posts immediately, as they type them, and given timestamped posts,
the record of the conversation is easily repaired when other posts are later
received. If network latency increases so that (18) no longer holds, then we
can either tolerate larger inconsistency, allowing $\bar{C}_{ij}$ to rise, or
tolerate higher unavailability, allowing $\bar{A}_{i}$ to rise. For the
bulletin board application, the former is probably the better choice under the
assumption that the network will eventually be repaired.

The CAL theorem gives a relation between inconsistency, unavailability, and
apparent latency, where causal consistency is ensured at the price of
processing offsets $O_{i}$. We will show in Section 5 how to derive these
offsets in Lingua Franca from program structure.

###  3.9 Ensuring Eventual Consistency

To achieve eventual consistency, if $\bar{C}_{ij}>0$, we have a bit more work
to do. To improve availability, we introduce logical delays $D_{ij}$, which
then determine the inconsistency $\bar{C}_{ij}$. To achieve eventual
consistency, all processes must eventually behave as if they had applied
writes in the same order, but the logical delays permit them to apply the
writes in different orders. For the bulletin board application, we can fix
this by modifying the merge operation to use the _original_ tag, before
incrementing by the logical delay $D_{ij}$, to perform the merge. That is, if
a node $i$ receives a message from node $j$ with timestamp $t_{j}$, it knows
that the original timestamp was $t_{j}-D_{ij}$. The receiving end can use this
original tag to sort updates, thereby ensuring that all replicas eventually
have the same sequence of posts to the bulletin board. We call this style of
merge operation a sorted append.

When the merge operation is replace, we can similarly modify the merge to
become a sorted replace which becomes associative and commutative. At the
receiving end, we can use (13) to compute the original timestamp
$\mathcal{T}(g_{j})=\mathcal{T}(g_{i})-D_{ij}$. However, we need a bit more,
because the order of the updates is determined by their _tags_ , not their
timestamps. Hence, it will be necessary to either convey the original tag in a
message or ensure that there is a way to calculate it at the receiving end
(for example, in Lingua Franca, by conveying the microstep). We can compare
this original tag against the tag of the latest local update to determine
whether the received value should overwrite the previously applied local
update. This strategy enforces a last writer wins policy in a rigorous, well-
defined way.

###  3.10 Max-Plus Formulation

Summarizing where we are, the CAL theorem gives an algebraic relation between
inconsistency, unavailability, and latency. This relation shows that if
network latency becomes unbounded (the network becomes partitioned), then one
of inconsistency and unavailability must also become unbounded, and hence the
CAP theorem is a special case of the CAL theorem. The relation involves
processing offsets, representing the cost of enforcing causal consistency. How
to derive these is somewhat implementation dependent. In Section 4, we show
how to use the Lingua Franca coordination language to specify such distributed
programs in a way that enables calculation of the offsets, and in Section 5,
we describe two specific distributed implementations that we have realized in
Lingua Franca. But first, we give a very general form of the CAL theorem.

The form of the CAL theorem given in (17) can be made much more elegant by
observing that the mathematical operations of maximization and addition on
numbers form an algebra called a max-plus algebra with very useful properties
[2]. In this algebra, addition (written $\oplus$) is maximization,

$a\oplus b=\max(a,b),$

and multiplication (written $\otimes$) is addition,

$a\otimes b=a+b.$

In max-plus, the additive identity is $-\infty$, and the multiplicative
identify is zero.

Let $N$ be the number of nodes, and define an $N\times N$ matrix $\Gamma$ such
that its elements are given by

$\Gamma_{ij}=\mathcal{L}_{ij}-\bar{C}_{ij}-O_{j}.$ (19)

That is, from (6), the $i$, $j$-th entry in the matrix is an assumed bound on
$X_{ij}+L_{ij}+E_{ij}$ (execution time, network latency, and clock
synchronization error), adjusted downwards by the specified tolerance for
inconsistency.

Let $\bm{A}$ be a column vector with elements equal to the unavailabilities
$\bar{A}_{i}$, and $\bm{O}$ be a column vector with elements equal to the
processing offsets $O_{i}$. Then the CAL theorem (17) can be written as

$\bm{A}=\bm{O}\oplus\Gamma\bm{O},$ (20)

where the matrix multiplication is in the max-plus algebra. This can be
rewritten as

$\bm{A}=(\bm{I}\oplus\Gamma)\bm{O},$ (21)

where $\bm{I}$ is the identity matrix in max-plus, which has zeros along the
diagonal and $-\infty$ everywhere else. Hence, unavailability is a simple
linear function of the processing offsets, where the function is given by a
matrix that depends on the network latencies, clock synchronization error,
execution times, and specified inconsistency in a simple way.

When the processing offsets are zero, as they are for the bulletin board
application, then the unavailability for each node is the sum (in max-plus) of
the corresponding row of $\bm{I}\oplus\Gamma$ which is either zero (if all
elements in the row of $\Gamma$ are less than or equal to zero) or the maximum
element in the row of $\Gamma$.

###  3.11 Pessimistic Evaluation of Processing Offsets

The processing offsets $O_{i}$ and $O_{j}$ are physical time delays incurred
on nodes $i$ and $j$ before they can begin handling events. Specifically, node
$i$ can begin handling a user input (specifically a write event) with tag
$g_{i}$ at physical time $T_{i}=\mathcal{T}(g_{i})+O_{i}$. In the absence of
any further information about a program, we can use our $\Gamma$ matrix to
calculate these offsets. However, we will find that the result is pessimistic
and can be refined with further information about the program structure.

First, consider nodes that have the possibility of new events appearing
asynchronously with timestamps given by the local physical clock, like those
in our bulletin board application. In such a node $i$, it is generally not
safe to process an event with tag $g_{i}$ until the physical clock $T_{i}$
exceeds $\mathcal{T}(g_{i})$, i.e., $T_{i}>\mathcal{T}(g_{i})$. Otherwise,
there is a possibility of processing events out of order. If a node $j$ is
purely functional and only reacts to network inputs, then there is no such
constraint. Define a column vector $\bm{Z}$ such that

$Z_{i}=\begin{cases}0,&\text{if node $i$ has local physically timestamped
inputs,}\\\ -\infty,&\text{otherwise}.\end{cases}$ (22)

With this, we require at least that

$\bm{O}\geq\bm{Z}.$

In addition, to ensure that node $i$ processes events in tag order, it is
sufficient to ensure that node $i$ has received all network input events with
tags less than or equal to $g_{i}$ before processing any event with tag
$g_{i}$. With this (conservative) policy,

$O_{i}\geq\max_{j}(\mathcal{L}_{ij}-\bar{C}_{ij}).$

The smallest processing offsets that satisfy these two constraints satisfies

$\bm{O}=\bm{Z}\oplus\Gamma\bm{O}.$ (23)

This is a system of equations in the max-plus algebra. From Baccelli, et al.
[2] (Theorem 3.17), if every cycle of the matrix $\Gamma$ has weight less than
zero, then the unique solution of this equation is

$\bm{O}=\Gamma^{*}\bm{Z},$ (24)

where the Kleene star is (Theorem 3.20 [2])

$\Gamma^{*}=\bm{I}\oplus\Gamma\oplus\Gamma^{2}\oplus\cdots.$

Baccelli et al. show that this reduces to

$\Gamma^{*}=\mathbf{I}\oplus\Gamma\oplus\cdots\oplus\Gamma^{N-1},$

where $N$ is the number of processes.

The requirement that the cycle weights be less than zero is intuitive, but
overly restrictive. It means that along any communication path from a node $i$
back to itself, the sum of the logical delays $D_{jk}$ must exceed the sum of
the execution times, network latencies, and clock synchronization errors along
the path. This means that we have to tolerate a non-zero inconsistency
somewhere on each cycle. For the bulletin board application, where every node
sends messages to every other node, every pair of nodes requires a non-zero
inconsistency in order to satisfy this cycle-mean constraint.

In practice, programs may have zero or positive cycle means. Theorem 3.17 of
Baccelli, et al. [2] shows that if all cycle weights are non-positive, then
there is a solution, but the solution may not be unique. If there are cycles
with positive cycle weights, there is no finite solution for $\bm{O}$. When
the inconsistency is zero, there are no logical delays at all, and all cycle
weights become positive. In this case, the only solution to (23) sets all the
processing offsets to $\infty$. Every node must wait forever before handling
any user input. This is, of course, the ultimate price in availability.

Why is (24) pessimistic? Absent further information about the application
logic, we must assume that any network input at node $i$ with tag $g_{i}$ can
causally affect any network output with tag $g_{i}$ or larger. Moreover, we
have no knowledge about where network inputs may originate. We next show how,
by constructing the application in a language that exposes more information
about causal relationships, we can derive much less pessimistic processing
offsets while still preserving causal and eventual consistency.

##  4 Lingua Franca Realization

Lingua Franca (or LF, for short) is a coordination language developed jointly
at UC Berkeley, TU Dresden, UT Dallas, and Kiel University [32]. We show here
that LF supports a full range of explicit tradeoffs between availability and
consistency for a wide variety of applications. Lingua Franca is a polyglot
coordination language that orchestrates concurrent and distributed programs
written in any of several target languages (as of this writing, C, C++,
Python, and TypeScript are supported, and Rust is in progress).

In this section, we give a small collection of complete Lingua Franca programs
that illustrate these tradeoffs.222Download these programs from
https://cal.lf-lang.org/. To keep the programs short but complete, we develop
an example borrowed from Kuhn [23] that has similar properties to the bulletin
board considered earlier, but does not require memory management for a growing
text and hence results in simpler code that fully illustrates the principles
of interest.

Kuhn’s application maintains bank balances in a distributed database and
accepts deposits and withdrawals at distributed locations like ATM machines.
Suppose that a customer shows up and wants to withdraw $w$ dollars. This needs
to be compared against the current balance $x$ in the account. However, there
may be near-simultaneous withdrawals occurring at other locations. Strong
consistency requires agreement on the order in which these withdrawals occur.
If all locations agree on this order, the bank can assure that the balance
never drops below zero by denying any withdrawal that would make it so.
Moreover, if the bank policy allows the balance to drop below zero, all
branches will agree on the number of times that this has occurred, so they
will all agree on what overdraft charges to apply.

When network latency gets large, however, such a strong consistency policy
reduces availability by making the user wait. In the extreme case of network
partitioning, the ATM may simply deny the service rather than wait for the
network to be repaired. We can define “network partitioning” to be
“sufficiently large network latency upon which communication retries are
abandoned.” The threshold of network partitioning, therefore, like the
tradeoff between availability and consistency, is defined by the application.

Kuhn’s banking example is a good one because we can image a whole range of
design choices. Ultimately, the choices are business decisions, not technical
ones. Is it acceptable for an ATM to deny dispensing cash because of a
temporary network failure? How would customers react? How big is the risk to
the bank if the ATM dispenses cash on the basis of possibly inconsistent data?
Banks routinely allow accounts to go into overdraft, and then heap fees on
their customers. Then again, a distributed attack that simultaneously makes a
large number of near-simultaneous withdrawals after launching a distributed
denial of service attack on a bank’s networks is not hard to imagine. The
tradeoff between risk and customer service is a business decision. We will
show software tools that make these choices clear and enable a whole range of
solutions.

This banking example also has the nice property that the database operations
are associative and commutative if overdrafts are allowed, and not otherwise.
Hence, we can explore designs with a variety of merge operations (associative,
commutative, both, or neither).

###  4.1 Brief Introduction to Lingua Franca

⬇

1target L;

2reactor ReactorClass {

3 input name:type;

4 ...

5 output name:type;

6 ...

7 state name:type(init);

8 ...

9 ... timers, actions, if any ...

10 ...

11 reaction(trigger, ...) -> effect, ... {=

12 ... code in language L ...

13 =}

14 ... more reactions ...

15}

16...

17federated reactor {

18 instance = new ReactorClass();

19 ...

20 instance.name -> instance.name;

21 ...

22}

Figure 3: Structure of a federated Lingua Franca program for target language
$L$.

Lingua Franca 333https://repo.lf-lang.org is a coordination language where
applications are defined as concurrent compositions of components called
reactors [31, 30]. Fig. 3 outlines the structure of a Lingua Franca program.
One or more reactor classes are defined with input ports (line 3), output
ports (line 5), state variables (line 7), and timers and actions. We will not
need timers here and will elaborate on actions later. If a reactor class is
instantiated within a federation, as shown on line 18, then the instance is
called a federate, and tagged inputs will arrive from the network at the input
ports and be handled in tag order. Inputs are handled by reactions, as shown
on line 11. Reactions declare their triggers, as on line 11, which can be
input ports, timers, or actions. If a reaction lists an output port among its
effects, then it can produce tagged output messages via that output port. The
routing of messages is specified by connections, as shown on line 20. The
syntax and semantics will become clearer as we develop our specific
applications.

⬇

1target C;

2reactor CAReplica {

3 input local_update:int;

4 input remote_update:int;

5 input query:bool;

6

7 state balance:int(0);

8

9 output response:int;

10

11 reaction(local_update, remote_update) {=

12 if (local_update->is_present) {

13 self->balance += local_update->value;

14 }

15 if (remote_update->is_present) {

16 self->balance += remote_update->value;

17 }

18 =}

19

20 reaction(query) -> response {=

21 SET(response, self->balance);

22 =}

23}

Figure 4: Lingua Franca code defining a reactor class that is a replica in a
replicated database storing a bank balance and accepting queries and updates.
This version is commutative and associative.

###  4.2 Commutative and Associative Replica

We begin with a version that has an associative and commutative merge
operation. The first part of this version is shown in Fig 4. This defines a
reactor class, a software component that is described as an “actor revisited”
[33]. The first line defines the target language, which is the language of the
program that the LF code generator will produce and the language in which the
business logic of the software component is written. To minimize dependencies,
we give our examples here with C as the target language.

The second line declares a new reactor class called CAReactor (‘CA’ for
Commutative and Associative). This reactor has three inputs, defined on lines
3 through 5. The first input accepts a local update, an integer that is
positive for a deposit and negative for a withdrawal. The second input accepts
a remote update, which will come from some other machine somewhere on the
network (we will generalize this later to accept an arbitrary number of remote
updates). The third input accepts a query for the current balance.

Line 7 defines a local state variable, an integer that is the local copy of
the balance. Line 9 defines an integer output, which will be a response to a
query for the balance. In a strongly consistent design (which we will see how
to construct), this balance will be agreed upon by all replicas at each tag.

Lines 11 through 18 give the business logic of how to handle local or remote
updates. The code between the delimiters {= ... =} is ordinary C code making
use of mechanisms provided by the LF code generator to access the inputs and
state. This code simply checks to see which inputs are present and adjusts the
balance accordingly. Because the operation is commutative and associative, it
does not matter in which order these inputs are handled.

The figure to the right is automatically generated by the Lingua Franca IDE
called Epoch.444Epoch is available for download at https://releases.lf-
lang.org.,555The diagram synthesis feature was created by Alexander Schulz-
Rosengarten of Kiel University using the graphical layout tools from the
KIELER Lightweight Diagrams framework [40] (see https://rtsys.informatik.uni-
kiel.de/kieler). The chevrons in the figure represent reactions, and their
dependencies on inputs and their ability to produce outputs is shown using
dashed lines.

###  4.3 User

The code in Fig. 5 defines a Lingua Franca reactor that stands in for an ATM
machine through which a customer can make deposits or withdrawals. This
component listens for the user to type a number on a terminal, and then
produces that number on its output port. To keep things simple, there is no
authentication and not much error checking. While obviously critical to a real
ATM design, those aspects are irrelevant to our discussion here, so we leave
them out.

⬇

1target C;

2reactor UserInput {

3 preamble {=

4 // Define a function to read user input.

5 void* get(void* r) {

6 int amt;

7 char buf[20];

8 while(1) {

9 // Read a character input.

10 char* ln = fgets(buf, 20, stdin);

11 // Exit if no more input.

12 if (ln == NULL) return NULL;

13 // Parse an input integer.

14 int n = sscanf(ln, "%d", &amt);

15 // If one integer was parsed...

16 if (n == 1) {

17 // Schedule an event.

18 schedule_int(r, 0, amt);

19 } else {

20 // Request another input.

21 printf("Please enter a number.\n");

22 }

23 }

24 }

25 =}

26 input balance:int;

27 output deposit:int;

28

29 physical action r:int;

30

31 reaction(startup) -> r {=

32 pthread_t id;

33 pthread_create(&id, NULL, &get, r);

34 =}

35 reaction(r) -> deposit {=

36 SET(deposit, r->value);

37 =}

38 reaction(balance) {=

39 printf("Balance: %d\n", balance->value);

40 =}

41}

Figure 5: Lingua Franca component that gets user input to provide deposits and
make withdrawals.

Upon startup, on line 33, this reactor creates a thread that executes
concurrently with the Lingua Franca program. That thread, defined on lines 5
to 24, repeatedly blocks on line 10 waiting for the user to type something. If
the user input is a valid number, then on line 18, the thread calls a built-in
thread-safe function schedule, passing it a pointer to the physical action
named ‘r’ and the amount entered by the user (the 0 argument is irrelevant to
the current discussion).

The physical action r, declared on line 29, is a Lingua Franca construct for
providing external, asynchronous inputs to an LF program. The key is that when
schedule is called, a tag $g$ based on a local measurement $T$ of physical
time, such that $\mathcal{T}(g)=T$, is assigned to the event, which is then
injected into the program to be handled in tag order.

The LF program reacts to the event created by the call to schedule by
executing the reaction given on lines 35 through 37. This sets the output
named deposit to the amount entered by the user.

The physical action declared on line 29 of Fig. 5 deserves more scrutiny.
First, Lingua Franca, by default, uses the system clock on the machine that
runs each federate to assign the timestamp part $\mathcal{T}(g)$ of the tag.
When a federated program is started, it performs a clock synchronization round
using the technique of Geng et al. [16]. This ensures that even if the system
clock is set manually to some arbitrary value, when the distributed program
starts up, all nodes will agree on the current physical time within a few
milliseconds. Lingua Franca also provides a facility for performing ongoing
clock synchronization that can correct for clock drifts, but in many systems,
it is not really necessary to enable this. We can rely instead on a built-in
NTP realization, if that is sufficiently precise for the application.

Once a tag is assigned, the handling of the update throughout the distributed
system is a deterministic function of that tag. This gives a clear semantics
to the behavior of the system when the actual order of events originating
throughout the system is unknown, unknowable, or ambiguous. Moreover, it
enables rigorous regression testing, where the UserInput reactor of Fig. 5 is
replaced by event generators driven by logical clocks. Those logical clocks
can generate events on distributed nodes that are deterministically ordered or
even simultaneous.

###  4.4 Composition

⬇

1target C;

2import CAReplica from "CAReplica.lf";

3import UserInput from "UserInput.lf";

4

5reactor ATM {

6 input update:int;

7 output publish:int;

8 u = new UserInput();

9 r = new CAReplica();

10 (u.deposit)+ -> r.query, r.local_update;

11 r.response -> u.balance;

12 update -> r.remote_update;

13 reaction(u.deposit) -> publish {=

14 if (u.deposit->value != 0) {

15 SET(publish, u.deposit->value);

16 }

17 =}

18}

19federated reactor ConsistencyFirst {

20 a = new ATM();

21 b = new ATM();

22 b.publish -> a.update;

23 a.publish -> b.update;

24}

Figure 6: Federated Lingua Franca program with two ATM machines that can
provide deposits and make withdrawals.

We can now put together the components to get a complete, executable program.
In Fig 6, we define an ATM reactor that contains one instance each of
UserInput from Fig. 5 and CAReplica from Fig. 4. We then define a federated
reactor named ConsistencyFirst that creates two instances of ATM and connects
them. Hopefully, the program, with the help of the diagram, is self-
explanatory, with the possible exception of the new syntax on line 10. To the
left of the arrow, the output port u.deposit is surrounded with ( ... )+,
which, in Lingua Franca syntax, indicates to use the port as many times as
necessary to satisfy all the destinations given to the right of the arrow. In
other words, it is a compact syntax for multicast. The reaction defined on
lines 13 through 17 simply prevents publishing zero-valued deposits. Hence, a
deposit equal to zero can be used to query the current balance and will not
generate network traffic.

###  4.5 Execution

When the top-level reactor in a Lingua Franca program is federated, as it is
in Fig. 6, then the code generator, instead of producing a single program,
produces as many programs as there are instances of reactors within the top
level reactor, plus one additional program that manages certain coordination
functions (more about this later). In this case, there are two reactors within
the top level, so a total of three programs will be generated.

Figure 7: An execution of the Lingua Franca program in Fig. 6 (slightly
elaborated to display (elapsed) logical time $\mathcal{T}(g)$ (in
nanoseconds).

An execution of the program in Fig. 6 is shown in Fig. 7, where there is one
terminal for each of the two ATM instances. The top line of each window shows
the command that starts each instance. In the lower window, user $b$ begins by
querying the current balance by entering ‘0’. The balance is zero. User $b$
then deposits 100 dollars. User $a$ then withdraws 20 dollars twice. User $b$
then queries the balance again, discovering that it is now 60.

The deposits and withdrawals are handled by both ATM federates in the same
order, defined by tags that are assigned when schedule is called on line 18 in
Fig. 5. If two deposits occur with the same tag, then the reported balance by
each user will reflect the aggregate of the two operations. That is, the two
deposits are semantically simultaneous.

This simultaneity feature is hard to test with an interactive program like
this, but in Lingua Franca, it is easy to create a regression test that
replaces the UserInput with timer-driven inputs, which gives precise control
over the tags. The ability to construct such deterministic, distributed
regression tests is one of the key advantages of Lingua Franca.

###  4.6 Variants of the Running Example

Lest the reader conclude that we are only talking about one rather
oversimplified example, we will now point out several variants of this design
that are easy to build.

First, the example in Fig. 6 has only two federates. We have extended Lingua
Franca with a convenient syntax shown in Fig. 8 for scaling this program to
any number of federates. The top-level ConsistencyFirstN reactor has a
parameter N defined on line 5 (with default value 4) that specifies how many
instances of the ATM reactor to create. Those instances are created on line 7,
and each instance is assigned the value N to its own parameter, defined on
line 11, which happens to also have the name “N.”

⬇

1target C;

2import UserInput from "UserInput.lf";

3

4federated reactor ConsistencyFirstN(

5 N:int(4)

6) {

7 bank = new[N] ATM(N = N);

8 (bank.publish)+ -> bank.updates;

9}

10reactor ATM(

11 N:int(2)

12) {

13 input[N] updates:int;

14 output publish:int;

15 u = new UserInput();

16 r = new CAReplicaN(N = N);

17 u.deposit -> r.query;

18 r.response -> u.balance;

19 updates -> r.updates;

20 reaction(u.deposit) -> publish {=

21 if (u.deposit->value != 0) {

22 SET(publish, u.deposit->value);

23 }

24 =}

25}

26reactor CAReplicaN(

27 N:int(2)

28) {

29 input[N] updates:int;

30 input query:int;

31 output response:int;

32 state balance:int(0);

33

34 reaction(updates) {=

35 for (int i = 0; i < updates_width; i++) {

36 if (updates[i]->is_present) {

37 self->balance += updates[i]->value;

38 }

39 }

40 =}

41 reaction(query) -> response {=

42 SET(response, self->balance);

43 =}

44}

Figure 8: Federated Lingua Franca program with any number of ATM machines that
can provide deposits and make withdrawals.

The CAReplicaN reactor is similar to CAReplica in Fig. 4, with the only
difference being its multiport input, defined on line 29, which can accept N
input connections. The reaction on lines 34 through 40 iterates over these
inputs and adds to the balance any values it finds.

For the example in Fig. 8, it does not matter in what order simultaneous
updates are applied because the updates are commutative and associative and no
replica reads the result until all updates have been applied. Many distributed
applications with shared data, however, do not naturally have commutative and
associative merge operations.

⬇

1target C;

2import UserInput from "UserInput2.lf";

3

4federated reactor ReplicatedDataStore(N:int(4)) {

5 nodes = new[N] Node(N = N);

6 (nodes.publish)+ -> nodes.updates;

7}

8reactor Node(N:int(2)) {

9 input[N] updates:int;

10 output publish:int;

11 u = new UserInput();

12 r = new ReplicaN(N = N);

13 u.update -> r.query;

14 u.update -> publish;

15 r.response -> u.current_value;

16 updates -> r.updates;

17}

18reactor ReplicaN(N:int(2)) {

19 input[N] updates:int;

20 input query:int;

21 output response:int;

22 state record:int(0);

23

24 reaction(updates) {=

25 for (int i = 0; i < updates_width; i++) {

26 if (updates[i]->is_present) {

27 // Overwrite simultaneous updates on lesser input channels.

28 self->record = updates[i]->value;

29 }

30 }

31 =}

32 reaction(query) -> response {=

33 SET(response, self->record);

34 =}

35}

Figure 9: Distributed Lingua Franca program with any number of nodes that can
update a shared data value.

Fig. 9 shows a variant where, on line 28, an update _overwrites_ the shared
value. Such an operation is associative but not commutative. Here, each user
update is broadcast to all nodes and, as before, applied before any query for
the value is processed. If two updates are logically simultaneous, then both
updates will appear _in deterministic order_ at the multiport input (line 19)
of the replica instance. Because the order in which these updates appear is
deterministic, a priority scheme can be used to determine which update
prevails. In this implementation, a bank of $N$ nodes is created on line 5,
and each node is assigned a unique bank_index parameter ranging from 0 to
$N-1$, where $N$ is the updates_width variable referenced on line 25. Nodes
with higher bank indices have priority over nodes with lower ones simply
because the iteration on line 25 reads inputs from other members of the bank
in the same order as their bank_index. Any write by a node with a higher
bank_index will overwrite a write by a node with a lower bank_index that is
simultaneous. Hence, simultaneous updates yield deterministic results
prioritized by bank index.

###  4.7 Trading Off Consistency and Availability in Lingua Franca

Unavailability is a measure of the time it takes for a system to respond to
user requests. If it takes a long time (or it never responds), then the system
is unavailable, whereas if responses are instantaneous, then the system is
highly available.

In the ATM application in Fig. 8, a user request at the $i$-th ATM is assigned
a tag $g_{i}$ on line 18 of Fig. 5 based on the local physical clock. Hence,
$\mathcal{T}(g_{i})$ is a good measure of the physical time at which the user
has initiated a request. The user’s request turns into a tagged deposit output
from the UserInput reactor, which gets sent to the query input of the
CAReplicaN reactor. That reactor sends back the value of the shared variable
at the tag $g_{i}$, which, in this design, reflects all updates throughout the
system with tags $g_{i}$ or less.

Note that a read of the value is handled locally on each node. However, this
read will have latency that depends on the time it takes for updates to
traverse the network. In Fig. 8, notice that the reaction to query on line 41
is defined _after_ the reaction to local and remote updates. In LF semantics,
this ensures that the reaction to query is not invoked at tag $g_{i}$ until
after all local _and remote_ updates with tags $g_{i}$ or less have been
processed. It is this property that gives this program strong consistency.

The key question becomes, when can the reaction to the query input on line 41
of Fig. 8 be executed? The physical time between $\mathcal{T}(g_{i})$ and the
time of that reaction invocation becomes our measure of unavailability.

This scenario matches exactly the scenario leading to the CAL theorem in
Section 3. Hence, the unavailability is given by (10). This result is
intuitive. We will shortly show that the processing offsets can be zero in
this case, so assume $O_{i}=O_{j}=0$. If execution times are negligible, then
$\mathcal{L}_{ij}$ is just the sum of the network latency and clock
synchronization error, and the unavailability at node $i$ due to possible
updates at node $j$ is

$\bar{A}_{ij}=\max(L_{ij}+E_{ij},0).$ (25)

Recall that even though clock synchronization error can be negative, the above
maximization ensures that the unavailability is non-negative. If we further
assume that clock synchronization errors are negligible compared to network
latency, then (25) tells us that the unavailability at $i$ due to possible
updates at $j$ is equal to the network latency from $j$ to $i$, a satisfyingly
intuitive result.

We can easily modify the program to improve availability at the cost of
consistency. Specifically, if we replace line 8 of Fig. 8 with this:

⬇

8 (bank.publish)+ -> bank.updates after 100 msec;

then the inconsistency is specified to be $\bar{C}=100$ milliseconds. The
after keyword specifies a logical time offset between the sender’s tag and the
receiver’s tag. In other words, it specifies a logical delay between the
initiation of an update by a user and the recording of that update in a state
variable of each replica. This is exactly the tag manipulation considered in
Section 3, so, from Theorem 3.1, the unavailability at node $i$ becomes

$\bar{A}_{i}=\max(0,\max_{j\in N}(\mathcal{L}_{ij}-\bar{C}_{ij})),$

where we have again assumed the processing offsets are zero. Again, if clock
synchronization error and execution times are negligible compared to network
latencies, this states that the unavailability is the largest difference
between network latency and logical delays. With the choice of $\bar{C}=100$
ms, if the network latency is less than 100 msec, then unavailability becomes
zero. The system can respond instantaneously to user requests.

Even with the logical delay, this design assures eventual consistency because
all updates are applied in the same order at all nodes. Note that even _local_
updates are logically delayed, and hence the same design can be applied even
if the merge operation is not associative and commutative, as in the example
in Fig. 9.

The price for improving availability in this ATM example is that all queries
for the value of the shared variable yield a result that is (logically) 100
msec old. This means that a query for $x$ may not even reflect a recent
_local_ update. If the operations are commutative and associative, however,
then local updates need not be delayed. We can use the structure of Fig. 6 and
apply the logical delay only on the connections that broadcast local updates.
We leave it as an exercise for the reader to modify the programs in Figs. 6
and 8 to accomplish this form of bounded inconsistency by inserting after
delays. If we were to change the design to apply local updates immediately in
situations where the merge operation is neither associative nor commutative,
such as the program in Fig. 9, then we would have to do some additional work
to ensure eventual consistency, using the sorted replace described in Section
3.

What happens if the apparent latency exceeds 100 ms? There are two
possibilities. We can delay handling of events, thereby increasing
unavailability, or we can proceed with processing events as if the inputs are
absent, thereby increasing inconsistency. In Section 5, we describe two
coordination mechanisms that we have implemented in Lingua Franca, one of
which emphasizes consistency and the other of which emphasizes availability.

Another alternative is to remove the bound on inconsistency altogether while
still preserving the property that if the network is repaired, we get eventual
consistency. A one-line change in the Lingua Franca program of Fig. 8 can
realize this strategy. If we change line 8 to this subtly different version:

⬇

8 (bank.publish)+ $\small\mathtt{\sim}$>bank.updates;

then there is no upper bound on the inconsistency $\bar{C}_{ij}$. The subtle
change is to replace the logical connection -> with a physical connection
$\small\mathtt{\sim}$>. In Lingua Franca, this is a directive to assign a new
tag $g_{i}$ at the receiving end $i$ based on a local measurement of physical
time $T_{i}$ when the message is received such that
$\mathcal{T}(g_{i})=T_{i}$. The original tag is discarded. If all connections
between federates are physical connections, then the federation no longer has
any need for clock synchronization. However, the price we pay is that the
order in which updates are applied is now dependent on apparent latencies. We
preserve eventual consistency only if the merge operation is associative and
commutative. Using physical connections is a draconian measure because it also
sacrifices determinacy. This makes it much harder to define regression tests
because a correct execution of the program admits many behaviors. The use of
logical delays, together with the coordination mechanisms given in Section 5,
offers more control.

###  4.8 Determinism, Idempotence, and Causal Consistency

A Lingua Franca program that has only logical connections and no physical
connections has deterministic semantics, in the sense that once tags are
assigned, there is exactly one correct execution of the program. The runtime
infrastructure is responsible for ensuring that every reactor is presented
with inputs in tag order, that messages are delivered to reactors exactly
once, and that reactions to messages with identical tags are invoked according
to the order specified by the code. Moreover, because reactors have explicit
input and output ports, Lingua Franca has a notion of a communication channel,
a connection between two ports. Each port is guaranteed to have at most one
message at any tag. These properties, taken together, make it much easier to
design consistent distributed programs, and to trade off consistency against
availability. These properties automatically deliver what Bailis and Ghodsi
call CALM, meaning consistency as logical monotonicity [3].

Lingua Franca also ensures causal consistency, preventing the scenario at
$r_{5}$ in Fig. 2. The runtime infrastructure, described next in Section 5,
uses the topology of interconnection between reactors together with tags to
ensure that no reaction that reads an input or a state variable is invoked
until all precedent reactions have been invoked. Note that this _does not_
require that messages be globally ordered! It only requires that _each
component_ (each reactor in LF) see messages in tag order, and that
simultaneous messages (those with the same tag) are handled in precedence
order. We discuss in Section 5 how this is achieved in Lingua Franca.

##  5 Implementation

We now give two distributed coordination mechanisms, which we have implemented
as an extension of the Lingua Franca coordination language, that support
arbitrary tradeoffs between consistency and availability as network latency
varies. With centralized coordination, inconsistency remains bounded by a
chosen numerical value at the cost that unavailability becomes unbounded under
network partitioning. With decentralized coordination, unavailability remains
bounded by a chosen numerical quantity at the cost that inconsistency becomes
unbounded under network partitioning. Our centralized coordination mechanism
is an extension of techniques that have historically been used for distributed
simulation, an application where consistency is paramount. Our decentralized
coordination mechanism is an extension of techniques that have been used in
distributed databases when availability is paramount.

###  5.1 Centralized Coordination

Centralized coordination is based on the High-Level Architecture (HLA) [22]
and other distributed simulation frameworks [15, 42], with significant
extensions that we describe here. Distributed simulation is a relevant problem
because, usually, consistency trumps availability. A distributed
implementation of a simulation is expected to yield the same results as a non-
distributed version, only faster. The HLA is designed for distributed
simulation of discrete-event systems, where events have timestamps, and hence
addresses a similar problem.

We face two complications, however, that are not present in distributed
simulation applications. The first is that, in our context, unlike simulation,
events may materialize out of nowhere with tags derived from the local
physical clock. Our context, in other words, has users _interacting_ with the
system, and hence availability becomes a concern. Simulation has no such
users. In Lingua Franca, we use physical actions to realize asynchronous
stimulus from users. A second problem is that the programs in Figs. 8 and 9
have cycles without logical delays, which are not allowed in HLA.

The HLA, like other distributed simulation frameworks, uses a centralized
controller called the runtime infrastructure (RTI). Each node that wishes to
process a tagged event consults with the RTI, which grants permission to
advance its current tag to that tag only when the RTI can assure the node that
no event with a lesser tag will later appear. The existence of physical
actions and zero-delay cycles in Lingua Franca complicates this assurance and
requires extending the protocols used in HLA. Once a node has advanced its
current tag to $g$, it is no longer able to handle any events with tag less
than $g$.

Our RTI, like those in distributed simulation frameworks, realizes a mechanism
similar to vector clocks [28]. Schwartz and Mattern show that any mechanism
that preserves causal consistency fundamentally has a complexity of at least
that of vector clocks [41]. However, because Lingua Franca exposes information
about which federates communicate with which, we have realized significant
optimizations. Our RTI keeps track of the tag to which each federate has
advanced, and uses that information, together with network topology
information, to regulate the advancement of the current tag at _downstream_
federates based on the activity of their upstream federates. A federate that
has no network inputs, for example, can advance its current tag without
consulting the RTI because there is no risk of later seeing an incoming
message that has a tag less than the tag to which it has advanced. A federate
with network inputs, however, must receive an assurance from the RTI, called a
tag advance grant (TAG) before it can advance its current tag.

Our first extension over HLA supports zero-delay cycles by introducing a
provisional tag advance grant (PTAG), where the RTI assures a federate that
there will be no future message with tags _less than_ some $g$, but makes no
promises about messages with tags _equal to $g$_. This permits a federate to
advance its current tag to $g$ and execute any reactions with no dependence,
direct or indirect, on network inputs. When it has executed such reactions and
the next reaction in the reaction sequence depends on network inputs, then the
federate is required to block until it either receives messages on those
network inputs or receives an assurance that no message is forthcoming with
tag $g$. Such an assurance is similar to the null messages of Chandy and Misra
[8].

In some cases, providing such an assurance is easy. If the upstream federates
have all advanced their own current tag beyond $g$, and have informed the RTI
of this fact, then the RTI can provide the required assurance to the
downstream federate. As long as that assurance message is sent along the same
order-preserving message channel as tagged messages, then when a federate
receives the assurance, it knows it has received all relevant tagged messages
and hence can proceed. However, if there are cycles between federates that
lack logical delays, a federate may need to send a null message, an indicator
that no message with tag $g$ is forthcoming, even before it has completed
processing of all events with tag $g$. It can send such a null message as soon
as it has executed or chosen not to execute all reactions that are capable of
producing the relevant network output. Such null messages are similar to those
of Chandy and Misra [8], but are only needed in particular circumstances.

When there are physical actions, however, things are still a bit more
complicated. Consider the program in Fig. 9, focusing particularly on the
graphical rendition at the bottom. This example instantiates four federates,
each an instance of the Node reactor class. Each federate has four input
channels on its updates input port. Under centralized coordination, a federate
cannot advance its current tag to $g$ until it receives either a TAG or a PTAG
from the RTI with value $g$. It also cannot advance to $g$ until its physical
clock exceeds $\mathcal{T}(g)$ because it has a physical action.

When can the RTI provide a TAG or PTAG message? This depends on how each
federate produces network outputs. Each federate has a physical action in its
UserInput reactor that triggers a network output on the publish port. The tag
$g$ of that output will have $\mathcal{T}(g)$ taken from the local physical
clock. Hence, as soon as the physical clock of a federate exceeds
$\mathcal{T}(g)$, downstream federates can be assured that there will be no
forthcoming message with timestamp $g$ or less.

Unlike decentralized coordination (see Section 5.2 below), centralized
coordination does not rely on clock synchronization except to give a meaning
to tags originated by distributed physical actions. Instead of relying on
clock synchronization, in centralized coordination, each federate that has a
physical action that can result in network outputs must notify downstream
federates as its physical clock advances. This is done by periodically sending
to the RTI a time advance notice (TAN) message with a time $t$; this message
is a promise to not produce future messages with tags $g$ where
$\mathcal{T}(g)\leq t$, and hence is also similar to the null messages of
Chandy and Misra [8].

Unlike Chandy and Misra’s technique, in Lingua Franca, null messages are only
required when communication between federates forms a cycle without logical
delays and when physical actions trigger network outputs. Unfortunately, all
of the applications considered in Section 4 have such cycles and physical
actions and therefore require null messages. Lingua Franca provides a
mechanism to control the frequency of the TAN messages, thus controlling the
overhead, but as the frequency of messages decreases, the cost in
unavailability increases. This overhead is avoided in decentralized
coordination, explained below in Section 5.2, but at the cost that consistency
is sacrificed under network partitioning.

###  5.2 Decentralized Coordination

Decentralized coordination extends a mechanism first described by Lamport
[24], first applied to explicitly timestamped distributed systems in PTIDES
[43], and reinvented at Google to form the core of Google Spanner [10]. All
three of these use timestamps to define the logical ordering of events and
physical clocks to determine when it is safe to process timestamped events.
The physical clocks are assumed to be synchronized with a bound on the clock
synchronization error. All three also assume a bound on network latency. If
these assumptions are met at run time, then all messages will be processed in
timestamp order without any centralized coordination.

Relying on physical clocks has a key advantage with respect to availability.
Specifically, we can assume that even in the presence of complete network
partitioning, physical clocks continue to advance. If progress is governed by
the advancement of physical clocks, then unavailability can be bounded even
with no network connectivity. This contrasts with centralized coordination,
where inconsistency can be bounded, but loss of network connectivity leads to
loss of availability.

####  5.2.1 Safe to Advance (STA) Offset.

In our implementation of decentralized coordination in Lingua Franca, each
federate can have an optional safe-to-advance (STA) offset given by the
programmer. The meaning of the STA offset is that if a federate has an
earliest pending event with tag $g$, then it can advance its current tag to
$g$ when current physical time $T$ satisfies $T\geq\mathcal{T}(g)+\text{STA}$.
Hence, to handle a user request that gets assigned tag $g$, the federate needs
to wait at least until physical time exceeds $\mathcal{T}(g)$ by the STA
offset. Put another way, the STA offset is a time interval beyond
$\mathcal{T}(g)$ that a federate needs to wait before it can assume that it
will not later receive any input messages with tags less than $g$. By default,
STA = 0. The STA offset is closely related to the safe-to-process offset of
Zhao et al. [43], but is more provisional. It gives a time threshold for
_committing_ to a tag advance, but not necessarily fully _processing_ that tag
advance. Put another way, it gives a time threshold at which the federate can
assume it has seen all messages with tags _less than_ $g$, but it cannot
necessarily assume it has seen all messages with tags _equal to_ $g$. This
distinction turns out to be important for the replicated data store examples
we have seen.

Obviously, the STA offset affects availability and is clearly closely related
to the processing offset of Definition 3. There is, however, a subtle but
important distinction. The processing offset of Definition 3 is a property of
a _trace_ , an actual execution, whereas the STA is a _specification_. The
Lingua Franca code generator generates code where, when executed, every trace
will have the property that

$O_{i}\geq\text{STA}_{i}$ (26)

for each federate $i$. If the STA offset is not sufficiently large for a
particular program, then the consistency requirements of the program will not
be met. Our task, therefore, is to determine sufficiently large STA offsets
such that, if the observed apparent latencies are within our assumed
constraints, the program will process all events in tag order, thereby
achieving the desired consistency. Only when the apparent latencies exceed our
assumed constraints will the program sacrifice consistency in order to
maintain availability.

The STA offsets depend on assumed bounds on apparent latency, and vice versa,
the assumed bounds on apparent latency depend on the STA offsets, which brings
us to a second subtlety. In Definition 4, apparent latency is also a property
of a trace, whereas, to use it to derive the STA offsets, we need to use it as
a bound on _all reasonable traces_. Any assumed bound may be exceeded in
practice (e.g., the network becomes partitioned), and the strategy of
decentralized coordination is to sacrifice consistency rather than
availability when this occurs. This condition will be detectable, and Lingua
Franca supports specification of fault handlers for such conditions (see
Section 5.4).

A third subtlety is that, in Definition 4, the apparent latency is a property
of a pair of _processes_ , sequential procedures where one sends updates to
another. Lingua Franca, however, is a more richly structured language.
Federates themselves may be concurrent, running in parallel on multicore
machines, for example, and communication between federates is mediated by
input and output ports that, pairwise, give specific communication channels
over which messages with monotonically increasing tags flow. As a consequence,
apparent latency between one federate and another may vary depending on which
communication channel between the two is used. Moreover, each pair of send-
receive ports may have a different logical delay.

####  5.2.2 Causality.

To analyze a Lingua Franca program, we need to redo that analysis of Section 3
using the structure of the program. Specifically, it is possible to tell by
looking at the program whether an event at one port can result in an event at
another port, and we can find bounds on the relationship between the tags of
these two events. This analysis must be done carefully, however, because we
have to distinguish whether an event at one port can _cause_ an event at
another from whether it can _influence_ an event at another port.

To determine whether a message with a certain tag can exist, we need to
analyze the counterfactual causality properties of the Lingua Franca program.
Counterfactual causality [39] is a relation between events $e_{1}$ and $e_{2}$
where $e_{2}$ would _not occur_ were it not for the occurrence of $e_{1}$. We
distinguish this from causal influence, where event $e_{1}$ can causally
affect $e_{2}$ [26]. In a Lingua Franca program, any input to a reactor with
tag $g$ can causally affect any output with tag larger than $g$ and some
outputs with tag equal to $g$ because a reaction to that input can change the
state of the reactor. However, only some inputs counterfactually cause
particular outputs, which then in turn counterfactually cause other inputs.

Specifically, consider a reactor like this:

⬇

1reactor DirectFlowThrough {

2 input in:int;

3 output out:int;

4 reaction(in) -> out {=

5 ...

6 =}

7}

Because of the reaction signature, we assume that an event at input named in
can counterfactually cause an event at the output named out. We do not need to
analyze the body of the reaction (which is written in the target language) to
determine this fact. In contrast, consider:

⬇

1reactor CausalInfluence {

2 input in1:int;

3 input in2:int;

4 output out:int;

5 state s:int(0);

6 reaction(in1) {=

7 ...

8 =}

9 reaction(in2) -> out {=

10 ...

11 =}

12}

Here, input in1 causally influences output out (because the first reaction can
change the state, and the second reaction can use that updated state), but it
does not counterfactually cause the output. For the output to occur, a message
must arrive on in2.

In the above example, an input with tag $g$ on in1 can causally influence any
output with tag $g$ or larger. If the reactions were given in the opposite
order, then it would only be able to causally influence an output with tag
_larger than_ $g$. This is because, given simultaneous inputs, reactions of a
reactor are invoked in the order that they are declared. This distinction
proves important when analyzing the distributed replicated databases
considered earlier.

Using actions, a reactor can declare a logical delay:

⬇

1reactor IndirectFlowThrough {

2 input in:int;

3 output out:int;

4 logical action a:int(10 msec); // minimum delay of 10 msec.

5 reaction(a) -> out {=

6 ...

7 =}

8 reaction(in) -> a {=

9 ...

10 =}

11}

In this example, the logical action has a minimum delay property (set to 10
msec). The pair of reactions, taken together, reveal that the input with tag
$g$ can counterfactually cause an output with tag $g^{\prime}$, where
$\mathcal{T}(g^{\prime})$ is larger than $\mathcal{T}(g)$ by at least 10 msec.
This introduces a logical delay on the path from in to out. Consider:

⬇

1reactor Composition {

2 a = new IndirectFlowThrough();

3 b = new DirectFlowThrough();

4 a.out -> b.in;

5}

Because of the connection, we can infer that output a.out can counterfactually
cause input b.in with no logical delay. Moreover, because of the minimum delay
property, we can infer that input a.in can counterfactually cause input b.in
with logical delay of at least 10 msec.

The connection may also have a logical delay (written with the after keyword),
as in:

⬇

4 a.out -> b.in after 20 msec;

Now, the program reveals that input a.in can counterfactually cause input b.in
with logical delay of at least 30 msec.

####  5.2.3 Safe-to-Assume Absent (STAA).

Similar to the STA offset (which is, essentially, found in Lamport, PTIDES,
and Spanner), we extended Lingua Franca to allow specification of a safe-to-
assume-absent (STAA) offset associated with a network input port. The STAA
offset is used to constrain when a reaction that depends on an input port can
be invoked. Specifically, it asserts that the invocation of any reaction at
tag $g$ that depends, directly or indirectly, on a network input port $p_{i}$
is delayed until either an input is received on port ${p_{i}}$ with tag $g$ or
the physical clock $T_{i}$ at $i$ satisfies

$T_{i}\geq\mathcal{T}(g)+\text{STA}_{i}+\text{STAA}_{p_{i}}.$ (27)

At this physical time, federate $i$ assumes it has seen all inputs at port
$p_{i}$ with tags less than _or equal_ to $g$ (vs. the STA offset alone, when
it can assume it has seen all inputs with tags less than $g$). If no message
has arrived with tag $g$, the federate assumes there is no message with tag
$g$. It would be an error, to be handled as a fault condition, to later
receive a message with tag $g$.

A positive STAA offset causes the federate to block execution of reactions in
the relevant reactor until either physical time advances sufficiently or a
message arrives. In this circumstance, a null message could be used to reduce
the amount of blocking, but, unlike with centralized coordination, no null
message is required to make progress. It is sufficient for physical time to
advance. Using a null message would just be an optimization that may allow
progress sooner.

As we will see below, the STA and STAA offsets together ensure that any event
that causally influences another is processed first. To determine the STA, we
need to consider causal influence, but to determine STAA, we only need to
consider counterfactual causality.

####  5.2.4 Decentralized Coordination for the Replicated Data Store.

Consider now the replicated data store in Fig. 9. The pessimistic analysis of
Section 3.11, for this program, yields an infinite processing offset for all
federates. We will now show that, by leveraging the semantics of Lingua Franca
and our extensions to its runtime, the program can be executed correctly with
finite STA and STAA offsets. We show how to determine these offsets for each
of the federates and their input ports.

First, it will shortly become obvious that we need to separate ReplicaN and
UserInput into distinct federates, even if they run on the same host, so that
they can independently advance their current tags.666In the current
implementation of Lingua Franca, an entire federate, with all its component
reactors, advances the current tag together. In principle, some future
implementation of Lingua Franca could allow component reactors to
independently advance their current tags. This could be accomplished using
mechanisms similar to LF’s federated execution. But for now, the only
available mechanism to permit independent advancement of tags is to separate
the reactors into distinct federates. Sometimes, however, it is not possible
to create such a separation. The Lingua Franca code generator assumes that any
two reactions of the same reactor share state. It does not analyze the target
code to check whether this is the case. As a consequence, if the reactions of
UserInput and ReplicaN in Fig. 9 were instead reactions of the same reactor,
then it would not be possible to separate them into distinct federates nor to
independently advance their tags.  The refactored program is shown in Fig. 10.
Given such a separation, note that each of the four instances of ReplicaN
receive inputs from each of the four instances of UserInput, including the one
running on the same host.

⬇

1target C {

2 coordination: decentralized

3}

4

5import UserInput from "UserInput2.lf";

6import ReplicaN from "ReplicatedDataStore.lf";

7

8federated reactor (

9 N:int(4)

10) {

11 u = new[N] UserInput();

12 r = new[N] ReplicaN(N = N);

13 u.update -> r.query;

14 r.response -> u.current_value;

15 (u.update)+ -> r.updates;

16}

Figure 10: Version of the replicated data store of Fig. 9 suitable for
decentralized coordination, where UserInput and ReplicaN have been split into
separate federates so that they can independently advance their current tags.
Figure 11: Minimal version of the replicated data store of Fig. 10 with ports
renamed to all be unique.

To further simplify the explanation, we have reduced the program to that shown
in Fig. 11, which is a minimal version that avoids the compact bank and
multiport syntax of Lingua Franca and renames all the ports so that they have
unique names. There are a total of four federates and six input ports now, so
we need to determine four STA offsets and six STAA offsets.

A key property of our execution policy for Lingua Franca programs is a
federate advances its current tag to $g$ only after it has completed handling
all events with lesser tags, and then it completes handling of all events with
tag $g$ before advancing to another larger tag. I.e., even if the federate is
executing reactions concurrently (e.g., on a multicore machine), it performs a
barrier synchronization with each tag advance. There is no such barrier
synchronization across federates, but we need the barrier synchronization
within a federate, as will become obvious.

Consider first federate $f_{1}$, the UserInput at the upper left of Fig. 11.
Suppose that the physical action (depicted as a triangle with a “P”) triggers
and is assigned tag $g_{1}$ using the local physical clock. The question now
is, when can the federate advance its tag to $g_{1}$? It has to ensure that it
has seen all inputs with lesser tags, including events that may have been sent
to port $p_{5}$. For the first triggering of the physical action, it is
evident from the program structure that there is no event at $p_{5}$ with a
lesser tag because all events at $p_{5}$ are ultimately counterfactually
caused by this same physical action. Therefore, the federate can safely
advance to tag $g_{1}$ and invoke its reaction 2 with no delay. Thus, it seems
that $STA_{1}=0$ could work for federate $f_{1}$, at least for this first
event.

Once federate $f_{1}$ has advanced to $g_{1}$, it will block any further
advances until physical time advances past
$\mathcal{T}(g_{1})+\text{STA}_{1}+\text{STAA}_{p_{5}}$. Assuming it does this
correctly, then for the next triggering of the physical action with tag
$g_{2}>g_{1}$, the federate will not even face this question of whether to
advance its tag to $g_{2}$ until it has completed processing events with tag
$g_{1}$. More generally, each time the physical action triggers with tag
$g_{n}$, $n>1$, by the time the federate is considering advancing its tag to
$g_{n}$, it will have completed processing all events with tag $g_{n-1}$.
Therefore, it does not need to wait for physical time to further advance.
Hence, by induction, $\text{STA}_{1}=0$ for all tag advances. The same
argument applies for federate $f_{2}$, yielding

$\displaystyle\text{STA}_{1}$ $\displaystyle=$ $\displaystyle 0$ (28)
$\displaystyle\text{STA}_{2}$ $\displaystyle=$ $\displaystyle 0.$ (29)

To determine $\text{STAA}_{p_{5}}$, we can follow all physical time lags that
might occur on the path from the original source. We now make a critical
assumption that is required to ensure finite offsets:

###### Assumption 1

Reaction 2 of $f_{1}$ is invoked exactly at
$\mathcal{T}(g_{1})+\text{STA}_{1}$ or negligibly thereafter.

We will see in Section 5.3 how this assumption can be enforced by the Lingua
Franca program, but first, we determine the consequences of this assumption.
With it, any $\text{STAA}_{p_{5}}$ that satisfies the following will suffice:

$\text{STAA}_{p_{5}}\geq\text{STA}_{3}+\max(\text{STAA}_{p_{7}},\text{STAA}_{p_{9}})+X_{31}+X_{32}+L_{13}+E_{13},$

where $X_{ij}$ is an execution time bound on reaction $j$ of federate $f_{i}$,
$L_{ij}$ is a communication latency bound on messages from $f_{j}$ to $f_{i}$,
and $E_{ij}$ is a bound on the clock synchronization error from $f_{j}$ to
$f_{i}$. The maximization and the presence of $X_{31}$ is a consequence of the
Lingua Franca semantics that requires that if reactions 1 and 2 of the same
reactor are both enabled at any tag $g$, then reaction 1 must run to
completion before reaction 2 is invoked.

Let us make a simplifying assumption to manage the complexity of this (this
assumption, unlike Assumption 1, is not necessary, but drastically simplifies
our example). Specifically, let’s assume that execution time bounds are
negligible compared to communication latencies. With this assumption, we get

$\text{STAA}_{p_{5}}\geq\text{STA}_{3}+\max(\text{STAA}_{p_{7}},\text{STAA}_{p_{9}})+L_{13}+E_{13}.$
(30)

We can write a similar inequality for $\text{STAA}_{p_{6}}$,

$\text{STAA}_{p_{6}}\geq\text{STA}_{4}+\max(\text{STAA}_{p_{8}},\text{STAA}_{p_{10}})+L_{24}+E_{24}.$
(31)

So far, we have $\text{STA}_{1}=\text{STA}_{2}=0$ and these two inequalities.
Let us now look at $\text{STA}_{3}$.

The question is, given an event with tag $g_{3}$ that federate $f_{3}$ wishes
to process, how much physical time should it wait before advancing to tag
$g_{3}$? First, this federate has no local sources of events (actions or
timers), so the event must be an input on either port $p_{7}$ or $p_{9}$. In
either case, in order to advance to $g_{3}$, the federate needs to be assured
that it has seen all inputs earlier than $g_{3}$ _on the other port_ in order
to ensure causality. (Lingua Franca assumes that messages on each channel are
delivered in tag order.)

If the input has arrived on $p_{7}$, then it requires

$\text{STA}_{3}\geq\text{STA}_{2}+X_{22}+L_{32}+E_{32},$

which is obtained by following the counterfactual causality chain upstream
from $p_{9}$. Using $\text{STA}_{2}=0$ and the negligible execution time
assumption,

$\text{STA}_{3}\geq L_{32}+E_{32}.$

If the input with tag $g_{3}$ has arrived instead on $p_{9}$, then we require

$\text{STA}_{3}\geq\text{STA}_{1}+X_{12}+L_{31}+E_{31}.$ (32)

Combining these and ignoring execution times, we get

$\text{STA}_{3}\geq\max(L_{32}+E_{32},L_{31}+E_{31}).$

There are no further constraints on $\text{STA}_{3}$, so we can simply set

$\text{STA}_{3}=\max(L_{32}+E_{32},L_{31}+E_{31}).$ (33)

Similarly,

$\text{STA}_{4}=\max(L_{41}+E_{41},L_{42}+E_{42}).$ (34)

Similar reasoning leads to

$\displaystyle\text{STAA}_{p_{7}}$ $\displaystyle=$ $\displaystyle
L_{32}+E_{32}-\text{STA}_{3}=\min(0,L_{32}+E_{32}-L_{31}-E_{31})$
$\displaystyle\text{STAA}_{p_{9}}$ $\displaystyle=$ $\displaystyle
L_{31}+E_{31}-\text{STA}_{3}=\min(0,L_{31}+E_{31}-L_{32}-E_{32})$
$\displaystyle\text{STAA}_{p_{8}}$ $\displaystyle=$ $\displaystyle
L_{41}+E_{41}-\text{STA}_{4}=\min(0,L_{41}+E_{41}-L_{42}-E_{42})$
$\displaystyle\text{STAA}_{p_{10}}$ $\displaystyle=$ $\displaystyle
L_{42}+E_{42}-\text{STA}_{4}=\min(0,L_{42}+E_{42}-L_{41}-E_{41}).$

There is no point in having a negative STAA offset, so

$\text{STAA}_{p_{7}}=\text{STAA}_{p_{9}}=\text{STAA}_{p_{8}}=\text{STAA}_{p_{10}}=0.$
(35)

Finally, from (30) and (31), we can set

$\displaystyle\text{STAA}_{p_{5}}$ $\displaystyle=$
$\displaystyle\text{STA}_{3}+\max(\text{STAA}_{p_{7}},\text{STAA}_{p_{9}})+L_{13}+E_{13}.$
(36) $\displaystyle=$
$\displaystyle\max(L_{32}+E_{32},L_{31}+E_{31})+L_{13}+E_{13}$
$\displaystyle\text{STAA}_{p_{6}}$ $\displaystyle=$
$\displaystyle\text{STA}_{4}+\max(\text{STAA}_{p_{8}},\text{STAA}_{p_{10}})+L_{24}+E_{24}$
(37) $\displaystyle=$
$\displaystyle\max(L_{41}+E_{41},L_{42}+E_{42})+L_{24}+E_{24}.$

As a sanity check, let’s simplify further by assuming that $f_{1}$ and $f_{3}$
are mapped to the same host, so that $E_{31}=E_{13}=E_{42}=E_{24}=0$, and
$L_{31}$, $L_{13}$, $L_{42}$ and $L_{24}$ are all negligible. Under these
assumptions, we get the following total results:

$\displaystyle\text{STA}_{1}$ $\displaystyle=$ $\displaystyle 0$
$\displaystyle\text{STA}_{2}$ $\displaystyle=$ $\displaystyle 0$
$\displaystyle\text{STA}_{3}$ $\displaystyle=$
$\displaystyle\max(L_{32}+E_{32},0)$ $\displaystyle\text{STA}_{4}$
$\displaystyle=$ $\displaystyle\max(L_{41}+E_{41},0)$
$\displaystyle\text{STAA}_{p_{5}}$ $\displaystyle=$
$\displaystyle\max(L_{32}+E_{32},0)$ $\displaystyle\text{STAA}_{p_{6}}$
$\displaystyle=$ $\displaystyle\max(L_{41}+E_{41},0)$
$\displaystyle\text{STAA}_{p_{7}}$ $\displaystyle=$ $\displaystyle 0$
$\displaystyle\text{STAA}_{p_{9}}$ $\displaystyle=$ $\displaystyle 0$
$\displaystyle\text{STAA}_{p_{8}}$ $\displaystyle=$ $\displaystyle 0$
$\displaystyle\text{STAA}_{p_{10}}$ $\displaystyle=$ $\displaystyle 0.$

Further, let’s assume that clock synchronization error is negligible compared
to network latencies. Then, we get:

$\displaystyle\text{STA}_{1}$ $\displaystyle=$ $\displaystyle 0$
$\displaystyle\text{STA}_{2}$ $\displaystyle=$ $\displaystyle 0$
$\displaystyle\text{STA}_{3}$ $\displaystyle=$ $\displaystyle L_{32}$
$\displaystyle\text{STA}_{4}$ $\displaystyle=$ $\displaystyle L_{41}$
$\displaystyle\text{STAA}_{p_{5}}$ $\displaystyle=$ $\displaystyle L_{32}$
$\displaystyle\text{STAA}_{p_{6}}$ $\displaystyle=$ $\displaystyle L_{41}$
$\displaystyle\text{STAA}_{p_{7}}$ $\displaystyle=$ $\displaystyle 0$
$\displaystyle\text{STAA}_{p_{9}}$ $\displaystyle=$ $\displaystyle 0$
$\displaystyle\text{STAA}_{p_{8}}$ $\displaystyle=$ $\displaystyle 0$
$\displaystyle\text{STAA}_{p_{10}}$ $\displaystyle=$ $\displaystyle 0.$ (38)

These results are intuitive. They show that, at the UserInput federates, when
a physical action triggers with tag $g$, the federate can immediately advance
its current tag to $g$, so reaction 2 can be immediately invoked, resulting in
a network output. Whether to invoke reaction 3 the UserInput federates cannot
be determined until physical time exceeds $\mathcal{T}(g)$ by a bound on the
network latency from the other host, a satisfyingly intuitive result because
that is where a remote update may occur.

At the Replica federates, when they receive an input with tag $g$, they can
advance to tag $g$ only when physical time exceeds $\mathcal{T}(g)$ by the
bound on the network latency from the other host. This too is intuitive
because only at that physical time can they be sure there is no forthcoming
message from the other host with a lesser tag.

For this particular example, at the UserInput federates, as long as the
assumptions on network latency and clock synchronization are satisfied, there
_will_ be a network input with tag $g$, and reaction 3 will be invoked. But
the LF infrastructure cannot be sure that this is the case without imposing
further constraints on the target code in reaction 2 of UserInput and reaction
2 of Replica. Those reactions are free to choose to not produce an output.

As of this writing, in Lingua Franca, the STA and STAA offsets must be derived
by hand and provided as part of the specification of the program. We leave it
to future work to derive these thresholds automatically given assumptions
about apparent latency. This will require performing analysis of the structure
of the program and rejecting programs that result in infinite values for these
offsets. The analysis is simple for this program, but it could be quite
challenging in general. For example, if UserInput had a second physical action
and there were a logical delay $D$ somewhere along the path from its update
output back to its current_value input, then $\text{STA}_{i}=0$ is not
necessarily any longer valid.

Why did we have to separate UserInput and Replica into distinct federates?
Were they in the same federate, then $\text{STA}_{3}=\text{STA}_{1}$. From
(32), we have the constraint that $\text{STA}_{3}>\text{STA}_{2}$.
Correspondingly, $\text{STA}_{4}>\text{STA}_{1}$ and
$\text{STA}_{4}=\text{STA}_{2}$. Combining these, we get
$\text{STA}_{1}>\text{STA}_{2}>\text{STA}_{1}$, a constraint which is not
satisfiable.

###  5.3 Unavailability in the Replicated Data Store

Using the most simplified result, given by (5.2.4), we can see the
consequences of the CAL theorem for the replicated data store example. This
program is strongly consistent. There are no logical delays, so each replica
will agree on the value of the shared variable at every tag. A user who issues
a query gets a reply when reaction 3 of UserInput is invoked. From (5.2.4), we
see that $\text{STAA}_{p_{5}}=L_{32}$ and $\text{STAA}_{p_{6}}=L_{41}$, which
means that the time it takes to respond to a user query is at most the network
latency between nodes.

This is intuitive and not surprising. However, there is more subtle
consequence. Recall Assumption 1, that reaction 2 of $f_{1}$ is invoked
exactly at $\mathcal{T}(g_{1})+\text{STA}_{1}$ or negligibly thereafter.
Because of the barrier synchronization for advancement of the current tag in a
federate, this assumption _may not be met_ if the physical action triggers too
closely after its previous trigger, specifically within $L_{32}$. If the
physical action triggers while UserInput1 is waiting on port $p_{5}$, then
there will be a delay in the invocation of reaction 2 and the derived STA and
STAA offsets are no longer assured to be valid. A fault condition may occur.

Fortunately, Lingua Franca provides mechanisms to prevent such eventualities.
First, a physical action can have a minimum spacing parameter, a minimum
logical time interval between tags assigned to events. When the environment
tries to violate this constraint by issuing requests too quickly, the
programmer can specify one of three policies: drop, replace, or defer. The
drop policy simply ignores the event. The replace policy replaces any
previously unhandled event, or if the event has already been handled, defers.
The defer policy assigns at tag $g$ to the event with timestamp
$\mathcal{T}(g)$ that is larger than the previous event by the specified
minimum spacing. This feature of the language can be used to help protect a
system against denial of service attacks that might otherwise trigger fault
conditions.

While the minimum spacing parameter ensures that tags are sufficiently spaced,
it does not, by itself, ensure that the scheduler will prioritize execution of
reaction 2 so as to satisfy Assumption 1. Lingua Franca provides a mechanism
to ensure this, a deadline that can be associated with a reaction. The syntax
for this is as follows:

⬇

38 physical action r;

39 reaction(r) {=

40 ... normal case

41 =} deadline(1 msec) {=

42 ... exception case

43 =}

The semantics of a Lingua Franca deadline is that if the reaction to an event
with tag $g$ is invoked at a physical time $T>\mathcal{T}(g)$, then instead of
invoking the “normal case” reaction, the “exception case” reaction will be
invoked. This provides a mechanism to handle overload conditions, but, more
importantly, the deadline provides a hint to the scheduler to prioritize
invocation of this reaction. Indeed, Lingua Franca uses an earliest-deadline-
first (EDF) scheduling policy, thereby ensuring, for sufficiently simple
programs, that Assumption 1 will be met.

###  5.4 Handling Fault Conditions

Whether we use centralized or decentralized coordination, faults can occur.
Centralized coordination bounds inconsistency, and when faults make it
impossible to proceed without exceeding those bounds, program execution must
pause or stop. In this case, the RTI is a single point of failure, so a fault-
tolerant system will need to have a mechanism to elect a new RTI when one
fails. Progress at a federate can also be stopped by failure of an upstream
federate. A mechanism for restarting federates could help mitigate this risk,
but the time it takes to perform such a restart will inevitably decrease
availability. A network partition will also stop progress. If communication
from a federate to the RTI is lost, no progress is possible either for the
federate or for federates downstream of it. There is no mitigation for this
loss of availability that keeps inconsistency bounded, as shown by the CAL
theorem.

When we want to bound unavailability instead of inconsistency, we should use
decentralized coordination. Decentralized coordination may also be more
efficient, because it does not require null messages to handle physical
actions, but it requires clock synchronization, which also increases network
traffic.

For decentralized coordination to work, we need to assume a bound on apparent
latency $\mathcal{L}$, which means bounding network latency, execution time
overhead, and clock synchronization error. Lingua Franca provides a convenient
mechanism for handling faults where these assumptions have been violated.
Specifically, in LF, the program can associate with each reaction that is
triggered by a network input a fault handling reaction that will be invoked
instead of the regular reaction whenever a network input arrives that has an
unexpectedly early tag such that the local current tag has already advanced
past that tag. In this case, there is no way to correctly execute the program.
What the fault handler does is application dependent. A database application,
such as Spanner, may, for example, overlay a transaction schema on top of the
mechanisms provided by decentralized coordination and reject a transaction
when such a fault occurs. A bulletin board application, in contrast, could
provide a more graceful recovery that simply re-orders the posts.

Decentralized coordination guarantees a bound on unavailability. The Lingua
Franca program specifies a bound on inconsistency, but the CAL theorem tells
us that if apparent latency exceeds some threshold, the bound on inconsistency
can no longer be achieved. If and when this happens, if specified, a fault
handler will be invoked.

Centralized coordination, in contrast, guarantees a bound on inconsistency. Is
there a way in this case to specify a bound on unavailability? In Lingua
Franca, the programmer can specify a deadline for a particular reaction
invocation. Suppose we replace lines 38 through 40 in Fig. 5 with this:

⬇

38 reaction(balance) {=

39 printf("Balance: %d\n", balance->value);

40 =} deadline(100 msec) {=

41 printf("Apologies for the delay! Your balance is %d\n", balance->value);

42 =}