José Ribeiro

Nuno Machado

Francisco Maia

Miguel Matos

Totally Ordered Replication for Massive Scale Key-Value Stores

Scalability is one of the most relevant features of today's data management systems. In order to achieve high scalability and availability, recent distributed key-value stores refrain from costly replica coordination when processing requests. However, these systems typically do not perform well under churn. In this paper, we propose DataFlagons, a large-scale key-value store that integrates epidemic dissemination with a probabilistic total order broadcast algorithm. By ensuring that all replicas process requests in the same order, DataFlagons provides probabilistic strong data consistency while achieving high scalability and robustness under churn.

Introduction

Distributed key-value stores are widely used nowadays for managing the evergrowing volume of data produced and accessed by Internet services. Key-value stores are appealing due to their high performance, scalability, and availability, but typically do not cope well with churn (i.e., the arrival and departure of system nodes of the system) [START_REF] Cooper | Pnuts: Yahoo!'s hosted data serving platform[END_REF][START_REF] Decandia | Dynamo: amazon's highly available key-value store[END_REF][START_REF] Lakshman | Cassandra: a decentralized structured storage system[END_REF]. The ability to handle dynamism is extremely relevant though, as has been already observed in large Internet of Things and blockchain deployments [START_REF] Eyal | Bitcoin-ng: A scalable blockchain protocol[END_REF].

The main reason why churn heavily affects current key-value systems is that they are built upon structured distributed hash tables (DHTs). DHTs organize nodes in the system into a structured logical overlay that, despite allowing for fast data access on deployments with hundreds of nodes, is not resilient enough to handle high dynamism on larger scale scenarios [START_REF] Rhea | Handling churn in a DHT[END_REF]. In contrast, unstructured peer-to-peer (P2P) approaches do not impose any pre-defined structure and hence are robust in the presence of heavy churn.

In the past, we proposed DataFlasks [START_REF] Maia | Dataflasks: epidemic store for massive scale systems[END_REF], an unstructured key-value store based on epidemic protocols. In DataFlasks, nodes self-divide into groups without the need of global synchronization. Each group is then responsible for handling a certain range of keys, ensuring that queries to those keys are served quickly and that the corresponding data is replicated among the nodes of the group. When peers join or leave the system, the other nodes adjust the group membership in an autonomous fashion and based on local information alone.

Unfortunately, similarly to other NoSQL key-value stores, the performance and scalability benefits of DataFlasks come at the cost of weaker data consistency guarantees. In detail, it assumes that conflicts are handled at the client side and does not offer any concurrency control mechanism. This means that two concurrent writes to the same key, with the same version but different values, may lead to an inconsistent scenario in which nodes of the same group store distinct data. In other words, the replicas diverge.

This paper aims at moving a step forward towards the design of a large-scale data management system capable of simultaneously offering high scalability, robustness, and data consistency. We propose a novel key-value store architecture that stems from the integration of DataFlasks with EpTO [START_REF] Matos | EpTO: An epidemic total order algorithm for large-scale distributed systems[END_REF], a probabilistic total order broadcast algorithm. We first show that this architecture can be easily adapted to provide different trade-offs between performance and consistency. Then, we explore a particular instance of such trade-offs with DataFlagons 3 . In a nutshell, DataFlagons is a key-value store that shares the scalability, availability, and flexibility benefits of DataFlasks while providing strongly consistent and reliable replication (as given by EpTO). Furthermore, DataFlagons obviates the need to control versioning at the application level, thus offering a simpler programming interface. Our experimental evaluation based on simulations shows that DataFlagons is scalable, resilient to churn, and that it successfully overcomes the data replication issues observed in DataFlasks, albeit at the cost of 4.7x throughput slowdown.

The rest of the paper is organized as follows. Section 2 overviews the related work, as well as DataFlasks and EpTO. Section 3 describes the design and implementation of DataFlagons. Section 4 presents the experimental evaluation. Finally, Section 5 concludes the paper by summarizing its main findings.

Background and Motivation

State of the Art

Over the last decade, there has been a large body of work on distributed data stores with a key-value model. The most relevant examples are arguably Cassandra [START_REF] Lakshman | Cassandra: a decentralized structured storage system[END_REF], Dynamo [START_REF] Decandia | Dynamo: amazon's highly available key-value store[END_REF]/Voldemort4 , Bigtable [START_REF] Chang | Bigtable: A distributed storage system for structured data[END_REF]/HBase5 , and PNUTS [START_REF] Cooper | Pnuts: Yahoo!'s hosted data serving platform[END_REF]. These systems explore different compromises between consistency, partition tolerance, and availability, although most choose to sacrifice strong consistency guarantees for higher performance and scalability. For data storage and retrieval, Cassandra and Dynamo rely on a DHT-based structure, PNUTS uses a modified B+ tree, and Bigtable relies on a set of master nodes. This structured design, despite scaling up to hundreds of nodes, typically lacks robustness in environments subject to churn. Here, robustness can be defined as the ability to cope with errors during execution and becomes particularly relevant as the system scales and the workload increases [START_REF] Lourenço | Choosing the right nosql database for the job: a quality attribute evaluation[END_REF]. On the other hand, DataFlasks [START_REF] Maia | Dataflasks: epidemic store for massive scale systems[END_REF] performs well even in highly dynamic massive scale settings by operating on top of unstructured peer-to-peer protocols. However, it has the drawback of lacking consistency guarantees during data replication, which can cause replicas to diverge.

Figure 1 sketches a generic comparison of the aforementioned systems in terms of their robustness and consistency properties [START_REF] Lourenço | Choosing the right nosql database for the job: a quality attribute evaluation[END_REF]. 6 From the figure, one can observe that there is currently a gap in terms of robust solutions, capable of coping with high churn, that ensure data consistency. In this paper, we aim at filling this gap with a novel key-value store architecture that combines DataFlasks with EpTO [START_REF] Matos | EpTO: An epidemic total order algorithm for large-scale distributed systems[END_REF], a probabilistic total order broadcast algorithm. The resulting architecture benefits from the resilience and scalability of DataFlasks, but is able to provide stronger consistency guarantees via total order delivery. Next, we describe DataFlasks and EpTO.

DataFlasks

DataFlasks [START_REF] Maia | Dataflasks: epidemic store for massive scale systems[END_REF] is a distributed key-value store that stores tuples of format (K, V, v), where K indicates the key mapping to a value V with version v. Clients can add tuples to the system by invoking put(K, V, v), and retrieve the value V using the operation get(K, v). Any DataFlasks node can handle client requests. Figure 2 depicts the architecture of DataFlasks in more detail. As shown in the figure, DataFlasks is composed of four main modules: request handler, group construction, anti-entropy, and dissemination, which we describe next.

Request handler. Upon receiving a put request, a node first checks whether the group it belongs to is responsible for handling that particular key. This is done by checking whether the tuple's key falls within the group's key range. If so, the node stores the tuple and forwards the request to its group peers for replication purposes. Otherwise, the request is forwarded to other groups via epidemic dissemination.

In turn, when a node receives a get of a given key and version, it immediately replies to the client in case it holds the corresponding value. Otherwise, the node simply disseminates the request to the other groups, which will be in turn responsible for sending the value to the client.

Group construction. This component is responsible for organizing nodes into groups of at least the size of the desired replication factor. This is done by continuously estimating the number of groups that should be maintained according to the system size. Under churn, nodes leaving the system cause the estimated number of groups to decrease, while nodes joining the system lead to the creation of new groups, thus maintaining the stability of the group size. Anti-entropy. This module periodically contacts another node from the same group to ensure that all tuples were correctly replicated and no data is missing.

Dissemination. This component is responsible for node communication via gossip. Concretely, each node in DataFlasks resorts to a peer sampling service (PSS), namely Cyclon [START_REF] Voulgaris | Cyclon: Inexpensive membership management for unstructured p2p overlays[END_REF], to obtain a partial view of the system, containing references to neighbor peers. The union of all nodes' partial views creates a logical overlay that evolves with time and allows efficient dissemination of information.

Local storage is a component that abstracts how data is persisted. A more thorough description of DataFlasks' design can be found in [START_REF] Maia | Dataflasks: epidemic store for massive scale systems[END_REF].

The case for total order. DataFlasks has no concurrency control mechanism. Instead, it relies on the clients to version each operation. If clients do not take this into account, scenarios of replica state divergence may occur. To provide further evidence of this issue, we implemented the algorithm of DataFlasks in Python to perform an experiment. 7 It consisted in having two clients concurrently inserting different values -v1 and v2 -by executing put(k, v1, 0) and put(k, v2, 0), respectively. Note that both tuples have the same key and version, thus represent the same data item in DataFlasks. For this experiment, DataFlasks ran on 300 nodes distributed among 32 groups.

Figure 3a reports the number of replicas of each value existing in the group responsible for handling key k, for a simulation of 8000 time units and observations performed every 125 time units. The figure shows that the two values were successfully replicated by disjoint subsets of nodes of the group and that no state changes were observed until the end of the simulation. Since DataFlasks employs a first-write-wins approach, once a node receives the first put operation referring to key k and version 0, it will ignore subsequent writes for the same key and version. Figure 3b confirms this fact by plotting the instant in which each put arrived to two distinct nodes, denoted X and Y . Given that replicas X and Y 0 1,000 2,000 3,000 4,000 5,000 6,000 7,000 0 receive v1 and v2 in opposite orders, both values end up being persisted and the state of the system diverges. This inconsistency is problematic because it allows read operations for the same tuple to return different values. On the other hand, ensuring that the requests have different versions (i.e. the requests are ordered) would suffice to overcome this drawback of DataFlasks. DataFlagons follows this approach with the help of a total order broadcast algorithm, which we describe next.

EpTO

EpTO [START_REF] Matos | EpTO: An epidemic total order algorithm for large-scale distributed systems[END_REF] is a scalable total order algorithm with probabilistic guarantees of total order delivery. This is done by having nodes agree (with high probability) on the order in which events received should be delivered to the application. Unlike traditional agreement algorithms though, EpTO achieves this property in a fully decentralized and coordination-free manner, which allows for high scalability and robustness in face of churn and message losses. EpTO relies on an adaption of the balls-and-bins [START_REF] Koldehofe | Simple gossiping with balls and bins[END_REF] algorithm to epidemic dissemination, where nodes are represented as bins and messages as balls. The main goal of the algorithm is then to send as many balls as necessary such that all bins receive at least a ball with high probability. To meet this goal, a node wishing to disseminate a ball (i.e. a set of messages) across the system simply sends it to K of its neighbors, chosen uniformly at random. In the following rounds, the nodes that received the ball will propagate it to another K random neighbors, thus quickly spreading the rumor through the network. After a few rounds, all the nodes have received all the messages with high probability. Hence, EpTO can then ensure total order delivery by simply sorting the messages according to a deterministic factor.

EpTO exposes two primitives to the application, namely EpTO-broadcast and EpTO-deliver, which are guaranteed to satisfy the following properties:

-Integrity. For every message m, each node EpTO-delivers m at most once, iff m was previously EpTO-broadcasted. -Validity. If a correct node (i.e. a node that did not fail or left the system) EpTO-broadcasts a message m, then it will eventually EpTO-deliver m.

EpTO Node

-Total Order. If two nodes P and Q EpTO-deliver two messages m and m , then P EpTO-delivers m before m iff Q EpTO-delivers m before m .

-Probabilistic Agreement. If a node EpTO-delivers a message m, then, with high probability, all correct nodes will eventually EpTO-deliver m.

Figure 4 illustrates EpTO's architecture. It comprises a dissemination component, which disseminates events, and an ordering component, which delivers events to the application in total order. The total order is satisfied by the ordering component, whereas the probabilistic agreement is satisfied by the dissemination component. The properties of integrity and validity are guaranteed by the two components in conjunction.

The dissemination component operates in periodic rounds of δ time units and aims at broadcasting balls across the network in a gossip-based fashion. As in DataFlasks, nodes in EpTO rely on an underlying PSS to obtain a fresh and randomized partial view of the system. This view is assumed to contain at least K correct nodes to which balls are propagated, denoted as the node's fanout. Each ball should be relayed for a time to live (TTL) number of rounds, with TTL being a configurable parameter of the system. Each node stores the messages to be forwarded in a set denoted nextBall.

When a node wishes to broadcast a message, it sets the message's ttl to 0, assigns a timestamp (i.e. a logical clock), and appends it to the nextBall set. In turn, when a node receives a ball of messages, it compares each message's ttl against the T T L configuration parameter. If ttl < T T L, the incoming message is added to nextBall for future retransmission. In case there is a message with the same id already present, EpTO keeps the one with higher ttl.

Independently to message broadcast and reception, at the beginning of every round (i.e. every δ time units), each node i) increments by one the ttl of the messages stored in nextBall, and ii) sends this set as a ball to K random neighbors. EpTO's execution flow then proceeds to the ordering component.

To ensure total order delivery, the ordering component maintains two lists of messages: one storing the messages received up to that moment and the other containing the messages that were already delivered to the application. The ordering component's main task is then to move messages from the list of received to the list of delivered, while avoiding duplicates and preserving the global order. Messages in the received list that were already delivered to the application are discarded, whereas the remaining ones are only added to the delivered list if their ttl already reached the T T L value. For this latter case, EpTO then delivers the messages to the application in ascending order of their timestamp, using the id of the broadcaster node to break ties.

Note that the probabilistic nature of the agreement property offered by EpTO permits a rare yet possible scenario in which some messages are not delivered at all by a subset of nodes. These nodes, despite exhibiting holes in the sequence of messages agreed upon, are guaranteed to preserve the total order property of message delivery to the application. In other words, the total order is ensured but the agreement might be violated (with arbitrarily low probability).

A keen observation of the architectures of DataFlasks and EpTO (in Figure 2 and Figure 4, respectively) is that the two systems exhibit interesting similarities: both organize nodes into an unstructured, logical overlay and both resort to a PSS to disseminate events. These similarities paved the way for the design of DataFlagons and its properties, as described in the next section.

DataFlagons

DataFlagons is designed as an integration of DataFlasks with EpTO, inheriting components and properties from both systems. This integration yields a synergy that allows not only obtaining the best of both solutions but also addressing the shortcomings of each one of them individually. On the one hand, the total order property of EpTO solves the replica divergence issue of DataFlasks discussed before. As all nodes in DataFlagons execute all operations by the same order, it becomes impossible for different replicas to store distinct values for the same key and version. Alongside, using total order in DataFlagons strengthens the poor consistency guarantees originally offered by DataFlasks.

On the other hand, DataFlasks' anti-entropy mechanism helps to cope with the potential holes existing in EpTO's delivery. In the following, we describe the architecture of DataFlagons and discuss the trade-offs between consistency and throughput that it enables.

System Overview

DataFlagons is a scalable, robust, and strongly consistent key-value store. To the best of our knowledge, DataFlagons is the first system capable of providing these properties in massive scale environments subject to churn. Figure 5 illustrates DataFlagons' architecture, indicating the components from both DataFlasks and EpTO. As shown, DataFlagons inherits the request handler, the group construction, the anti-entropy, and the storage components from the former. From the latter, DataFlagons inherits the ordering and dissemination modules. As a concrete example of the modus operandi of DataFlagons, consider the diagram of Figure 6, representing the execution flow of a put in our system. 81 A client wishing to add new tuples to the data store has simply to issue a put request indicating the key and the value. This contrasts to DataFlasks, where it was necessary to specify the version in addition to the key and the value. Likewise, for performing a get operation in DataFlagons, the client needs only to indicate the key it wants to read, instead of the key and the version. 2 The incoming request is processed by DataFlagons' request handler. In practice, this corresponds to encapsulating the operation in a ball to be disseminated through the network via EpTO-broadcast.

DataFlagons Node

3 Following EpTO's algorithm, the invocation of the EpTO-broadcast primitive results in the propagation of the ball to K random neighbors of the broadcasting node. The neighbors receiving the ball will later forward it to another K peers of their own view. [START_REF] Decandia | Dynamo: amazon's highly available key-value store[END_REF] In the following rounds, the node will receive, with high probability, balls containing the message previously broadcast. All messages received are forwarded to the ordering component.

5 Upon the reception of a message with a ttl equal to the T T L of the system, the ordering component checks if the event's timestamp is the next in line to be delivered according to the total order. 9 If it is, the ordering component executes the EpTO-deliver primitive. However, unlike in EpTO, the component notified by this primitive in DataFlagons is the request handler and not the client, as the processing of the actual operation (i.e. the put) has not been concluded yet.

6 When the request handler receives an operation via total order delivery, it proceeds as in DataFlasks and queries the group construction component in order to assess whether the tuple should be stored locally. [START_REF] Lakshman | Cassandra: a decentralized structured storage system[END_REF] The group construction, in turn, checks whether the key belongs to the group's key range. If so, the tuple is persisted in the storage; otherwise, it is dis- carded. Since the dissemination mechanism of EpTO already delivers the events to all nodes with high probability, DataFlagons no longer needs to propagate the operation to other groups in case the key is not the responsibility of the node's group, as opposed to DataFlasks.

8 Finally, DataFlagons acknowledges the client that the operation was successfully performed.

Note that the anti-entropy module is not included in the diagram of Figure 6. The reason is that anti-entropy operates orthogonally to the common request handling procedure, ensuring that all nodes of the group have a replica of the data despite potential holes in EpTO's dissemination. Anti-entropy is also particularly helpful to secure data durability and improve system availability during periods of churn. When a new node joins DataFlagons, this module ensures that it quickly receives, from another peer of the same group, a replica of the data for which it is responsible.

Design Trade-offs

Recall the robustness-consistency spectrum depicted in Figure 1. The design of DataFlagons has the advantage of allowing us to easily support various consistency levels while maintaining the system highly robust. This is achieved by tuning the module through which get and put requests are propagated across the network. Table 1 summarizes these design choices and their respective consistency level.

As shown in the table, using DataFlasks alone to disseminate the two types of requests naturally causes DataFlagons to lose any data consistency guarantees. In contrast, if we rely on DataFlasks to only propagate gets and use EpTO to broadcast puts, the consistency level increases to eventual. A system is eventually consistent if it guarantees that all replicas will eventually return the last updated value of an item, assuming that no new updates are made to that item [START_REF] Vogels | Eventually consistent[END_REF]. In other words, there is a time window in which two concurrent reads on different replicas can return different values for the same key, although both replicas will eventually converge to the same value.

Finally, delivering both get and put requests in total order via EpTO renders DataFlagons probabilistically strongly consistent. Strong consistency states that any read happening after an update will return the updated value [START_REF] Vogels | Eventually consistent[END_REF]. Although we believe that DataFlagons is strongly consistent for the vast majority of scenarios, we claim that it offers probabilistic strong consistency due to the very unlikely yet possible event where a subset of nodes does not deliver all the messages. This event is equivalent to not applying all operations, which would allow a small window of inconsistency. This issue can be solved but it is out of the scope of the present paper.

For space constraints, in this paper, we only explore propagation done exclusively through EpTO which provides stronger consistency. As advocated by prior work [START_REF] Corbett | Spanner: Google's globally distributed database[END_REF], strong consistency has the advantage of significantly simplifying the implementation of applications and reduce the occurrence of bugs. Nevertheless, preliminary results about the other design choices show that customizable consistency is a promising approach.

The next section presents an experimental evaluation of DataFlagons with probabilistic strong consistency by comparing it against a vanilla version of DataFlasks.

Evaluation

We implemented a prototype of DataFlagons in Python and evaluated it by means of simulations. In particular, our experimental analysis focuses on answering the following questions:

-Consistency: Does DataFlagons ensure consistent replication? ( §4.1) -Performance: How does DataFlagons compare to DataFlasks in terms of response latency, throughput, and the number of messages when subject to different churn levels? ( §4.2)

The experiments were conducted on a machine with a 3.4 GHz Dual-Core Intel i3 Processor, 8 GB of RAM and a 7200 RPMs SATA disk. The discrete simulator used was that of EpTO's evaluation [START_REF] Matos | EpTO: An epidemic total order algorithm for large-scale distributed systems[END_REF], which allows easily tuning the number of nodes in the system, network latency, churn, message loss, and the duration of the simulation. 10 The passage of time is modeled as simulation ticks by means of a priority queue and a monotonically increasing integer.

Regarding the experimental setting, the network size was varied between 300 and 1000 nodes, and the end-to-end latency was parameterized according to real-world PlanetLab traces as in Matos et al. [START_REF] Matos | EpTO: An epidemic total order algorithm for large-scale distributed systems[END_REF].

For the configuration of DataFlagons, we used the values suggested in Matos et al. [START_REF] Matos | EpTO: An epidemic total order algorithm for large-scale distributed systems[END_REF] and Maia et al. [START_REF] Maia | Dataflasks: epidemic store for massive scale systems[END_REF]: EpTO's fanout and T T L were set as 18 and 25 respectively, while DataFlasks' group construction boundaries were set as [START_REF] Koldehofe | Simple gossiping with balls and bins[END_REF][START_REF] Vogels | Eventually consistent[END_REF] (i.e., each group much have at least 6 replicas and at most 12 replicas). In all experiments, we initialize the PSS of each node with a view containing 20 references to other nodes, chosen uniformly at random, and with a shuffling period of 125 simulation ticks. This allows DataFlagons to be bootstrapped on top of a random overlay. The anti-entropy mechanism of DataFlasks and the relaying procedure of EpTO also execute every 125 ticks.

Finally, at the beginning of each simulation, we first let DataFlagons run until convergence, meaning that all nodes have the correct estimation of both the total number of groups in the system and to which group they belong. After that, the experiment's respective workload is executed.

Consistency Results

To thoroughly assess DataFlagons' data consistency guarantees, we conducted an experiment similar to that of Section 2.2, in which two clients concurrently write different values for the same key. However, in this case, the experience is more elaborate in that the two clients perform 20 requests each (80% gets and 20% puts) for churn levels of 10%, 20%, and 30%. The goal of the experiment is twofold: i) check whether DataFlagons delivers the requests in order, and ii) confirm that strong consistency is guaranteed under churn.

To ensure that the operations are handled by the system despite node failures, each request is sent to three nodes randomly chosen. For get requests, the value read is considered to be that of the first response.

Churn is implemented by replacing an existing node with a fresh one, ensuring that the distribution of nodes leaving/entering the system is balanced across all groups. As such, a test case with a churn level of 10% indicates that each group is subject to 10% of churn. The churn period was configured in order to allow the convergence of the system in-between membership variations. When a new node joins the system, it also receives a random partial view with references to the nodes alive at that moment.

Figure 7 reports the results of the experiments for a system with 300 nodes and simulations of 32K ticks. In particular, the figure depicts the total order of requests performed by DataFlagons (note that each put has its own unique version) and the version of the key returned by each get, which corresponds to a unique value. In this context, a violation of the strong consistency guarantees would be observed as a mismatch between the version read by a get request and the version of the latest put delivered by DataFlagons. The results in the figure show that, for all churn levels, no consistency violations occurred and that DataFlagons successfully replicated the values across all nodes of the group Simulation Ticks (K)

Key Version

Client 0 PUT Client 0 GET Client 1 PUT Client 1 GET Fig. 7. Request delivery order and consequent versioning given by DataFlagons for churn of 10%, 20%, and 30% (top to bottom) applied to a system of 300 nodes. No violations of strong consistency were observed, as get operations always returned the value written by the latest put.

without state divergence. This outcome corroborates the effectiveness of EpTO in delivering events in total order.

Performance Results

We now assess the performance of DataFlagons, by comparing its latency, throughput, and number of messages exchanged against those of DataFlasks. The workload used for these experiments consists of a number of clients (namely, 30 and 100 clients respectively for the scenarios with 300 and 1000 nodes) systematically performing requests until the end of the simulation, with a get/put ratio of 80%/20%. Naturally, the operations on DataFlasks are associated with a version, as opposed to the operations on DataFlagons. Clients issue each request to a randomly chosen node and receive acknowledgments from all the nodes belonging to the group responsible for handling that key. Since the way one treats the acknowledgments poses a trade-off between throughput and data durability, we opted for conducting tests with two different versions of DataFlasks exploring different compromises:

-DFlasks-1/3. A request is only considered complete after the client receives acknowledgments from one-third of the average number of replicas of the group. For the group construction interval [START_REF] Koldehofe | Simple gossiping with balls and bins[END_REF][START_REF] Vogels | Eventually consistent[END_REF] considered in the experiments, it means that the client has to receive 3 acknowledgments before being able to issue another request ((6 + 12)/2/3 = 3).

-DFlasks-All. A request is only considered complete after the client receives acknowledgments from a number of replicas equal to the lower bound of the group construction interval. For the interval [START_REF] Koldehofe | Simple gossiping with balls and bins[END_REF][START_REF] Vogels | Eventually consistent[END_REF], it means that the client has to receive 6 acknowledgments before being able to issue another request.

Note that, for DataFlagons, the client only needs to receive a single acknowledgment to achieve durability, because the request is guaranteed to have been previously broadcast through the network via EpTO-broadcast.

Our experiments also considered different churn levels, namely 0%, 10%, 20%, and 30%. As in the previous section, churn is implemented by replacing an existing node by a fresh one, ensuring that the distribution of nodes leaving/entering the system is balanced across all groups. The results of the experiments, averaged from five runs, are as follows.

Response Latency. Figure 8 depicts the cumulative distribution function (CDF) of the response latencies of DataFlagons, DFlasks-1/3, and DFlasks-All, measured for the different testing scenarios. The main observation stemming from the results is that DataFlagons incurs significantly higher latency than both DataFlasks configurations, being one order of magnitude slower. The slowdown in DataFlagons is due to the time required by EpTO to achieve probabilistic agreement and, consequently, enable total order delivery. In contrast, replicas in DataFlasks can reply to the client immediately after receiving the request.

Regarding the robustness and scalability of DataFlagons, the results indicate that there is a slight increase in the response latency as the churn and system size grow (of up to 10% and 27%, respectively). Since the slowdown is not substantial, we argue that the experiments reveal that DataFlagons is robust and scalable. Throughput. Figure 9 shows the throughput of DataFlagons, DFlasks-1/3, and DFlasks-All observed in the experiments. These results follow the same trend as those for response latency: DFlasks-1/3 exhibits the highest throughput, closely followed by DFlasks-All, while DataFlagons processes an order of magnitude fewer requests. Figure 9 also demonstrates that increasing churn barely affects DataFlagons, whereas DFlasks-All is heavily hampered. Since this configuration of DataFlasks requires all replicas to answer to the client, having more nodes leaving the network will necessarily reduce the system's ability to process requests, potentially leading to clients becoming blocked when the system fails to maintain the minimum number of replicas (for 1000 nodes with 30% churn).

Message Cost. We measured the average message cost per request for the different systems, which includes the messages exchanged by the anti-entropy mechanism, the peer sampling service, and EpTO's relaying (for DataFlagons). For the scenario with 300 nodes, DFlasks-1/3, DFlasks-All, and DataFlagons required, on average, 200K, 200K, and 4M messages to process a request, respectively. In turn, for 1000 nodes, the message overhead increased respectively to 2M, 2M, and 60M. Finally, we did not observe significant variations of the number of messages exchanged when increasing churn.

Improving DataFlagons' performance by parameter tuning. The aforementioned experiments reveal that the stronger consistency guarantees offered by DataFlagons come at the cost of around an order of magnitude performance slowdown comparing to DataFlasks. However, since the performance of Data-Flagons is directly related to the conservativeness of EpTO's parameters, we performed a sensitivity analysis in order to assess the throughput improvements achievable by our system when varying the fanout and T T L. The analysis is omitted due to space constraints, but the outcomes indicate that, for the same message overhead as DataFlasks and without compromising strong consistency, DataFlagons can improve its performance using a fanout of 11 and a T T L of 8. Figure 10 plots the throughput of the new configuration for 300 nodes.

The results in the figure show that fine-tuning the parameters of EpTO leads to an average throughput increase of 58% with respect to the original configuration of DataFlagons. Again, it should be noted that this parameter tuning did not affect the consistency guarantees offered by DataFlagons.

We also note that the flexible design of DataFlagons permits further performance improvements through the exploration of weaker data consistency guarantees (see Section 3.2). For instance, by applying EpTO's dissemination solely to put requests within the nodes of the same group while propagating gets via DataFlasks' traditional gossip mechanism, one could achieve substantial throughput benefits at the cost of lowering the consistency level from strong to eventual. We believe that the exploration of these trade-offs on DataFlagons poses an interesting research direction, which we intend to pursue in the future.

Conclusions

This paper addresses the issue of efficiently storing and managing data in massivescale environments subject to churn. In particular, we propose DataFlagons, a scalable and probabilistically strongly-consistent distributed data store that stems from the combination of DataFlasks (a large-scale key-value store) with EpTO (a probabilistic total order broadcast algorithm). The experimental evaluation of DataFlagons based on simulations shows that it is able to address the consistency issues observed in DataFlasks with the same scalability and availability, albeit reducing the performance on 4.7x on average. This is, to the best of our knowledge, the first system ever to offer simultaneously strong consistency guarantees coupled with the ability to handle high levels of churn for systems with thousands of nodes.

Fig. 1 .

 1 Fig. 1. Qualitative comparison of prior key-value stores in terms of (singletuple) consistency and robustness.

Fig. 3 .

 3 Fig. 3. (a) Node states in DataFlasks when concurrently storing values v1 and v2 for the same K, v. (b) Detail with request arrival order at two different nodes that leads to state divergence.

Fig. 4 .

 4 Fig. 4. EpTO architecture. m represents a message being broadcast/delivered by EpTO.

Fig. 5 .

 5 Fig.5. DataFlagons architecture as integration of components from DataFlasks (solid line boxes) and EpTO (dashed line boxes).

Fig. 6 .

 6 Fig. 6. Execution flow of a put operation in DataFlagons. This operation is abstracted as a message m disseminated via EpTO.

Fig. 9 .

 9 Fig.9. Throughput of the three systems for scenarios with 300 nodes (top) and 1000 nodes (bottom), and churn levels of 0%, 10%, 20%, and 30%.

Fig. 10 .

 10 Fig. 10. Throughput of DataFlagons after fine-tuning the fanout and T T L versus that of a non-optimized version, for a scenario with 300 nodes.

Table 1 .

 1 Consistency levels offered depending on the dissemination module used by DataFlagons. The shaded row indicates the design choice explored in this paper.

	Consistency	Dissemination Module
	Level	get	put
	No guarantees	DataFlasks DataFlasks
	Eventual	DataFlasks	EpTO
	Strong (probabilistic)	EpTO	EpTO

  Response latency of the three systems for scenarios with 300 nodes (top row) and 1000 nodes (bottom row), and churn levels of 0%, 10%, 20%, and 30%. For readability purposes, the results for each simulation were divided into two different plots (depicted in the same row), corresponding to the relevant portions of the experiment.

		100										100
		80							DF-1/3 0% churn			80
	CDF (%)	40 60							DF-1/3 10% churn DF-1/3 20% churn DF-1/3 30% churn DF-All 0% churn DF-All 10% churn DF-All 20% churn	CDF (%)	40 60			DFlagons 0% churn DFlagons 10% churn DFlagons 20% churn DFlagons 30% churn
		20							DF-All 30% churn		20
		0	20	40	60	80 Latency (simulation ticks) 100 120	140	160	180	200		600	700	800	900	1,000 1,100 1,200 1,300 1,400 1,500 1,600 Latency (simulation ticks)
		100										100
		80							DF-1/3 0% churn			80
	CDF (%)	40 60							DF-1/3 10% churn DF-1/3 20% churn DF-1/3 30% churn DF-All 0% churn DF-All 10% churn DF-All 20% churn	CDF (%)	40 60			DFlagons 0% churn DFlagons 10% churn DFlagons 20% churn DFlagons 30% churn
		20							DF-All 30% churn		20
		0	20	40	60	80 Latency (simulation ticks) 100 120	140	160	180	200		1,000 1,100 1,200 1,300 1,400 1,500 1,600 1,700 1,800 1,900 2,000 Latency (simulation ticks)
	Fig. 8.

Our system was named DataFlagons to convey an improvement over DataFlasks, as flagons are arguably more robust and consistent containers than flasks.

http://www.project-voldemort.com/voldemort/

https://hbase.apache.org

Note that this is a high-level quantitive comparison based on prior studies[START_REF] Lourenço | Choosing the right nosql database for the job: a quality attribute evaluation[END_REF], as a thorough experimental analysis for robustness is currently lacking in the literature.

Our implementation of DataFlasks, as well as that of the simulator used in the experiments, are detailed in Section 4.

We omit a description of the flow for a get request, as it is similar to that of a put.

For messages with the same logical clock, DataFlagons orders them in ascending order of the ids of their broadcasting nodes.

Acknowledgments. The authors would like to thank the anonymous reviewers for their valuable feedback. This work was partially supported by Project "TEC4Growth -Pervasive Intelligence, Enhancers and Proofs of Concept with Industrial Impact" (NORTE-01-0145-FEDER-000020), financed by the North Portugal Regional Operational Programme (NORTE 2020), under the POR-TUGAL 2020 Partnership Agreement, by the ERDF -European Regional Development Fund through the Operational Programme for Competitiveness and Internationalisation -COMPETE 2020 Programme within project "POCI-01-0145-FEDER-006961", and by National Funds through the Portuguese funding agency, FCT -Fundação para a Ciência as part of project "UID/EEA/50014/2013".


Saverio Giallorenzo

Fabrizio Montesi

Maurizio Gabbrielli

Choreographies

de niveau recherche, publiés ou non, émanant des établissements d'enseignement et de recherche français ou étrangers, des laboratoires publics ou privés.

Introduction

Background. In Choreographic Programming, programs are choreographies of communications used to synthesise correct implementations through an End-Point Projection (EPP) procedure [START_REF] Montesi | Choreographic Programming[END_REF]. The generated code is guaranteed to follow the behaviour specified in the choreography and to be deadlock-free [START_REF] Carbone | Deadlock-freedom-by-design: multiparty asynchronous global programming[END_REF]. For these reasons, the communities of business processes and Service-Oriented Computing (SOC) widely adopted choreographies, using them in standards (e.g., WS-CDL [START_REF] W3c | WS-CDL version 1.0[END_REF] and BPMN [START_REF] Omg | Business Process Model and Notation[END_REF]), languages [START_REF] Lanese | Bridging the gap between interaction-and process-oriented choreographies[END_REF][START_REF] Basu | Deciding choreography realizability[END_REF][START_REF] Honda | Scribbling interactions with a formal foundation[END_REF][START_REF] Dalla Preda | Dynamic choreographies[END_REF][START_REF]Pi4soa[END_REF][START_REF]Savara[END_REF][START_REF]Chor Programming Language[END_REF][START_REF]AIOCJ framework[END_REF], and type systems and logics [START_REF] Honda | Multiparty asynchronous session types[END_REF][START_REF] Carbone | Structured communication-centered programming for web services[END_REF][START_REF] Carbone | Choreographies, logically[END_REF][START_REF] Carbone | Multiparty session types as coherence proofs[END_REF].

Example 1. Below, we give a representative example of a choreography. In the example, we implement a simple business protocol among a client process c, a seller service located at l S and a bank service located at l B (locations are abstractions of network addresses, or URIs). At line 1, the client c asks the seller and the bank services to create two new processes, respectively s and b. The three processes c, s, and b can now communicate over a multiparty session k, intended as in Multiparty Session Types [START_REF] Honda | Multiparty asynchronous session types[END_REF]: when a session is created, each process gets ownership of a statically-defined role, which identifies a message queue that the process uses to asynchronously receive messages from other processes. For simplicity, at line 1, we assign role C to process c, S to s, and B to b. At line 2, over session k, the client c invokes operation buy of the seller s with the name of a product it wishes to buy, which the seller stores in its local variable x. As usual, processes have local state and run concurrently. At line 3, the seller uses its internal function mkOrder to prepare an order (e.g., compute the price of the product) and sends it to the bank on operation openTx, for opening a payment transaction. At line 4, the client sends its credit card information cc to the bank on operation pay. Then, at line 5, the bank makes an internal choice on whether the payment can be performed (with internal function closeTx, which takes the local variables cc and order as parameters). The bank then notifies the client and the seller of the final outcome, by invoking them both either on operation ok or ko. Motivation. In previous definitions of EPP, both the choreography language and the target language abstract from how real-world frameworks support communications [START_REF] Qiu | Towards the theoretical foundation of choreography[END_REF][START_REF] Lanese | Bridging the gap between interaction-and process-oriented choreographies[END_REF][START_REF] Carbone | Structured communication-centered programming for web services[END_REF][START_REF] Carbone | Deadlock-freedom-by-design: multiparty asynchronous global programming[END_REF][START_REF] Carbone | Choreographies, logically[END_REF], by modelling communications as synchronisations on names (cf. [START_REF] Milner | A Calculus of Communicating Systems[END_REF][START_REF] Milner | A calculus of mobile processes, I and II[END_REF]). Thus, the implementations of choreographic programming [START_REF]Chor Programming Language[END_REF][START_REF]AIOCJ framework[END_REF] significantly depart from their respective formalisations [START_REF] Carbone | Deadlock-freedom-by-design: multiparty asynchronous global programming[END_REF][START_REF] Dalla Preda | Dynamic choreographies[END_REF]. In particular, the implemented EPPs realise channel creation and message routing with additional data structures and message exchanges [START_REF] Montesi | Choreographic Programming[END_REF][START_REF] Giallorenzo | Real-World Choreographies[END_REF]. The specific communication mechanism used in these implementations is message correlation. Correlation is the reference message routing technology in Service-Oriented Computing (SOC)-the major field of application of choreographies-and it is supported by mainstream technologies (e.g., WS-BPEL [START_REF]WS-BPEL[END_REF], Java/JMS, C#/.NET). The gap between formalisations and implementations of choreographic programming can compromise its correctness-by-construction guarantee.

Contributions. We reduce the gap between choreographies and their implementations by developing Applied Choreographies, a choreographic framework consisting of three calculi: the Frontend Calculus (FC), which offers the wellknown simplicity of abstract channel semantics to programmers; the Backend Calculus (BC), which formalises how abstract channels can be implemented on top of message correlation; and the Dynamic Correlation Calculus (DCC), an abstract model of Service-Oriented Computing where distributed services communicate through correlation. Differently from BC, DCC has no global view on the state of the system (which is instead distributed), and there are no multiparty synchronisation primitives.

Our main contribution is the definition of a behaviour-preserving compiler from choreographies in FC to distributed services in DCC, which uses BC as intermediate representation. This is the first correctness result of an end-to-end translation from choreographies to an abstract model based on a real-world communication mechanism. Our compiler proceeds in three steps:

it projects (EPP) a choreography, describing the behaviours of many participants, into a composition of modules called endpoint choreographies, each describing the behaviour of a single participant; -it generates the data structures needed by BC to support the execution of the obtained endpoint choreographies using message correlation; -it synthesises a correct distributed implementation in DCC, where the multiparty synchronisations in choreographies are translated to correct distributed handshakes (consisting of many communications) on correlation data.

Full definitions and proof sketches are in [START_REF] Gabbrielli | Applied choreographies[END_REF].

Frontend Calculus

We start by presenting the Frontend Calculus (FC), which follows the structure of Compositional Choreographies [START_REF] Montesi | Compositional choreographies[END_REF]. The main novelty is in the semantics, which is based on a notion of deployment and deployment effects that will ease the definition of our translation. Remarkably, it also yields a new concise formalisation of asynchrony in choreographies.

Syntax. In the syntax of FC (Fig. 1), C denotes an FC program. FC programs are choreographies, as in Example 1, and we often refer to them as Frontend choreographies. A choreography describes the behaviour of some processes. Processes, denoted p, q ∈ P, are intended as usual: they are independent execution units running concurrently and equipped with local variables, denoted x ∈ Var . Message exchanges happen through a session, denoted k ∈ K, which acts as a communication channel. Intuitively, a session is an instantiation of a protocol, where each process is responsible for implementing the actions of a role defined in the protocol. We denote roles with A, B ∈ R. A process can create new processes and sessions at runtime by invoking service processes (services for short). Services are always available at fixed locations, denoted l ∈ L, meaning that they can be used multiple times (in process calculus terms, they act as replicated processes [START_REF] Sangiorgi | The π-calculus: a Theory of Mobile Processes[END_REF]). Locations are novel to choreographies: they are addresses of always-available services able to create processes at runtime (cf. public channels in [START_REF] Carbone | Deadlock-freedom-by-design: multiparty asynchronous global programming[END_REF][START_REF] Montesi | Compositional choreographies[END_REF]). Processes communicate by exchanging messages. A message consists of two elements: i ) a payload, representing the data exchanged between two processes; and ii ) an operation, which is a label used by the receiver to determine what it should do with the message-in object-oriented programming, these labels are called method names [START_REF] Pierce | Types and Programming Languages[END_REF]; in service-oriented computing, labels are typically called operations as in here. Operations are denoted o ∈ O.

In the rest of the section, we comment the syntax of FC and present its semantics. However, before doing that, we dedicate the next paragraph to illustrate with an example the purpose and relationship between complete and partial actions.

C ::= η; C (seq) | C 1 | C 2 (par ) | if p.e {C 1 } else {C 2 } (cond ) | 0 (inact) | def X = C in C (rec) | X (call ) | k : A -> q[B].{o i (x i ); C i } i∈I (recv ) η ::= k : p[A].e -> q[B].o(x) (com) | start k : p[A] <=> l.q[B] (start) | k : p[A].e -> B.o (send ) | req k : p[A] <=> l.B (req) | acc k : l.q[B] (acc)
Fig. 1. Frontend Calculus, syntax.

Complete and Partial Actions. As in [START_REF] Montesi | Compositional choreographies[END_REF], FC supports modular development by allowing choreographies, say C and C , to be composed in parallel, written C | C . A parallel composition of choreographies is also a choreography, which can thus be used in further parallel compositions. Composing two choreographies in parallel allows the processes in the two choreographies to interact over shared location and session names. In particular, we distinguish between two kinds of terms inside of a choreography: complete and partial actions. A complete action is internal to the system defined by the choreography, and thus does not have any external dependency. By contrast, a partial action defines the behaviour of some processes that need to interact with another choreography in order to be executed. Therefore, a choreography containing partial actions needs to be composed with other choreographies that provide compatible partial actions. To exemplify the distinction between complete and partial actions, let us consider the case of a single communication between two processes.

Complete interaction

Composed partial actions

k : c[C].product -> s[S].buy( x ) k : c[C].product -> S.buy | k : C -> s[S].buy( x )
Above, on the left we have the communication statement as written at line 2 of Example 1. This is a complete action: it defines exactly all the processes that should interact (c and s). On the right, we implement the same action as the parallel composition of two modules, i.e., choreographies with partial actions. At the left of the parallel we write a send action, performed by process c to role S over session k, at the right of the parallel we write the complementary reception from a role C and performed by process s over the session k. More specifically, we read the send action as "process c sends a message as role C with payload product for operation buy to the process playing role S on session k". Dually, we read the receive action as "process s receives a message for role S and operation buy over session k and stores the payload in variable x". The compatible roles, session, and operation used in the two partial actions make them compliant. Thus, the choreography on the left is operationally equivalent to the one on the right. Observe that partial actions do not mention the name of the process on the other end-for example, the send action performed by process c does not specify that it wishes to communicate with process s precisely. This supports some information hiding: a partial action in a choreography can interact with partial actions in other choreographies, independently from the process names used in the latter. Expressions and variables used by senders and receivers are also kept local to statements that define local actions.

By equipping FC with both partial and complete actions, we purposefully made FC non-minimal. However, while a minimal version of FC (i.e., equipped with either partial or complete actions) can capture well-known choreographic models like [START_REF] Lanese | Bridging the gap between interaction-and process-oriented choreographies[END_REF][START_REF] Basu | Deciding choreography realizability[END_REF][START_REF] Honda | Scribbling interactions with a formal foundation[END_REF][START_REF] Carbone | Deadlock-freedom-by-design: multiparty asynchronous global programming[END_REF][START_REF] Dalla Preda | Dynamic choreographies[END_REF], we included both kinds of actions to describe a comprehensive language for programmers. On the one hand, complete actions offer a concise syntax to express closed systems, as found in other choreographic models. On the other hand, partial actions support compositionality in Frontend choreographies, allowing developers to write FC modules separately and possibly reuse the same module in multiple compositions.

Complete Actions. We start commenting the syntax of FC from complete actions, marked with a shade in Fig. 1. In term (start), process p starts a new session k together with some new processes q (q is assumed non-empty). Process p, called active process, is already running, whereas each process q in l.q, called service process, is dynamically created at the respective location l. Each process is annotated with the role it plays in the session. Term (com) models a communication: on session k, process p sends to process q a message for its operation o; the message carries the evaluation of expression e (we assume expressions to consist of standard computations on local variables) on the local state of p whilst x is the variable where q will store the content of the message. Partial Actions. Partial actions correspond to the terms obtained by respectively splitting the (start) and (com) complete terms into their partial counterparts. In term (req), process p requests some external services respectively located at l, to create some external processes and start a new session k. By "external", we mean that the behaviour of such processes is defined in a separate choreography module, to be composed in parallel. Role annotations follow those of term (start). Term (acc) is the dual of (req) and defines a choreography module that provides the implementation of some service processes. We assume (acc) terms to always be at the top level (not guarded by other actions), capturing always-available choreography modules. In term (send ), process p sends a message to an external process that plays B in session k. Dually, in term (recv ), process q receives a message for one of the operations o i from an external process playing role A in session k, and then proceeds with the corresponding continuation (cf. [START_REF] Coppo | Global progress for dynamically interleaved multiparty sessions[END_REF]).

Other Terms. In a conditional (cond ), process p evaluates a condition e in its local state to choose between the continuations C 1 and C 2 . Term (par ) is standard parallel composition, which allows partial actions in two choreographies C 1 and C 2 to interact. Respectively, terms (def ), (call ), and (inact) model the definition of recursive procedures, procedure calls, and inaction. Some terms bind identifiers in continuations. In terms (start) and (acc), the session identifier k and the process identifiers q are bound (as they are freshly created). In terms (com) and (recv ), the variables used by the receiver to store the message are bound (x and all the x i , respectively). In term (req), the session identifier k is bound. Finally, in term (def ), the procedure identifier X is bound. In the remainder, we omit 0 or irrelevant variables (e.g., in communications with empty messages). Terms (com), (send ), and (recv ) include role annotations only for clarity reasons; roles in such terms can be inferred, as shown in [START_REF] Montesi | Choreographic Programming[END_REF]. Semantics. The semantics of FC follows the standard principles of asynchronous multiparty sessions (cf. [START_REF] Honda | Multiparty asynchronous session types[END_REF]) and it is formalised in terms of reductions of the form D, C → D , C , where D is a deployment. Deployments store the states of processes and the message queues that support message exchange in sessions. We start by formalising the notion of deployment. Deployment. As in multiparty session types [START_REF] Honda | Multiparty asynchronous session types[END_REF], we equip each pair of roles in a session with a dedicated asynchronous queue to communicate in each direction. Formally, let Q = K × R × R be the set of all queue identifiers; we write k[A B] ∈ Q to identify the queue from A to B in session k. Now, we define D as an overloaded partial function defined by cases as the sum of two partial functions f s : P Var Val and f q : Q Seq(O × Val ) whose domains and co-domains are disjoint:

D = f s (z) if z ∈ P, f q (z) if z ∈ Q.
Function f s maps a process p to its state, which is a partial function from variables x, y ∈ Var to values v ∈ Val . Function f q stores the queues used in sessions. Each queue is a sequence of messages m = m 1 :: . . . :: m n | ε (ε is the empty queue), where each message m = (o, v) ∈ O × Val contains the operation o it was received on and the payload v.

Programmers do not deal with deployments. We assume that choreographies without free session names start execution with a default deployment that contains empty process states. Let fp(C) return the set of free process names in C and, with abuse of notation, ∅ be the undefined function for any given signature.

Definition 1 (Default Deployment). Let C be a choreography without free session names. The default deployment of D for C is defined as the function mapping all free names in C to empty states, i.e.,

D[p → ∅ | p ∈ fp(C)].
Reductions. In our semantics, choreographic actions have effects on the state of a system-deployments change during execution. At the same time, a deployment also determines which choreographic actions can be performed. For example, a communication from role A to role B over session k requires a queue k[A B] to exist in the deployment of the system.

We formalise the notion of which choreographic actions are allowed by a deployment and their effects using transitions of the form D δ -→ D , read "the deployment D allows for the execution of δ and becomes D as the result". Actions δ are defined by the following grammar:

start k : p[A] <=> l.q[B], session start, k : p[A].e -> B.o, send in session, k : A -> q[B].o(x), receive in session.
We report the rules defining D δ -→ D in the top-half of Fig. 2. Rule D | Start states that the creation of a new session k between an existing process p and new processes q results in updating the deployment with: a new (empty) state for each of the new processes q in q ( q → ∅ | q ∈ {q} ); and a new (empty) queue between each pair of distinct roles in the session ( k[C E] → ε | {C, E} ⊆ {A, B} ). Rule D | Send models the effect of a send action. In the premise, we use the auxiliary function ↓ to evaluate the local expression e in the state of process p, obtaining the value v to use as message payload. Then, in the conclusion, we append a message with its payload-(o, v)-to the end of the queue from the sender's role to the receiver's role (k[A B]). We assume that function ↓ always terminates-in practice, this can be obtained by using timeouts. Rule D | Recv models the effect of a reception. If the queue k[A B] contains in its head a message on operation o, it is always possible to remove it from the queue and to store its value v under variable x in the state of the receiver.

D | Start D start k: p[A] <=> l.q[B] --------------→ D q → ∅ | q ∈ {q} k[C E] → ε | {C, E} ⊆ {A, B} D | Send e ↓ D(p) v ⇒ D k[A B] → m k: p[A].e -> B.o ----------→ D k[A B] → m :: (o, v) D | Recv D k[A B] → (o, v) :: m k: A -> q[B].o(x) ----------→ D k[A B] → m, q → D(q)[x → v] C | Start δ = start k : p[A] <=> l.q[B] ∧ D#k ,r ∧ D δ[k /k][r/q] -------→ D ⇒ D, δ; C → D , C[k /k][r/q] C | Send η = k : p[A].e -> B.o ∧ D η -→ D ⇒ D, η; C → D , C C | Recv δ = k : A -> q[B].o j (x j ) ∧ j ∈ I ∧ D δ -→ D ⇒ D, k : A -> q[B].{o i (x i ); C i } i∈I → D , C j C | Eq R ∈ { ≡C , C } ∧ C 1 R C 1 ∧ D, C 1 → D , C 2 ∧ C 2 R C 2 ⇒ D, C 1 → D , C 2
Using deployment transitions, we can now define the rules for reductions

k : p[A].e -> q[B].o(x); C ≡ C k : p[A].e -> B.o; k : A -> q[B].{o(x); C}
unfolding complete communications into the corresponding send and receive terms.

Typing. We equip FC with a typing discipline based on multiparty session types [START_REF] Honda | Multiparty asynchronous session types[END_REF][START_REF] Coppo | Global progress for dynamically interleaved multiparty sessions[END_REF], which checks that partial actions composed in parallel are compatible. Our typing discipline is a straightforward adaptation of that presented for Compositional Choreographies [START_REF] Montesi | Compositional choreographies[END_REF], so we omit most details here (reported in [START_REF] Gabbrielli | Applied choreographies[END_REF]). However, the type system also gives us important information that will be critical in the compilation that we will develop later. Specifically, we mainly use types to keep track of the location that each process has been created at, the types of variables, the roles played by processes in open sessions, the role played by the processes spawned at each location in a choreography, and the status of message queues. Keeping track of this information is straightforward; hence, for brevity, we simply report the definition of the main elements of the typing environment Γ that we use to store it.

Γ ::= Γ, l : G A| B| C | Γ, k[A] : T | Γ, p : k[A] | Γ, p.x : U | Γ, p@l | k[B A] : T | ∅
In Γ , a service typing l : G A| B| C types with the (standard, from [START_REF] Coppo | Global progress for dynamically interleaved multiparty sessions[END_REF][START_REF] Montesi | Compositional choreographies[END_REF]) global type G any session started by contacting the services at the locations l. Role A is the role of the active process, whereas roles B are the respective roles of the service processes located at l. (The roles C are used to keep track of whether the implementation of some roles is provided by external choreography modules.) A session typing k[A] : T defines that role A in session k follows the local type T . An ownership typing p : k[A] states that process p owns the role A in session k. A location typing p@l states that process p runs at location l. The typing p.x : U states that variable x has type U in the state of process p. Finally, the buffer typing k[A B] : T states that the queue k[A B] contains a sequence of messages typed by the local type T .

A typing judgement Γ D, C establishes that D and C are typed according to Γ . If such a Γ exists, we say that D, C is well-typed.

Thanks to Γ , we can define a formal notion of coherence co, useful to check if a given set of Frontend modules can correctly execute a typed session:

Definition 2 (Coherence). co(Γ ) holds iff ∀ k ∈ Γ , ∃ G s.t. -l : G A| B| C ∈ Γ ∧ C = B and -∀ A ∈ roles(G), k[A] : T ∈ Γ ∧ T = G A ∧ ∀ B ∈ roles(G) \ {A}, Γ k[A B] : G A B
Coherence checks that i ) all services needed to start new sessions are present and ii ) all the roles in every open session are correctly implemented by some processes. To do this, given a global type G, co uses function G A to extract the local type of a role A in G and function G A B to extract the local type of the queue where role B receives from role A in G.

Well-typed Frontend choreographies that contain all necessary modules never deadlock.

Theorem 1 (Deadlock freedom). Γ D, C and co(Γ ) imply that either (i) C ≡ 0 or (ii) there exist D and C such that D, C → D , C .

Backend Calculus

We now present the Backend Calculus (BC). Formally, the syntax of programs in BC is the same as that of FC. The only difference between BC and FC is in the semantics: we replace the notions of deployment and deployment effects with new versions that formalise message exchanges based on message correlation, as found in Service-Oriented Computing (SOC) [START_REF]WS-BPEL[END_REF]. Deployments in BC. Deployments in BC capture communications in SOC, where data trees are used to correlate messages to input queues. We first informally introduce correlation.

Message Correlation. Processes in SOC run within services and communicate asynchronously: each process can retrieve messages from an unbound number of FIFO queues, managed by its enclosing service. To identify queues, services use some data, called correlation key. When a service receives a message from the network, it inspects its content looking for a correlation key that points one of its queues. If a queue can be found, the message is enqueued in its tail. As noted in the example, messages in SOC contain correlation keys as either part their payload or in some separate header. As in [START_REF] Montesi | Programming services with correlation sets[END_REF], we abstract from such details.

Data and Process state. Data in SOC is structured following a tree-like format, e.g., XML or JSON. We use trees to represent both the payload of messages and the state of running processes (as in, e.g., BPEL [START_REF]WS-BPEL[END_REF] and Jolie [START_REF] Montesi | Service-oriented programming with Jolie[END_REF]). Formally, we consider rooted trees t ∈ T , where edges are labelled by names, ranged over by x, and ∅ is the empty tree. We assume that all outgoing edges of a node have distinct labels and that only leaves contain values of type Val ∪ L, i.e., basic data (int, str, . . . ) or locations. Variables x are paths to traverse a tree: x, y ::= x.x | ε , where ε is the empty path (often omitted). Given a path x and a tree t, x(t) is the node reached by following x in t; otherwise, x(t) is undefined. Abusing notation, when x(t) is a leaf, then x(t) denotes also the value of the node. To manipulate trees, we will use the (total) replacement operator t ( x, t ). If x(t) is defined, t ( x, t ) returns the tree obtained by replacing in t the subtree rooted in x(t) by t . If x(t) is undefined, t ( x, t ) adds the smallest chain of empty nodes to t such that x(t) is defined and it stores t in x(t).

Backend Deployment. We can now define the notion of deployment for BC, denoted D. Formally, D is an overloaded partial function defined by cases as the sum of three partial functions g l : L Set(P), g m : L × T Seq(O × T ), and g s : P

T whose domains and co-domains are disjoint:

D(z) = g l (z) if z ∈ L, g m (z) if z ∈ L × T , and g s (z) if z ∈ P.
Function g l maps a location to the set of processes running in the service at that location. Given l, we read D(l) = {p 1 , . . . , p n } as "the processes p 1 , . . . , p n are running at the location l" (we assume processes to run at most at one location). Function g m maps a couple location-tree to a message queue. This reflects message correlation as informally described above, where a queue resides in a service, i.e., at its location, and is pointed by a correlation key. Given a couple l : t, we read D(l : t) = m as "the queue m resides in a service at location l and is pointed by correlation key t". The queue m is a sequence of messages m ::= m 1 :: • • • :: m n | ε and a message of the queue is m ::= (o, t), where t is the payload of the message and o is the operation on which the message was received. Function g s maps a process to its local state. Given a process p, the notation D(p) = t means that p has local state t. Deployment Effects in BC. In BC, we replace FC deployment effects (i.e., the rules for D δ -→ D ) with the ones reported in Fig. 3, commented below. Rule D | Start simply retrieves the location of process p (the one that requested the creation of session k) and uses Rule D | Sup to obtain the new deployment D that supports interactions over session k. Namely, D is an updated version of D with: i ) the newly created processes for session k and ii ) the queues used by the new processes and p to communicate over session k. In addition, in D , iii ) the new processes and p contain in their states a structure, rooted in k and called session descriptor, that includes all the information (correlation keys and the locations of all involved processes) to support correlation-based communication in session k. Formally, this is done by Rule D | Sup where we 1 ○ retrieve the starter process, here called q 1 , which is the only process already present in D. Then, given a tree t, we ensure it is a proper session descriptor for session k, i.e., that: 2 ○ t contains the location l i of each process. The location is stored under path B i .l, where B i is the role played by the i-th process in the session;

p ∈ D(l) D sup( l.p[A], l.q.[B] ) -----------→ D D start k: p[A] <=> l.q[B] --------------→ D D | Start q 1 ∈ D 1 ○ j ∈ I \ {i} B i .l(t) = l i 2 ○ B i .B j (t) = t ij 3 ○ l j : t ij ∈ D 4 ○ D 1 = D l i → D(l i ) ∪ {q i } 5 ○ D 2 = D 1 l i : t ij → ε 6 ○ D sup( li.qi[Bi] i∈I ) ------------→ D 2 q 1 → D 2 (q 1 ) ( k, t ) 7 ○ q h → {k : t} | h ∈ I \ {q 1 } 8 ○ D | Sup l = k.B.l( D(p) ) tc = k.A.B( D(p) ) e ↓ D(p) tm D k: p[A].e -> B.o ----------→ D [ l : tc → D(l : tc) :: (o, tm) ] D | Send tc = k.A.B( D(q) ) q ∈ D(l) D(l : tc) = (o, tm) :: m D = D[ l : tc → m ] D k: A -> q[B].o(x) ----------→ D [ q → D (q) ( x, tm ) ] D | Recv
3 ○ t contains a correlation key t ij for each ordered couple of roles B i , B j under path B i .B j , such that 4 ○ there is no queue in D at location l j pointed by correlation key t ij ; Finally, we assemble the update of D in four steps:

5

○ we obtain D 1 by adding in D the processes q 2 , . . . , q n at their respective locations;

6

○ we obtain D 2 by adding to D 1 an empty queue ε for each couple l j : t ij ;

7

○ we update D 2 to store in the state of (the starter) process q 1 the session support t under path k;

8 ○ we further update D 2 such that each new created process (q 2 , . . . , q n ) has in its state just the session descriptor t rooted under path k.

We deliberately define in D | Sup the session descriptor t with a set of constrains on data, rather than with a procedure to obtain the data for correlation. In this way, our model is general enough to capture different methodologies for creating correlation keys (e.g., UUIDs or API keys).

Rule D | Send models the sending of a message. From left to right of the premises: we retrieve the location l of the receiver B from the state of the sender p; we retrieve the correlation key t c in the state of p (playing role A) to send messages to role B; we compute the payload of the message by evaluating the expression e against the local state of the sender p. Then we obtain the updated deployment by adding message (o, t m ) in the queue pointed by l : t c that we found via correlation.

Rule D | Recv models a reception. From left to right of the premises: we find the correlation key t c for the queue that q (playing role B) uses to receive from A in session k; we retrieve the location l of q; we access the queue pointed by l : t c and retrieve message (o, t m ); we obtain a partial update of D in D removing (o, t m ) from the interested queue; we obtain the updated deployment by storing the payload t m in the state of q under path x. Encoding FC to BC. Runtime terms D, C in FC can be translated to BC simply by encoding D to an appropriate Backend deployment, since the syntax of choreographies is the same. For this translation, we need to know the roles played by processes and their locations, which is not recorded in D. We extract this information from the typing of C. Commenting the algorithm of D Γ , at line 2 it includes in D all (located) processes present in D (and typed in Γ ) and instantiate their empty states. At line 3 it translates the state (i.e., the association Var iable-Val ue) of each process in D to its correspondent tree-shaped state in D. At lines 4-7, for each ongoing session in D (namely, for each couple of processes, each playing a role in a given session), it sets the proper correlation keys and queues in D and, for each queue, it imports and translates the sequence of related messages.

The encoding from FC to BC guarantees a strong operational correspondence.

Theorem 2 (Operational Correspondence (FC ↔ BC)). Let Γ D, C. Then: 1. (Completeness) D, C → D , C implies D Γ , C → D Γ , C for some Γ s.t. Γ D , C ; 2. (Soundness) D Γ , C → D, C implies D, C → D , C and D = D Γ for some Γ s.t. Γ D , C .

Dynamic Correlation Calculus

We now introduce the Dynamic Correlation Calculus (DCC), the target language of our compilation. To define DCC, we considered a previous formal model for Service-Oriented Computing, based on correlation [START_REF] Montesi | Programming services with correlation sets[END_REF]. However, we found the calculus in [START_REF] Montesi | Programming services with correlation sets[END_REF] too simple for our purposes: there, each process has only one message queue, while here we need to manage many queues per process (as in our Backend deployments). Hence, to define DCC, we basically extend the calculus in [START_REF] Montesi | Programming services with correlation sets[END_REF] to let processes create and receive from multiple queues. Beside the requirement of this work, many languages for SOC (e.g., BPEL [START_REF]WS-BPEL[END_REF]) let processes create and receive from multiple queues, which makes DCC a useful reference calculus in general.

Syntax. The syntax of DCC, reported in Fig. 4, comprises two layers: Services, ranged over by S, and Processes, ranged over by P.

In the syntax of services, term (srv ) is a service, located at l, with a Start Behaviour B s and running processes P (both described later on) and a queue map M. The queue map is a partial function M : T Seq(O × T ) that, similarly to function g m in BC deployments, associates a correlation key t to a message queue. We model messages as in BC: a message is a couple (o, t) where o is

Behaviours B ::= ?@e 1 (e 2 ); B (reqst) | i [o i (x i ) from e] {B i } (choice) | def X = B in B (def ) | ν x; B (newque) | x = e; B (assign)
| o@e 1 (e 2 ) to e 3 ; B (output) the operation on which the message has been received and t the payload of the message. Services are composed in parallel in term (net).

| if e {B 1 } else {B 2 } (cond ) | 0 (inact) | X (call )
Concerning behaviours, in DCC we distinguish between start behaviours B s and process behaviours B. Process behaviours define the general behaviour of processes in DCC, as described later on. Start behaviours use term !(x) to indicate the availability of a service to generate new local processes on request. At runtime, the start behaviour B s of a service is activated by the reception of a dedicated message that triggers the creation of a new process. The new process has (process) behaviour B, which is defined in B s after the !(x) term, and an empty state. The content of the request message is stored in the state of the newly created process, under the bound path x. As in BC, also in DCC paths are used to access process states. Operations (o), procedures (X), variables (x, which are paths), and expressions (e, evaluated at runtime on the state of the enclosing process) are as in BC. Terms (choice) and (output) model communications. In a (choice), when a message can be received from one of the operations o i from the queue correlating with e, the process stores under x i the received message, it discards all other inputs, and executes the continuation B i . When only one input is available in a (choice), we use the contracted form o(x) from e; B. Term (output) sends a message on operation o, with content e 2 , while e 1 defines the location of the service where the addressee (process) is running and e 3 is the key that correlates with the receiving queue of the addressee. Term (reqst) is the dual of (acpt) and asks the service located at e 1 to spawn a new process, passing to it the message in e 2 . Term (cqueue) models the creation of a new queue that correlates with the key contained in variable x. Other terms are standard. Semantics. In Fig. 5, we report a selection of the rules defining the semantics of DCC, a relation → closed under a (standard) structural congruence ≡ that e ↓t tc j ∈ I M(tc) = (o j , tm) :: supports commutativity and associativity of parallel composition. To enhance readability, in rules we omit irrelevant elements with the place-holder -. We comment the rules. Rule DCC | Recv models message reception: if the queue correlating with t c (obtained from the evaluation of expression e against the state of the receiving process) has a message on operation o j , we remove the message from the queue and assign the payload to the variable x j in the state of the process. Rule DCC | Newque adds to M an empty queue (ε) correlating with a key stored in x. As for BC in rule D | Sup , we do not impose a structure for correlation keys, yet we require that they are distinct within their service. Rule DCC | Send models message delivery between processes in different services: the rule adds the message from the sender at the end of the correlating queue of the receiver. Rule DCC | Start accepts the creation of a new processes in a service upon request from an external process. The spawned process has B 2 as its behaviour and an empty state, except for x that stores the payload of the request.

m P = B j . t ( x j , tm ) -, & i∈I ([o i (x i ) from e] {B i }) . t | -, M -→ -, P | -, M[tc → m] - DCC | Recv tc ∈ dom(M) -, ν x; B . t | -, M -→ -, B . t ( x, tc ) | -, M[tc → ε] - DCC | Newque P = o@e 1 (e 2 ) to e 3 ; B . t e 1 , e 2 , e 3 ↓t l, tm, tc tc ∈ dom(M) -, P | -, --| -, -, M l → -, B . t | -, M -| -, -, M[tc → M(tc) :: (o, tm)] l DCC | Send Q = B 2 . ∅ ( x, t ) e 1 ↓t l e 2 ↓t t -, ?@e 1 (e 2 ); B 1 . t | -, --| !(x); B 2 , -, -l → -, B 1 . t | -, --| !(x); B 2 , Q | -, -l DCC | Start

Compiler from FC to DCC and Properties

We now present our main result: the correct compilation of FC programs into networks of DCC services. Given a term D, C in FC, our compilation consists of three steps: 1) it projects C into a parallel composition of endpoint choreographies, each describing the behaviour of a single process or service in C; 2) it encodes D to a Backend deployment; 3) it compiles the Backend choreography, result of the two previous steps, into DCC programs.

Step 1: Endpoint Projection (EPP). Given a choreography C, its EPP, denoted C , returns an operationally-equivalent composition of endpoint choreographies. Intuitively, an endpoint choreography is a choreography that does not contain complete actions-terms (start) and (com) in FC-describing the behaviour of a process, which can be either a service process or an active one. Our definition of EPP is a straightforward adaptation of that presented in [START_REF] Montesi | Compositional choreographies[END_REF], so we omit it here (see [START_REF] Gabbrielli | Applied choreographies[END_REF] for the full definition). First, we define a process projection to derive the endpoint choreography of a single process p from a choreography C, written C p . Then, we formalise the EPP of a choreography as the parallel composition of i ) the projections of all active processes and ii ) the merging of all service processes accepting requests at the same location. In the definition below, we use two standard auxiliary operators: the grouping operator C l returns the set of all service processes accepting requests at location l, and the merging operator C C returns the service process whose behaviour is the merge of the behaviours of all the service processes accepting requests at the same location.

Definition 4 (Endpoint Projection). Let C be a choreography. The endpoint projection of C, denoted C , is:

C = p ∈ fp(C) C p | l   p ∈ C l C p   Example 2.
As an example, let C be lines 5-8 of Example 1. Its EPP C is the parallel composition of the endpoint choreographies of processes c, s, and b, let them be respectively C c , C s , and C b , then

C = C c | C s | C b C b = if b.closeTx(cc, order) { k : b[B] -> C.ok; k : b[B] -> S.ok } else { k : b[B] -> C.ko; k : b[B] -> S.ko } C c = k : B -> c[C].{ ok(), ko() } C s = k : B -> s[S].{ ok(), ko() }
As shown above, the projection of the conditional is homomorphic on the process (b) that evaluates it. The projection of (com) terms results into a partial (send ) for the sender -as in the two branches of the conditional in C b -and a partial (recv ) for the receiver -as in C c and C s . Note that the EPP merges branching behaviours: in C c and C s the two complete communications are merged into a partial reception on either operation ok or ko.

Below, C ≺ C is the standard pruning relation [START_REF] Carbone | Structured communication-centered programming for web services[END_REF], a strong typed bisimilarity such that C has some unused branches and always-available accepts. In FC, the EPP preserves well-typedness and behaviour: Step 3: from BC to DCC. Given Γ D, C, where C is a composition of endpoint choreographies as returned by our EPP, we define a compilation D, C Γ into DCC by using the deployment encoding from FC to BC. To define D, C Γ , we use:

-C| l , to return the endpoint choreography for location l in C (e.g., acc k : l.p[A]; C ); -C| p , to return the endpoint choreography of process p in C; -C Γ , to compile an endpoint choreography C to DCC, using the (selected) rules in Fig. 6; -l ∈ Γ , a predicate satisfied if, according to Γ , location l contains or can spawn processes; -D| l returns the partial function of type T Seq(O × T ) that corresponds to the projection of function g m in D with location l fixed.

Definition 5 (Compilation). Let Γ D, C, where C is a composition of endpoint choreographies, and D = D Γ . The compilation D, C Γ is defined as

D, C Γ = l ∈ Γ C| l Γ , p ∈ D(l) C| p Γ. D(p), D| l l
Intuitively, for each service B s , P, M l in the compiled network: i ) the start behaviour B s is the compilation of the endpoint choreography in C accepting the creation of processes at location l; ii ) P is the parallel composition of the compilation of all active processes located at l, equipped with their respective states according to D = D Γ ; iii ) M is the set of queues in D corresponding to location l. We comment the rules in Fig. 6, where the notation is the sequence of behaviours i∈[1,n] (B i ) = B 1 ; . . . ; B n . Requests. Function start defines the compilation of (req) terms: it compiles (req) terms to create the queues and a part of the session descriptor of a valid session support (mirroring rule D | Sup ) for the starter. Given a session identifier k, the located role of the starter (l A .A), and the other located roles in the session ( l B .B), function start returns DCC code that: (s 1 ) includes in the session descriptor all the locations of the processes involved in the session. In (s 2 ) it adds all the keys correlating with the queues of the starter for the session, it requests the creation of all the service processes for the session, and it waits for them to be ready using the reserved operation sync. Finally, (s 3 ) it sends to them the complete session descriptor obtained after the reception (in the sync step) of all correlation keys from all processes. Accepts. Term (acc) defines the start behaviour of a spawned process at a location. Given a session identifier k, the role B of the service process, and the service typing G A| C| D of the location, function accept compiles the code that: (a 1 ) accepts the request to spawn a process, (a 2 ) creates its queues and keys, updates the session descriptor received from the starter, and sends it back to the latter (a 3 ). Finally with (a 4 ) the new process waits to start the session. We omit to report B B , which is similar to B S .

Properties. We report the main properties of our compilation to DCC.

In our definition, we use the term projectable to indicate that, given a choreography C, we can obtain its projection C . Theorem 4 defines our result, for which, given a well-typed, projectable Frontend choreography, we can obtain its correct implementation as a DCC network. By Theorem 1 and Theorem 4, deadlock-freedom is preserved from well-typed choreographies to their final translation in DCC. We say that a network S in DCC is deadlock-free if it is either a composition of services with terminated running processes or it can reduce.

Corollary 2. Γ D, C and co(Γ ) imply that D, C

Γ is deadlock-free.

Related Work and Discussion

Choreography Languages. This is the first correctness result of an end-to-end translation from choreographies to an abstract model based on a real-world communication mechanism. Previous formal choreography languages specify only an EPP procedure towards a calculus based on name synchronisation, leaving the design of its concrete support to implementors. Chor and AIOCJ [START_REF]Chor Programming Language[END_REF][START_REF] Dalla Preda | AIOCJ: A choreographic framework for safe adaptive distributed applications[END_REF] are the respective implementations of the models found in [START_REF] Carbone | Deadlock-freedom-by-design: multiparty asynchronous global programming[END_REF] and [START_REF] Dalla Preda | Dynamic choreographies[END_REF]. However, the implementation of their EPP significantly departs from their respective formalisation, since the former are based on message correlation. This gap breaks the correctness-by-construction guarantee of choreographies-there is no proof that the implementation correctly supports synchronisation on names. Implementations of other frameworks based on sessions share similar issues. For example, Scribble [START_REF] Honda | Scribbling interactions with a formal foundation[END_REF] is a protocol definition language based on multiparty asynchronous session types [START_REF] Honda | Multiparty asynchronous session types[END_REF] used to statically [START_REF] Hu | Session-based distributed programming in java[END_REF] and dynamically [START_REF] Demangeon | Practical interruptible conversations: distributed dynamic verification with multiparty session types and python[END_REF][START_REF] Neykova | Multiparty session actors[END_REF] check compliance of interacting programs. Our work can be a useful reference to formalise the implementation of these session-based languages. Delegation. Delegation supports the transferring of the responsibility to continue a session from a process to another [START_REF] Honda | Multiparty asynchronous session types[END_REF] and it was introduced to choreographies in [START_REF] Carbone | Deadlock-freedom-by-design: multiparty asynchronous global programming[END_REF]. Introducing delegation in FC is straightforward, since we can just import the development from [START_REF] Carbone | Deadlock-freedom-by-design: multiparty asynchronous global programming[END_REF]. Implementing it in BC and DCC would be more involved, but not difficult: delegating a role in a session translates to moving the content of a queue from a process to another, and ensuring that future messages reach the new process. The mechanisms to achieve the latter part have been investigated in [START_REF] Hu | Session-based distributed programming in java[END_REF], which would be interesting to formalise in our framework. Correlation keys. In the semantics of BC, we abstract from how correlation keys are generated. With this loose definition we capture several implementations, provided they satisfy the requirement of uniqueness of keys (wrt to locations).

As future work, we plan to implement a language, based on our framework, able to support custom procedures for the generation of correlation keys (e.g., from database queries, cookies, etc.).

1

  start k : c[C] <=> l S .s[S], l B .b[B]; 2 k : c[C].product -> s[S].buy(x); 3 k : s[S].mkOrder(x) -> b[B].openTx(order); 4 k : c[C].cc -> b[B].pay(cc); 5 if b.closeTx(cc, order) 6 { k : b[B] -> c[C].ok(); k : b[B] -> q[S].ok() } 7 else 8 { k : b[B] -> c[C].ko(); k : b[B] -> q[S].ko() }

Fig. 2 .

 2 Fig. 2. FC: top-half deployment transitions, bottom-half semantics (selected).

  D, C → D , C . We call a configuration D, C a running choreography. The reduction relation → for FC is the smallest relation closed under the (excerpt of) rules given in the bottom-half of Fig. 2. Rule C | Start creates a new session, by ensuring that both the new session name k and new processes r are fresh wrt D (D#k ,r). We use the fresh names in the continuation C, by using a standard substitution C[k /k][r/q]. Rule C | Send reduces a send action, if it is allowed by the deployment. Rule C | Recv reduces a message reception, if the deployment allows for receiving a message on one of the branches in the receive term (j ∈ I). Recalling the corresponding rule D | Recv , this can happen only if the deployment D has a message for operation o j in the queue k[A B]. Rule C | Eq closes → under the congruences ≡ C and C . Structural congruence ≡ C is the smallest congruence supporting α-conversion, recursion unfolding, and commutativity and associativity of parallel composition. The swap relation C is the smallest congruence able to exchange the order of non-interfering concurrent actions. Rule C | Eq also enables the reduction of complete communications on (com) terms with the equivalence

Fig. 3 .

 3 Fig. 3. Deployment effects for Backend Choreographies.

Definition 3 (

 3 Encoding FC in BC). Let D, C be well-typed: Γ D, C. Then, its Backend encoding is defined as D Γ , C, where D Γ is given by the algorithm in Listing 1.1.

1 D

 1 Γ = D := ∅; 2 f o r e a c h p@l i n Γ. D := D [ l → D(l) ∪ {p} ] ; D := D [ p → ∅ ] 3 f o r e a c h p.x : U i n Γ. D := D[ p → D(p) ( x, D(p)(x) ) ]; 4 f o r e a c h { p : k[A] q : k[B], q@l} i n Γ. t := fresh(D, l); 5 D := D[ l : t → D(k[A B]) ]; 6 D := D[ p → D(p) ( k.A.B, t ) ]; D := D[ p → D(p) ( k.B.l, l ) ]; 7 D := D[ q → D(q) ( k.A.B, t ) ]; D := D[ q → D(q) ( k.B.l, l ) ] 8 r e t u r n D Listing 1.1. Encoding Frontend to Backend Deployments.

Fig. 4 .

 4 Fig. 4. Dynamic Correlation Calculus, syntax.

Fig. 5 .

 5 Fig. 5. Dynamic Correlation Calculus, semantics (selected).

Theorem 3 (

 3 EPP Theorem). Let D, C be well-typed. Then, 1. (Well-typedness) D, C is well-typed. 2. (Completeness) D, C → D , C implies D, C → D , C and C ≺ C . 3. (Soundness) D, C → D , C implies D, C → D , C and C ≺ C .

Step 2 : 1 : 1 .

 211 Encoding to BC. After the EPP, we use our deployment encoding to obtain an operationally-equivalent system in BC. From Theorems 2 and 3 we derive Corollary Corollary Let Γ D, C. Then: 1. (Completeness) D, C → D , C implies D Γ , C → D Γ , C for some Γ s.t. Γ D , C and C ≺ C ; 2. (Soundness) D Γ , C → D, C implies D, C → D , C for some Γ s.t. Γ D , C , D Γ = D and C ≺ C .

(s 2 )(s 3 )(a 4 )Fig. 6 .

 2346 Fig. 6. Compiler from Endpoint Choreographies to DCC.

Example 3 .

 3 We compile the first two lines of the choreography C in Example 1. D, C Γ = 0, P c lC | B S , 0 lS | B B , 0 lB where P c = .l = l S ; k.B.l = l B ; ν k.S.C; ?@k.S.l(k); sync(k) from k.S.C; ν k.B.C; ?@k.B.l(k); sync(k) from k.B.C; start@k.S.l(k) to k.C.S; start@k.B.l(k) to k.C.B; /* end of start-request */ buy@k.S.l(product) to k.C.S; ... and B S = !(k); ν k.C.S; ν k.B.S; sync@k.C.l(k) to k.S.C; start(k) from k.C.S; /* end of accept */ buy(x) from k.C.S; ...

Theorem 4 (

 4 Applied Choreographies).Let D, C be a Frontend choreography where C is projectable and Γ D, C for some Γ . Then:1. (Completeness) D, C → D , C implies D Γ , C Γ → + D Γ , CΓ and C ≺ C and for some Γ , Γ D , C . 2. (Soundness) D Γ , C Γ → * S implies D, C → * D , C and S → * D Γ , C Γ and C ≺ C and for some Γ , Γ D , C .