In recent times, considerable emphasis has been given to two apparently disjoint research topics: data-parallel and eventually consistent, distributed systems. In this paper we propose a study on an eventually consistent, dataparallel computational model, the keystone of which is provided by the recent finding that a class of programs exists that can be computed in an eventually consistent, coordination-free way: monotonic programs. This principle is called CALM and has been proven by Ameloot et al. for distributed, asynchronous settings. We advocate that CALM should be employed as a basic theoretical tool also for data-parallel systems, wherein computation usually proceeds synchronously in rounds and where communication is assumed to be reliable. We deem this problem relevant and interesting, especially for what concerns parallel workflow optimization, and make the case that CALM does not hold in general for dataparallel systems if the techniques developed by Ameloot et al. are directly used. In this paper we sketch how, using novel techniques, the satisfiability of the if direction of the CALM principle can still be obtained, although just for a subclass of monotonic queries.
Recent research has explored ways to exploit different levels of consistency in order to
improve the performance of distributed systems w.r.t. specific tasks and network
configurations, while maintaining correctness [
Rsync is a common setting in modern data-parallel frameworks - such as
MapReduce - in which computation is usually performed in rounds, where each task is blocked
and cannot start the new round until a synchronization barrier is reached, i.e., every
other task has completed its local computation. In this work we consider
synchronization (barrier) and coordination as two different, although related entities: the former is
a mechanism enforcing the rsync model, the latter a property of executions.
Identifying under what circumstances eventually consistent, coordination-free computation can
be employed over rsync systems would enable us to “stretch” the declarativeness of
parallel programs, freeing execution plans of the restriction to follow predefined
(synchronous) patterns. In fact, all recent high-level data-parallel languages suffer from this
limitation, for instance both Hive [
Recently, a class of programs that can be computed in an eventually consistent,
coordination-free way has been identified: monotonic programs [
In this paper we expect the reader to be familiar with the basic notions of database
theory and relational transducer (networks). In this section we use some example to set
forth our notation, which is close to that of [
We employ a transducer (resp. a transducer network) as an abstraction modeling the behavior of a single computing node (resp. a network of computing nodes): this abstract computational model permits us to make our results as general as possible without having to rely on a particular framework, since transducers and transducer networks can be easily imposed over any modern data-parallel system. We consider each node to be equipped with an immutable database and a memory used to store useful data between any two consecutive computation steps. In addition, a node can produce an output for the user and can also communicate some data to other nodes (the concept of data communication in a transducer network will appear clearer in Section 3). Finally, an internal time, and system data are kept mainly for configuration purposes. Every node executes a program that operates on (input) instances of the database, the memory and the communication channel, and produces new instances that are either saved in memory, or directly output to the user, or addressed to other nodes.
Example 1. A first example of relational transducer is the following UCQ-transducer T , with schema , that computes the ternary relation Q as the join between two binary relations R and T :
Schema: db = fR(2); T (2)g; mem = ;; com = ;; out = fQ(3)g
Program: Qout(u; v; w) R(u; v); T (v; w): Let I be an initial instance over which we want to compute the join. Then, let us define Idb = I as an instance over the database schema db. A transition I ! J for T is such that I = I [ Isys, Ircv and Jsnd are empty (no communication query exists), and J = I [ Iout [ Isys, where Iout is the result of the query Qout, i.e., the join between R and T . Note that the subscript in Qout means that this is an output query, that is, it specifies the final result of the whole computation. 3
In order to allow query evaluation in parallel settings, we will sketch a novel transducer
network [
Example 2. Assume we want to compute a distributed version of the join of Example 1. We can implement it using a broadcasting synchronous transducer network which emits one of the two relations, say T , and then joins R with the received facts over T . Note that the sent facts will be used just starting from the successive round, and the program will then employ two rounds to compute the distributed join. UCQ is again expressive enough. The transducer network can be written as follows – where Ssnd denotes a communication query and this time schema com is non-empty because communication is needed:
Schema: db = fR(2); T (2)g; com = fS(2)g; out = fQ(3)g
T (u; v): Qout(u; v; w)
R(u; v); S(u; w):
Lemma 1 Let L be a language containing UCQ and contained in DATALOG:. Every query expressible in L can be distributively computed in 2 rounds by a broadcasting L-transducer network.
The above lemma permits us to draw the following conclusion: under the rsync
semantics, monotonic and non-monotonic queries behave in the same way: two rounds
are needed in both cases. This is due to the fact that, contrary to what happens in the
asynchronous case of [
The CALM conjecture [
The concept of coordination suggests that all the nodes in a network must exchange
information and wait until an agreement is reached about a common property of interest.
Following this intuition, Ameloot et al. established that a specification is
coordinationfree if communication is not strictly necessary to obtain a consistent final result.
Surprisingly enough, under this definition of coordination-freedom, CALM does not hold
in rsync settings under the broadcasting communication model:
Example 3. Let Qout be the “emptiness” query of [
Schema: db = fR(0)g; mem = fReady(0)g; com = fS(0)g; out = fT (0)g: Program: Ssnd() R():
Readyins()
:Ready(): Tout()
:S(); Ready():
One can show [
The result we have is indeed interesting although expected: when we move from the
general asynchronous model to the more restrictive rsync setting, we no longer have a
complete understanding of which queries can be computed without coordination, and
which ones, instead, do require coordination. It turns out that both the communication
model and the definition of coordination proposed in [
Broadcasting specifications are not really convenient from a practical perspective.
Following other parallel programming models such as MapReduce, in this section we are
going to introduce hashing transducer networks: i.e., synchronous networks of
relational transducers equipped with a content-based communication model founded on
hashing. Under this new model, the node to which an emitted fact must be addressed is
derived using a hash function applied to a subset of its terms called keys.
Example 4. This program is the hashed version of Example 2, where every tuple
emitted over S and U is hashed on the first term (this is specified by the schema definition
S(1;2) and U (1;2), where the pair (1; 2) means that the related relation has arity 2 and
the first term is the key-term). In this way we are assured that, for each pair of joining
tuples, at least a node exists containing the pair. This because S and U are joined over
their key-terms, and hence the joining tuples are addressed to the same node.
We have seen in Section 3.1 that, for rsynch systems, a particular notion of
coordinationfreedom is needed. In fact we have shown that, under such model, certain
nonmonotonic queries – Example 3 – requiring coordination under the asynchronous model
can be computed in a coordination-free way. The key-point is that, as observed in [
Definition 4.1. Given a run , we say that two points ( i; t), ( j ; t0) are related by a direct potential causality relation !, if one of the following is true: 1. t0 = t + 1 and i = j; 2. t0 t + 1 and node i sent a tuple at time t addressed to j; 3. there is a point ( k; t00) s.t. ( i; t) ! ( k; t00) and ( k; t00) ! ( j ; t0). Note that direct dependencies define precisely Lamport’s happen-before relation – and hence we maintain the same signature !.
Differently from asynchronous systems, we however have that a point on node j
can occasionally indirectly depend on another point on node i even if no fact addressed
to j is actually sent by i. This is because j can still draw some conclusion simply
as a consequence of the bounded delay guarantee of synchronous systems. That is,
each node can use the common knowledge that every sent tuple is received at most
after a certain bounded delay to reason about the state of the system. The bounded
delay guarantee can be modelled as an imaginary N U LL fact, like in [
Definition 4.2. Given a run , we say that two points ( i; t), ( j ; t0) are related by an indirect potential causality relation 99K, if i 6= j, t0 t + 1 and a N U LLiR fact addressed to node j has been (virtually) sent by node i at round t.
An interesting fact about the bounded delay guarantee is that it can be employed to specify when negation can be safely applied to a predicate. In general, negation can be applied to a literal R(u) when the content of R is sealed for what concerns the current round. In local settings, we have that such condition holds for a predicate at round t0 if its content has been completely generated at round t, with t0 > t. In distributed settings, we have that if R is a communication relation, being in a new round t0 is not enough, in general, for establishing that its content is sealed. This is because tuples can still be floating, and therefore, until we are assured that every tuple has been delivered, the above condition does not hold. The result is that negation cannot be applied safely. We can reason in the same way also for every other negative literal depending on R. We will then model the fact that the content of a communication relation R is stable because of the bounded delay guarantee, by having every node i emit a fact N U LLiR at round t, for every communication relation R, which will be delivered at node j exactly by the 1 Note that a point in a synchronous system is what Lamport defines as an event in an asynchronous system. next round. We then have that the content of R is stable once j has received a N U LLiR fact from every node i contained in the set N of nodes composing the network. The sealing of a communication relation at a certain round is then ascertained only when jN j N U LLR facts have been counted. Recall that not necessarily the N U LLiR facts must be physically sent. This in particular is true under our rsync model, where the strike of a new round automatically seals all the communication relations. Example 5 shows one situation in which this applies.
Example 5. Consider the hashing version of the program of Example 3. Let I be an initial instance. At round t + 1 we have that the relation S is stable, and hence negation can be applied. Note that if R is empty in the initial instance, no fact is sent. Despite this, every node can still conclude at round t + 1 that the content of S is stable. In this situation we clearly have an indirect potential causality relation.
We are now able to introduce the definition of syncausality: a generalization of Lamport’s happen-before relation which considers not only the direct information flow, but also the flow generated by indirect dependencies.
Definition 4.3. Let be a run. The syncausality relation ; is the smallest relation s.t.: 1. if ( i; t) ! ( j ; t0), then ( i; t) ; ( j ; t0); 2. if ( i; t) 99K ( j ; t0), then ( i; t) ; ( j ; t0); and 3. if ( i; t) ; ( j ; t0) and ( j ; t0) ; ( k; t00), then ( i; t) ; ( k; t00). 4.2
We next propose the predicate-level syncausality relationship, modeling causal relations at the predicate level. That is, instead of considering how (direct and indirect) information flows between nodes, we introduce a more fine-grained relationship modelling the flows between predicates and nodes.
Definition 4.4. Given a run , we say that two points ( i; t), ( j ; t0) are linked by a relation of predicate-level syncausality ;R, if any of the following holds: 1. i = j, t0 = t + 1 and a tuple over R 2 mem [ out has been derived by a query in Qins [ Qout at time t0; 2. R 2 com and node i sends a tuple over R at time t addressed to node j, with t0 t + 1; 3. R 2 com and node i (virtually) sends a N U LLiR fact at time t addressed to node j, with t0 t + 1; 4. there is a point ( k; t00) s.t. ( i; t) ;R ( k; t00) and ( k; t00) ;R ( j ; t0). We are now able to specify a condition for achieving coordination. Informally, we have that coordination exists when all the nodes of a network reach a common agreement that some event happened. But the only way to reach such an agreement is that a (direct or indirect) information flow exists between the node in which the event actually occurs, and every other node. This is a sufficient and necessary condition because of the reliability and bounded-delay guarantee of rsync systems. Formalizing this intuition by means of the (predicate level) syncausality relationship we have that: ( i; t) ;R ( j ; t0).
Definition 4.5. Let N be a set of nodes. We say that a synchronous relational
transducer network manifests the coordination pattern if, for all possible initial instances
I 2 inst( db), whichever run we select, a point ( i; t) and a communication
relation R exist so that 8j 2 N there is a predicate-level syncausality relation such that
We call node i the coordination master. A pattern with a similar role has been named
broom in [
Remark: The reader can now appreciate to which extent coordination was already
“baked” inside the broadcasting synchronous specifications of Section 3. Note that
broadcasting, in rsync, brings coordination. This is not true in asynchronous systems.
Intuitively, the coordination master is where the event occurs. If a broadcasting of
(informative or non-informative) fact occurs, then such event will become common
knowledge [
The original version of the CALM principle is not satisfiable in rsync systems because a monotonic class of queries exists—i.e., unchained queries, introduced next—which is not coordination-free. Informally, a query is chained if every relation is connected through a join-path with every other relation composing the same query. Definition 5.6. Let body(qR) be a conjunction of literals defining the body of a query qR. We say that two different positive litteral occurrences Ri(ui), Rj (uj ) 2 body(qR) are chained in qR if either: – ui \ uj 6= ;; or – a third relation Rk 2 qR different from Ri, Rj exists such that Ri is chained with
Rk, and Rk is chained with Rj .
Definition 5.7. A query Qout is said chained if, for every rule qR 2 Qout, each relation occurrence Ri 2 body(qR) is chained with every other relation occurrence Rj 2 body(qR).
Example 6. Assume two relations R(2) and T (1), and the following query Qout returning the full R-instance if T is nonempty.
Q(u; v)
R(u; v); T ( ): The query is clearly monotonic. Let T be the following broadcasting UCQ-transducer program computing Qout.
Assume now we want to make the above transducer a hashing one. We have that, whichever key we chose, the related specification might be no more consistent. Indeed, consider an initial instance I and a set of keys spanning all the terms of S and U . Assume I such that adom(IR) adom(IT ), and a network composed by a large number of nodes. In this situation, it may happen that a nonempty set of facts over R is hashed to a certain node i, while no fact over T is hashed to i. This because a constant may exist in adom(IR) that is not in adom(IT ) and for which the hashing function returns a node i not returned by hashing any constant in adom(IT ). Hence no tuple emitted to i will ever appear in the output, although they do appear in Qout(I). Thus this transducer is not eventually consistent.
From the above example we can intuitively see that, for rsync, a final consistent
result can be obtained without coordination only for queries that are chained and
monotonic. That is, the following restricted version of the CALM conjecture holds for rsync
systems:
Theorem 1 A query can be parallelly computed by a coordination-free transducer
network if it is chained and monotonic [
We will leave for future works the investigation on whether every monotone and chained query is also coordination-free.
Remark: For the readers familiar with the works [
In this paper the CALM principle is analyzed under synchronous and reliable settings. By exploiting CALM, in fact, we would be able to break the synchronous cage of modern parallel computation models, and provide pipelined coordination-free executions when allowed by the program logic. In order to reach our goal, we have introduced a new abstract model emulating BSP computation, and a novel interpretation of coordination with sound logical foundations in distributed knowledge reasoning. By exploiting such techniques, we have shown that the if direction of the CALM principle indeed holds also in rsync settings, but just for the subclass of monotonic queries defined as chained.