Logic-based languages, such as Datalog and Answer Set Programming, have been recently put forward as a data-centric model to e ectively specify and implement network services and protocols, seeing them as dynamic systems of distributed computational nodes where each node evolves an internal database and exchanges data with the other nodes of the network. This approach provides the basis for declarative distributed computing. However, a rigorous, comprehensive characterization of the decidability and complexity of veri cation in declarative distributed systems is yet to come. This paper charts the decidability border of the veri cation of convergence properties, considering the case where the network is a xed connected graph, nodes can incorporate fresh data from the external world into the system, and can communicate asynchronously by means of reliable but unordered channels.
In the past years we have seen how declarative database query languages, such as
Datalog, can naturally be used to specify and implement network services and
protocols [
There are several variants of concrete languages for specifying ddss [
Commons License Attribution 4.0 International (CC BY 4.0).
passed between nodes are snippets of DBs, providing a close correspondence
between the programs and a formal speci cation in logic of their computation. This
facilitates the development of program analysis tools [
In the next Section 2 we introduce the dds model, provide its execution semantics, and de ne the veri cation problem over convergence properties. In Section 3 we chart the decidability boundary of veri cation of convergence. Finally, Section 4 provides the conclusions. 2
The general computational model of ddss can be described as a network of
uniquely identi ed nodes, each running an input/output state machine, where
outputs of some machines become inputs of others [
A state in the state machine of a node is represented by a (relational) DB. As customary, a DB de nes a set of relations conforming to a given schema. A DB schema is a nite set of relation schemas R=n, with a relation name R and an arity n, representing the number of components (or attributes) it contains. By a slight abuse of notation, sometimes we drop the arity and interchangeably use the same symbol to indicate the relation schema and its name. We x a countable data domain denoting an in nite set of constants. An instance of a relation schema R=n is a nite set of n-tuples over , and a DB is the union of relation instances over its schema. Given a DB instance I, we denote by adom(I) the active domain of I, that is the ( nite) set of constants explicitly appearing in I.
State transitions occur when the node receives inputs, in the form of DBs, from external components (such as applications running in the node, or humans), and/or from state machines running in other nodes. A state change can also produce an output in the form of a DB that can be delivered to another node fact by fact. Thus, every node has an input schema I, a state schema S, and a transport schema T . The latter is used to communicate between nodes. Let I, S, and T denote all possible instances of I; S, and T , respectively. The state transition mapping in a node is a subset of S T I S T: given a pair (S ; T ) of state and transport DBs, and an input DB I , the transition results in a new pair (Sn; Tn) of state and transport DBs.
Speci cally, that mapping is speci ed by means of d2c, a declarative
programming language introduced in [
H if L1; : : : ; Ln; prev Ln+1; : : : ; Lm; C Like in Datalog, H is the head atom, but in d2c it is possibly annotated with a term of the form @t, where t is a variable or constant from a xed domain in the language used to identify nodes (concretely, such domain would correspond to objects such as IP addresses or URLs). The Lis are literals (i.e., atoms or negated atoms), again possibly annotated with a term of the form @t. Finally, C is a set of (in)equality constraints over variables and constants.
The predicate names follow the standard correspondence of predicates and DB tables made by Datalog. All variables appearing in H, in C, or in any negated atom inside a rule must also appear in a non-negated atom among the Lis of the same rule3. The name of annotated atoms must correspond to relation names in the transport schema T of the node, and only names that correspond to transport or state tables can appear in H. Informally, a ground instance h if l1; : : : ; ln; prev ln+1; : : : ; lm; c of a rule says that h is in the current state or is an output of the current state if l1; : : : ; ln are true in the current state, ln+1; : : : ; lm were true in the previous state, and c is valid. As in Datalog with negation, we make the usual assumption on the strati cation of negated atoms in l1; : : : ; ln w.r.t. h, so that the transition mapping can be computed using the standard Datalog xpoint evaluation of Datalog. Notice that the literals in the prev scope do not concur to determine strati cation, since they are over a xed database already computed in the previous computation step. Also annotated predicates are not involved in strati cation, since they address incoming messages received in the form of an extensional DB fact. The meaning of an annotation depends on whether it appears in the head or the body of a rule: in the former case, it indicates the destination of the transport tuple, in the latter it points to the source. 3 We can relax this requirement for negated atoms containing anonymous variables.
Consider in particular a negated atom not P (X~ ; : : : ; ) appearing in a rule where variables X~ also appear in positive atoms. Such an atom can be replaced by not N (X~ ), where N is a fresh predicate, provided we introduce the additional rule N (X~ ) if P (X~ ; ; : : : ; ).
Example 1. Consider the input predicate echo=2, state predicate alive=1, and transport predicate say=1. Rule say(M)@D if echo(M; D) models that the node sends m to the node with ID d (if it is a node neighbor) when the corresponding echo(m; d) fact is given as input. M and D are variables, which in this case are bound to constants m and d respectively. Rule alive(S) if say(M)@S models that the node adds to its state the information that node s is alive, whenever the transport tuple say(m) is received from node s. Rule say(M)@Src if say(M)@Src \echoes" the say tuple back to the source node. tu
To support also non-deterministic transitions, D2C features the special
predicate choice by Sacca and Zaniolo [
A dds relies on an underlying network N of communicating nodes, each running a d2c program. In distributed computing, N is typically represented as a graph hV; Ai, where V are the computation nodes, and the arcs in A re ect the ability to communicate according to the physical network con guration. Directed arcs denote non-symmetric communication.
There is a complex spectrum of di erent network models, depending on the topology of the graph, the degree of mobility of nodes, and on which extent the network may vary over time. Here we consider networks with (i) xed topology; (ii) bidirectional communication channels; (iii) strongly connected nodes; (iv) the ability for every node to communicate with itself. This class of networks can be represented by xed undirected, connected graphs, where each node has a self-loop. From now on, we always assume that the network is in this class. To make nodes aware of their own name and that of their neighbors, each node has the following rules: my name(M) if prev my name(M): neighbor(N) if prev neighbor(N): This information is read-only, so no other rule can use my name and neighbor in its head. 2.3
We now formalize ddss adopting homogeneous nodes, i.e., all nodes run the same program. This also means that the local DB schemas are the same for all nodes. Still, the behavior of di erent nodes might diverge over time, depending on: (i) their location in the network, (ii) the presence of non-deterministic choices in the program, and (iii) the data obtained from the interaction with other nodes and with the external world. A dds M is a tuple hN ; I; T ; S; P; D0i, where: N is the network graph (obeying to the assumptions of Section 2.2); I, T , and S respectively denote the input, transport, and state schemas of every node in M; P is the d2c program run by every node in M; D0 is a local state DB over S representing the initial state of each node, and so that it assigns no tuples to my name and neighbor (these are in fact implicitly set di erently for each node, depending on the network topology); its active domain is denoted by 0, while its extension with node names is denoted by 0;M
Commonly, the execution semantics of dynamic systems over relational data
is given in terms of a relational transition system (RTS), i.e., a transition
system where each state is labeled by a DB, representing the con guration of the
current memory of the system in that state [
To account for those local aspects, we reorganize the global memory into many local ones attached to nodes and channels. Technically, given a dds hN ; I; T ; S; P; D0i with N = hV; Ai, and given a data structure CT over T for communication, the execution semantics of the dds under CT is given via a so-called distributed RTS (DRTS) of the form hN ; ; S; T ; ; 0; ndb; nch; )i, where:
is a (possibly in nite) set of states; 0 2 is the initial state; ndb is a function that, given a state corresponding DB instance of S over and a node n 2 V , returns a nch is a function that, given a state 2 and two nodes n1; n2 2 V such that hn1; n2i 2 A, returns an instance of CT storing the pending messages from n1 to n2; )
is a transition relation between states.
is serial if, for every state 2 , there exists 0 2 such that ) 0. Notice that the input schema and databases are not mentioned in the general DRT S de nition, but will be used later to de ne concrete transition relations. Moreover, in case the instantiations of CT can be represented by means of a DB, it is possible to boil down a DRTS to a standard RTS by labeling states with the disjoint union of all the state and channel DBs mentioned in the DRTS states.
To build a DRTS capturing the execution semantics of a dds M, one has to choose which communication model is used by M to handle the exchange of messages in the network, and how does M interact with the external users that exchange data with the computation nodes. Speci cally, we consider the relevant setting where communication is reliable and asynchronous, i.e., messages are never lost and message exchanges occur independently from each other. Consistently with the fact that each transport atom in a D2C program is attached to a sender/receiver, each message consists of a single transport fact.
As for user interaction, nodes may receive a new input DB when they start processing an incoming message. We call ddss obeying to that input-policy interactive ddss (iddss). Coupling the processing of input DBs with that of incoming messages is without loss of generality, since a node may send dummy messages to itself just to signal that it is ready to process a user input. 2.4
Due to the aforementioned asynchronicity and reliability, we can assume that, at each computation step, only one node reacts to the delivery of an incoming message. Within the computed result, there may be transport facts labeled with corresponding destination nodes; these are all simultaneously emitted, and will be asynchronously received by their respective recipient nodes fact by fact. Since we assume no guarantees on the order in which messages are received, communication can be abstractly captured by equipping each node with one message multiset per neighbor. Hence, from now on, we always assume that the data structure for communication channel is multiset, and make use of the usual operators to manipulate and inspect multisets. Relying on multisets instead of sets is important to capture the fact that two distinct messages exchanged between the same two nodes and carrying exactly the same payload may be both on their way to the recipient node.
The dds evolution is then captured by iterating through these steps: a non-empty multiset is non-deterministically picked, non-deterministically extracting a message M . the destination node performs a computation step triggered by M , possibly considering also external input data; the node state is updated and the produced messages are inserted in the corresponding destination multiset.
Formally, given a program P , an input DB I, a previous state DB S, and a labeled transport tuple t@n, we denote by state(P; I; S; t@n) the new state database computed by the program P over S[I[ft@ng, by transp(P; I; S; t@n; d) the set of computed output tuples (over the trasnport signature) labeled by @d, and by transp # (P; I; S; t@n; d) the set of tuples in transp(P; I; S; t@n; d) where the label @d has been dropped. Let M = hN ; I; T ; S; P; D0i be a dds system whose network is N = hV; Ai. To formalize the execution semantics, we introduce a relation c-stepM that substantiates the generic state transition mapping introduced in Section 2.1: c-stepM
Y S n2V
Y C a2A
A I
Y S n2V
Y C a2A Speci cally, given S; S0 2 Qn2V S, and C; C0 2 Qn2V C, a channel (s; d) 2 A, and an input database I 2 I, we have that hS; C; (s; d); I; S0; C0i 2 c-stepM if and only if there exists a message tuple t 2 C(s;d) such that:
Sn0 = (state(P; I; S; t@s) if n = d
Sn otherwise >8C(s;d) n ftg C(0n;m) = <C(d;m) [ transp # (P; I; Sn; t@d; m) if n = d >:C(n;m) otherwise if n = s and m = d Finally, we de ne the transition system of M, written int, as the DRTS M hN ; ; S; T ; ; 0; ndb; nch; )i, where: { Qn2V S Qa2A C and, for each 2 of the form
ndb( ; n) = Sn and nch( ; s; d) = C(s;d); { 0 = ((D0)n2V ; (Cc)c2A), where C(s;d) = ; if s 6= d and C(s; d) = fstartg otherwise, where start is a special 0-ary transport tuple used to guarantee at least one computation step to each node; { The extensions of and ) are de ned by simultaneous induction as follows: 1. 0 2 ; 2. if (S; C) 2 , then, for each hS; C; (s; d); I; S0; C0i 2 c-stepM, it is true that (S0; C0) 2 and (S; C) ) (S0; C0). 3. if (S; C) 2 and, for each c 2 A, Cc = ;, then (S; C) ) (S; C). = ((Sn)n2V ; ((Cc)c2A)), Notice that ) is guaranteed to be serial. 2.5
Convergence is a generalization of termination for ddss, indicating that the distributed computation run by the whole dds eventually reaches a stable situation where all nodes are quiescent, i.e., do not change anymore their state DBs. This typically occurs when no messages are exchanged. However, we want to rule out those cases in which quiescence is reached because one or more nodes stop their computation due to an error (e.g., because the node has received an unexpected message/input, or has entered an undesired state). We assume that a node declares that it is faulty by inserting the special ag error in its state. Under this assumption, d2c programs employ error to indicate under which circumstances a node becomes faulty.
Convergence properties are then de ned by mixing two dimensions: (i) number of faulty nodes in a run, with the two extreme cases of total (no faulty node) vs partial (at least one non-faulty node) correctness; (ii) quanti cation over runs, considering the case in which the dds sometimes (i.e., for at least one run) vs always (i.e., for all runs) converges. Given a dds M, this gives rise to four variants of convergence: { M sometimes converges with total correctness if there exists a run of M eventually reaching a state where all nodes are quiescent, and none is faulty. { M sometimes converges with partial correctness if there exists a run of M eventually reaching a state where all nodes are quiescent, and at least one is not faulty. { M always converges with total correctness if every run in which all nodes of
M stay non-faulty eventually converges. { M always converges with partial correctness if every run in which at least one node of M stays non-faulty eventually converges.
Moreover, convergence properties can be formulated in sophisticated logics that allow to specify properties over the data owing in the DDS and the DDS temporal behavior. A suitable language over RTSs is FO-CTL, which mixes rstorder (FO) logic with the computation three logic (CTL). It is possible to adapt that logic to DRTSs, called DDS CT L. That can be achieved by exploiting some D2C rule to transfer the previous state con guration into a copy in the current one, using the veri cation formula to check that the two available state con gurations are identical and, nally, using temporal operators to propagate that condition in time. Thus, the next undecidability results apply also to that DDS-CTL, but also the decidability proofs can be extended to address the full language. 3
While input, state, and transport schemas are xed in advance, the extension
of such relations is not, and could grow unboundedly over time. If no bound
is imposed on the data manipulated by the dds, veri cation of convergence
properties is undecidable even for a single-node dds [
Taking inspiration from the well-studied notion of state-boundedness [
A bound independent from the network size ts with the idea that the declarative programs run by nodes are not tailored to a speci c network. Moreover, state-boundedness does not interfere with the information each node has about its neighbors, since our networks are xed, thus the relation neighbor is trivially bounded by the xed number of nodes (cf. Section 2.2). In fact, the next decidability results deal with network complexity, i.e., the complexity of the global initial con guration projected over the neighbor and my name predicates. Thus, the data complexity of the initial con guration incorporates the network complexity.
Additionally, a channel b-bounded (channel-bounded ) condition imposes a similar constraint on the channel multiset cardinality, i.e., in any reachable conguration there are at most b facts laying on any channel. In channel-bounded ddss the instantiations of the multiset channel data-structure can be represented by means of a nite DB, since the multiplicity of messages is bounded and a nite domain of constants su ce to represent them. Thus, in this case, the respective DRTSs can be translated into standard RTSs. We also say that the whole dds is bounded if all of its information sources are so, i.e., it is input-, state-, transport-, and channel-bounded at the same time.
In our analysis, a key observation is that the notion of uniformity [
and 0 ) D. Making use of uniformity, we get: Theorem 1. Veri cation of convergence properties over bounded i ddss is decidable in pspace in the network size.
Proof (sketch). First of all, it can be easily proven that ddss are uniform.
Uniformity comes from the fact that: (i) d2c is based on Datalog, which, as virtually
all DB query languages, enjoys genericity; (ii) external inputs are provided
\uniformly", i.e., whenever an input DB is delivered to a node, there is an alternative
execution in which the node receives an isomorphic variant of the same input.
For a uniform dynamic system, with a pre-de ned bound on the size of its DBs,
thus including the RTS version of the DRTSs of bounded-dds, veri cation of
FO-CTL properties, including translations of convergence, is decidable with a
pspace (tight) bound in data complexity [
We next show that bounding the channel multisets is necessary towards decidability of veri cation over iddss.
Theorem 2. Veri cation of convergence over input- and state-bounded i ddss is undecidable, even when the dds employs: (i) a single-node network; (ii) a single unary, 1-bounded input relation; (iii) 0-ary state relations; (iv) two 2-ary transport relations.
Proof (sketch). The proof is via a reduction from the undecidable halting
problem of deterministic 2-counter machines [
At the rst computation step the node, say named me, sends to himself two messages: counter1(me; me) and counter2(me; me). Then, the program triggers the increment and conditional decrement instructions according to the current dds state- ag.
To increment a counter, say counter 1, the node extracts a single constant from the current input DB, checks whether it is fresh with respect to the counter, and, in that case, puts the fresh constant in the appropriate cyclic graph. To perform the freshness test, the node expects to receive (and then sends it back on the channel), fact by fact, the counter1 cyclic-graph in the right order, i.e. from counter1(me; ) to counter1( ; me). In case the input constant was not available in the input DB, i.e., the DB was empty, or it already appeared in the incoming message, i.e., it was not fresh, or the incoming message is a counter2 or an unexpected counter1 message, the node enters in an error state.
To perform a conditional decrement on a counter, say again counter 1, the node expects to receive a counter1 message. If it is a counter1(me; me) message, then the node sends back the message and detects that the counter is zero. If the message is a di erent counter1 message, the node stores it in the state, expects to receive the next counter1 edge and then sends back only one edge where the middle constant has been dropped. Again, in case unexpected messages are delivered, the node enters in an error state.
After performing each increment or decrement instruction, the node transitions to the next state. In case the error state is reached, the node starts sending at each step a foo=0 message, whose reception triggers a random state transition (excluding the nal state of M and the dds current state- ag). Thus, M terminates if and only if the dds sometimes (always) converges with total (partial) correctness.
Considering Theorem 2, in the following we assume that ddss are channelbounded. Notice that channel-boundedness implies transport-boundedness, since at each computation step the whole transport DB is always sent in the reliable channel. This means that it makes no sense to study the veri cation of ddss that are channel-bounded but not transport-bounded. Thus, we consider now what happens when the dds is input- and transport-bounded, but have unconstrained state.
Theorem 3. Veri cation of convergence over input- and channel-bounded i ddss is undecidable, even when i dds employs: (i) a single computation node; (ii) a single unary, 1-bounded input relation; (iii) Two 1-ary state relations. (iv) a single 0-ary transport relation; Proof (sketch). The proof is similar to that of theorem 2, but the counters of M are now encoded in the state of M by means of two state predicates counter1=1 and counter2=1. The node continually sends to himself a wakeup=0 message, unless it reaches the nal state of M. Thus, the channel gets empty, causing the dds to trivially converge, if and only if the dds reaches the nal state of M.
In this case, freshness and zero checks can be done in one step by means of a couple of D2C rules that inspect the state DB. In case the freshness check fails or the input constant is not available, the node enters in a state error as above, triggering a run that never converges. Hence, M terminates if and only if the dds sometimes (always) converges with total (partial) correctness.
We investigate now what happens when the state DBs and channels are bounded, but the input is not. This is a subtle case: unboundedly many input data can be delivered to a node, but not inserted into its state/transport. In fact: Theorem 4. Veri cation of convergence properties over state- and channelbounded i ddss is in pspace in the network size.
Proof. Let M be an iddss, with DRTS hN ; ; S; T ; ; 0; ndb; nch; )i, whose state and transport are, together, bounded by b, and let ' be a convergence property. While could be in nite, we can specify and build a nite DRTS 0 equivalent to under '. Thus, to check whether j= ' amounts to check whether 0 j= '.
Let vars(') and Con be the set of variables and constants, respectively,
occurring in '. Fix a set C of 2b + jvars(')j + jConj constants, disjoint
from those in 0M, i.e., the initial state active domain, and call 0 = 0;M [ C.
Then, 0 is the DRTS hN ; ; S; T ; 0; 0; ndb0; nch0; )0i such that: (i) 0 is the
set of all the states 2 such that the set of constants mentioned in are at
most b and all contained in 0; (ii) ndb0, nch0, and )0 are the restrictions to
0 of ndb, nch, and ) respectively. Since both the schemas S and T , and the
domain 0 are nite, also 0 is nite, so that 0 is a nite DRTS. Moreover, since
j 0j 2b + jConj + jvars(')j and is uniform (see proof 1), it is possible to
adapt Th. 3.18 in [
Given a rule in a D2C program, we call input-, state-, and transport-variables those variables occurring only in input-, state-, and transport-predicates respectively. While the semantics of D2C requires a preliminary full grounding of the rules under the state and the input DBs active domains, we start by grounding only the state- and transport-variables with constants in the ( nite) full domain 0 of 0, resulting in the program P 0. Its variables occur only in input-predicates, hence the heads are fully grounded in 0. Nevertheless, the resulting program is suitable to compute all the transitions in 0 since, by construction, each state in 0 has a DB over 0.
While we should ground also the input-variables with constants in the unconstrained input domains, we can avoid it by noting that they act like variables in a Boolean query over the input, independent of the grounded part of the program. Speci cally, for each semi-grounded rule in P 0 with at least one variable, consider the existential closure q of the conjunction of all input literals in . These qi form a nite family of existential sentences. The e ect of an input DB I is just to discard those rules such that I 6j= q . To capture the e ects of all possible input DBs at once, we initialize a table with one column for each q and populate it with all the distinct rows rj containing a possible sequence of the symbols > and ?. For each row rj in the table, consider the conjunction of: (i) all q such that the corresponding cell contains >, and (ii) of all :q such that the corresponding cell contains ?. Written in negative normal form, this is a conjunction of existential and universal sentences, which can be transformed into a prenex formula j of the form 9 n8 n, where n is the number of all variables in P 0 times the number of queries q . This is a fragment of the Bernays-Schon nkel class enjoying a nite satis ablity problem in nexptime in the length of the sentence, but in O(1) in the network size. If the sentence j is not satis able in the nite, then there is no input DB enabling the e ects described by rj , thus it has to be deleted from the table. Otherwise, it has to be retained. After this pruning, for each row rj consider the fully-grounded subprogram Pj0 of P 0 containing only those rules , pruned of the input predicates, such that the corresponding cell in rj contains >. If the body results empty, ll it with true.
Given a state 0 in 0, the nite family of programs Pj0, with no input predicates, capture the e ects of all the in nitely many input DBs. Thus, to compute a successor of a given con guration in 0, it is su cient to compute a stable model of a program Pj0 over the con guration DB, which can be done in ptime in data complexity. Thus, by applying on-the- y veri cation techniques for nite RTSs against FO-CTL, which can express convergence translated over those RTSs, we can check 0 in npspace in data complexity, which amounts to pspace. 4
In the wide spectrum of declarative distributed computing, we have formalized and studied veri cation of convergence in the important case of reliable communication with unordered asynchronous communication and interactive input policy. While the problem is undecidable in general, decidability can be regained by imposing boundedness conditions on the channels and state DBs.
We foresee two main lines of future research. First, we plan to build on our
foundational results to implement veri cation techniques for dds cases with
decidable veri cation . To this aim, we will rely on existing ASP techniques for d2c,
and in particular on the implementation in [
We are also interested in exploring the complexity of weaker properties besides convergence, like safety and liveness.