Nicoló Rivetti
email: nicolo.rivetti@gmail.com

Yann Busnel
email: yann.busnel@imt-atlantique.fr

Leonardo Querzoni
email: querzoni@diag.uniroma1.it

Load-Aware Shedding in Stream Processing Systems

Keywords: Load-Shedding, Stream Processing, Data Streaming, Distributed systems

Distributed stream processing systems are today gaining momentum as a tool to perform analytics on continuous data streams. Load shedding is a technique used to handle unpredictable spikes in the input load whenever available computing resources are not adequately provisioned. In this paper, we propose Load-Aware Shedding (LAS), a novel load shedding solution that, unlike previous works, does not rely neither on a pre-defined cost model nor on any assumption on the tuple execution duration. Leveraging sketches, LAS efficiently estimates the execution duration of each tuple with small error bounds and uses this knowledge to proactively shed input streams at any operator to limiting queuing latencies while dropping as few tuples as possible. We provide a theoretical analysis proving that LAS is an (ε, δ)-approximation of the optimal online load shedder. Furthermore, through an extensive practical evaluation based on simulations and a prototype, we evaluate its impact on stream processing applications.

Introduction

Distributed stream processing systems (DSPS) and Complex Event Processing (CEP) are today considered as a mainstream technology to build architectures for the real-time analysis of big data. An application running in a DSPS, or a query executed by a CEP engine, is typically modeled as a directed acyclic graph (a topology) where data operators, represented by nodes, are interconnected by streams of tuples containing data to be analyzed, the directed edges. The success of such systems can be traced back to their ability to run complex applications at scale on clusters of commodity hardware or in the cloud.

Correctly provisioning computing resources for DSPS or CEP engines however is far from being a trivial task. System designers need to take into account several factors: the computational complexity of the operators, the overhead induced by the framework, and the characteristics of the input streams. This latter aspect is often the most critical, as input data streams may unpredictably change over time both in rate and in content. Over-provisioning is not economically sensible, thus system designers are today moving toward approaches based on elastic scalability [START_REF] Heinze | Cloud-based data stream processing[END_REF], where an underlying infrastructure can tune at runtime the available resources in response to changes in the workload. This represents a desirable solution when coupled with on-demand provisioning offered by many cloud platforms, but still may be affected by transient overloads [START_REF] Borkowski | Minimizing cost by reducing scaling operations in distributed stream processing[END_REF], caused for example by unexpected load spikes, that could temporarily degrade performance below the desired SLA.

Bursty input load represents a problem for both DSPS and CEP engines as it may create unpredictable bottlenecks within the system that lead to an increase in queuing latencies, pushing the system in a state where it cannot deliver the expected quality of service (typically expressed in terms of tuple completion latency). Load shedding is generally considered a practical approach to handle bursty traffic. It consists of dropping a subset of incoming tuples as soon as a bottleneck is detected in the system. As such, load shedding is a solution that is complementary [START_REF] Stanoi | Whitewater: Distributed processing of fast streams[END_REF] and must coexist with resource shaping techniques (like elastic scaling), rather than being an alternative.

Existing load shedding solutions either randomly drop tuples when bottlenecks are detected [START_REF] Abadi | Aurora: a new model and architecture for data stream management[END_REF] or apply a pre-defined model of the application and its input that allows them to deterministically take the best shedding decision [START_REF] Tatbul | Staying fit: Efficient load shedding techniques for distributed stream processing[END_REF]. In any case, all the existing solutions assume that incoming tuples all impose the same computational load. However, such assumption does not hold for many practical use cases; tuple execution duration, in fact, may depend on the tuple content itself. This is often the case whenever the receiving operator implements a logic with branches where only a subset of the incoming tuples travels through every single branch. If the computation associated with each branch generates different loads, then the execution duration will change from tuple to tuple. A tuple with a large execution duration may delay the execution of subsequent tuples in the same stream, thus increasing queuing latencies. If further tuples are enqueued with large execution durations, this may bring to the emergence of a bottleneck.

As an example, consider the reach of a tweet, i.e., the number of users that may receive the re-tweets of a given tweet. This computation entails counting the number of users that have a direct and un-direct follower relationship (until a given depth) with the tweet author. Then, depending on the size of the subgraph rooted in the author node, the execution times vary. For instance, in our experiments the execution time belongs to the interval [0.01, 70] ms, the most frequent execution time was 65 ms, while the average per execution time was 20 ms. In the experimental evaluation, we provide a second use-case exhibiting the same phenomena.

Based on this simple observation, we introduce Load-Aware Shedding (LAS), a novel solution for load shedding in DSPS (or CEP engines) engines. LAS gets rid of the aforementioned assumptions and provides efficient shedding aimed at matching given queuing latency targets while dropping as few tuples as possible.

To reach this goal LAS leverages a smart combination of sketch data structures to efficiently collect at runtime information on the time needed to compute tuples. This information is used to build and maintain, at runtime, a cost model that is then exploited to take decisions on when input tuples must be shed. LAS has been designed as a flexible solution that can be applied on a per-operator basis, thus allowing developers to target specific critical stream paths in their applications. The proposed solution provides predictable per operator queuing latencies, an extremely important feature in several application scenarios where the stream processing system is expected to deliver results to users in a quasi-real-time fashion. Furthermore, LAS implements an efficient load shedding solution that perfectly fits the characteristics of settings where scarce resources are available (e.g. fog-computing). Finally, LAS can be complemented by an output quality model that allows blocking it from dropping tuples that may significantly degrade the final output quality.

The contributions provided by this paper are:

-the introduction of LAS, the first solution for load shedding in DSPS (or CEP engines) that proactively drops tuples to avoid bottlenecks without requiring a predefined cost model and without any assumption on the distribution of tuples; -a theoretical analysis of LAS that points out how it is an ( , δ)-approximation of the optimal online shedding algorithm; -an experimental evaluation that illustrates how LAS can provide predictable queuing latencies that approximate a given threshold while dropping a small fraction of the incoming tuples.

Below, the next section states the system model we consider. Afterward, Section 3 details LAS whose behavior is then theoretically analyzed in Section 4. Section 5 reports on our experimental evaluation and Section 6 analyzes the related works. Finally, Section 7 concludes the paper.

System Model and Problem Definition

We consider a distributed stream processing system (DSPS) or Complex Event Processing (CEP) engine deployed on a cluster where several computing nodes exchange data through messages sent over a network. The stream processing application (or query) executed by the DSPS (or CEP engine) can be represented by a topology: a directed acyclic graph interconnecting operators, represented by vertices, with data streams (DS), represented by edges. Each topology contains at least a source, i.e., an operator connected only through outbound DSs, and a sink, i.e., an operator connected only to inbound DSs. Data injected by the source is encapsulated in units called tuples (or events) and each data stream is an unbounded sequence of tuples. Without loss of generality, here we assume that each tuple t is a finite set of key/value pairs that can be customized to represent complex data structures. To simplify the discussion, in the rest of this work, we deal with streams of unary tuples each representing a single non-negative integer value.

For the sake of clarity, and without loss of generality, here we restrict our model to a topology with an operator LS (load shedder ) that decides which tuples of its outbound DS σ consumed by a downstream operator O shall be dropped. The actual positioning of LS within a real topology may be tuned, depending on where bottlenecks are expected to appear within the topology itself. Nevertheless, we assume that LS is never deployed as a source or sink in any topology. Tuples in σ are drawn from a large universe [n] = {1, . . . , n} and are ordered, i.e., σ = t 1 , . . . , t m . Therefore [m] = 1, . . . , m is the index sequence associated with the m tuples contained in the stream σ. Both m and n are unknown. We denote with f t the unknown frequency of tuple t, i.e., the number of occurrences4 of t in σ.

We assume that the execution duration of tuple t on operator O, denoted as w(t), depends on the content of the tuple t. We simplify the model assuming that w depends on a single, fixed and known attribute value of tuple t. Cases in which this assumption does not hold, e.g. w depends on multiple attributes can be simply treated by concatenating their values and considering them as a sin-gle multiplexed attribute [START_REF] Dobra | Sketch-based multi-query processing over data streams[END_REF][START_REF] Cormode | Sketch techniques for approximate query processing[END_REF][START_REF] Kammoun | Enhancing Stream Processing and Complex Event Processing Systems[END_REF]. The probability distribution of such attribute values, as well as the function w are unknown, may differ from operator to operator and may change over time. However, we assume that subsequent changes are interleaved by a large enough time frame such that an algorithm may have a reasonable amount of time to adapt. On the other hand, the input throughput of the stream may vary, even with a large magnitude, at any time.

Let q(i) be the queuing latency of the i-th tuple of the stream, i.e., the time spent by the i-th tuple in the inbound buffer of operator O before being processed. Let us denote as D(j) ⊆ [j], j ≤ m, the set of dropped tuples in a stream of length m, i.e., dropped tuples are thus represented in D(j) by their indices in [j] ⊆ [m]. Moreover, let d(j) ≤ j ≤ m be the number of dropped tuples in a stream prefix of length j, i.e., d(j) = |D(j)|. Then we can define the average queuing latency as:

Q(j) = i∈[j]\D(j) q(i)/(j -d(j)) for all j ∈ [m].
The goal of the load shedder is to maintain at any point in the stream the average queuing latency smaller than a given threshold τ by dropping as few tuples as possible. The quality of the shedder can be evaluated both by comparing the resulting Q(j) against τ and by measuring the number of dropped tuples d(j). More formally, the load shedding problem can be defined as follows 5 .

Problem 1 (Load Shedding). Given a data stream σ = t 1 , . . . , t m , find the smallest set D(j) such that

∀j ∈ [m] \ D(j), Q(j) ≤ τ.

Load Aware Shedding

This section introduces the Load-Aware Shedding algorithm by first providing an overview, then detailing some background knowledge, and finally describing the details of its functioning.

Overview

Load-Aware Shedding (LAS) is based on a simple, yet effective, idea: if we assume to know the execution duration w(t) of each tuple t on the operator, then we can foresee the queuing time for each tuple of the operator input stream and then drop all tuples that will cause the queuing latency threshold τ to be violated. However, the value of w(t) is generally unknown. A possible solution to this problem is to build a static cost model for tuple execution duration and then use it to proactively shed load. However, building an accurate cost model usually requires a large amount of a priori knowledge on the system. Furthermore, once a model has been built, it can be hard to handle changes in the system or input stream characteristics at runtime. LAS overcomes these issues by building and maintaining at run-time a cost model for tuple execution durations. It takes shedding decision based on the estimation C of the total execution duration of the operator: C = i∈[m]\D(m) w(t i ). To do so, LAS computes an estimation ŵ(t) of the execution duration w(t) of each tuple t. Then, it computes the sum of the estimated execution durations of the tuples assigned to the operator, i.e., C = i∈[m]\D(m) ŵ(t). At the arrival of the i-th tuple, subtracting from C the (physical) time elapsed from the emission of the first tuple provides us with an estimation q(i) of the queuing latency q(i) for the current tuple.

To enable this approach, LAS builds a sketch on the operator (i.e., a memory efficient data structure) that will track the execution duration of the tuples it processes. Using a sketch allows LAS to efficiently track this data independently from the amount of possibly different tuples handled by the operator. When a change in the stream or operator characteristics affects the tuples execution durations w(t), i.e., the sketch content changes, the operator will forward an updated version to the load shedder, which will then be able to (again) correctly estimate the tuples execution durations. This solution does not require any a priori knowledge on the stream or system and is designed to continuously adapt to changes in the input stream or on the operator characteristics.

Shedding tuples from an incoming stream has in general a negative impact on the stream processing output quality. LAS approach is focussed on discarding tuples whose contribution to operator overload is larger, independently from their content. This approach is meaningful only under the assumption that the contribution to the stream output is the same for each input tuple. Applications, where this assumption does not hold, can be managed in LAS by building up a model for output degradation caused by shedding and then using this model to check for any candidate tuple if its contribution to the output quality is compatible with a given constraint.

Background

2-Universal Hash Functions -Our algorithm uses hash functions randomly picked from a 2-universal hash functions family. A collection H of hash functions h : {1, . . . , n} → {0, . . . , c} is said to be 2-universal if for every two different items x, y ∈ [n], for any h ∈ H, P{h(x) = h(y)} ≤ 1 c , which is the probability of collision obtained if the hash function assigned truly random values to any x ∈ [n]. Carter and Wegman [START_REF] Carter | Universal classes of hash functions[END_REF] provide an efficient method to build large families of hash functions approximating the 2-universality property.

Count Min sketch algorithm -Cormode and Muthukrishnan have introduced in [START_REF] Cormode | An improved data stream summary: The count-min sketch and its applications[END_REF] the Count Min sketch that provides, for each item t in the input stream an (ε, δ)-additive-approximation ft of the frequency f t . The Count Min sketch consists of a two-dimensional matrix F of size r × c, where r = log 1 it updates each row: ∀i ∈

[r], F[i, h i (t)] ← F[i, h i (t)] + 1.
Thus, the cell value is the sum of the frequencies of all the items mapped to that cell. Upon request of f t estimation, the algorithm returns the smallest cell value among the cells associated with t:

ft = min i∈[r] {F[i, h i (t)]}.
Fed with a stream of m items, the space complexity of this algorithm is O( 1 ε log 1 δ (log m + log n)) bits, while update and query time complexities are O(log 1/δ). The Count Min algorithm guarantees that the following bound holds on the estimation accuracy for each item read from the input stream: P{| ft -

f t | ≥ ε(m -f t )} ≤ δ, while f t ≤ ft is always true.
This algorithm can be easily generalized to provide (ε, δ)-additive-approximation of point queries on a stream of updates, i.e., a stream where each item t carries a positive integer update value v t . When the Count Min algorithm reads the pair t, v from the input stream, the update routine changes as follows: ∀i ∈

[r], F[i, h i (t)] ← F[i, h i (t)] + v.

LAS design

The operator stores two Count Min sketch matrices (Figure 1.A): the first one, denoted as F, tracks the tuple frequencies f t ; the second one, denoted as W, tracks the tuple cumulated execution durations W t = w(t) × f t . Both Count Min matrices share the same sizes, controlled by parameters ε and δ, and hash functions. The latter is the generalized version of the Count Min (Section 3.2) where the update value is the tuple execution duration when processed by the instance (i.e., v = w(t)). The operator updates (Listing 3.1 lines 24-27) both matrices after each tuple execution.

The operator is modeled as a finite state machine (Figure 2) with two states: START and STABILIZING. The START state lasts as long as the operator has executed N tuples, where N is a user defined window size parameter. The transition to the STABILIZING state (Figure 2.A) triggers the creation of a new snapshot S. A snapshot is a matrix of size r × c where ∀i ∈ [START_REF] Kammoun | Enhancing Stream Processing and Complex Event Processing Systems[END_REF][START_REF] Katsipoulakis | Concept-driven load shedding: size and error of voluminous and variable data streams[END_REF]. We say that the F and W for i = 1 to r do 26:

[r], j ∈ [c] : S[i, j] = W[i, j]/F[i, j] (Listing 3.1 lines
F[i, hi(t)] ← F [i, hi(t)] + 1 27: W[i, hi(t)] ← W[i, hi(t)] + l 28:
end for 29: end function matrices are stable when the relative error η between the previous snapshot and the current one is smaller than a parameter µ, i.e.,

η = ∀i,j |S[i, j] -W[i,j] F [i,j]) | ∀i,j S[i, j] ≤ µ (1)
is satisfied. Then, each time the operator has executed N tuples (Listing 3.1 lines 17-23), it checks whether Equation 1 is satisfied. (i) In the negative case S is updated (Figure 2.B). (ii) In the positive case, the operator sends the F and W matrices to the load shedder (Figure 1.B), resets their content, and moves back to the START state (Figure 2.C). This approach allows to limit the amount of data sent from the operator to LS, and resembles what was proposed in [START_REF] Ilarri | A query processor for prediction-based monitoring of data streams[END_REF].

There is a delay between any change in w(t) and when LS receives the updated F and W matrices. This introduces a skew in the cumulated execution duration estimated by LS. To compensate this skew, we introduce a synchronization mechanism that kicks in whenever the LS receives a new pair of matrices from the operator. The LS (Figure 1.C) maintains the estimated cumulated execution duration of the operator C and a pair of initially empty matrices F, W . LS is modeled as a finite state machine (Figure 3) with three states: NOP, SEND, and RUN. The LS executes the code reported in Listing 3.2. In particular, every time a new tuple t arrives at the LS, the function shed is executed. The LS starts in the NOP state where no action is performed (Listing 3.2 lines 15-17). Here we assume that in this initial phase, i.e., when the topology has just been deployed, no load shedding is required. When LS receives the first pair F, W of matrices (Figure 3.A), it moves into the SEND state and updates its local pair of matrices (Listing 3.2 lines 7-9). While being in the SEND states, LS sends to O the current cumulated execution duration estimation C (Figure 1.D) piggybacking it with the first tuple t that is not dropped (Listing 3.2 lines 22-24) and moves in the RUN state (Figure 3.B). This information is used to synchronize the LS with O and remove the skew between O's cumulated execution duration C and the estimation C at LS. O replies to this request (Figure 1.E) with the difference ∆ = C -C (Listing 3.1 lines 11-13). When the load shedder receives the synchronization reply (Figure 3.C) it updates its estimation C + ∆ (Listing 3.2 lines 11-13).

In the RUN state, the load shedder computes, for each tuple t, the estimated queuing latency q(i) as the difference between the operator estimated execution duration C and the time elapsed from the emission of the first tuple if state = NOP then 16:

return f alse 17: q ← C-elapsed time from first tuple 18:

if CheckLatency(q) ∧ CheckUtility(t) then 19:

return true 20:

i ← arg min i∈[r] {F[i, hi(t)]} 21: C ← C + (W[i, hi(t)]/F[i, hi(t)]) × (1 + ε) 22: if state = Send then Figure 3.B 23:
piggyback C to operator on t 24:

state ← Run 25:

return f alse 26: end function 27: function CheckLatency(q) 28:

if (Q + q)/ > τ then 29:

return true 30:

Q ← Q + q 31:
← + 1 32:

return f alse 33: end function (Listing 3.2 line 17). It then checks if the estimated queuing latency for t satisfies the CheckLatency method (Listing 3.2 line 18). This method encapsulates the logic for checking if a desired condition on queuing latencies is violated or not. In this paper, as stated in Section 2, we aim at maintaining the average queuing latency below a threshold τ . Then, Check-Latency tries to add q to the current average queuing latency (Listing 3.2 lines 28). If the result is larger than τ (i), it simply returns true; otherwise (ii), it updates its local value for the average queuing latency and returns f alse (Listing 3.2 lines 30-32). Note that different goals, based on the queuing latency, can be defined and encapsulated within CheckLatency, e.g., maintain the absolute per-tuple queuing latency below τ , or maintain the average queuing latency calculated on a sliding window below τ [START_REF] Rivetti | Efficiently summarizing data streams over sliding windows[END_REF].

Function CheckUtility evaluates the impact the output quality would incur by dropping t. This function encapsulates optional requirements on the maximum acceptable quality drop as defined by the semantics of the application. Considering that the quality definition is application dependent, we don't provide here a specific implementation. However, we assume that, independently of the implementation, it will return true if the t can be dropped with an acceptable quality loss.

If both CheckLatency(q) and CheckUtility(t) return true (i) the load shedder returns true as well, i.e., tuple t must be dropped. Otherwise (ii), the operator estimated execution duration C is updated with the estimated tuple execution duration ŵ(t), increased by a factor 1 + ε to mitigate potential underestimations 6 , and the load shedder returns f alse (Listing 3.2 line 25), i.e., the tuple must not be dropped. Finally, if the load shedder receives a new pair F, W of matrices (Figure 3.D), it will update its local pair of matrices and move to the SEND state (Listing 3.2 lines 7-9). Now we will discuss the complexity of LAS. 7Theorem 1 (Time complexity of LAS).

For each tuple read from the input stream, the time complexity of LAS for the operator and the load shedder is O(log 1/δ).

Theorem 2 (Space Complexity of LAS).

The space complexity of LAS for the operator and load shedder is

O 1 ε log 1 δ (log m + log n) bits.

Theorem 3 (Communication complexity of LAS).

The communication complexity of LAS is of O m N messages and

O m N 1 ε log 1 δ (log m + log n) + log m bits.
Note that the communication cost is low with respect to the stream size since the window size N should be chosen such that N 1 (e.g., in our tests we have N = 1024).

Theoretical Analysis

This section provides an analysis of the quality of the shedding performed by LAS in two steps. First, we study the correctness and optimality of the shedding algorithm, under full knowledge assumption (i.e., the shedding strategy is aware of the exact execution duration w t for each tuple t). Then, in Section 4.2, we provide a probabilistic analysis of the mechanism that LAS uses to estimate the tuple execution durations. For the sake of simplicity, in both sections, we assume CheckUtility always returns true. The proofs of the theorem are available in Appendix A.

Correctness of LAS

We suppose that tuples cannot be preempted, that is they must be processed uninterruptedly on the available operator instance. As mentioned before, in this analysis we assume that the execution duration w(t) is known for each tuple t. Finally, given our system model, we consider the problem of minimizing d, the number of dropped tuples, while guaranteeing that the average queuing latency Q(t) will be upper-bounded by τ , ∀t ∈ σ. The solution must work online, thus the decision of enqueueing or dropping a tuple has to be made only resorting to knowledge about tuples received so far in the stream.

Let OPT be the online algorithm that provides the optimal solution to Problem 1. We denote with D σ OP T (resp. d σ OP T ) the set of dropped tuple indices (resp. the number of dropped tuples) produced by the OPT algorithm fed by stream σ (cf., Section 2). We also denote with d σ LAS the number of dropped tuples produced by LAS introduced in Section 3.3 fed with the same stream σ.

Theorem 4 (Correctness and Optimality of LAS). For any σ, we have

d σ LAS = d σ OP T and ∀t ∈ σ, Q σ LAS (t) ≤ τ .
This theorem establishes that LAS is optimal, given that its execution time is the same as that of the optimal OPT algorithm. Moreover, it is correct in the sense of the Definition 1 proposed in Section 2, namely that its average queuing latency will not exceed the predetermined threshold τ .

Execution Time Estimation

In this section, we analyze the approximation made on execution duration w(t) for each tuple t when the assumption of full knowledge is removed. LAS uses two matrices, F and W, to estimate the execution time w(t) of each tuple submitted to the operator. By the Count Min sketch algorithm (cf., Section 3.2) and Listing 3.1, we have that for any t ∈ [n] and each row i ∈ [r],

F[i][h i (t)](m) = f t + n u=1,u =t f u 1 {hi(u)=hi(t)} , and
W[i][h i (t)](m) = f t w t + n u=1,u =t f u w u 1 {hi(u)=hi(t)} .
Let us denote respectively by w min and w max the minimum and the maximum execution time of the items. For sake of clarity in the following equations, we denote the ratio

V i,t = W[i][h i (t)]/F[i][h i (t)].
We have trivially

w min ≤ V i,t ≤ w max .
We define S = n =1 w . We then have Theorem 5.

E{V i,t } = S -w t n -1 - k(S -nw t ) n(n -1) 1 -1 - 1 k n .
The proof of this theorem is available in appendix. First, it important to note that this result does not depend on m. Moreover, we easily understand that the formula proposed in this last theorem may seem rather uninformative. Thus, we propose to present a numeric application of it to take the measure of the potential use of it for an end-user.

We take for instance k = 55, n = 4096 and the distinct values of w u equal to 1, 2, 3, . . . , 64, each item being present 64 times in the input stream, we get for t = 1, . . . , 64, E{V i,t } ∈ [32.08, 32.92]. Note also from above that we have

1 ≤ V i,t = W[i][h i (t)]/F[i][h i (t)] ≤ 64.
From the Markov inequality, we have, for every x > 0,

P{V i,t ≥ x} ≤ E{V i,t } x .
By taking x = 64a, with a ∈ [0.6, 1), we obtain

P{V i,t ≥ 64a} ≤ E{V i,t } 64a ≤ 33 64a .
Recall that r denotes the number of rows of the system; we then have by the independence of the h functions, P{ min i=1,...,r

(V i,t ) ≥ 64a} = (P{V i,t ≥ 64a}) r ≤ 33 64a r .
By taking for instance a = 3/4 and r = 10, we get P{ min i=1,...,r

(V i,t )) ≥ 48} ≤ 11 16
10 ≤ 0.024.

Experimental Evaluation

In this section, we evaluate the performance obtained by using LAS to perform load shedding. We first describe the general setting used to run the tests and then discuss the results obtained through simulations (Section 5.2) and with a prototype of LAS integrated within Apache Storm (Section 5.3).

Setup

Datasets -In our tests we consider both synthetic and real datasets. Synthetic datasets are built as streams of integer values (items) representing the values of the tuple attribute driving the execution duration when processed on the operator. We consider streams of m = 32, 768 tuples, each containing a value chosen among n = 4, 096 distinct items. Streams have been generated using the Uniform and Zipfian distributions with different values of α ∈ {0.5, 1.0, 1.5, 2.0, 2.5, 3.0}, denoted respectively as Zipf-0.5, Zipf-1.0, Zipf-1. Let W be the average execution duration of the stream tuples, then the stream maximum theoretical input throughput sustainable by the setup is equal to 1/W . When fed with an input throughput smaller than 1/W the system will be over-provisioned (i.e., possible underutilization of computing resources). Conversely, an input throughput larger than 1/W will result in an underprovisioned system. We refer to the ratio between the maximum theoretical input throughput and the actual input throughput as the percentage of underprovisioning that, unless otherwise stated, was set to 25%.

To generate 100 different streams, we randomize the association between the w n execution duration valuees and the n distinct items: for each of the w n execution duration values, we pick uniformly at random n/w n different values in [n] that will be associated to that execution duration value. This means that the 100 different streams we use in our tests do not share the same association between execution duration and item as well as the association between frequency and execution duration (thus each stream has also a different average execution duration W ). Each of these permutations has been run with 50 different seeds to randomize the stream ordering and the generation of the hash functions used by LAS. This means that each single experiment reports the mean outcome of 5, 000 independent runs.

We considered two types of constraints defined on the queuing latency:

ABS(τ ): requires that the queuing latency per tuple does not exceed τ milliseconds: ∀i ∈ [m] \ D, q(i) ≤ τ . AVG(τ ): requires that the total average queuing latency does not exceed τ mil-

liseconds: ∀i ∈ [m] \ D, Q(i) ≤ τ .
While not being a realistic requirement, the straightforwardness of the ABS(τ ) constraint allowed us to grasp a better insight of the mechanisms of the algorithm. However, in this section, we only show results for the AVG(6.4) constraint as is it a much more sensible requirement with respect to a real setting.

The LAS operator window size parameter N , the tolerance parameter µ and the number of rows of the F and W matrices δ were set to N = 1024, µ = 0.05 and δ = 0.1 (i.e., r = 4 rows) respectively. By default, the LAS precision parameter (i.e., the number of columns of the F and W matrices) was set to ε = 0.05 (i.e., c = 54 columns), however in one of the test we evaluated LAS performance using several values: ε ∈ [0.001, 1.0]. To evaluate LAS performance without other external factors, in all our experiments we set CheckUtility to always return true.

For the real data, we used a dataset containing a stream of preprocessed tweets related to the 2014 European elections. Among other information, the tweets are enriched with a field mention containing the entities mentioned in the tweet. These entities can be easily classified into politicians, media, and others. We consider the first 500, 000 tweets, mentioning roughly n = 35, 000 distinct entities and where the most frequent entity has an empirical probability of occurrence equal to 0.065.

Tested Algorithms -We compare LAS performance against three other algorithms:

Base Line The Base Line algorithm takes as input the percentage of underprovisioning and drops at random an equivalent fraction of the tuples. Straw-Man The Straw-Man algorithm uses the same shedding strategy of LAS, however, it uses the average execution duration W as the estimated execution duration ŵ(t) for each tuple t. Full Knowledge The Full Knowledge algorithm uses the same shedding strategy of LAS, however, it feeds it with the exact execution duration w t for each tuple t as they were provided by an omniscient oracle.

Evaluation Metrics -The evaluation metrics we used are:

-the dropped ratio: α = d/m.

-the ratio of tuples dropped by algorithm alg with respect to Base Line: λ = (d algd Base Line )/d Base Line . In the following, we refer to this metric as shedding ratio. -the average queuing latency:

Q = i∈[m]\D q(i)/(m -d).
-the average completion latency, i.e., the average time it takes for a tuple from the moment it is injected by the source in the topology, till the moment operator O concludes its processing.

Whenever applicable we provide the maximum, mean, and minimum figures over the 5, 000 runs.

Simulation Results

In this section, we analyze, through a simulator built ad-hoc for this study, the sensitivity of LAS while varying several characteristics of the input load. The simulator faithfully simulates the execution of LAS and the other algorithms and simulates the execution of each tuple t on O doing busy waiting for w(t) milliseconds. Input Throughput -Figure 4 shows the average queuing latency Q (top) and dropped ratio α (bottom) as a function of the percentage of under-provisioning ranging from 90% to -10% (i.e., the system is 10% overprovisioned with respect to the average input throughput). As expected, in this latter case all algorithms perform at the same level as load shedding is superfluous. In all the other cases both Base Line and Straw-Man do not shed enough load and induce a huge amount of exceeding queuing latency. On the other hand, LAS average queuing latency is quite close to the required value of τ = 6.4 milliseconds, even if this threshold is violated in some of the tests. Finally, Full Knowledge always abide by the constraint and is even able to produce a much lower average queuing latency while dropping no more tuples that the competing solutions. Comparing the two plots we can see that the resulting average queuing latency is strongly linked to which tuples are dropped. In particular, Base Line and Straw-Man shed the same amount of tuples, LAS slightly more and Full Knowledge is in the middle. This result corroborates our initial claim that dropping tuples based on the load they impose allows designing more effective load shedding strategies. Threshold τ -Figure 5 shows the average queuing latency Q (top) and shedding ratio λ (bottom) as a function of the τ threshold. Notice that with τ = 0 we do not allow any queuing, while with τ = 6.4 we allow at least a queuing latency equal to the maximum execution duration w max . In other words, we believe that with τ < 6.4 the constraint is strongly conservative, thus representing a difficult scenario for any load shedding solution. Since Base Line does not take into account the latency constraint τ it always drops the same amount of tuples and achieves a constant average queueing latency. For this reason, Figure 5b reports the shedding ratio λ achieved by Full Knowledge, LAS, and Straw-Man against Base Line. The horizontal segments in Figure 5a represent the distinct values for τ . As the graph shows Full Knowledge always perfectly approaches the latency threshold, but for τ ≥ 12.8 where it is slightly smaller. Straw-Man performs reasonably well when the threshold is very small, but this is a consequence of the fact that it drops a large number of tuples when compared with Base Line as can be seen by Figure 5b. However, as τ becomes larger (i.e., τ ≥ 0.8) Straw-Man average queuing latency quickly grows and approaches the one from Base Line as it starts to drop the same amount of tuples. LAS, in the same setting, performs largely better, with the average queuing latency that for large values of τ approaches the one provided by Full Knowledge. While delivering this performance LAS drops a slightly larger amount of tuples compared to Full Knowledge, to account for the approximation in calculating tuple execution durations.

Maximum execution duration value w max -Figure 6 shows the average queuing latency Q (top) and dropped ratio λ (bottom) as a function of the maximum execution duration value w max . Notice that in this test we varied the value for τ setting it equal to w max . Accordingly, Figure 6a shows horizontal lines that mark the different thresholds τ . As the two graphs show, the behavior for LAS is rather consistent while varying w max ; this means that LAS can be employed in widely different settings where the load imposed by tuples in the operator is not easily predictable. The price paid for this flexibility is in the shedding ratio that, as shown in Figure 6b, is always positive.

Frequency Probability Distributions -7 shows the average queuing latency Q (top) and dropped ratio λ (bottom) as a function of the input frequency distribution. As Figure 7a shows Straw-Man and Base Line perform invariably bad with any distribution. The span between the best and worst performance per run increases as we move from a uniform distribution to more skewed distributions as the latter may present extreme cases where tuple latencies match their frequencies in a way that is particularly favorable or unfavorable for these two solutions. Conversely, LAS performance improves the more the frequency distribution is skewed. This result stems from the fact that the sketch data structures tracing tuple execution durations perform at their best on strongly skewed distribution, rather than on uniform ones. This result is confirmed by the shedding ratio (Figure 7b) that decreases, on average, as α for the distribution increases.

Precision parameter ε -Figure 8 shows the average queuing latency Q (top) and dropped ratio α (bottom) as a function of the precision parameter ε. This parameter controls the trade-off between the precision and the space complexity of the sketches maintained by LAS. As a consequence, it has an impact on LAS performance. In particular, for large values of ε (left side of the graph), the sketch data structures are extremely small, thus the estimation ŵ(t) is extremely unreliable. The corrective factor 1 + ε (see Listing 3.2 line 21) in this case is so large that it pushes LAS to largely overestimate the execution duration of each tuple. As a consequence LAS drops a large number of tuples while delivering average queuing latencies that are close to 0. By decreasing the value of ε (i.e., ε ≤ 0.1), sketches become larger and their estimation more reliable. In this configuration LAS performs at its best delivering average queuing latencies that are always below or equal to the threshold τ = 6.4 while dropping a smaller number of tuples. The dotted lines in both graphs represent the performance of Full Knowledge and are provided as a reference.

Time Series -Figure 9 shows the average queuing latency Q (top) and dropped ratio α (bottom) as the stream unfolds (x-axis). Both metrics are computed on a jumping window of 4.000 tuples, i.e., each dot represents the mean queuing latency Q or the dropped ratio α computed on the previous 4.000 tuples. Notice that the points for Straw-Man, LAS and Full Knowledge related to the same value of the x-axis are artificially shifted to improve readability. In this test, we set τ = 64 milliseconds. The input stream is made of 140, 000 tuples and is divided into phases, from a A through G, each lasting 20, 000 tuples. At the beginning of each phase we inject an abrupt change in the input stream throughput and distribution, as well as in w(t) as follows:

phase A : the input throughput is set according to the provisioning (i.e., 0% underprovisioning); phase B : the input throughput is increased to induce 50% of underprovisioning; phase C : same as phase A; phase D : we swap the most frequent tuple t with a less frequent tuple t such that w(t ) = w max , inducing an abrupt change in the tuple values frequency distribution and in the average execution duration W ; phase E : the input throughput is reduced to induce 50% of overprovisioning; phase F : the input throughput is increased back to 0% underprovisioning and we also double the execution duration w(t) for each tuple, simulating a change in the operator resource availability; phase G : same as phase A.

As the graphs show, during phase A the queuing latencies of LAS and Straw-Man diverge: while LAS quickly approaches the performance provided by Full Knowledge, Straw-Man average queuing latencies quickly grow. In the same timespan, both Full Knowledge and LAS drop slightly more tuples than Straw-Man. All the three solutions correctly manage phase B: their average queuing latencies see slight changes, while, correctly, they start to drop larger amounts of tuples to compensate for the increased input throughput. The transition to phase C brings the system back in the initial configuration, while in phase D the change in the tuple frequency distribution is managed very differently by each solution: both Full Knowledge and LAS compensate this change by starting to drop more tuples, but still maintaining the average queuing latency close to the desired threshold τ . Conversely, Straw-Man cannot handle such change, and its performance incurs a strong deterioration as it drops still the same amount of tuples. In phase E the system is strongly overprovisioned, and, as it was expected, all three solutions perform equally well as no tuple needs to be dropped. The transition to phase F is extremely abrupt as the input throughput is brought back to the equivalent of 0% of underprovisioning, but the cost to handle each tuple on the operator is doubled. At the beginning of this phase, both Straw-Man and LAS perform badly, with queuing latencies that are largely above τ . However, while the phase unfolds LAS quickly updates its data structures and converges toward the given threshold, while Straw-Man diverges as tuples continue to be enqueued on the operator worsening the bottleneck effect. Bringing back the tuple execution durations to the initial values in phase G has little effect on LAS, while the bottleneck created by Straw-Man cannot be recovered as it continues to drop an insufficient number of tuples.

Prototype

To evaluate the impact of LAS on real applications we implemented it as a bolt within the Apache Storm [START_REF]Apache Storm[END_REF] framework. We have deployed our cluster on Microsoft Azure cloud service, using a Standard Tier A4 VM (4 cores and 7 GB of RAM) for each worker node, each with a single available slot. The test topology is made of a source (spout) and two operators (bolts) LS and O. The source generates (reads) the synthetic (real) input stream and emits the tuples consumed by bolt LS. Bolt LS uses either Straw-Man, LAS or Full Time Series -In this test we ran the simulator using the same synthetic load used for the time series discussed in the previous section. The goal of this test is to show how our simulated tests capture the main characteristic of a real run. Notice, however, that plots in Figure 10 report the average completion latency per tuple instead of the queuing latency. This is due to the difficulties in correctly measuring queuing latencies in Storm. Furthermore, the completion latency is, from a practical point of view, a more significant metric as it can be directly perceived on the output. From this standpoint, the results, depicted in Figure 10, report the same qualitative behavior already discussed with Figure 9. Two main differences are worth to be discussed: firstly, the behaviors exposed by the shedding solution in response to phase transitions in the input load are in general shifted in time (with respect to the same effects reported in Figure 9) as a consequence of the general overhead induced by the software stack. Secondly, several data points for Straw-Man are missing in phases E and G. This is a consequence of failed tuples that start to appear as soon as the number of enqueued tuples is too large to be managed by Storm. While this may appear as a sort of "implicit" load shedding imposed by Storm, we decided not to consider these tuples in the metric calculation as they have not been dropped as a consequence of a decision taken by the Straw-Man load shedder.

Simple Application with Real Dataset -In this test we pretended to run a simple application on a real dataset: for each tweet of the twitter dataset mentioned in Section 5.1 we want to gather some statistics and decorate the outgoing tuples with some additional information. However, the statistics and additional information differ depending on which class the entities mentioned in each tweet belong. We assumed that this leads to a long execution duration for media (e.g., possibly caused by access to an external DB to gather historical data), an average execution duration for politicians and a fast execution duration for others (e.g., possibly because these tweets are not decorated). We modeled execution durations with 25 milliseconds, 5 milliseconds, and 1 millisecond of busy waiting respectively. Each of the 500, 000 tweets may contain more than one mention, leading to w n = 110 different execution duration values from w min = 1 millisecond to w max = 152 milliseconds, among which the most frequent (36% of the stream) execution duration is 1 millisecond. The average execution time W is equal to 9.7 millisecond, the threshold τ is set to 32 milliseconds and the under-provisioning is set to 0%. Figure 11 reports the average completion latency (top) and dropped ratio λ (bottom) as the stream unfolds. As the plots show, LAS provides completion latencies that are extremely close to Full Knowledge, dropping a similar amount of tuples. Conversely, Straw-Man completion latencies are at least one order of magnitude larger. This is a consequence of the fact that in the given setting Straw-Man does not drop tuples, while Full Knowledge and LAS drop on average a steady amount of tuples ranging from 5% to 10% of the stream. These results confirm the effectiveness of LAS in keeping close control on queuing latencies (and thus provide more predictable performance) at the cost of dropping a fraction of the input load.

Related Work

Aurora [START_REF] Abadi | Aurora: a new model and architecture for data stream management[END_REF] is the first stream processing system where shedding has been proposed as a technique to deal with bursty input traffic. Aurora employs two different kinds of shedding, the first and better detailed being random tuple dropping at strategic places in the application topology to satisfy QoS constraints.

A large number of works proposed solutions aimed at reducing the impact of load shedding on the quality of the system output. These solutions fall under the name of semantic load shedding, as drop policies are linked to the significance of each tuple with respect to the computation results. Tatbul et al. first introduced in [START_REF] Tatbul | Load shedding in a data stream manager[END_REF] the idea of semantic load shedding. Babcock et al. in [START_REF] Babcock | Load shedding for aggregation queries over data streams[END_REF] provided an approach tailored to aggregation queries. Tatbul et al. in [START_REF] Tatbul | Staying fit: Efficient load shedding techniques for distributed stream processing[END_REF] ported the concept of semantic load shedding in the realm of DSPS. GrubJoin [START_REF] Gedik | Grubjoin: An adaptive, multi-way, windowed stream join with time correlation-aware cpu load shedding[END_REF] is a solution tailored for shedding load in multiway windowed stream joins while minimizing output degradation. Finally, Kalyvianaki et al. in [START_REF] Kalyvianaki | Themis: Fairness in federated stream processing under overload[END_REF] contextualized the problem to the realm of federated DSPS, and provided a solution for shedding fairness. Several solutions assume that the utility of an event depends on the event type and its frequency in the input event stream [START_REF] Tatbul | Load shedding in a data stream manager[END_REF], i.e. they assume a static model for quality degradation; other works propose solutions to build and maintain at runtime a model for event utility [START_REF] Olston | Adaptive filters for continuous queries over distributed data streams[END_REF][START_REF] Katsipoulakis | Concept-driven load shedding: size and error of voluminous and variable data streams[END_REF]. All the previous works are based on the same goal, i.e., to reduce the impact of load shedding on the semantics of the queries deployed in the stream processing system, while avoiding overloads. We believe that avoiding excessive degradation in the performance of the DSPS and in the semantics of the deployed query output are two orthogonal facets of the load shedding problem. In our work, we did not consider the latter and focused on the former while including in our solution the possibility to limit output quality degradation.

A different approach has been proposed in [START_REF] Reiss | Data triage: An adaptive architecture for load shedding in TelegraphCQ[END_REF], with a system that builds summaries of dropped tuples to later produce approximate evaluations of queries. The idea is that such approximate results may provide users with useful information about the contribution of dropped tuples. A similar approach is adopted in StreamApprox [START_REF] Quoc | Streamapprox: Approximate computing for stream analytics[END_REF] where the authors designed an online stratified reservoir sampling algorithm to produce approximate output with rigorous error bounds. A similar approach was also adopted in [START_REF] Tok | A stratified approach to progressive approximate joins[END_REF].

A classical control theory approach based on a closed control loop with feedback has been considered in [START_REF] Kalyvianaki | Overload management in data stream processing systems with latency guarantees[END_REF][START_REF] Tu | Load shedding in stream databases: a control-based approach[END_REF][START_REF] Zhang | A novel adaptive load shedding scheme for data stream processing[END_REF]. In all these works the focus is on the design of the loop controller, while data is shed using a simple random selection strategy. In all these cases the goal is to reactively feed the stream processing engine system with a bounded tuple rate, without proactively considering how much load these tuples will generate.

Finally, a few works have recently appeared that address the problem of shedding load in Complex Event Processing (CEP) applications [START_REF] He | On load shedding in complex event processing[END_REF][START_REF] He | On load shedding in complex event processing[END_REF][START_REF] Slo | eSPICE: Probabilistic load shedding from input event streams in complex event processing[END_REF][START_REF] Slo | pSPICE: Partial match shedding for complex event processing[END_REF][START_REF] Zhao | Load shedding for complex event processing: Input-based and state-based techniques[END_REF]. While these solution leverage techniques similar to those discussed in the previous paragraphs, they provide specific adaptations to the CEP context where intput load can be shed both in the form of events and partial pattern matches.

Conclusions

In this paper, we introduced Load-Aware Shedding (LAS), a novel solution for load shedding in DSPS. LAS exploits a characteristic of many stream-based applications, i.e., the fact that load on operators depends both on the input rate and on the content of tuples, to smartly drop tuples and avoid the appearance of performance bottlenecks. In particular, LAS leverages sketch data structures to efficiently collect at runtime information on the operator load characteristics and then use this information to implement a load shedding policy aimed at maintaining the average queuing latencies close to a given threshold. Through a theoretical analysis, we proved that LAS is an ( , δ)-approximation of the optimal algorithm. Furthermore, we extensively tested LAS both in a simulated setting, studying its sensitivity to changes of several characteristics of the input load, and with a prototype implementation integrated within the Apache Storm DSPS. Our tests confirm that by taking into account the specific load imposed by each tuple, LAS can provide performance that closely approaches a given target, while dropping a limited number of tuples.

A Theoretical Analysis

Data streaming algorithms strongly rely on pseudo-random functions that map elements of the stream to uniformly distributed image values to keep the essential information of the input stream, regardless of the stream elements frequency distribution.

This appendix extends with the proofs the theoretical analysis of the quality of the shedding performed by LAS in two steps provided in Section 4 as well as the complexities presented in Section 3.

First we study the correctness and optimality of the shedding algorithm, under full knowledge assumption (i.e., the shedding strategy is aware of the exact execution duration w t for each tuple t). Then, in A.3, we provide a probabilistic analysis of the mechanism that LAS uses to estimate the tuple execution durations.

A.1 Time, Space and Communication Complexities

In this section we provide the proofs of the time, space and communication complexities presented in Section 3.

Theorem 1 [Time complexity of LAS] For each tuple read from the input stream, the time complexity of LAS for the operator and the load shedder is O(log 1/δ).

Proof. By Listing 3.1, for each tuple read from the input stream, the algorithm increments an entry per row of both the F and W matrices. Since each has log 1/δ rows, the resulting update time complexity is O(log 1/δ). By Listing 3.2, for each submitted tuple, the scheduler has to retrieve the estimated execution duration for the submitted tuple. This operation requires to read entry per row of both the F and W matrices. Since each has log 1/δ rows, the resulting query time complexity is O(log 1/δ).

Theorem 2 [Space Complexity of LAS] The space complexity of LAS for the operator and load shedder is O 1 ε log 1 δ (log m + log n) bits.

Proof. The operator stores two matrices of size log( 1 δ ) × e ε of counters of size log m. In addition, it also stores a hash function with a domain of size n. Then the space complexity of LAS on the operator is O 1 ε log 1 δ (log m + log n) bits. The load shedder stores the same matrices, as well as a scalar. Then the space complexity of LAS on the load shedder is also

O 1 ε log 1 δ (log m + log n) bits.

A.2 Correctness of LAS

We suppose that tuples cannot be preempted, that is they must be processed uninterruptedly on the available operator instance. As mentioned before, in this analysis we assume that the execution duration w(t) is known for each tuple t. Finally, given our system model, we consider the problem of minimizing d, the number of dropped tuples, while guaranteeing that the average queuing latency Q(t) will be upper-bounded by τ , ∀t ∈ σ. The solution must work online, thus the decision of enqueueing or dropping a tuple has to be made only resorting to knowledge about tuples received so far in the stream.

Let OPT be the online algorithm that provides the optimal solution to Problem 1. We denote with D σ OP T (resp. d σ OP T ) the set of dropped tuple indices (resp. the number of dropped tuples) produced by the OPT algorithm fed by stream σ (cf., Section 2). We also denote with d σ LAS the number of dropped tuples produced by LAS introduced in Section 3.3 fed with the same stream σ. Let us define a as the largest natural integer such that ∀ ≤ a, i = j (i.e., i 1 = j 1 , . . . , i a = j a ). Thus, we have i a+1 = j a+1 .

-Assume that i a+1 < j a+1 . Then, according to Section 3.3, the i a+1 -th tuple of σ has been dropped by LAS as the method Check returned true. Thus, as i a+1 / ∈ D σ OP T , the OPT run has enqueued this tuple violating the constraint τ . But this is in contradiction with the definition of OPT.

-Assume now that i a+1 > j a+1 . The fact that LAS does not drop the j a+1 tuple means that Check returns false, thus that tuple does not violate the constraint on τ . However, as OPT is optimal, it may drop some tuples for which Check is false, just because this allows it to drop an overall lower number of tuples. Therefore, if it drops this j a+1 tuple, it means that OPT knows the future evolution of the stream and takes a decision on this knowledge. But, by assumption, OPT is an online algorithm, and the contradiction follows.

Then, we have that i a+1 = j a+1 . By induction, we iterate this reasoning for all the remaining indices from a+1 to d σ OP T . We then obtain that D σ OP T ⊆ D σ LAS . As by assumption d σ OP T < d σ LAS , we have that ∃ ∈ D σ LAS \ D σ OP T such that has been dropped by LAS. This means that, with the same tuple index prefix shared by OPT and LAS, the method Check returned true when evaluated on , and OPT would violate the condition on τ by enqueuing it. That leads to a contradiction. Then, D σ LAS \ D σ OP T = ∅, and d σ OP T = d σ LAS . Furthermore, by construction, LAS never enqueues a tuple that violates the condition on τ because Check would return true.

Consequently, ∀t ∈ σ, Q σ LAS (t) ≤ τ , which concludes the proof.

A.3 Execution Time Estimation

In this section, we analyze the approximation made on execution duration w(t) for each tuple t when the assumption of full knowledge is removed. LAS uses two matrices, F and W, to estimate the execution time w(t) of each tuple submitted to the operator. By the Count Min sketch algorithm (cf., Section 3.2) and Listing 3.1, we have that for any t ∈ [n] and for each row i ∈ [r], Let us denote respectively by w min and w max the minimum and the maximum execution time of the items. We have trivially

F[i][h i (t)](m) =
w min ≤ W[i][h i (t)] F[i][h i (t)] ≤ w max .
We define S = n =1 w . We then have Theorem 5

E{W[i][h i (t)]/F[i][h i (t)]} = S -w t n -1 - k(S -nw t ) n(n -1) 1 -1 - 1 k n .
It important to note that this result does not depend on m. Proof.

For any t = 1, . . . , n, = 0, . . . , n -1 and A ∈ U (t), we introduce the event B(t, , A) defined by B(t, ,A) = {h i (u) = h i (t), ∀u ∈ A and h i (u) = h i (t), ∀u ∈ {1, . . . , n} \ (A ∪ {t})}.

From the independence of the hash function h i , we have

P{B(t, , A)} = 1 k 1 - 1 k n-1- .
Let us consider the ratio

V i,t = W[i][h i (t)]/F[i][h i (t)].
For any i = 0, . . . , n, we define

R (t) = f t w t + u∈A f u w u f t + u∈A f u , A ∈ U (t) .
We have R 0 (t) = {w t }. We introduce the set R(t) defined by

R(t) = n-1 =0 R (t).
Thus with probability 1,

W[i][h i (t)]/F[i][h i (t)] ∈ R(t).

Let x ∈ R(t). We have

P{V i,t = x} = n-1 =0 A∈U (t)
P{V i,t = x | B(t, , A)}P{B(t, , A)}

= n-1 =0 1 k 1 - 1 k n-1- A∈U (t)
1 {x=X(t,A)} .

where X(t, A) is the fraction:

X(t, A) = f t w t + u∈A f u w u f t + u∈A f u

Fig. 1 .

 1 Fig. 1. Load-Aware Shedding design with r = 2 (δ = 0.25), c = 4 (ε = 0.70).

Fig. 2 .Fig. 3 .

 23 Fig. 2. Operator finite state machine.

  ← 0r,c, 0r,c zero matrices pair of size r × c 4: Same hash functions h1 . . . hr of the operator 5: state ← NOP 6: end init 7: upon F , W do Figure 3.A and 3.D 8: state ← Send 9: F, W ← F , W 10: end upon 11: upon ∆ do Figure 3.C 12: C ← C + ∆ 13: end upon 14: function shed(tuple: t) 15:

Fig. 4 .

 4 Fig. 4. LAS performance varying the amount of underprovisioning.

(b) Shedding ratio λFig. 5 .

 5 Fig. 5. LAS performance varying the threshold τ .

(b) Shedding ratio λFig. 6 .

 6 Fig. 6. LAS performance varying the maximum execution duration value wmax.

Fig. 7 .

 7 Fig. 7. LAS performance varying the frequency probability distributions.

Fig. 9 .

 9 Fig. 9. Simulator time-series.

Fig. 10 .

 10 Fig. 10. Prototype time-series

Fig. 11 .

 11 Fig. 11. Prototype use case

Theorem 3 [ 1 ε log 1 δ 1 ε log 1 δ

 31111 Communication complexity of LAS] The communication complexity of LAS is of O m N messages and O m N (log m + log n) + log m bits. Proof. After executing N tuples, the operator may send the F and W matrices to the load shedder. This generates a communication cost of O m N (log m + log n) bits via O m N messages. When the load shedder receives these matrices, the synchronization mechanism kicks in and triggers a round trip communication (half of which is piggybacked by the tuples) with the operator. The communication cost of the synchronization mechanism is O m N messages and O m N log m bits. Note that the communication cost is low with respect to the stream size since the window size N should be chosen such that N 1 (e.g., in our tests we have N = 1024).

Theorem 4 [

 4 Correctness and Optimality of LAS] For any σ, we have d σ LAS = d σ OP T and ∀t ∈ σ, Q σ LAS (t) ≤ τ . Proof. Given a stream σ, consider the sets of indices of tuples dropped by respectively OPT and LAS, namely D σ OP T and D σ LAS . Below, we prove by contradiction that d σ LAS = d σ OP T . Assume that d σ LAS > d σ OP T . Without loss of generality, we denote i 1 , . . . , i d σ LAS the ordered indices in D σ LAS , and j 1 , . . . , j d σ OP T the ordered indices in D σ OP T .

n u=1 f u 1

 1 {hi(u)=hi(t)} =f t + n u=1,u =t f u 1 {hi(u)=hi(t)} .andW[i][h i (t)](m) = f t w t + n u=1,u =t f u w u 1 {hi(u)=hi(t)} ,

Table 1 .

 1 Symbols used in the text.

	Symbol Description
	t	Tuple
	σ	Stream of tuples
	[n]	Universe of possible tuples
	ft	Number of occurrences of t in σ
	w(t)	Execution duration of tuple t on operator O
	q(i)	Queuing latency of the i-th tuple of the stream
	D(j)	Set of dropped tuples
	d(j)	Number of dropped tuples
	Q(j)	Average queuing latency
	τ	Average queuing latency threshold
	C	Estimation of the total operator execution duration
	F	Count Min sketch that tracks tuple frequencies
	W	Count Min sketch that tracks tuple cumulated execution durations
	N	Window size parameter
	S	Snapshot
	η	Relative error between consecutive snapshots
	µ	Error threshold

  5, Zipf-2.0, Zipf-2.5, and Zipf-3.0. We define w n as the number of distinct execution duration values that the tuples can have. These w n values are selected at a constant distance in the interval [w min , w max ]. We ran experiments with w n {1, 2, • • • , 64}, however, due to space constraints, we only report results for w n = 64, and with w

max ∈ {0.1, 0.2 • • • , 51.2} milliseconds. Tests performed with different values for w n did not show unexpected deviations from what is reported in this section. Unless otherwise specified, the frequency distribution is Zipf-1.0 and the stream parameters are set to w n = 64, w min = 0.1 ms and w max = 6.4 ms; this means that the w n = 64 execution durations are picked in the set {0.1, 0.2, • • • , 6.4} ms.

In the data streaming literature, the frequency is the number of occurrences not divided by time, which differs from the classical (physics) definition[START_REF] Muthukrishnan | Data Streams: Algorithms and Applications[END_REF].

This is not the only possible definition of the load shedding problem. Other variants are briefly discussed in section

This correction factor derives from the fact that ŵ(t) is a (ε, δ)-approximation of w(t) as shown in Section 4.

For readability reasons, proofs of these theorems are available in A.

This work has been partially funded by the MIUR SCN-00064 project RoMA and by Sapienza University of Rome through the project RM11916B75A3293D. A preliminary short version of this work appeared in the Proceedings of the 10th

Let us assume that all the f u are equal, that is for each u, we have f u = m/n. The experimental evaluation tends to show that the worst case scenario of input streams is exhibited when all the items show the same number of occurrences in the input stream. We get } that concludes the proof.


Emmanuel Moulay

Stabilization via homogeneous feedback controls

Keywords: Control Lyapunov function, Feedback stabilization, Homogeneous system

In this paper, we provide an explicit homogeneous feedback control with the requirement that a control Lyapunov function exists for an affine control system and satisfies an homogeneous condition. We use a modified version of the Sontag formula to achieve our main goal. Moreover, we prove that the existence of an homogeneous control Lyapunov function for an homogeneous affine system leads to an homogeneous closed-loop system by using the previous feedback control.

Introduction

The weighted homogeneity, which has been introduced by Rothschild and Stein in [START_REF] Rothschild | Hypoelliptic differential operators and niipotent groups[END_REF], plays an important role in nonlinear control theory. This notion was extensively used:

(1) for the construction of a nilpotent approximating system which is small time local controllable and homogeneous (see [START_REF] Hermes | Nilpotent and high-order approximations of vector field systems[END_REF]), (2) for homogeneous feedback stabilization in [START_REF] Kawski | Homogeneous stabilizing feedback laws[END_REF][START_REF] Kawski | Homogeneous feedback stabilization[END_REF] when searching an homogeneous closed-loop system, (3) for finite time stabilization (see [START_REF] Bhat | Geometric homogeneity with applications to finite-time stability[END_REF]).

The stabilization of control systems by homogeneous feedback control has also been developed in [START_REF] Coron | Adding an integrator for the stabilization problem[END_REF] by using backstepping technics and in [START_REF] Grüne | Homogeneous state feedback stabilization of homogeneous systems[END_REF] by using homogeneous control Lyapunov function for controllable systems.

The paper deals with finding homogeneous continuous feedback laws by using control Lyapunov functions. We develop a modified version of the Sontag formula in order to obtain an explicit homogeneous feedback control. In a first way, it is interesting to find an homogeneous feedback control even if homogeneous feedback stabilization is not possible. Indeed, homogeneous functions possess very interesting properties which allow to restrict their study to compact manifolds (see [START_REF] Rosier | Homogeneous Lyapunov function for homogeneous continuous vector field[END_REF]Lemma 1]). In a second way, finding an homogeneous feedback control is often the first step when dealing with the research of an homogeneous closed-loop system. We can also mention that certain homogeneous systems cannot be stabilized by homogeneous feedback laws. This has been first pointed out in [START_REF] Rosier | Etude de Quelques Problèmes de Stabilisation[END_REF] and in [START_REF] Sepulchre | Stabilizability does not imply homogeneous stabilizability for controllable systems[END_REF] for affine systems.

The organization of this paper is as follow. Section 2 provides some usefull notations and definitions. In section 3 we develop our main result based on a modified version of the Sontag formula. We give an explicit homogeneous feedback control for affine systems which possess a control Lyapunov function satisfying an homogeneous condition. The closed-loop system is not homogeneous in general. In section 4, we give an explicit homogeneous feedback control for homogeneous affine systems which possess an homogeneous control Lyapunov function. In this case, the closed-loop system is homogeneous.

Notations and definitions

Throughout this paper, V will be a non empty neighborhood of the origin in R n . Let us consider the system

ẋ = f (x) , x ∈ R n (1)
where f : R n → R n is a continuous function. If V : V → R n is a continuously differentiable function, the derivative of V along the solutions of the system (1) is defined by

V (x) = ∇V (x), f (x) .
As it is customary in control theory, a Lyapunov function V : V → R n for the system (1) is a continuously differentiable positive definite function such that V is negative definite. The Lie derivative of V : R n → R along f : R n → R n is defined by:

L f V : R n → R x → ∇V (x), f (x) A function V : R n → R is homogeneous of degree d with respect to the weights (r 1 , . . . , r n ) ∈ R n >0 if V (∆ λ (x)) = λ d V (x 1 , . . . , x n )
for all λ > 0 where ∆ λ (x) = (λ r 1 x 1 , . . . , λ r n x n ). A vector field f is homogeneous of degree d with respect to the weights (r

1 , . . . , r n ) ∈ R n >0 if for all 1 ≤ i ≤ n, the i-th component f i is a homogeneous function of degree r i + d, that is f i (∆ λ (x)) = λ r i +d f i (x 1 , . . . , x n ) for all λ > 0. The system (1) is said homogeneous of degree d with respect to the weights (r 1 , . . . , r n ) ∈ R n >0 if f is homogeneous of degree d with respect to the weights (r 1 , . . . , r n ) ∈ R n >0 .

Stabilization by homogeneous feedback controls

Let m ∈ N * , and consider the following affine system

ẋ = f 0 (x) + m i=1 f i (x)u i , x ∈ R n , u ∈ R m (2)
where

f i ∈ C (R n ) for all 0 ≤ i ≤ m and f 0 (0) = 0 and the closed-loop system ẋ = f 0 (x) + m i=1 f i (x)u i (x), x ∈ R n . ( 3
)
Let us recall the definition of stabilization. The control system (2) is stabilizable (respectively continuously stabilizable) if there exists a non empty neighborhood of the origin V in R n and a feedback control law

u ∈ C 0 (V \ {0} , R m ) (respectively u ∈ C 0 (V, R m )) such that:
(1) u (0) = 0, (2) the origin of the system (3) is asymptotically stable.

We now recall some usual definitions. A continuously differentiable positive definite function

V : V → R ≥0 is acontrol Lyapunov function for the system (2) if for all x ∈ V \ {0}, inf u∈R m (a (x) + B (x) , u ) < 0.
where

a (x) = L f0 V (x), b i (x) = L fi V (x), 1 ≤ i ≤ m B (x) = (b 1 (x), . . . , b m (x)) .
As usual, such a control Lyapunov function satisfies the small control property if for each > 0, there exists δ > 0 such that, if x ∈ δB n , then there exists some u ∈ B m such that a (x) + B (x) , u < 0.

Remark 1 If m = 1, the small control property is equivalent to lim sup

x n →0 a(x) |B (x)| ≤ 0.
The limit may very well be -∞.

We set b (x) = B (x) 2 .
If we want to use a control Lyapunov function to obtain an homogeneous feedback control, we can use the following modified version of the Sontag feedback control.

Lemma 2 If there exists a continuously differentiable control Lyapunov function V : V → R ≥0 for the control system [START_REF] Coron | Adding an integrator for the stabilization problem[END_REF], then it is stabilizable under the feedback control u (x) = (u 1 (x) , . . . , u m (x)) defined by

u i (x) =    -b i (x) a(x)+(|a(x)| p +b(x) q ) 1 p b(x) if x ∈ V\ {0} 0 if x = 0 (4
) where p > 1, q > 1 are positive real numbers. If furthermore V satisfies the small control property, then the feedback control ( 4) is also continuous at the origin.

Proof. Suppose there exists a continuously differentiable control Lyapunov function

V : V → R ≥0 . Let E = (x, y) ∈ R 2 : x < 0 or y > 0 and ϕ a function defined on E by ϕ (x, y) =      x + (|x| p + |y| q ) 1 p y if y = 0 0 if y = 0 .
As ϕ is defined on E, we have to prove that lim y→0,x<0

x + (|x|

p + |y| q ) 1 p y = 0. lim y→0,x<0 x + (|x| p + |y| q ) 1 p y = lim y→0,x<0 x + |x| 1 + |y| q |x| p 1 p y = lim y→0,x<0 -x |y| q py |x| p = lim y→0,x<0 |y| q-1 sgn (y) p |x| p-1 = 0. ϕ is continuous on E. The fact that V is a control Lya- punov function implies that (a(x), b (x)) ∈ E for all x ∈ V\ {0}. Thus u (x) = (u 1 (x) , . . . , u m (x))
defined by

u i (x) = -b i (x) ϕ (a(x), b (x))
is continuous on V\ {0} and we obtain for all x ∈ V\ {0},

∇V (x) , f 0 (x) + m i=1 f i (x)u i (x) = -(|a(x)| p + b (x) q ) 1 p < 0.
V is a Lyapunov function for the closed-loop system (3)-( 4), and by using the Lyapunov theorem we know that the origin of the closed loop system (3)-( 4) is asymptotically stable.

The proof concerning the stabilization under the small control property is similar to the one given in [12, Theorem 1].

A less general version of this lemma was proved in [START_REF] Moulay | Finite time stability and stabilization of a class of continuous systems[END_REF]Lemma 19]. The Sontag control corresponds to the case (p, q) = (2, 2) (see [START_REF] Sontag | A universal construction of Arststein's theorem on nonlinear stabilization[END_REF]).

We now give a result concerning single input systems.

Theorem 3 Consider the single input system (2) with m = 1. If there exists:

(1) a continuously differentiable control Lyapunov function V : V → R ≥0 for the control system (2) such that a (x) is homogeneous of degree d 0 and B (x) is homogeneous of degree d 1 with respect to the weights (r 1 , . . . , r n ) ∈ R n >0 , (2) p > 1, q > 1 positive reals such that p = 2qd 1 d 0 , then the feedback control

u (x) =    -a(x)+(|a(x)| p +b(x) q ) 1 p B(x) if x ∈ V\ {0} 0 if x = 0 (5)
stabilizes the system [START_REF] Coron | Adding an integrator for the stabilization problem[END_REF] and is homogeneous of degree d 0 -d 1 with respect to the weights (r 1 , . . . , r n ). If furthermore V satisfies the small control property, then the feedback control ( 5) continuously stabilizes the system [START_REF] Coron | Adding an integrator for the stabilization problem[END_REF].

Proof. The fact that u (x) stabilizes the system ( 2) is given by Lemma 2.

As a (x) is homogeneous of degree d 0 and B (x) is homogeneous of degree d 1 with respect to the weights (r 1 , . . . , r n ) ∈ R n >0 , we have:

u (∆ λ (x)) = - λ d 0 a(x)+(λ pd 0 |a(x)| p +λ 2d 1 q b(x) q ) 1 p λ d 1 B(x) = - λ d 0 a(x)+(λ pd 0 (|a(x)| p +b(x) q )) 1 p λ d 1 B(x) = - λ d 0 a(x)+(|a(x)| p +b(x) q ) 1 p λ d 1 B(x) = λ d 0 -d 1 u (x) .

Let us give an example

Example 4 Consider the system ẋ1 = -x 3 1 + x 2 ẋ2 = -x 1 + u and the control Lyapunov function

V (x) = x 2 1 + x 2 2 2 . a (x) = -x 4
1 is homogeneous of degree 4 and B (x) = x 2 is homogeneous of degree 1 with respect to the weights

(1, 1). As a (x) |B (x)| = -x 4 1 |x 2 | ≤ 0
we deduce from Remark 1 that V satisfies the small control property. Theorem (3) with 4p = 2q, for instance (p, q) = 3 2 , 3 , implies that the system is continuously stabilizable by the feedback control

u (x) = - -x 4 1 +(x 6 1 +x 6 2 ) 2 3 x2
which is homogeneous of degree 3.

We now address a result for multi-input systems whose proof can be easily deduced from the previous one.

Theorem 5 Consider the system [START_REF] Coron | Adding an integrator for the stabilization problem[END_REF]. If there exists:

(1) a continuously differentiable control Lyapunov function V : V → R ≥0 for the control system (2) such that a (x) is homogeneous of degree d 0 and b i (x) are homogeneous of degree d 1 with respect to the weights

(r 1 , . . . , r n ) ∈ R n >0 for all 1 ≤ i ≤ m, (2) p > 1, q > 1 positive reals such that p = 2qd 1 d 0 ,
then the feedback control (4) stabilizes the system (2) and its components u i (x) are homogeneous of degree d 0 -d 1 with respect to the weights (r 1 , . . . , r n ) for all 1 ≤ i ≤ m.

If furthermore V satisfies the small control property, then the feedback control ( 4) continuously stabilizes the system (2).

Homogeneous feedback stabilization

By homogeneous stabilization, we mean that there exists an explicit feedback control such that the closedloop system is homogeneous. This is the purpose of this section. The following lemma can be found in [START_REF] Rosier | Homogeneous Lyapunov function for homogeneous continuous vector field[END_REF].

Lemma 6 If V is homogeneous of degree d 1 with respect to the weights (r 1 , . . . , r n ) ∈ R n >0 and if f is homogeneous of degree d 2 with respect to the weights (r 1 , . . . , r n ) ∈ R n >0 , then L f V is homogeneous of degree d 1 + d 2 with respect to the weights (r 1 , . . . , r n ) ∈ R n >0 .

We can now give a theorem for homogeneous feedback stabilization using an homogeneous control Lyapunov function. Nevertheless, it is more restrictive than Theorem 3.

Theorem 7 Consider the single input system (2) with m = 1 where f 0 (x) is homogeneous of degree d 0 and f 1 (x) homogeneous of degree d 1 with respect to the weights (r 1 , . . . , r n ) ∈ R n >0 . If there exists a continuously differentiable control Lyapunov function V : V → R ≥0 for the control system (2) homogeneous of degree d with respect to the weights (r 1 , . . . , r n ) ∈ R n >0 , then the feedback control [START_REF] Kawski | Homogeneous stabilizing feedback laws[END_REF] where

p = 2q (d + d 1 ) d + d 0 , p > 1, q > 1,
is homogeneous of degree d 0 -d 1 and the closed-loop system (3)-( 5) is homogeneous of degree d 0 with respect to the weights (r 1 , . . . , r n ). If furthermore V satisfies the small control property, then the feedback control (5) continuously stabilizes the system (2).

Proof. As V (x) is homogeneous of degree d, f 0 (x) is homogeneous of degree d 0 and f 1 (x) is homogeneous of degree d 1 with respect to the weights (r 1 , . . . , r n ) ∈ R n >0 , Lemma 6 implies that a (x) is homogeneous of degree d+ d 0 and B (x) is homogeneous of degree d + d 1 . Theorem 3 implies that the feedback control ( 5) is homogeneous of degree d 0 -d 1 . The closed loop system satisfies

(f 0 + f 1 u) (∆ λ (x)) = λ d 0 f 0 (x) + λ d 1 f 1 (x) λ d 0 -d 1 u (x) = λ d 0 (f 0 + f 1 u) (x) . Example 8 Consider the system ẋ1 = x 1 -x 2 ẋ2 = x 1 + x 2 u
and the control Lyapunov function

V (x) = x 2 1 + x 2 2 2
homogeneous of degree 2 with respect to the weights (1, 1).

f 0 (x) = x 1 -x 2
x 1 and f 1 (x) = 0

x 2 are homogeneous of degree 0 with respect to the weights (1, 1). Theorem [START_REF] Moulay | Finite time stability and stabilization of a class of continuous systems[END_REF] with p = 2q, for instance (p, q) = (4, 2), implies that the system is stabilizable by the feedback control u (x) = -

x 2 1 +(x 8 1 +x 8
2 )

1 4
x 2 2 which is homogeneous of degree 0 with respect to the weights [START_REF] Bhat | Geometric homogeneity with applications to finite-time stability[END_REF][START_REF] Bhat | Geometric homogeneity with applications to finite-time stability[END_REF]. Moreover, the closed-loop system is then homogeneous of degree 0 with respect to the weights (1, 1).

As Theorem 5 generalized Theorem 3 to multi-input systems, it is also possible to generalize Theorem 7 to multiinput systems.

Concluding remarks

The paper deals with the design of an homogeneous feedback control based on control Lyapunov functions satisfying an homogeneous condition. The control law is derived from the well known Sontag control. When the affine control system and the control Lyapunov functions are homogeneous, the closed-loop system is also homogeneous.

Acknowledgements

The author is extremely grateful to Sergej Celikovsky for the many discussions he has had about the notion of homogeneity.