Roberta Barbi
email: roberta.barbi@unine.ch

Vitaly Buravlev
email: vitaly.buravlel@imtlucca.it

Claudio Antares Mezzina

Valerio Schiavoni
email: valerio.schiavoni@unine.ch

Block placement strategies for fault-resilient distributed tuple spaces: an experimental study (Practical experience report)

The tuple space abstraction provides an easy-to-use programming paradigm for distributed applications. Intuitively, it behaves like a distributed shared memory, where applications write and read entries (tuples). When deployed over a wide area network, the tuple space needs to efficiently cope with faults of links and nodes. Erasure coding techniques are increasingly popular to deal with such catastrophic events, in particular due to their storage efficiency with respect to replication. When a client writes a tuple into the system, this is first striped into k blocks and encoded into n > k blocks, in a fault-redundant manner. Then, any k out of the n blocks are sufficient to reconstruct and read the tuple. This paper presents several strategies to place those blocks across the set of nodes of a wide area network, that all together form the tuple space. We present the performance trade-offs of different placement strategies by means of simulations and a Python implementation of a distributed tuple space. Our results reveal important differences in the efficiency of the different strategies, for example in terms of block fetching latency, and that having some knowledge of the underlying network graph topology is highly beneficial.

Introduction

We are currently observing a deluge of data originated by our personal devices. Distributed applications must be able to efficiently collect, store, process and expose data. When dealing with such applications, developers need to settle on a specific programming model, to i) facilitate the implementation of such systems and ii) retain userfriendliness and ability to scale, both horizontally and geographically. Distributed storage systems are one prominent example of such applications. They are typically operated across wide area networks, such as Amazon AWS, which currently spans across 15 geographical regions. 3 In such deployment scenarios, applications must transparently tolerate faults, a common threat for distributed systems.

A trivial strategy to tolerate faults is to rely on replication. Block replication obviously entails a huge storage overhead. A state-of-the-art solution to decrease such overhead while providing the same level of fault-tolerance is to use erasure coding techniques [START_REF] Macwilliams | The theory of error-correcting codes[END_REF]. With a systematic (n, k) linear code, each codeword (an element of the linear code) consists of n blocks: k source blocks for the original data, and nk redundant blocks. The storage overhead is n-k k , and if the code is Maximum Distance Separable (MDS) [START_REF] Macwilliams | The theory of error-correcting codes[END_REF], any k of the n blocks are necessary and sufficient to recover the original data.

From a fault tolerance point of view, it is optimal to place the n blocks of a codeword on different logical units (with respect to failures), so that the MDS code can tolerate up to nk failures. A logical unit can be a single node (in this case for the optimum it is sufficient to place different blocks of a codeword on different nodes), but it can also be a cluster of nodes (e.g. a set of machines physically hosted in a single room can go down at the same moment if the cooling system of the room fails). In this second scenario, one is tempted to spread different blocks of a codeword into separate and faraway clusters. Although being optimal with respect to fault tolerance, this solution affects negatively the latency to fetch the required blocks.

The case of distributed tuple spaces. A programming model can be made of two separate pieces: the computation model and the coordination model. The computation model allows programmer to build a single computational unit, while the coordination model is the glue that binds separate activities into an ensemble [START_REF] Gelernter | Coordination languages and their significance[END_REF]. The tuple space paradigm, based on this idea, offers a flexible technique to program parallel and distributed systems, by providing the abstraction of a shared space where all the processes can access. In this model, communication between processes is indirect and anonymous as it is done through the shared (distributed) space. Moreover, data exists in a tuple space and do not belongs to any process. Despite the simplicity of the model, very few implementations of tuple spaces offer fault tolerant facilities usually in the form of data replication ( [START_REF] Patterson | Construction of a fault-tolerant distributed tuple-space[END_REF][START_REF] Bessani | An efficient byzantine-resilient tuple space[END_REF]), with the drawbacks of space overhead and consistency maintenance. In this paper, we consider an extended, distributed tuple space system with erasure-coding capabilities. A tuple to be inserted in the tuple space is erasure-coded and its blocks are placed across the nodes joining the tuple space group.

Contributions. First, we study how to distribute the encoded blocks of single codewords over a large-scale network, in order to decrease the fetch latency. We do so by designing and evaluating several different block placement heuristics, over synthetic and real-world network topologies. Second, we evaluate how the proposed heuristics behave with respect to data loss when injecting faults into the topology. Third, we leverage the results of our simulations to identify two suitable placement strategies that we deploy atop a simple distributed tuple space system with the aim of evaluating their performance in a practical setting. This paper is organized as follows. First, we present the related work (Section 2). Next, Section 3 introduces the tuple space paradigm. In Section 4 we describe the block placement heuristics. Section 5 presents some modeling results that we leverage to drive the prototype implementation. Section 6 presents its implementation details and the extensions done to support both erasure-coding techniques and a pluggable mechanism to choose among the different placement strategies. We present the evaluation of the complete prototype in Section 7. We conclude in Section 8.

The tuple space coordination model is very appealing for distributed systems thanks to its space and time decoupling and its synchronization power. As a consequence, researchers have tried to add fault-tolerance and security to tuple spaces.

One recent result is DEPSPACE [START_REF] Bessani | Depspace: a byzantine faulttolerant coordination service[END_REF], a Byzantine fault-tolerant coordination service, which employs process replication for handling crashes and providing fault tolerance.

An alternative to process replication is block replication which entails the problem of block placement.

Block placement policies have been mainly studied in MapReduce contexts such as Hadoop [START_REF] Shvachko | The Hadoop Distributed File System[END_REF]. The main purpose of Hadoop's data placement policy is to provide good balance between reliability, write bandwidth, read performance and load balancing [START_REF] White | Hadoop: The definitive guide[END_REF]. Placing all replicas on a single node incurs the lowest write bandwidth penalty but it lacks redundancy: if the node fails, data is lost. On the other hand, placing replicas in different data centers maximize redundancy, but at the cost of bandwidth.

Hadoop's default strategy is to place the first replica on the same node as the client (for clients running outside the cluster, a node is chosen at random, although the system tries not to pick nodes that are too full or too busy). The second replica is placed on a different rack from the first, chosen at random. The third replica goes to the same rack as the second, but on a different node. Further replicas are placed on random nodes on the cluster, although Hadoop's block scheduler avoids placing too many replicas on the same rack. Our cluster-aware and distance-aware strategies share some similarity with this approach, in that they take into account zones of the system that are more sensitive to simultaneous failures. Several enhancement were introduced in Hadoop with respect to block placement policies, such as pluggable policies (since v0.21) or guarantees of even distributions across the cluster (since v0.17). We envision a similar technique to rebalance blocks of the tuple space according to the announced load ratio.

COHADOOP [START_REF] Eltabakh | CoHadoop: flexible data placement and its exploitation in Hadoop[END_REF] is a lightweight Hadoop extension that gives applications a finegrain control of data location. Similarly, our scheduling policies allow deployers to choose the destination of the blocks according to different performance criteria.

ADAPT [START_REF] Jin | Adapt: Availability-aware mapreduce data placement for non-dedicated distributed computing[END_REF] introduced a strategy to mitigate availability heterogeneity issues in non-dedicated distributed computing environments. ADAPT dynamically dispatches data blocks according to hosts' storage capacities. Through simulations, this strategy is shown to reduce the application runtime by more than 30%, increasing data locality and reducing data migration cost, even though the improvement of performances is less significant for environment with higher network connectivity.

Tuple Spaces in a Nutshell

The tuple space paradigm, made popular by Linda [START_REF] Gelernter | Generative communication in Linda[END_REF], is an abstraction of shared associative memory for parallel and distributed computing. A tuple space is a repository of tuples that processes can concurrently access via pattern-matching. Processes create new tuples (out or write operation), test the existence of a tuple (read) and consume a tuple (via the in operation). The simplicity of this coordination model makes this model intuitive and easy to use, also for distributed applications. In fact, some synchronization primitives (e.g. semaphores, synchronization barriers) can be easily implemented [START_REF] Carriero | How to Write Parallel Programs: A First Course[END_REF] leveraging this coordination model. Tuple space interaction model provides time and space decoupling, in that tuple producers and consumers remain anonymous with respect to each other [START_REF] Eugster | The Many Faces of Publish/Subscribe[END_REF]. Moreover a tuple has to survive its producer's termination, which can be caused by a node crash or due to the ending of the normal execution. In a distributed tuple space, each node writes tuples in its own local space, but it can read tuples also from remote ones. For example, in Figure 1 node D reads also the tuple produced by node C.

Despite the wide development of tuple space implementations [START_REF] Buravlev | Tuple spaces implementations and their efficiency[END_REF], very few of them offer support for distribution. While some systems use replication to guarantee data availability [START_REF] Patterson | Construction of a fault-tolerant distributed tuple-space[END_REF] or to be resilient to Byzantine faults [START_REF] Bessani | An efficient byzantine-resilient tuple space[END_REF], no existing system handle link or node faults to guarantee availability of data via erasure-coding. The extensions presented in Section 6 fill this gap.

Block Placement Strategies

In this section we describe several heuristics for block placement. Data is stored in the nodes of a graph representing a distributed storage system adding redundancy via standard [START_REF] Knight | The internet topology zoo. Selected Areas in Communications[END_REF][START_REF] Gelernter | Coordination languages and their significance[END_REF] Reed-Solomon code. The aim of the code is to map 10-blocks-inputs into 14-blocks-codewords in such a way that any 10 encoded blocks are sufficient to recover the original 10. In other words, this linear code can withstand loss of any 4 blocks of a codeword. Then the code provides the same level of fault-tolerance as 5 times replication while entailing a storage overhead of 40% only.

In this configuration, from a fault-tolerance point of view, it is optimal to place the 14 blocks of a codeword on units failing independently, such as geographically remote nodes. In reality, nodes hosted in the same data center have a higher likelihood to fail or being unreachable at the same time. Indeed there are several threats that can lead a data center to a power outage. We can mention cyber attacks, UPS system failures, air conditioner failures or human errors [START_REF] Institute | 2013 cost of data center outages[END_REF]. The proposed strategies must consider a trade-off between: -Latency efficiency: placing blocks apart from each other negatively affects the fetch latencies; -Failure resiliency: if related blocks are placed geographically close to each other, a failure affecting a wide geographical area will affect several blocks at once. With the aim of understanding experimentally this trade-off, we study 5 different placement heuristics. They take into account several structural graph properties (e.g. the clustering degree) with the objective of minimizing the latency for fetching blocks.

Round-robin (rr). The graph is divided into K clusters C 1 , . . . ,C K using K -means algorithm [START_REF] James | An introduction to statistical learning[END_REF]. We place the first block in a random node inside cluster C 1 , the second block in a random node in cluster C 2 and so on. We proceed until all blocks are placed.

Degree proportional (deg). This strategy places more blocks in nodes with higher degree. Intuitively, it let nodes with higher network capacity serve more blocks, irrespectively of their geographical location.

Cluster-aware (ca). This strategy assumes knowledge of the clustering of the network and places blocks in the cluster hosting the emitting node and two neighboring clusters. Using K -means, we divide the graph in K cluster C 1 , . . . ,C K . We say that cluster C i is at distance 1 from cluster C j if there is an edge of the graph with source/target in C i and target/source in C j . For each C i , we compute all clusters being at distance 1 from C i . We say that clusters C i and C j are at distance δ from each other if we must cross δ -1 clusters to go from C i to C j and this is the smallest number possible. For each C i , we compute all clusters being at distance 2 from C i .

In our simulations, we statically precompute the distances between clusters. We select a first cluster C at random for each codeword. Then, we extract at random 8 nodes from C, 4 different ones from a cluster at distance 1 from C, and finally 2 more from a cluster at distance 2. The chosen nodes receive the 14 blocks of the codeword. Notice that this heuristic needs at least 3 clusters to work.

Distance-aware (da). This strategy takes into account the distance between the node emitting the block and the other nodes in the graph. It assumes the knowledge of the diameter of the graph (d max), and proceeds as follows. First, 3 ranges of node-tonode distances (3 being a parameter of the algorithm) must be fixed: short (from the minimum to the 33 rd percentile of d max), mid (from the 33 rd to the 66 th percentile of d max), and long from the 66 th to d max. Then, for each codeword the algorithm picks a node N at random, and respectively 7 short-range nodes, 4 at mid-range and 2 from long-range nodes, for a total of 14 target nodes. Finally it places the 14 blocks of the codeword in such nodes. We report results for 3 ranges (da3), for 4 ranges (da4, for which the percentiles are 25 th , 50 th , 70 th and the number of blocks are 6, 4, 2, 1 for each range, respectively) and finally for 5 ranges (da5, using the percentiles 20 th , 40 th , 60 th , 80 th and the number of blocks are 5, 5, 1, 1, 1 for each range respectively). 4Random-Degree (drnd). This strategy combines a naive random strategy with deg. Each strategy contributes for the placement of half of the blocks. This section presents the results of our simulations with the aim of evaluating how the different placement strategies perform with respect to fetch-latency and data loss.

Load Balancing. We begin by studying how the strategies spread blocks on 4 different graph topologies. First, we consider a random graph of 1000 nodes, as depicted in Figure 2 on the left, where we highlight the 10 clusters computed by K -means using the Euclidean distance between nodes. The distribution of blocks among nodes is presented in Figure 2 (right). As expected, the rnd strategy produces a Gaussian distribution, while the other approaches tend to flatten and/or shift the bell.

Figure 3 shows topology and block distribution for a scale-free graph of 1000 nodes built using the preferential attachment method [START_REF] Barabási | Emergence of scaling in random networks[END_REF]. This topology closely maps a real Internet topology, yet is simple to study and analyze. We observe that deg and drnd produce a long-tail block distribution: several nodes have few blocks (right side of the figure), while few nodes store plenty of blocks (left side of the figure).

Finally, we consider two real-world topologies. The first is the Full European NREN network [START_REF] Knight | The internet topology zoo. Selected Areas in Communications[END_REF]. This graph has 1157 nodes and 1465 edges. When computing 10 clusters, we observe 1170 inter-cluster edges (i.e. source and destination nodes belong to different clusters). Topology and block distributions are presented in Figure 4. As an empirical confirmation that scale-free graphs are well-suited for representing Internet topologies, we underline the similarity between the two block distributions.

The second real-world topology, depicted in Figure 5, is the Cogent network [START_REF] Knight | The internet topology zoo. Selected Areas in Communications[END_REF]. It is smaller than the NREN topology (197 nodes, 245 edges) nevertheless it extends across Europe and US. This topology presents trans-oceanic links, with 13 edges to connect nodes across the Europe and North America. Different ranges in the block distribution with respect to other graphs are due to the much smaller number of nodes (while we distribute the same amount of data blocks).

Overall, block distributions generated by the da and rr strategies tend to be bellshaped, while dar and deg entail left-sided pick and long tail corresponding to few blocks in many nodes and few nodes hosting many blocks respectively.

For NREN and Cogent, we know the geographical coordinates of the nodes. To take into account of the curvature of the Earth and place more precisely the centroids of the clusters, we use the Haversine distance [START_REF] Robusto | The Cosine-Haversine Formula[END_REF] as K -means distance function.

We fix the number of cluster K = 10 in our simulations except for Cogent topology which is split in K = 2 clusters corresponding to USA and Europe. For the same reason, results of ca are not available for Cogent, since the heuristic requires at least 3 clusters.

Fetching latency. We continue by evaluating how the proposed strategies differ in terms of block recovery latency, as observed by the clients wishing to reconstruct matching tuples. We assume that the fetch-latency is proportional to the distance between nodes. Hence, we measure the length of the minimum paths between the node hosting the target block and the client.

We observe that a node storing a lot of blocks we necessarily need to fetch only few ones to reconstruct tuples. Hence, for each topology and each placement heuristic, we distinguish 3 types of clients based on the number of blocks they store. The lucky and the unlucky node stores the greatest and the smallest amount of blocks respectively. We use a representation based on stacked percentiles throughout the reminder of this section. The white bar at the bottom represents the minimum value, the pale gray on top the maximal value. Intermediate shades of gray represent the 25 th , 50 th -the median-and 75 th percentiles. We compare the results against a baseline rnd strategy that randomly places blocks across the graph. Figure 6 and 7 presents the case of the lucky and unlucky node respectively. These results validate the intuition that the number of blocks the client is storing greatly affects the observed fetch-latency. For instance, da3 performs better than other heuristics in 3 out of 4 topologies when the client is lucky. However, this is not the case for the unlucky case, where deg and rr perform better instead. These observations suggest that no strategy wins in all possible topologies, and that deployers need to carefully consider the different trade-offs for their applications and workloads.

Fetching latency under faults. Next, we perform a set of experiments that faults into the graph. For each graph, we select the most populated among the 10 clusters and we crash 1% of its nodes. This setting simulates a catastrophic event occurring to nodes geographically close to each other. Once the faults are injected, we use the lucky nodes (Figure 8) and unlucky nodes (Figure 9) to try to reconstruct all data stored. During these simulations, we did not observe any data loss. Hence, the heuristics are spreading blocks sufficiently apart from each other to tolerate crashes within the same cluster.

However, when injecting faults the fetch-latency highly depends on the particular failing nodes. In the case of the Cogent topology, the deg strategy greatly improves the results produced by the rnd placement, while on the scale-free graph performance degrades for the unlucky client. The da3 strategy outperforms the other heuristics in the NREN topology. More in general, distance-aware heuristics seem to be well-suited for the random graph.

Statistical analysis. Finally, to evaluate the statistical significance of the differences recorded by the simulations between the various heuristics, we perform two sets of t-tests [START_REF] James | An introduction to statistical learning[END_REF] on fault-free graphs. First, we build the dataset with one entry for every node. In this entry we compute the cumulative distance, that is, the sum of the length of all minimum paths covered to retrieve all data in the system from that particular client. We fix a topology and compare different heuristics against each other. We find the following p-values: scale-free graph random graph NREN Cogent t.test(rnd,deg) 0.2486 0.03055 0.00761 0.7828 t.test(rnd,da3) 0.4805 0.3242 0.3774 0.2203 These p-values answer the question: "what is the probability that the means of the cumulative distances covered by the two heuristics are equal?". For every graph we found an heuristic between da3 and deg such that the probability is less the 25%. We consider this a low evidence that the two means are the same but still such a value does not provide a decisive response.

For this reason, instead of using cumulative distances, we create a dataset of the distances covered to fetch every block by each node in the graph (e.g. in the case of the scale-free graph the dataset has 1000 entries times the number of blocks fetched, i.e. 3276000 entries). We run t-tests on random 1000-entries-samples from this dataset to compare different heuristics against each other. We find the following p-values: scale-free graph random graph NREN Cogent t.test(rnd,deg) 0.1661 6 • 10 -6 0.0004 0.6475 t.test(rnd,da3) 0.4215 0.0406 0.2936 0.1042 So for every topology we can find a heuristic between deg and da3 with support less than 16% for the hypothesis that the distance covered is the same as the one covered by rand. We take into account the modeling and statistical results to implement deg and da3 into in a real tuple space and evaluate how they perform in a practical setting.

Implementation

We implement and deploy three of the described blocks placement strategies (da3, deg and rnd) atop SIMPLETS, 5 a tuple-space implemented in Python (v3.4.0). The original implementation of SIMPLETS did not support remote tuple space nodes. Therefore, we first extended it to support a distributed scenario, leveraging PYRO (v4.0), 6 a remoting library for Python. Overall, our modifications to the SIMPLETS source code consist of only 250 additional lines of code.

To add erasure coding and block placement techniques, we extend the tuple space APIs with additional operations to properly handle writing, reading, and deletion of encoded tuples. For example, using a [START_REF] Knight | The internet topology zoo. Selected Areas in Communications[END_REF][START_REF] Gelernter | Coordination languages and their significance[END_REF] Reed-Solomon code, the out(t) operation that emits the tuple t in the tuple space, becomes out ec(t). This version encodes the tuple, splits it into 14 blocks and, according to the chosen strategy, distributes these blocks among the other nodes. To this end, from the original tuple a list of tuples of the following form is created: <tupleUID, blocksAndIndicesList, nodeList> where tupleUID is a unique identifier of the original tuple t, blocksAndIndicesList is a list of pairs (b i , i) indicating that b i is the i-th block of the codeword and nodeList is a list of nodes containing the remaining blocks. Figure 10 shows the extended version of SIMPLETS with erasure coding abilities.

In this configuration, reading a tuple only require to fetch 10 out of the 14 existing blocks. The tuple space programming paradigm requires the reading operations to operate via pattern-matching [START_REF] Gelernter | Generative communication in Linda[END_REF]. In the case of encoded tuples, the tuple space needs to decode the tuple. Therefore, this operation sequentially reads a tuple with blocks from the tuple space. Specifically, it leverages the nodeList index to discover and retrieve the missing blocks from other nodes in order to reconstruct the tuple. Then it checks whether the reconstructed tuples matches the template. Clearly, in the worst case to find a matching tuple the system has to decode the entire tuple space. We assume the existence of an up-to-date indexing service that serves the purpose of speeding up the process of discovering the location of the required blocks. In our evaluation, we assume to know the location of the nodes storing the blocks required to decode the tuple. It is out of the scope of this work how to efficiently maintain this index.

We implement both da3 and deg strategies on our tuple space and test them on a scale-free network made of 100 nodes. We emulate a large-scale network deployment using Docker (v1.13.1). We map each SIMPLETS node (with its local tuple space) to a standalone container. The latency between two nodes, say i and j, is proportional to their minimum distance on the graph. Latency (by mean of a sleep system call) is then interposed by the proxy interface of the Pyro service exposed by each tuple space process. In practice, when node i contacts node j to read (or write) a tuple, node j sleeps latency i, j milliseconds before replying. An alternative method is be to add latency at the OS level, e.g. by implementing a software router. 7

Prototype evaluation

This section presents our evaluation with the extended SIMPLETS system. Due to the lack of hardware resources, we are limited to a cluster of 100 node mimicking a scalefree network. Each node is executed by a SIMPLETS Docker container. In this evaluation, only communication delays among nodes are emulated.

Erasure-coding overhead. To evaluate the overhead of erasure coding, we execute an initial set of microbenchmarks for reading times. In this experiment, we vary the size of data stored in each tuple, from 1 byte to 512KB. At the beginning, we randomly distribute 1000 tuples across 100 nodes. Then, 10 random nodes read all the 1000 tuples. We measure the time for reading each tuple, and we report them as Cumulative Distribution Function (CDF). As shown in Figure 11 (left), the size of the tuple only modestly affects the reading time from the tuple space without encoding.

When erasure coding is enabled, Figure 11 (right), the reading time is more sensitive to the tuple size: it grows from milliseconds for the tuples containing 1 byte to several seconds for the size of 512KB. For bigger tuples, the time for encoding and decoding is significantly higher. We believe that a highly optimized erasure-code library, such as Intel ISA-L [START_REF]Intel's ISA-L[END_REF], would greatly reduce the overhead and make it more practical.

Experiments with different strategies. This experiment evaluates the performances of the tuple space using different block placement strategies. At the beginning, each of the 100 nodes writes 10 tuples. The tuples are encoded and split into blocks. Those are dispatched to remote nodes according to the given strategy. Finally, a node is chosen to fetch and reads the blocks of its own tuples. Figure 12 presents our results for write and read operations. 8 The plot shows the CDF of the timings to write/read the tuples into/from the tuple space. The da3 strategy achieves the best performances for writes, because the writing time depends on the number of nodes used to spread each tuple's blocks. Random placement offers the worst performance as it involves a high number of nodes. The reading time depends on the distribution of the blocks among the nodes. The distributions obtained reflect the ones shown in Figure 3 and we do not report them here due to lack of space. For the distanceaware and random strategies, distributions are more uniform and the times are low. For the degree-aware strategy, nodes with the higher degree have considerably more blocks and the reading time vary significantly. As consequence, the reading time depends also on the order in which tuples are written. In the case of SIMPLETS, the tuple space is implemented as a list, hence the reading time will be greater for the tuples which were written toward the end.

Conclusion

The problem of data block placement in a wide-area network setting is of paramount importance. Several distributed applications rely on a random strategy. In this paper we considered a scenario where distributed applications are implemented via the tuple space paradigm. These systems need to efficiently cope with network faults to avoid losing tuples, while at the same time being storage efficient and allow fast fetching time. We extended an open-source Python-based tuple-space implementation with distribution capabilities and erasure-coding features. We presented a study of several block placement strategies to dispatch blocks over the nodes of a distributed tuple space. We considered synthetic and real-world graph topologies, up to thousands of nodes. Our modeling, statistical analysis and system performance results, also based the evaluation of our full working prototype, shed some light on the trade-offs that one need to accept when deploying such systems. Our results reinforce the believe that it is important to gather structural informations about the underlying network topology to wisely choose the appropriate block placement heuristic.

In this work we considered the distributed tuple space as practical use-case. We stress that our strategies are general purpose and can be deployed in other distributed systems such as distributed key-value stores.

Fig. 1 :

 1 Fig. 1: Example of distributed tuple space: each node writes tuples in its own local tuple space (left) and read tuples from local and remote nodes (right).

Fig. 2 :Fig. 3 :

 23 Fig. 2: Left: random graph used in this experiment. Center and right: blocks distribution induced by the placement strategies under study.

Fig. 4 :

 4 Fig. 4: NREN topology and its blocks distribution.

Fig. 5 :

 5 Fig. 5: Cogent topology and its blocks distribution.

Fig. 6 :

 6 Fig. 6: Distance for fetching blocks (lucky node).

Fig. 8 :

 8 Fig.8: 1% crashes in one cluster, lucky node.

Fig. 9 :

 9 Fig.9: 1% crashes in one cluster, unlucky node.

Fig. 10 :

 10 Fig. 10: Distributed tuple space with erasure code: write ops. spread blocks apart driven by a specific strategy; read ops. fetch blocks from remote nodes.

Fig. 11 :

 11 Fig. 11: Distribution (CDF) of tuple's reading performance for increasing tuple size. Left: without erasure-coding. Right: with erasure-coding.

Fig. 12 :

 12 Fig. 12: Writing and the reading times for different strategies of blocks placement

http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/using-regions-availability-zones.html

The number of blocks assigned to each class of range nodes (da3, da4, da5) has been experimentally proved to work better in practice.

https://pythonhosted.org/Pyro4/

We report on our failed attempt in using Linux tc's traffic shaping (using delay.sh https://gist.github.com/arr2036/6598137) to emulate network latencies. In particular, the current Docker networking layer does not cope well with this approach, where all nodes in a given network class (such as all the Docker containers running in the same host) apply the same delay, preventing the emulation of more complex graph topologies.

We omit the results for withdrawing operations. They show similar trends to read results plus a small overhead due to the fetching of all the 14 blocks.

Acknowledgments

The authors are grateful to Hugues Mercier and Pascal Felber for invaluable discussions during the preliminary phases of this work. We are grateful to Rocco De Nicola for fruitful discussions around tuple spaces. This research was partially supported by the European Union's Horizon 2020 -The EU Framework Programme for Research and Innovation 2014-2020, under grant agreement No. 653884.


Tim Jungnickel
email: tim.jungnickel@tu-berlin.der.braeunlich@campus.tu-berlin.de

Ronny Bräunlich

formic: Building Collaborative Applications with Operational Transformation

Keywords: Consistency Control, Operational Transformation, Collaboration, Web Development

As part of the ongoing revolution of the way people work in distributed teams, the need of applications for real-time collaboration is increasing. Commercial products like Google Docs set the landmark for modern web-based collaboration. In this work we provide a library that utilizes the underlying technology, namely Operational Transformation, to simplify the development of collaborative web applications. Our library formic features a novel transformation function that enables simultaneous editing of JSON objects.

Introduction

The Internet has changed the way we work together and collaboration is no longer restricted to face-to-face meetings. Hence, working in geographically distributed teams will be the predominant part of our future work life [START_REF] Powell | Virtual teams: A review of current literature and directions for future research[END_REF]. Moreover, due to the increasing number of internet devices everyone uses, we will easily become collaborators with ourselves.

The only way to technically realize usable collaborative applications in high latency networks, like the Internet, is to use a local replica of the application state on every collaborating device. Hence, users can directly access and update the local replica on the device without any noticeable latency. All changes are propagated in the background. Unfortunately, having multiple replicas of the application state raises fundamental questions in distributed systems research. Most important is the need of a consistency control mechanism to ensure convergence among replicas [START_REF] Brewer | Towards robust distributed systems[END_REF].

Successful collaborative applications like Google Docs or Etherpad use Operational Transformation (OT) [START_REF] Ellis | Concurrency control in groupware systems[END_REF] to allow simultaneous editing of shared documents. We illustrate the mechanics of OT with the following simple text editing scenario. Two users u 1 and u 2 maintain their own replica of the character sequence abc. Both users simultaneously invoke edit operations on their local replica. The user u 1 inserts an X at position 0, resulting in Xabc. The user u 2 deletes the character b at position 1, resulting in ac. A naïve interchange of the invoked edit operations would result in diverging replicas: u 1 results in Xbc, whereas u 2 results in Xac. In OT, remote operations are transformed based on previously executed local operations. Hence, u 1 needs to transform the position of the remote delete operation to respect the effect of the local insert operation, i.e. u 2 's delete operation on position 1 needs to be transformed to a delete operation on position 2 to ensure convergence.

In modern web development, single-page applications based on JavaScript become more and more popular. In frameworks like AngularJS or React essential business logic is executed in the browser at client site, allowing fully responsive user interfaces, even if the network connection is unstable. The major challenge for building collaborative web applications is to combine the chosen web framework with a fitting consistency control mechanism like OT.

Contributions:

In this work we contribute to close the gap between the conceptual and formal descriptions of OT systems and real world web development by providing the missing details of an OT extension that supports simultaneous editing of JSON objects. Since JSON is the de facto standard data interchange format of the web, we expect that a combination of JSON with OT encourages the design of more complex collaborative web applications. Ultimately we present and evaluate a programming library that utilizes provably correct transformations and simplifies the development of collaboration systems.

Related Work: OT has been introduced by Ellis and Gibbs in 1989 [START_REF] Ellis | Concurrency control in groupware systems[END_REF], followed by multiple decades of research around the mechanism and very valuable contributions from various groups. Prominent example applications are Google's document editing suite Google Docs and the free competitor Etherpad. In recent work, Dang and Ignat showed, that the performance of both systems with a larger number of collaborators is limited [START_REF] Dang | Performance of real-time collaborative editors at large scale: User perspective[END_REF]. Hence, our library must be able to compete with such systems, which we evaluate in section 4.

We have seen notable work outside the academic community that aims to use OT for the development of new collaborative applications. The JavaScript library ShareDB [START_REF] Gentle | ShareDB[END_REF] offers the OT mechanism with support for various datatypes such as lists or JSON objects. However, especially the JSON datatype misses a verification of the necessary transformation property (namely TP1).

Apart from OT, other consistency control systems are interesting for collaborative applications, for example Conflict-Free Replicated Data Types [START_REF] Shapiro | Conflict-free replicated data types[END_REF]. Several benchmarks have been conducted to show the suitability of CRDTs for document editing [START_REF] Briot | High responsiveness for group editing crdts[END_REF][START_REF] Ahmed-Nacer | Evaluating crdts for real-time document editing[END_REF]. In recent work, Nédelec et al introduced a web-based collaborative editor CRATE that enables collaboration without the need of a central server [START_REF] Nédelec | Crate: Writing stories together with our browsers[END_REF]. Kleppmann and Beresford presented a very promising JSON CRDT [START_REF] Kleppmann | A conflict-free replicated JSON datatype[END_REF] which has, to the best of our knowledge, not been demonstrated in a collaborative application. In this work, however, we focus on OT systems and compare the performance of Google Docs and ShareDB against our library.

An alternative but noteworthy mechanism is Differential Synchronization by Fraser [START_REF] Fraser | Differential synchronization[END_REF], which is a state based synchronization mechanism based on diffing and patching. An implementation of Jan Monschke demonstrates the applicability to JSON documents [START_REF] Monschke | DiffSync[END_REF]. So far we have not seen much academic attention to it.

In earlier research, we presented the conceptual extension of OT to support operations on JSON objects [START_REF] Jungnickel | Simultaneous editing of json objects via operational transformation[END_REF]. In this paper, we deliver the missing implementation and evaluation.

Preliminaries

In general, OT is an operation based consistency control system, i.e. edit operations on the local replicas of the collaborators are propagated through the network and applied at remote sites. An OT system is composed of a control algorithm and a transformation function [START_REF] Sun | Achieving convergence, causality preservation, and intention preservation in real-time cooperative editing systems[END_REF]. The control algorithm determines the operations to be transformed and the transformation order. The transformation function determines how the operations are transformed to include the effects of previous operations.

We demonstrate an extract of a transformation function for operations on lists in listing 1.1, which was initially introduced by Ellis and Gibbs in [START_REF]Wave Protocol[END_REF] and improved by Ressel et al in [START_REF] Fraser | Differential synchronization[END_REF]. We recall the example from the introduction where u 1 inserts a character at position 0 and u 2 simultaneously deletes a character at position 1. According to line number 2 of Listing 1.1, the position of the delete operation must be increased by one. In this example, the transformation of the delete operation ensures convergence among replicas. One essential and necessary property of a transformation function is the Transformation Property 1 (TP1) [START_REF] Ressel | An integrating, transformation-oriented approach to concurrency control and undo in group editors[END_REF]. In essence, TP1 describes that the transformation function needs to repair the inconsistencies that occur if two operation instances are applied in different orders, loosely formalized as:

∀O 1 , O 2 . XFORM(O 1 , O 2 ) = (O 1 , O 2 ) ⇒ (O 2 • O 1 = O 1 • O 2 )
OT Systems are not restricted to operations on linear data structures such as lists. However, more complex data structures result in more complex transformation functions, which makes it difficult to prove that TP1 is satisfied. For the rest of this paper, we focus on operations on JSON objects, because we identified them to be most relevant for modern web applications.

JSON (JavaScript Object Notation) is a hierarchically structured data interchange format [START_REF] Json | JavaScript Object Notation[END_REF]. A JSON object is an unordered set of key/value pairs. Values can be of primitive type (such as string, number, or boolean) or complex structures, such as arrays or other objects. An array is an ordered list of values.

formic's JSON Transformation

Our goal is to provide the necessary mechanics to build collaborative web applications easier and based on solid formal guarantees. Therefore, we implemented formic [START_REF] Bräunlich | formic[END_REF], a free software library that features collaboration on list structures, ordered n-ary trees, and JSON. In this section we report on formic's architecture and provide the missing details of the transformation of operations on JSON objects.

Architecture: Our library formic utilizes the Wave OT control algorithm, which has been introduced by Google [START_REF]Wave Protocol[END_REF]. The algorithm provably ensures convergence among replicas as long as the used transformation function satisfies TP1. Our library implements the client and the server part of the Wave algorithm. Updates among server and client are sent via WebSockets, i.e. a communication protocol for bidirectional communication over a single TCP connection. In case of disconnection, a client can continue to operate offline based on the state of the local replica. If the client reconnects, all operations are exchanged and the replicas converge. In contrast to ShareDB and Etherpad, formic is, to the best of our knowledge, the first free software OT library that supports an offline mode.

In formic, we implemented transformation functions for operations on lists and n-ary trees that are proven1 to satisfy TP1. Moreover, we developed a transformation function for operations on arbitrary JSON objects.

JSON Transformation:

In formic, we introduce a novel transformation function, which features insert, delete, and replace operations on arbitrary JSON objects. Since JSON is a hierarchical format, we derive essential parts from the transformation function on n-ary trees, which is already proven to be TP1-valid. The major difference is, that the position parameter, which indicates where an item should be edited, is no longer an access path, i.e. a vector with numeric parameters. Since a position inside a JSON is identified with a vector of keys and positions inside an array, the operations must be translated to operations on trees accordingly. For example, the position parameter of an operation on the JSON object in Fig. 1, that aims to insert an item at position 0 of the array with the key "key5", would be translated in the following way:

["key3", "key5", 0] → [2, 0, 1, 0, 0]
The complete translation mechanism from a JSON object to an n-ary tree is documented in the formic repository [START_REF] Bräunlich | formic[END_REF]. The transformation function for replace operations can be found in the updated version of the technical report in [START_REF] Jungnickel | Tp1-valid transformation functions for operations on ordered n-ary trees[END_REF].

Evaluation

We evaluated the performance of our library by reusing the experiment of Dang and Ignat [START_REF] Dang | Performance of real-time collaborative editors at large scale: User perspective[END_REF], which was initially used to explore the performance of Google Docs {} "key1" "key2" "key3" [ ] "string" {} 42 true "key4" "key5"

null [ ] { "key1": [42, true], "key2": "string", "key3": { "key4": null, "key5": [] } } Fig. 1: Tree representation of a JSON object [START_REF] Jungnickel | Simultaneous editing of json objects via operational transformation[END_REF] at large scale. In their experiment, real users have been simulated with Selenium, a widely accepted web-based testing tool. The simulated users are divided into one Writer, one Reader, and up to 50 DummyWriters. The DummyWriters write random strings to a shared document. The Writer writes a specific string to the document and the Reader waits until the specific string is present and reports the delay. Dang and Ignat measured the delay with different numbers of Dum-myWriters and various type speed (1-8 keystrokes per second). We used a similar setup to evaluate formic by installing the server and the Selenium users on several virtualized machines on our local cluster (16 servers with 2 x Intel Xeon X5355 (2x4 cores), 32GB Memory).

Note that the original experiment design of Dang and Ignat is based on simple insertions of characters and strings to an empty document. Hence, no functionality of a rich text editor is utilized in this experiment. Therefore we decided to compare the performance measurement of Google Docs in [START_REF] Dang | Performance of real-time collaborative editors at large scale: User perspective[END_REF] to the performance of the transformation of list operations in formic. We present the results in Fig. 2.

In contrast to Google Docs, formic offers the OT mechanism in a way that web developers can enable simultaneous and collaborative editing of arbitrary objects, as long as the objects can be serialized into JSON. In order to evaluate the transformation of operations on JSON objects properly, we decided to compare the performance of formic to ShareDB in an collaborative JSON editing scenario. For this run, we modified the mentioned experiment design so that the DummyWriters are now invoking operations to a shared JSON object over a test website. To ensure comparability, we implemented an identical test website with formic and ShareDB and installed both systems on the same local cluster. We present the results in Fig. 3.

Results: In Fig. 2, we show a comparison between the results of Dang and Ignat and the performance of formic in an identical setup. We see that formic is able to compete with the performance of Google Docs and even shows a better performance at large scale. In Fig. 3 we show the comparison between formic and ShareDB in a JSON editing scenario. We see that ShareDB performs slightly (a) Google Docs from [START_REF] Dang | Performance of real-time collaborative editors at large scale: User perspective[END_REF] q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q 0 10 better. In contrast to the first experiment, the delay of both systems remains relatively low and does not exceed 10 seconds.

Discussion: With respect to text editing scenarios, we can confirm the finding of Dang and Ignat, that the performance of OT in collaborative web applications is limited at large scale. However, the performance of formic is comparable with Google Docs. We note that the used local cluster for the evaluation of formic is relatively old. Hence, we would expect even better results with modern hardware.

Unfortunately, Google provides no insight into the used infrastructure and the underlying OT implementation and it is therefore difficult to reason about the performance results of Google Docs.

In the JSON editing scenario, our library performs slightly worse than the competitor ShareDB. We explain the difference in the performance by the used optimizations in ShareDB which are not implemented in formic yet. For example, multiple operations on the local replica can be combined before they are sent to the server. This reduces the amount of necessary communication and leads to faster response times. One major bottleneck in formic is the mapping of a JSON object to an ordered n-ary tree. The mapping enforces a total order in every layer of the tree, which is technically not necessary for every JSON component. For example, key/value pairs inside a JSON object do not require ordering, whereas elements inside an array must be ordered. This issue can be solved by introducing a more complex data model, which leads, as mentioned in section 2, to more complex proofs. The most interesting solution would use a combination of different consistency control systems to best suit the JSON definition, e.g. a combination of the Observed-Remove Set CRDT [START_REF] Shapiro | Conflict-free replicated data types[END_REF] and OT.

Ultimately, formic is a considerable tool to develop collaborative web applications that are able to compete with established collaborative solutions. The very next step is to improve the accessibility of formic by providing examples and developer friendly integrations for commonly used web frameworks. As a first step, we implemented a collaborative Battleship game based on a shared q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q 0 10 (a) ShareDB q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q q 0 10 20 JSON object and published the source code along with the library [START_REF] Bräunlich | formic[END_REF]. Moreover, we plan to integrate the JSON transformation into a collaborative patient documentation system [START_REF] Jungnickel | Hotpi: Open-source collaborative patient documentation[END_REF].

Conclusion

With formic, we presented an open-source library that simplifies the development of web-based collaborative applications by providing a fully working OT system with implemented transformation functions for operations on lists, trees and JSON objects. Moreover, formic is composed of tested and verified components. Especially the used transformation functions are proven to be TP1-valid, which makes formic an attractive tool to solve the consistency-related challenges in collaborative applications.

The conducted experiment has demonstrated that our library is able to compete against Google Docs, the most successful collaborative application that utilizes OT. However, the underlying client-server architecture limits the performance at large scale.

Ultimately, the development of collaborative web applications will become more important in the future. The use of a single service for web-based collaboration, such as Google Docs, is highly questionable in terms of privacy and confidentiality, especially for sensitive data. Therefore, we expect to see more organizations, which use in-house applications for the collaborative work in distributed teams. The presented library simplifies the development and enables the design of collaborative applications that are not restricted to collaborative text editing, but rather fully flexible due to the JSON transformation. Future work includes the improvement of the accessibility and the development of more features, such as undo/redo or move operations.

Listing 1 . 1 :

 11 Pseudo code of the transformation function 1 function XFORM(insert(i, k1), delete(k2)): 2 if k1 < k2: return(insert(i, k1), delete(k2 + 1)) 3 if k1 > k2: return(insert(i, k1 -1), delete(k2)) 4 if k1 == k2: return(insert(i, k1), delete(k2 + 1))

Fig. 2 :

 2 Fig. 2: The performance of collaborative editing with one keystroke per second.

Fig. 3 :

 3 Fig.3: JSON editing with one modification per second.

For list operations, we implemented a proof in the theorem prover Isabelle[START_REF] Jungnickel | A proof of tp1 for transformations of list operations[END_REF]. For operations on n-ary trees, we refer to our technical report[START_REF] Jungnickel | Tp1-valid transformation functions for operations on ordered n-ary trees[END_REF].


Nikos Zacheilas
email: zacheilas@aueb.gr

Vana Kalogeraki

DIsCO: DynamIc Data COmpression in Distributed Stream Processing Systems

Supporting high throughput in Distributed Stream Processing Systems (DSPSs) has been an important goal in recent years. Current works either focus on automatically increasing the system resources whenever the current setup is inadequate or apply load shedding techniques discarding some of the incoming data. However, both approaches have significant shortcomings as they require on the fly application reconfiguration where the application needs to be stopped and re-uploaded in the cluster with the new configurations, and can lead to significant information loss. One approach that has not yet been considered for improving the throughput of DSPSs is exploiting compression algorithms to minimize the communication overhead between components especially in cases where we have large-sized data like live CCTV camera reports. This work is the first that provides a novel framework, built on top of Apache Storm, which enables dynamic compression of incoming streaming data. Our approach uses a profiling algorithm to automatically determine the compression algorithm that should be applied and supports both lossless and lossy compression techniques. Furthermore, we propose a novel algorithm for determining when profiling should be applied. Finally, our detailed experimental evaluation with commonly used stream processing applications, indicates a clear improvement on the applications' throughput when our proposed techniques are applied.

Introduction

In recent years we observe a growing need for supporting complex real-time processing of "big data". Many systems need to process large volumes of live data to detect events of interest in real-time. For example, in a traffic monitoring application it is necessary to inform the city's authorities for events like traffic congestion or accidents [START_REF] Zacheilas | Elastic Complex Event Processing exploiting Prediction[END_REF] as they occur. Similarly, healthcare applications [START_REF] Liu | Realtime Healthcare Services Via Nested Complex Event Processing Technology[END_REF] receive input from multiple sensors to detect unusual behavior in the patients' conditions. In order to be able to analyze such a high volume of data, novel distributed systems such as Storm [START_REF] Marz | s Storm[END_REF], Spark [START_REF] Zaharia | Discretized Streams: Fault Tolerant Streaming Computation at Scale[END_REF] and Flink [START_REF] Carbone | Apache flink: Stream and batch processing in a single engine[END_REF] have been proposed that enable us to perform scalable and low latency complex event detection.

One important challenge in such systems is to support high throughput during the applications' execution despite changes in the data size or the input rate. In the literature, diverse techniques such as elasticity [START_REF] Gedik | Elastic scaling for data stream processing. Parallel and Distributed Systems[END_REF] and load shedding [START_REF] Tatbul | Staying fit: Efficient load shedding techniques for distributed stream processing[END_REF] (a) Compressed image (0.01 JPEG quality) (b) Original Image Fig. 1. Image degradation due to JPEG compression.

have been proposed for solving this problem. Elasticity schemes like [START_REF] Gedik | Elastic scaling for data stream processing. Parallel and Distributed Systems[END_REF] and [START_REF] Zacheilas | Elastic Complex Event Processing exploiting Prediction[END_REF], automatically increase the amount of system resources (i.e., stream processing operators or components) in order to adapt to sudden load spikes. Such techniques have significant shortcomings: (a) often DSPSs do not support this feature so applications need to be stopped and re-uploaded in the cluster with the new configurations, and (b) approaches such as load shedding [START_REF] Tatbul | Staying fit: Efficient load shedding techniques for distributed stream processing[END_REF], [START_REF] Eberle | Online unsupervised state recognition in sensor data[END_REF], automatically drop incoming data when load spikes occur; this penalizes the results' accuracy as many tuples will not be processed and thus important events of interest may be lost.

A technique that has not been fully exploited in these settings is the use of data compression. Data compression can reduce the impact of large-sized data in the components' communication time and thus enables the system to process data faster, increasing the system's throughput. However, applying compression is not always beneficial as it creates additional processing overhead as tuples need to be compressed/decompressed before the operator processes them. In recent years, we observe a plethora of novel compression algorithms [START_REF] Bicer | Integrating online compression to accelerate large-scale data analytics applications[END_REF] or commonly used compression libraries such as LZ4 [13], Zip [22] and Snappy [15] which are applied in the distributed system's domain [START_REF] Chen | To compress or not to compress-compute vs. io tradeoffs for mapreduce energy efficiency[END_REF]. However, in current systems the compression algorithm must be manually provided by the users and cannot change during the application's lifetime. Furthermore, it is not trivial for the end-user to determine how useful a compression technique is, when it should be applied, and which compression algorithm is most appropriate to maximize the system throughput. In general, compression techniques can be divided in two major categories, lossless and lossy. The main difference is that lossless techniques enable the perfect reconstruction of the input data and therefore the decompressed data will be the same as the initial data. In contrast, lossy techniques include the class of data encoding methods that use inexact approximations and partial data discarding, in order to represent the content. So these techniques lead to data degradation and possibly to inaccurate results if they are applied in event detection applications.

In the following application example we demonstrate that the compression ratio of lossy techniques needs to be carefully chosen, taking into account the impact of the compression on the results' accuracy. Our application (described in more details in Section 4) receives as input CCTV camera images streamed in real-time in Dublin city and utilizes a simple image similarity algorithm [START_REF] Lux | Lire: lucene image retrieval: an extensible java cbir library[END_REF] against historical images which depict normal and abnormal traffic conditions. More specifically, for each incoming image we find the most similar historical image and based on the characterization of the latter (i.e., whether it depicts normal or abnormal traffic conditions) we alert the traffic authorities. In Figures 1(b), 1(a) we illustrate how the JPEG quality metric affects the image in terms of visibility. As it can observed when we use very low JPEG quality the image is not visible (i.e., in Figure 1(a) the obstacle that fell on the road is blurred) and therefore the application is not able to detect the actual accident that happened (i.e., depicted in Figure 1(b)).

So the JPEG quality affects the accuracy of the results and this can be clearly shown in Figure 2 where we illustrate the number of detected events and how many of them are false positives and false negatives. As it is shown in the figure, when we use low compression quality the false positives and negatives events increase due to the degradation of the images. However, when low compression quality is applied it increases the system's throughput due to the smaller image sizes. This is clearly illustrated in Figure 3 where we display the application's throughput using varying JPEG quality.

This trade-off between the applications' accuracy and the system's throughput needs to be taken into account when we determine whether a lossy compression technique like JPEG should be applied. Furthermore, it is not trivial for the enduser to understand when compression is beneficial and how different compression algorithms can affect these two metrics. So our aim in this work is to provide a framework that is able to provide answers to the following questions:

1. When is it beneficial to compress the incoming data of a DSPS? 2. Which algorithm should be used for the compression? 3. How the chosen algorithm will be applied efficiently in a distributed streaming environment where the overhead of compression should be minimal?

We propose a novel framework, DIsCO, that executes on top of Apache Storm and enables the automatic compression of incoming streaming data for the users' applications. Our approach monitors the application's performance and automatically adjusts the compression algorithm that should be applied taking into account the impact of the compression on the observed throughput and the results' accuracy. The key contributions of this work are as follows:

-We apply a profiling technique that aims to determine when compression is beneficial for Storm applications and which algorithm should be applied. In our framework we consider well-known compression algorithms like Zip, LZ4, Snappy and JPEG. Our technique has as goal to find the compression algorithm that balances the trade-off between the application's throughput and the results' accuracy. Our approach works adaptively, where the profiling algorithm is re-invoked when the performance of the application in terms of throughput and results' accuracy has significantly decreased. -We enhance the Storm system by adding special threads that are part of the Storm's components (i.e., spouts/bolts) and perform the compression/decompression procedure in parallel. This way we minimize the compression/decompression overhead (i.e., due to the time required for compressing and decompressing the tuples) on the application's throughput. -We conduct an extensive experimental evaluation in our local cluster using different applications that process both text and image data. More specifically, we used a traffic monitoring application that performs image similarity on Dublin's CCTV camera reports, an application that periodically crawls 5 major news sites and searches for traffic incidents and a Twitter First Detection application [START_REF] Zacheilas | Dynamic load balancing techniques for distributed complex event processing systems[END_REF] which processes the Twitter stream and detects first story events. Our experimental results indicate the benefits of our approach and illustrate that we can automatically determine the compression technique that should be applied during the applications' execution.

Preliminaries and System Model

In this section we provide a brief description of Apache Storm and provide the key parameters of our approach.

Preliminaries

Storm is one of the most widely used DSPS mainly because it provides low endto-end tuples' latency and is widely used in a wide range of application domains including traffic monitoring [START_REF] Zacheilas | Elastic Complex Event Processing exploiting Prediction[END_REF] and Twitter analysis [START_REF] Zacheilas | Dynamic load balancing techniques for distributed complex event processing systems[END_REF]. In Storm, the logic of a stream processing application is packaged into a Storm topology; a graph whose nodes are operators that encapsulate the processing logic and edges model data flows among operators. Storm uses a Master-workers architecture where the Master node (i.e., Nimbus) orchestrates the execution of topologies in the available workers. In the Storm terminology nodes are called components and the unit of information that is transferred between components is referred to as a tuple. Users can define two types of components: (i) spouts which are the input sources of the topology, and (ii) bolts that encapsulate the processing logic, performing operations such as filtering, correlating and transforming tuples.

In general, there are two main approaches for processing streaming data either on a per tuple basis or in mini-batches. The first approach has the lowest per tuple latency, as tuples are immediately forwarded to the downstream components. However, this approach can increase the communication cost and thus affects the system's throughput [START_REF] Chintapalli | Benchmarking streaming computation engines: Storm, flink and spark streaming[END_REF]. On the other hand, when mini-batches are used, tuples are stalled until the mini-batch is considered ready for further processing [START_REF] Zaharia | Discretized Streams: Fault Tolerant Streaming Computation at Scale[END_REF]. Usually, time (i.e., 1 second has elapsed since the creation of the mini-batch) or size (e.g., emit the mini-batch when it comprises 50 tuples) based criteria are applied for determining when the mini-batch should be forwarded.

Mini-batches are expected to improve the system's throughput but they add overhead on the per tuple's latency as tuples need to wait until the mini-batch is considered ready for processing. We have enhanced the Storm API to support the processing of mini-batches for two main reasons: (1) mini-batches improve the topology's throughput [START_REF] Das | Adaptive stream processing using dynamic batch sizing[END_REF], and (2) they enable us to parallelize the compression/decompression procedure and thus minimize its overhead on the application's throughput (i.e., see Section 3.2 for more details). Finally, we did not consider frameworks like Apache Spark [START_REF] Zaharia | Discretized Streams: Fault Tolerant Streaming Computation at Scale[END_REF] and Apache Flink [START_REF] Carbone | Apache flink: Stream and batch processing in a single engine[END_REF] that use only mini-batches for the data processing as we wanted to be able to support the per tuple processing offered by Storm to support applications with very strict time requirements like stock market applications.

System Model

In this section we define the parameters of our approach. For each Storm topology the user provides the set of Spouts that comprise the topology's input sources and the set of Bolts which are the topology's processing components. Furthermore, the user provides CompressAlgos which is the set of different compression algorithms that can be applied for compressing the streaming data that are exchanged between the topology's components. In our framework, we support some well-known algorithms (e.g., Snappy) but the user can provide also custom implementations. Finally, the user defines the batchT hr parameter which controls the frequency with which monitor reports will be sent to Nimbus. For example, if this parameter is set to 10 then every 10 processed mini-batches the tasks will send a monitor report to Nimbus.

When the topology processes batchT hr mini-batches, Nimbus receives a monitor report regarding the performance of the compression algorithm c that is currently utilized. More specifically, it receives the throughput c metric which depicts the number of processed tuples per second when c is applied. Based on the compression algorithm that is utilized we end up with different values for this metric as each algorithm has different effect on the communication cost and also on the time required for the compression/decompression procedure.

Furthermore, lossy techniques penalize the accuracy of the application as they modify the input data. For example, when JPG compression with low quality is applied, the image degrades and some of its characteristics are not visible. In order to take into account this fact, we also measure the number of false positive (i.e., f alseP os c ) and false negative (i.e., f alseN eg c ) events when batchT hr mini-batches have been processed by a lossy compression algorithm c. Moreover, we compute the number of true positive (i.e., trueP os c ) and true negative events (i.e., trueN eg c ). In our traffic monitoring application (i.e., described in Section 1), false positive events occur when we mistakenly report that we have traffic congestion. This happens because the image resolution has degraded and the similarity algorithm points out a historical image without traffic congestion as the most similar one. The inverse problem occurs with the false negative events. For lossless techniques, the f alseP os c and f alseN eg c metrics are equal to zero as these techniques guarantee the perfect reconstruction of the input data and thus when they are applied we detect the exact same events as with the case of no compression.

Methodology

In this section we describe the basic components of our framework.

Profiling

The first component of our approach is the use of profiling for determining if compression should be applied and which algorithm to use. Profiling techniques are commonly used for determining the appropriate configuration parameters to be utilized in distributed processing systems [START_REF] Das | Adaptive stream processing using dynamic batch sizing[END_REF]. For example profiling has been efficiently applied in the context of Spark applications to determine the appropriate number of nodes that should be allocated to the applications [START_REF] Venkataraman | Ernest: efficient performance prediction for large-scale advanced analytics[END_REF]. The benefit is that they can capture the system conditions (e.g., throughput) which can be extremely useful in streaming environments like the one we consider where conditions vary over time.

Our technique receives as input a list of the possible compression algorithms and a threshold on the number of mini-batches that will be processed by each algorithm. In this work we consider three well-known lossless compression techniques (i.e, Zip, Snappy and LZ4 ) and one lossy compression algorithm (i.e, JPEG). The JPEG algorithm is examined only when we have image data to be processed while the other techniques are used mainly in case of text data. Moreover, JPEG compression depends on the quality metric that determines the compression ratio, so when this compression algorithm is considered the profiling algorithm receives as input the quality metrics that need to be examined. For example, if we want to evaluate three possible values for the quality metric (i.e., 0.1, 0.5, 0.9) then the profiling algorithm will consider three different variations of the JPEG algorithm, one for each quality metric.

The basic idea of our algorithm is to execute for a fixed number of minibatches each possible configuration (i.e., compression algorithm), compute the throughput and the results' accuracy for each compression technique, and then choose the most appropriate technique. Moreover, the profiling technique also examines these two metrics when no compression algorithm is utilized in order to determine whether compression is beneficial. For the results' accuracy we take into account the fact that lossy compression algorithms can lead to false positive and false negative events (as we indicated in Section 1). More formally, we compute the results' accuracy via the following Formula: accc = trueP osc + trueN egc trueP osc + trueN egc + f alseP osc + f alseN egc , ∀c ∈ CompressAlgos (1)

The acc c metric captures the percentage of false positives and false negatives produced when c is utilized. When lossless techniques are applied those two metrics are equal to zero therefore we have accurate results as acc c equals 1. However, when lossy compression algorithms are considered we expect a decrease in the accuracy as more false positives and false negatives events will be reported and thus the denominator in Equation 1 will increase. In contrast, we expect better throughput when lossy techniques are applied as smaller-size data are transferred between the components. To balance the throughput with the results' accuracy, we introduce a utility score function as follows:

utilityc = throughputc * w + accc * (1 -w), ∀c ∈ CompressAlgos (2)
where w ∈ [0, 1] is a weight given by the user based on where he wants to put more emphasis, i.e., the application's throughput or the accuracy of the results. Furthermore, throughput c is normalized so that it is in the same range as acc c . The goal of our profiling algorithm is to identify the compression algorithm that maximizes Equation 2. The algorithm runs in the Nimbus node as it requires information from all the topology's components. More specifically, our technique consists of the following steps:

1. Initially the profiling algorithm sends to the topology's spouts all the compression algorithms that need to be examined and a threshold on the number of mini-batches to emit for each compression algorithm. 2. The algorithm waits for the monitor reports (i.e., described in Section 2.2) from the bolts that receive and process the data. 3. Upon receiving all reports, the profiling algorithm computes the utility scores of the different compression algorithms using Formula 2. Then it detects the compression algorithm that has the largest utility score (i.e., this indicates that the compression algorithm balances better the trade-off between throughput and accuracy) and informs the spouts that they should use this compression technique from now on.

In order to take into account dynamic changes in the system's condition, we keep monitoring the utility score and if we observe that its value drops significantly we re-apply our profiling algorithm. The technique we used for determining when the profiling algorithm should be re-applied is described in Algorithm 1. Nimbus continues to receive monitor reports and computes their utility scores. When a certain number of reports have been gathered, we compare them with a previous report in terms of utility scores. More specifically, we compute the tan of the line that is drawn between the utility scores of the new and the previous report (i.e., Line 5 in Algorithm 1). We use the tan metric because it provides an indication of how much the utility score has changed (decreased or increased) from the previous report [START_REF] Boutsis | Location privacy for crowdsourcing applications[END_REF]. If the computed tan is positive we have an increase Algorithm 1 Trigger Algorithm 1: Input: prevT an: the previous tan that we have computed, prevReport: the previous utility score, monitorReports: the monitor reports that will be checked for a significant decrease in their utility scores, θ: the threshold used for determining when we have a significant decrease. 2: Output: applyAlgorithm: a Boolean variable that will determine if the profiling algorithm should be re-applied. 3: applyAlgorithm ← true 4: for (report ∈ monitorReports) do 5:

tan ← report.getU tility()-prevReport.getU tility() report.getT ime()-prevReport.getT ime()

6: if (tan > 0 || | tan-prevT an prevT an | < θ) then 7:
applyAlgorithm ← f alse 8:

prevReport ← report 9:

prevT an ← tan 10:

return applyAlgorithm 11: prevT an ← tan 12: prevReport ← monitorReports.getLast() 13: return applyAlgorithm in the utility score and thus we stop the search as there is no point to re-apply the profiling algorithm. However, if it is negative (i.e., the utility score has decreased) we must examine whether the difference between the computed tan and the previous one (i.e., prevT an) exceeds a user-determined threshold θ. More formally, we examine whether the following condition is true:

| tan -prevT an prevT an | > θ (3)
If the condition is true for all the reports that Nimbus has received, then we re-apply the profiling algorithm. Essentially, the tan difference helps us identify whether the utility score has decreased from its previous value significantly (i.e., based on the θ parameter). The performance of the algorithm depends on the θ threshold and the number of monitor reports that we consider. Using a large threshold we expect fewer re-invocations of the profiling algorithm as it will be harder to satisfy the reinvocation condition (i.e., Equation 3). Furthermore, the number of reports that we use in Algorithm 1 also affects the re-invocations as when we use multiple reports it will be harder to satisfy the condition in all of them so we expect less re-invocations of the profiling algorithm. We evaluated how these two parameters affect the re-invocations in our experimental evaluation (i.e., see Section 4). In general, we observed that we need at least three reports to be sure that we have a significant change in the utility score and thus re-applying the profiling technique is beneficial. In contrast, using only one report leads to multiple invocations of the profiling algorithm due to short-term fluctuations of the utility score and this can penalize the application's throughput as when the profiling algorithm is re-applied we have to examine again all the compression algorithms.

Parallel Data Compression/Decompression

In Figure 4 we illustrate how the compression algorithms are applied on the mini-batches to be processed. The benefit of utilizing mini-batches is that they enable us to parallelize the compression/decompression procedure. We use multiple threads that run in parallel to perform the decompression whenever the processing component receives a mini-batch. The decompressed data are kept in a mini-batch which will be used for feeding the component with the input tuples. A similar procedure is followed when we compress the data after the output mini-batch is complete (i.e., reached the required batch size) and needs to be emitted to downstream components. In this case the mini-batch is forwarded to multiple concurrently running threads for the compression.

The number of compression/decompression threads can be provided by the user. In most cases the number of threads will be fewer than the number of tuples in the mini-batch so we have to distribute the tuples that comprise it to the available threads. In order to balance the work among the threads we follow a round-robin approach for assigning the tuples to the compression/decompression threads. More formally, assuming that we have N threads, T tuples in the mini-batch and each tuple t in the mini-batch has an id id t ∈ [0, T -1] then the thread that will compress tuple t is computed as: thrId = id t %N , where thrId will be the id of the thread that will be responsible for compressing (or decompressing) tuple t. This simple technique balances the load between the threads and minimizes the compression/decompression overhead.

Implementation and Evaluation

Implementation. We have implemented our framework1 as a module of Apache Storm and in Figure 4 we illustrate the components we have added in order to support dynamic data compression. More specifically, we have added two extra components on Nimbus that are responsible for auto-tuning the compression algorithm that is used by the topology's processing components. The P rof iling component is responsible for invoking the profiling algorithm described in Section 3.1 and for adjusting the compression algorithm of the Storm components' (spouts or bolts). The second component (i.e., M onitor in Figure 4) monitors the performance of all the topology's components. More specifically, M onitor receives reports from the bolts' tasks whenever the threshold of processed minibatches has been reached (i.e., batchT hr parameter in Section 2.2), computes the utility score (i.e., Equation 2) and informs the P rof iling component about the new value.

In order to be able to exploit the auto-compression features we offer, users must extend two abstract classes. More specifically, the CompressionSpout class must be extended by the users' spouts while the CompressionBolt class must be extended by the users' bolts. Each instance of these classes uses a special thread for receiving the compression algorithm that should be applied. Furthermore, these classes create the compression/decompression threads (i.e., see Section 3.2) that are used for minimizing the computation overhead. Users still have to provide the implementation of nextT uple (i.e., for spouts) and execute (i.e., for bolts) methods in order to be able to utilize their processing components in their topologies. Evaluation Setup. We have evaluated our approach in our local 8 nodes cluster. Each node had attached 8 CPU processors and 16 GB RAM. All nodes were connected to the same LAN and their clocks were synchronized using the NTP protocol. We implemented our proposals on top of Storm 0.10.2 and used a dedicated Nimbus node to avoid overloading one of the nodes. We considered the following applications for examining the performance of our approach:

-Traffic Monitoring Application: This application receives as input live CCTV data from Dublin city and detects events by invoking a simple image similarity algorithm (supplied by LIRE framework [START_REF] Lux | Lire: lucene image retrieval: an extensible java cbir library[END_REF]) against historical images depicting normal and abnormal traffic conditions. More specifically, for each incoming image we find the most similar historical image and based on the characterization of the latter we inform the traffic authorities. -URL Crawling: The application crawls web pages and detects keywords in them. More specifically, we crawl well-known sites (e.g., https://news. google.ie/) and try to detect events like accidents in Dublin city. -Twitter First Story Detection (Twitter FSD) [START_REF] Zacheilas | Dynamic load balancing techniques for distributed complex event processing systems[END_REF]: This application retrieves data from the Twitter Streaming API and detects tweets that correspond to new events (e.g., a traffic accident in Dublin city).

Our goal was to consider applications that process both image and text data. The difference between the two text processing applications is the size of the data. For the URL crawling application we have larger input data (i.e., approximately 600 kB) therefore compression may be beneficial while for the Twitter application the tweets are usually small-sized (i.e., less than 200 bytes) so compression may penalize the application's performance. Performance of parallel compression/decompression. In the first set of experiments we illustrate the benefits of performing the parallel compression/decompression procedure. More specifically, we compare our multi-threaded compression technique varying the number of threads. For the traffic monitoring application we used JPEG compression with quality 0.1 while for the other two applications we used Zip compression (similar results were observed with the other compression algorithms but we do not display them due to lack of space). As can be observed in Figures 5(a), 5(b) and 5(c) using more threads improves throughput as long as we do not exceed the available CPU cores on the nodes. We argue that using at maximum four threads for the compression/decompression procedure is a valid choice when we have 8 core nodes in the cluster. Furthermore, we wanted to evaluate how the compression overhead (i.e., in terms of execution time) is affected by the size of the data that we want to compress/decompress. We used synthetic test data for this experiment and we applied the Zip compression algorithm. As we illustrate in Figure 6, our multi-threaded technique (using 4 compression/decompression threads) is able to keep the overhead low (i.e., less than 6 seconds) even when we consider data larger than 8 MB.

Profiling re-invocation. In the second set of experiments (i.e., Figures 7 and8) we perform a sensitivity analysis on the two parameters that affect the performance of our proposed re-triggering mechanism (i.e., Algorithm 1 in Section 3.1). We considered as input the utility scores of the Twitter FSD application when no compression algorithm is applied (see the Default approach in Figure 9(c)) and we examined how the number of monitor reports and the θ parameter affect the number of times that the profiling algorithm will be re-applied. As we illustrate in Figures 7 and8 both parameters influence the number of times we re-invoke the profiling algorithm with the one that affects the most being the number of monitor reports to consider for the evaluation. In we set θ to 0.5 while in Figure 8 we used 2 reports. As it can be observed in Figure 7, if we use more than 4 reports the number of re-invocations is minimized. In contrast, when we use only one monitor report for the comparison, we end up invoking constantly the profiling algorithm. For the rest of the experiments we set θ to 0.5 and use 3 previous reports. Dynamic compression evaluation. In the last set of experiments we evaluated the applicability of our framework to detect the correct compression algorithm that should be utilized. In Figures 9(a), 9(b) and 9(c) we illustrate the utility score when our framework is applied against the default approach (i.e., Default in the Figures) that does not apply any compression algorithm. The weight of the utility score was set to 0.5. As it can be observed, DIsCO is able to maximize the utility score in all applications. For example, in case of the traffic monitoring application our framework exploits JPEG compression and it is able to guarantee that the impact on the results' accuracy will be minimal. DIsCO decides to compress the images using the JPEG compression algorithm with quality 0.3. So we have less than 200 false positives and negative events as it can be observed in Figure 2 in Section 1.

For the other two applications the utility score depends solely on the throughput metric as we considered only lossless compression techniques (i.e., for this reason the utility score in Figures 9(b) and 9(c) is not scaled between 0 and 1). The URL crawling application exploits the Snappy compression algorithm which is able to achieve better throughput than the Default approach. In contrast, in the Twitter FSD application, DIsCO detects that it is not beneficial to use a compression algorithm as the improvements in the communication cost are negligible compared to the time required for compressing/decompressing the tweets. However, because DIsCO samples all the possible compression algorithms, in this case the Default approach has better throughput.

Related Work

In recent years we have observed a plethora of stream processing frameworks including Spark [START_REF] Zaharia | Discretized Streams: Fault Tolerant Streaming Computation at Scale[END_REF] and Flink [START_REF] Carbone | Apache flink: Stream and batch processing in a single engine[END_REF]. Despite their popularity, there has been little work in exploiting data compression. One recent proposal comes from the Spark community with the implementation of a novel distributed filesystem called Succint [START_REF] Agarwal | Succinct: Enabling queries on compressed data[END_REF] which keeps the data in a compressed form using Arrays of Suffixes (AoS) and enables the processing of the data in this form. However, the wide adoption of such system requires time as the majority of the data are stored in either HDFS or distributed databases like MongoDB or Cassandra. So an approach like ours that performs on-the-fly compression on the data streams is beneficial to the system's performance.

Previous work exploiting compression for improving the energy efficiency in Hadoop clusters was done in [START_REF] Chen | To compress or not to compress-compute vs. io tradeoffs for mapreduce energy efficiency[END_REF]. Our work differs in, that, we examine the problem in a distributed stream processing setting trying to maximize the observed system throughput. Authors in [START_REF] Eberle | Online unsupervised state recognition in sensor data[END_REF] propose the use of an unsupervised learning technique for detecting patterns in the incoming data flow and minimize the amount of emitted tuples by not transmitting tuples that will not contribute on the query's results. In our approach we decided to avoid such load shedding techniques to minimize the information loss. Novel frameworks like [START_REF] Yang | Druid: a real-time analytical data store[END_REF] and [START_REF] Hu | Elf: efficient lightweight fast stream processing at scale[END_REF] have been proposed for performing analysis on streaming data using a compressed representation of the input dataset. Authors in [START_REF] Yang | Druid: a real-time analytical data store[END_REF] propose the use of dictionaries for keeping the incoming data and process them in this compressed form while in [START_REF] Hu | Elf: efficient lightweight fast stream processing at scale[END_REF] the authors describe the use of compressed buffer trees (CBTs) for keeping the in-memory data. We could exploit such techniques in DIsCO as users can easily plugin their custom compression algorithms.

Conclusions

In this paper we present DIsCO, a novel auto-tuning framework on top of the Apache Storm whose goal is to balance the trade-off between application's throughput and results' accuracy by dynamically deciding whether data compression would be useful in the streaming applications and which compression algorithm would be the most appropriate. Our approach is able to efficiently adjust the compression algorithm to be utilized during the application's execution, and by exploiting the use of mini-batches it can further maximize the system's throughput and minimize the compression/decompression overhead. In our experimental evaluation on our local cluster using well-known stream processing applications, we demonstrate the benefits of our approach and illustrate a clear improvement on the applications' performance.

Fig. 2 . 8 Fig. 3 .

 283 Fig. 2. JPEG's quality metric impact on the results' accuracy.

Fig. 4 .

 4 Fig. 4. Implementation Details.

Fig. 6 .Fig. 7 .

 67 Fig.6. Impact of data size on compression/decompression time.

Fig. 8 .

 8 Fig. 8. Impact of θ on the number of profiling algorithm's invocations.

Fig. 9 .

 9 Utility scores comparison.

http://rtds.aueb.gr/index.php/software/

Acknowledgment

This research has been financed by the European Union through the FP7 ERC IDEAS 308019 NGHCS project and the Horizon2020 688380 VaVeL project.